Wave A: events, observers, hooks, evals and routing (design)

Status: design

This page is the design for Wave A of the product redesign, which ships as release 0.7. Each pull request in the PR plan adds a "Built" note to the sections it builds, as Phase 1 and Phase 2 did; A0 (the foundation), E1, H1, H2, V1, E2, V2, E4, S1 and D1 are built. Phase 1 is Configuration v2 and the extension registry; Phase 2 is Phase 2: test bed, storage, providers, policies and adapter SDK v2. Wave A is what those pages called "Phase 3", plus the cross-database routing work ("Phase R") and three smaller tracks.

flowchart LR
  A0["A0. Foundation<br/>event types, bus, run scope, config shapes,<br/>error codes, result fields, registry kinds"]
  A0 --> E["Track 1: events and observers<br/>E1 to E4"]
  A0 --> H["Track 2: hooks and approval<br/>H1, H2"]
  A0 --> V["Track 3: evals<br/>V1 to V3"]
  A0 --> R["Track 4: Phase R routing<br/>R1 to R3"]
  E --> S["Smaller tracks<br/>D1 drift, S1 sanity, M1 MCP"]
  H --> S
  V -->|"R3 needs V1's scorers"| R
  S --> REL["Release 0.7"]
  E --> REL
  H --> REL
  V --> REL
  R --> REL

# nl2sql_sdk.events (pydantic and the standard library only, D14)
class Event(BaseModel):
    model_config = ConfigDict(frozen=True)
    name: str                    # "node.finished", from the catalog
    at: datetime                 # UTC
    trace_id: str | None = None  # the run's id; None for an engine event (an index build)
    tenant_id: str | None = None
    data: dict[str, Any] = {}    # JSON-ready; the catalog's model for this name, dumped

    def sse(self) -> str:        # "event: <name>\ndata: <json>\n\n"
        ...

class Observer(Protocol):
    interface_version: ClassVar[int]           # 1
    config_model: ClassVar[type[BaseModel]]
    events: ClassVar[frozenset[str] | None]    # the names it wants; None means every event
    @classmethod
    def from_config(cls, cfg: BaseModel, *, folder: Path) -> "Observer": ...   # folder: storage.path
    def on_event(self, event: Event) -> None: ...
    def close(self) -> None: ...

# nl2sql/events/bus.py
class EventBus:
    def subscribe(self, fn: Callable[[Event], None], names: Iterable[str] | None = None,
                  *, fail: Literal["open", "raise"] = "open") -> Subscription: ...
    def publish(self, name: str, data: BaseModel | Mapping[str, Any] = {}, *,
                trace_id: str | None = None) -> Event: ...
    def close(self) -> None: ...               # closes every observer it built

Built: the event, the bus, the catalog and the run scope (A0)

nl2sql_sdk/events.py holds Event (with sse(), whose data line is the whole event as JSON) and the Observer protocol. nl2sql/events/ holds EventBus (bus.py; publish also takes tenant_id), the catalog with a strict model per event and catalog() returning EventInfo (catalog.py, which also defines RunStatus), and RunScope, current_scope() and the module publish (run.py). Runtime.bus is built with the runtime and closed with it. QueryAPI.open_scope(question, datasource_id=, execute=, user_context=) makes the scope, with the run's usage collector, and calls start() (which publishes run.started); QueryAPI.close_scope(scope, result) closes it with finish(result, tables=...), or fail(message, code) when the summary cannot be built, and a result built without a graph run is accepted. run_query and stream_query open their own scope, or take an open one as scope= (internal; Wave C's turn opens it before its rewrite call) and close it. RunScope.callbacks() is the usage collector and the LangChain handlers attached (the budget, from E3): a model call made in the run but outside the graph passes them, inside entered(), so it counts in the run's usage and carries its trace_id. run_with_graph takes scope=, uses its trace_id and its usage collector, enters it beside policy_engine.run() in _invoke (submitted with the caller's copy_context()) and adds its callbacks() to the graph's. That is all of Wave C's A2 need (wave_c.md); E1 keeps scope= when it moves stream_query to Events. The scope sets the context TraceContextFilter reads, so log lines written during a run carry its trace_id and tenant_id. attach takes a function or an object with on_event (subscribed to this run's events with fail="raise"), a LangChain callback handler (added to the graph's callbacks) and anything with close (closed with the scope); a closed scope refuses attach with RuntimeError. Closing detaches every subscription and calls every close even when an attached piece raised on the terminal event. run_query and stream_query close the scope with run.failed on any BaseException (a KeyboardInterrupt too) and re-raise it. The scope holds its lock while the bus delivers each of its events, so run.finished is always last; the cost is that a run's events reach subscribers one at a time and finish (a timed-out run's too) waits for the delivery in flight, so an observer must never block (the Observer and RunScope docstrings say so; E2's audit and otel hand slow work to a queue). run.finished's usage and result_manifest copy only the catalog models' own fields, so a field added to UsageTotals or ResultSummary never fails a run. run.finished's tables come from each sub-query's plan, or a verified query's SQL, as datasource.schema.table. engine.events and engine.evals are empty namespaces; E1 adds subscribe and catalog.

Built: run events, the stream and engine.events (E1)

pipeline/events.py's RunEvents(publish) is a LangChain callback that publishes node.started, node.finished (each with its sub_query_id, from the sub-query subgraph's subgraph_id and inherited by the runs inside it), result.manifest (results only; the event's trace_id is the request_id) and answer.delta. QueryAPI.open_scope attaches one to every scope (scope.attach(RunEvents(scope.publish))), so every run publishes its steps, streamed or not, and scope.callbacks() is now the usage collector, the RunEvents, then whatever was attached. QueryEvent is deleted, with its export. The allow-list is one definition: each catalog entry (EventSpec) has stream and audit flags, and nl2sql.events.STREAM_EVENTS (D17) and AUDIT_EVENTS (D19, which E2's audit observer reads; of hook.decision it keeps the decisions the entry's description names) are derived from them; EventInfo reports both. A later event joins a list by its own catalog entry (Wave B's events stay off the stream). The stream: QueryAPI.stream_query(..., scope=None, names=STREAM_EVENTS) yields Events. RunScope.follow(fn, names) replays what the scope already published (RunScope.events, a new list of every event the scope published, in order) and attaches fn, both under the scope's lock, so an event published meanwhile is neither lost nor doubled; the stream follows its scope with names minus run.finished, and puts its own run.finished: the summary close_scope (which now returns the published event) returned, plus data.result. So a scope opened before the stream (Wave C's turn, need A2) still gives the stream its run.started and anything the turn published in it, and names (internal) is where Wave C's conversations.stream adds conversation.rewritten (need A4); it appends conversation.turn after the stream's run.finished itself. The stream ends with exactly one terminal event whatever breaks: when no run.failed reached it (the scope could not open, was already closed, or a collector raised on its terminal event), it ends with a run.failed built in place, with the scope's trace_id when there is a scope. A KeyboardInterrupt ends it the same way. scope= is kept on run_query and stream_query; a closed scope= is a ValueError at the call. engine.events.subscribe(fn, names=None) subscribes to Runtime.bus (fail open) and catalog() returns catalog(). The trace document has events (the scope's events but answer.delta); since run_with_graph writes the trace before run.finished is published, the list ends before it. That reads scope.events in pipeline/runtime.py (one argument to write_run_trace), not tracing/recorder.py, a small difference from the file list above. nl2sql run, the playground (app.py and web/playground/src/stream.js, rebuilt) and nl2sql-api's stream route read the new names; each SSE data line is the whole event. The invariant Events Carry No Rows, and the Stream Carries No Operator Data is in invariants.md.

Built: the Phase 2 events, published (E4)

Engine events go through ctx.bus explicitly, never a context variable, because an index build and a purge run outside any RunScope; each carries the engine's tenant_id. rebuild_index wraps its on_datasource callback, so every IndexProgress it reports started publishes index.started (with ctx.namespace) and every failed one index.failed (a datasource that is unavailable, unknown, locked out by another build, or whose build raised). IndexingOrchestrator keeps ctx.bus and ctx.tenant_id: index_datasource publishes one index.column_failed per IndexIssue before the commit (table, column, stage, message: the issue, never the column's statistics or sample values, D15), and _commit publishes index.swapped after the pointer switch, with counts by entry type and the replaced pointer's build_id and schema_version (None on a first build), so a curated-only refresh publishes it too. Runtime._published(status) publishes datasource.failed for a failed DatasourceStatus; check_datasource returns through it, and try_datasource is now _published around _try_datasource (the old body), which is the whole runtime.py change. StorageAPI.purge publishes retention.purged with the counts it returns (audited, so the audit observer writes it to the current day's file after deleting the old ones). Run events use the module publish: structured()'s retry loop publishes llm.parse_retry before each retry (attempt 1 is the first retry; errors are pydantic's path: message lines, or the exception's type name, never the model's answer). policy.decision is published at the three enforcement points only (D16 amended): LogicalValidatorNode._validate_policy per refused table (step: logical_validator, schema the table's or the scope's default), CuratedMatcherNode._check_access on the refused table (step: curated_matcher; it now asks scope.rule_for, which allows wraps), and the resolver when the role check refuses every datasource it was asked about: every registered one before the routing search (R1's scope filter), or the explicit or only one (_publish_refusals, one per datasource, schema and table None, step: datasource_resolver, rule naming the roles or the policy check). The resolver's re-check after datasources.after hooks publishes nothing; the hook's own hook.decision covers it. After review: datasource.failed (message, fix), index.failed and index.column_failed run their text through nl2sql.tracing.trace.redact_for(ctx, text) (the trace's Redactor with collect_secrets(ctx); its patterns now also catch a bare user:password@host). For the values and profile stages, index.column_failed's message is a fixed sentence plus the new IndexIssue.error_type (SDK, optional, set by the SQLAlchemy base), so a cast error quoting a cell value never reaches the bus. The rule comes from the new Scope.datasource_rule() (reaches_datasource is datasource_rule() is None), which names a failed policy check, and the validator publishes once per (schema, table, rule). Test fakes that build an orchestrator or call rebuild_index need bus and tenant_id (tests/unit/index_helpers.py's TestIndex has them). Tests: tests/unit/test_phase2_events.py.

# nl2sql.yaml
observers: observers.yaml

# observers.yaml
- type: log
  level: INFO                     # node.started and answer.delta are logged only at DEBUG
- type: audit
  max_age_days: 90
- type: otel                      # needs nl2sql-engine[otel]
  endpoint: http://collector:4317
  metrics: true
  traces: true
- type: corp_siem                 # a plugin; its config_model checks the rest
  url: "${env:SIEM_URL}"

from nl2sql import Config, ObserverConfig
Config(..., observers=[ObserverConfig(type="log"), MyObserver()])   # a {type: ...} block or an object

Built: the observer kind and the section (A0)

The observer registry kind (add_observer, @registry.observer) and the observers section (ObserverConfig, or an object with on_event) are accepted and validated; nothing builds an observer yet (E2). Config.observers is None when the section is absent, which E2 reads as log and audit. settings.observability and its readers (nl2sql run's configure_metrics, the CLI's and nl2sql-api's switch to JSON logs) are gone. nl2sql --log-format text|json and NL2SQL_LOG_FORMAT choose the log format; configure_logging(json_format=None) reads the variable.

Built: the observers (E2)

nl2sql/observers/ holds build_observers (__init__.py, with DEFAULT_OBSERVERS, log then audit), log.py (LogObserver, LogConfig), audit.py (AuditObserver, AuditConfig) and otel.py (OtelObserver, OtelConfig), declared in the nl2sql.observers entry point group. Runtime.components builds them after every other component (a problem is one more line of the one ConfigError, and a built observer is closed again), keeps them as Runtime.observers and subscribes them to Runtime.bus in config order; engine.close() closes them through the bus. The registry checks an observer's requires like an LLM provider's, so otel without the extra is a ConfigError ending in pip install "nl2sql-engine[otel]". What is audited is the catalog's: EventSpec.audit, and a new EventSpec.audit_only (a field name to the values written; hook.decision's is decision: deny, needs_approval, approved, error), applied by EventSpec.audited(data) and reported in EventInfo.audit_only. The observer reads the audited names from the catalog when it is built, so Wave C's datasource.saved, api_key.created, api_key.revoked and mcp.denied join the log with audit=True on their own entries. run.failed gained user_id and roles (two lines in RunScope.fail), so a failed run's line names its user too. The audit observer queues each line and writes on its own daemon thread, started by the first audited event, opening and closing the day file per batch. Its queue is bounded (MAX_QUEUE, 10,000): a line that does not fit is dropped, counted in dropped and warned about once. close writes what is queued, and what arrived while it closed; an atexit hook (through a weak reference, unregistered by close) does the same for a process that never closes its engine. RunScope.fail passes message through redact_keys (D15). engine.storage.purge() adds audit (day files older than max_age_days) when the engine runs an audit observer. otel records in memory on its own MeterProvider and TracerProvider (a BatchSpanProcessor, a PeriodicExportingMetricReader) and never sets the global ones, and close bounds the providers' flush at 5 seconds; nl2sql.token.usage lost its datasource_id attribute, which llm.call does not carry. TokenUsageCallback publishes llm.call per finished call through nl2sql.events.publish (with the client's nl2sql_agent metadata as agent), PlanCache.get publishes plan_cache.lookup, and Runtime.replace_llm publishes llm.changed on ctx.bus with the engine's tenant_id; common/metrics.py and common/event_logger.py are deleted, and opentelemetry is the otel extra (CI's unit and integration jobs install it). One test differs from the plan: test_building_an_engine_loads_no_opentelemetry cannot hold, because chromadb (a core dependency) installs the OpenTelemetry SDK and langsmith, under langchain, imports it whenever it is installed; test_building_an_engine_loads_no_otel_observer checks what the engine controls, beside the two AST tests on nl2sql/observers/otel.py. The page is Observers.

Built: log correlation (E1)

run_query and stream_query's thread run inside scope.entered() as well as the graph's worker, so every log line written during a run, on the caller's thread or a LangGraph worker thread, carries its trace_id and tenant_id (test_every_log_line_in_a_run_has_its_trace_id, test_log_lines_on_worker_threads_have_the_trace_id). docs/observability/events.md shows a library caller adding TraceContextFilter.

# nl2sql/llm/prices.py
PRICES_CHECKED_ON = "2026-09-21"
PRICES: dict[str, ModelPrice] = {...}            # dated and sourced; updated by hand

def price_for(model: str, overrides: Mapping[str, Price]) -> ModelPrice | None:
    """The config's llm.prices row, else the table's, by exact name; None when neither has it."""

# settings.yaml
pipeline:
  timeout_s: 60
  max_llm_calls: 60       # model calls per question, parse retries included; null: no cap
  max_cost_usd: 0.50      # dollars per question; null (the default): no cap

Built: the price table, cost and caps (E3)

nl2sql/llm/prices.py holds PRICES_CHECKED_ON, PRICES (the five rows moved unchanged), ModelPrice (a NamedTuple: input, cached_input, cache_write, output, verified, source; a config row has source="config" and may leave the two cache rates None), price_for(model, overrides) (the llm.prices row, else the table's, by exact name), call_cost(price, *, input_tokens, output_tokens, cached_input_tokens=None, cache_write_input_tokens=None, cache_write_multiplier=1.0) (None when the input or output count is unknown; a cache count not reported is not split out of input; a config row's missing cache_write is input times the wire's multiplier, so Anthropic's 1.25x from #193 holds) and effective_prices(overrides). evaluation/prices.py is deleted with no alias; evaluation/run.py's check_prices(configs, prices=None) and question_cost(usage, cfg, prices=None) take llm.prices as overrides (run_evals passes the engine's), and the UnknownPriceError fix names llm.prices. TokenUsageCallback(prices=llm.prices) prices every call with price_for by the served name, then the configured one, and gained spent() (the dollars of the calls with a known cost). Unknown counts are None, never 0 (an owner requirement after the design): the wires' usage() takes Optional counts (opt_int, sum_known in wires/base.py; as_int is gone), so an unreported cache read, cache write (OpenAI reports none) or reasoning count is None, and a call with no usage (a failed call) has every count None and a null cost. LLMCallUsage, UsageTotals and the catalog's LLMCall and UsageTotalsData have Optional[int] token fields; a total is the sum of what was reported, None when nothing was; the otel observer skips a None count. engine.llm.prices() returns {"checked_on", "models": {name: {input, cached_input, cache_write, output, verified, source, citation}}} (source is table or config; citation the table row's source). nl2sql/limits.py holds Budget, LimitReached, describe and stop_answer beside RunStop. Budget(max_llm_calls=None, max_cost_usd=None, usage=None, token=None, setting="settings.pipeline") is a LangChain callback (raise_error = True): on on_chat_model_start and on_llm_start it refuses the call when the calls it let start have reached max_llm_calls or usage.spent() has reached max_cost_usd; on the first refusal it sets budget.stop, cancels every bound token with the RunStop (CancellationToken.cancel(reason=None) keeps the first reason in token.reason), publishes run.limit_reached through the module publish (nothing outside a run), and raises LimitReached(stop, setting) (not a ValueError, so the parse-retry loop never swallows it); every later call is refused with the same stop. Work that is not a question (Wave B's profiler, V3's seeding) opens its own Budget(..., setting="evals"), makes its calls with budget.callbacks() (the budget, then its usage collector), and reads budget.stop or catches LimitReached. Budget.for_question(pipeline, usage) is None when both caps are off. The budget reads the usage collector directly, not the llm.call events, because a budget outside a question has no scope to read events from (a difference from the file list's "usage collector on llm.call"). QueryAPI.open_scope attaches the question's budget after the RunEvents, so scope.callbacks() is the usage collector, the RunEvents, the budget, then anything attached later; run_with_graph binds every attached Budget to the run's token (budget.bind(token)), and a token cancelled with a RunStop maps to _stopped: stop, the one error RUN_LIMIT_REACHED ("The question reached its limit of 60 model calls (settings.pipeline.max_llm_calls)."), the answer "This question reached its limit of 60 model calls (settings.pipeline.max_llm_calls)." (or "its cost limit of $0.50 (settings.pipeline.max_cost_usd)"), and the trace outcome stopped. Runtime.components and Runtime.replace_llm raise a ConfigError per unpriced model (llm.model, llm.agents.<agent>.model; a model object by its model_name) when max_cost_usd is set, with the fix add llm.prices.<model> .... Tests: tests/unit/test_cost_and_caps.py. After review (PR #326): a call is priced by price_for_call(served, configured, overrides), every llm.prices override (served name, then configured name) before the table for either name; Price's four rates are NonNegativeFloat; a timeout that fires while a budget stop unwinds reports stopped; mode full counts a call with no token counts in EvalReport.unknown_cost_calls (new unknown_cost_calls(usage), one warning per run) instead of adding 0; the OpenAI rows are marked for the owner to re-verify.

flowchart TD
  Q["question"] --> P1(["run.before"])
  P1 --> CM["curated matcher"]
  CM -->|"no match"| RES["datasource resolver"]
  RES --> P2(["datasources.after"])
  P2 --> DEC["decomposer"]
  DEC --> PL["planner, or the plan cache"]
  PL --> P3(["plan.after"])
  P3 --> VAL["logical validator"]
  VAL --> GEN["generator"]
  GEN --> P4(["execute.before"])
  CM -->|"verified query"| P4
  P4 --> EX["executor"]
  EX --> P5(["execute.after"])
  P5 --> AGG["aggregator"]
  AGG --> ANS["answer synthesizer"]
  ANS --> P6(["answer.after"])
  P6 --> R["QueryResult"]

# nl2sql/hooks/__init__.py (engine: the context holds engine types, D14)
HookPoint = Literal["run.before", "datasources.after", "plan.after",
                    "execute.before", "execute.after", "answer.after"]

class HookContext(BaseModel):          # frozen; fields beyond the point's are None
    point: HookPoint
    trace_id: str
    question: str
    user: UserContext
    datasource_id: str | None
    datasource_ids: list[str] | None
    sub_query: SubQuery | None
    plan: PlanModel | None
    sql: str | None
    params: dict[str, Any] | None
    tables: list[str] | None           # datasource.schema.table, read by the SQL
    result: ResultSummary | None
    final_answer: dict | None
    approvals: list[Approval]          # what the run was re-run with (section 7)

class Allow(BaseModel): ...
class Modify(BaseModel):  question: str | None = None; datasource_ids: list[str] | None = None; plan: PlanModel | None = None
class Deny(BaseModel):    reason: str; scope: Literal["run", "sub_query"] = "run"
class NeedsApproval(BaseModel): reason: str
Decision = Allow | Modify | Deny | NeedsApproval

class Hook(Protocol):
    interface_version: ClassVar[int]           # 1
    config_model: ClassVar[type[BaseModel]]
    points: ClassVar[frozenset[HookPoint]]     # where it may be attached
    @classmethod
    def from_config(cls, cfg: BaseModel) -> "Hook": ...
    def __call__(self, ctx: HookContext) -> Decision: ...

# nl2sql.yaml
hooks: hooks.yaml

# hooks.yaml (a list; order matters)
- type: corp_sql_review           # a registered hook
  points: [execute.before]        # a subset of the hook's own points; all of them by default
  on_error: closed                # closed (the default) | open
  approver_roles: [dba]           # who may approve its needs_approval; default policies.admin_roles
  url: "${env:REVIEW_URL}"        # the hook's own keys, checked by its config_model
- type: question_rewriter
  points: [run.before]

from nl2sql import Config, HookConfig
Config(..., hooks=[HookConfig(type="corp_sql_review", points=["execute.before"]), MyHook()])

Built: the hook kind and the section (A0)

The hook registry kind and the hooks section are accepted and validated: HookConfig (points, on_error, approver_roles, and the hook's own keys) or an object with __call__. HookPoint, the six point names, is defined in nl2sql/config/models.py, where the config and the hook.decision event both read it, and re-exported from nl2sql.hooks, where H1 imports it.

Built: the runner, the six points and the boundary test (H1)

nl2sql/hooks/: the interface in __init__.py (HookContext, Allow, Modify, Deny, NeedsApproval, Decision, Hook; HookPoint is re-exported from nl2sql.config.models), points.py (HOOK_POINTS, a PointRule per point: decisions, modifies, per_sub_query) and runner.py. The shapes differ from the sketch above in four places:

• HookRunner(bindings) takes a list of HookBinding(name, hook, points, on_error, approver_roles) from the runtime (Runtime._build_hooks, build_hook per entry, in config order) and reads no config, so a built-in hook (Wave B's policy.cost) goes first with its own approver_roles. ctx.hooks is the engine's runner; hooks_of(ctx) gives NO_HOOKS for a stand-in runtime.
• HookRunner.call(point, **fields) builds the context from deep copies of fields (a hook changing the plan or the caller in place changes nothing) and returns a HookVerdict: ctx after every Modify, modified, and the decision and hook that stopped the point (allowed, scope, error(node), the PipelineError; a hook error's message names only the hook and the point). With no hook at a point, nothing is validated. HookRunner.attached(point) says whether any hook runs there: while one runs at answer.after the answer is not streamed (no answer.delta), RunEvents(..., hold_manifest=True) holds result.manifest until the answer is allowed (dropped on a deny), and result_from_state drops the rows, artifact_refs and result_manifest of a run a hook ended.
• HookContext adds conversation_id: str | None (Wave C's A3), execute (run.before) and results (the run's ResultSummarys at answer.after, the "sub-query summaries" of the table).
• NeedsApproval at execute.before stops the SQL with NEEDS_APPROVAL (status error) until H2 adds the request and the re-run; so NEEDS_APPROVAL left DECLARED_AHEAD here with HOOK_DENIED and HOOK_FAILED.

The call sites: run.before in run_with_graph's worker; datasources.after in DatasourceResolverNode._after_hooks (the one addition to the resolver; R1 rebases on it); plan.after in ASTPlannerNode.after_plan(state, plan, plan_source), which the planner's and the cache's plan go through and which a supplied plan (H2's approved, Wave B's chosen, Wave C's supplied) calls the same way; ASTPlannerNode.check_plan(state, plan) is one plan's verdict, for Wave B's alternatives. ASTPlannerResponse.planner_plan is the plan before the hooks, and wrap_subgraph caches it. execute.before and execute.after in ExecutorNode and CuratedMatcherNode (each a self.hooks.call("<point>", **self._hook_fields(...)); a deny on the verified path ends the run, as _ended, never a fallback); answer.after in AnswerSynthesizerNode. A sub-query a hook dropped has SubgraphOutput.status == "dropped" and a HOOK_DENIED warning; build_scan_layer_router (in pipeline/routes.py, a small addition the file list missed) ends the run on a hook's error and still aggregates when a sub-query was dropped; AggregationService.execute(..., dropped=) drops a join or compare group whole; the answer's warnings name what was dropped. SubgraphExecutionState.user_query (in pipeline/state.py, also not in the file list) carries the question to the sub-query's hooks. The architecture test is tests/architecture/test_hooks_boundary.py: every runner call names its point as a literal; its graph check asks that every path from ast_planner to generator pass logical_validator, so Wave B's clarifier may sit between the validator and the generator.

class ApprovalRequest(BaseModel):          # on QueryResult.approvals_needed
    id: str                                # sha256 over every field below except reason and approver_roles, hex
    question: str
    asker_user_id: str | None              # the run's UserContext
    asker_tenant_id: str | None
    hook: str
    point: Literal["execute.before"]
    datasource_id: str
    decomposition: DecomposerResponse | None   # generated path: the run's decomposition
    sub_query: SubQuery | None                 # generated path: the sub-query the SQL answers
    plan: PlanModel | None                     # generated path: its validated plan
    verified: VerifiedRef | None               # verified path: the query id and its bound params
    sql: str
    sql_sha256: str
    reason: str
    approver_roles: list[str]              # for display; the engine checks the hook's configured roles

class Approval(BaseModel):                 # passed back on the re-run
    request: ApprovalRequest
    by: UserContext                        # the approver

Built: the shapes on the result (A0)

ApprovalRequest, Approval and VerifiedRef (id, params) are in nl2sql/hooks/approval.py, and RunStop in nl2sql/limits.py; all are exported from nl2sql. QueryResult.status is a RunStatus (success, plan_only, error, needs_approval, stopped, and needs_clarification, reserved for Wave B), or None when nothing ran and nothing failed (0.6's ""). result_from_state reads the state's approvals_needed and stop, and sets the status from them before looking at errors, so H2 and E3 put those keys in the state and touch no result code. The error codes HOOK_DENIED, HOOK_FAILED, NEEDS_APPROVAL, RUN_LIMIT_REACHED and QUESTION_TOO_BROAD (never retried), JOIN_KEY_TYPE_MISMATCH and JOIN_FAN_OUT (retried) and JOIN_KEY_UNCONVERTED and ALL_NULL_COLUMN (warnings) are in ErrorCode; DECLARED_AHEAD in test_boundary_tidy_ups.py lists each with the PR that raises it, which takes it off the list.

Built: approval and the pinned-plan entry (H2)

• The entry. nl2sql/pipeline/pinned.py: PinnedSource, SuppliedPlan, PinnedRun (source, decomposition, plans keyed by sub-query id, or verified with verified_datasource_id), PinnedRun.supplied(plan), pinned_from_approvals(approvals, question=, user=) and PinnedEntryNode. build_graph(ctx, execute, pinned=True) builds a pinned run's graph, whose entry is pinned_entry (with no curated matcher); every other run's graph is unchanged and has no such node. run_with_graph builds the pinned graph for a supplied pin, or, for an approval's pin (known only after run.before), inside the worker. The pinned_entry node scope-checks every datasource of the decomposition (SECURITY_VIOLATION otherwise, routes.pinned_route ends the run) and goes to the layer router. build_scan_payload hands each sub-query its pinned_plan and pinned_source; the planner returns it through after_plan(state, plan, source). The SQL agent reads a pinned sub-query's retries as spent, so neither a validator nor a database error is retried, and the executor's value check never asks for a retry on it. The schema retriever gives a pinned plan every table the caller may read, with no search, for the validator. check_pinned(pinned, approvals, datasource_id) refuses, with a ValueError at the call (QueryAPI.run_query and stream_query, before the run's scope opens), a chosen pin with approvals and a datasource_id that is not the pinned plan's datasource. QueryAPI.run_query(..., pinned=) (internal) takes a pin of another source; Wave B passes its chosen plan there.
• The request. ApprovalRequest.build(...) computes sql_sha256 and id; expected_id() recomputes it: SHA-256 of the canonical JSON of the fields in REQUEST_ID_FIELDS only, each dumped without the values it holds at their defaults. The rule for Wave D's store: a field added later, to the request or with a default to a nested model, keeps the ids of stored requests; changing a default or the meaning of a listed field changes them, and a test pins one id (GOLDEN_ID) so that never happens by accident. wrap_subgraph builds the request for a sub-query that ended NEEDS_APPROVAL (the error's details now carry the reason) into GraphState.approvals_needed; the curated matcher does the same with a VerifiedRef. One difference from the sketch above: the request pins the plan as plan.after received it (planner_plan), not the plan the hooks let through, because the re-run passes it through plan.after again; pinning the hooked plan would apply a hook's Modify twice and the SQL would never match the hash. What the approver approves is the SQL, and it is the same.
• The check. HookRunner(bindings, admin_roles=) (the runtime passes policies.admin_roles), HookRunner.approver_roles(hook). On a NeedsApproval the runner looks for an approval in ctx.approvals with no approval_problem(...): unedited, this run's question (after run.before) and asker, this hook and datasource, the hash of the SQL just generated, the same verified values, and an approver holding one of the hook's configured roles. A match publishes hook.decision with decision: "approved" and the approver, and the next hook runs. request_problem refuses an edited request, a request whose sql is not the SQL of its sql_sha256 (so the SQL an approver reads is the SQL whose hash is checked), and another question or asker. pinned_from_approvals(approvals, question=, user=, hooks=) pins only an approval with no request_problem, for a configured hook, whose by holds one of that hook's approver roles; so a forged approval, an approver without the role, or a request the asker built with no approval hook configured, pins nothing and covers nothing: the run plans as usual and asks again, and plan_source is never "approved" for it.
• The verified path. A pinned verified request takes the default entry; the curated matcher runs that query with its values (_run_pinned), with no search and no model call, through the same access checks and hooks. If it is no longer available, the matcher falls back as usual.
• Not done: the result's final_answer stays None on a needs_approval run (the error's message is the answer text), as on every run a hook ended; nothing is stored (Wave D).

# nl2sql.yaml
evals: evals.yaml

# evals.yaml: the section
suites: [evals/chinook.yaml, evals/routing.yaml]
scorers: [result_match, outcome, faithfulness, retrieval_recall, routing]   # the default: all five
curated: true                       # D12
max_cost_usd: 5.00                  # across a run in mode full; required for it
gate:
  max_regressions: 2
  alpha: 0.05
  max_cost_increase: 0.20

# evals/chinook.yaml: a suite
version: 1
cases:
- id: chinook_001
  question: How many customers do we have, by country?
  datasources: [chinook]
  status: approved                  # draft | approved | retired; only approved cases run
  origin: hand                      # hand | seeded
  difficulty: easy
  tags: [aggregation, rbac-denial]
  paraphrase_group: null
  outcome: {admin: allowed, analyst: allowed, viewer: refused}
  tables: [Customer]
  columns: [Customer.Country, Customer.CustomerId]
  order_matters: false
  sql: |
    SELECT Country, COUNT(CustomerId) AS customer_count
    FROM Customer GROUP BY Country ORDER BY customer_count DESC, Country
  rows: [...]                       # written by regenerate, never by hand
  alt_sql: []                       # reviewed alternatives
  alt_rows: []                      # written by regenerate
  plan: {...}                       # optional: the hand-written gold plan, for mode plans

Built: the section and the settings (A0)

EvalsConfig (with EvalGate) is Config.evals, from evals.yaml; suites are paths relative to nl2sql.yaml, resolved by V1. The settings the other tracks read are in place with their defaults: settings.pipeline.max_llm_calls, max_cost_usd and sanity_checks, settings.routing, and settings.retrieval.table_k, planning_k and decomposer_tables; eval_seeder is in AgentName.

# nl2sql/evaluation/scorers/__init__.py (engine: it sees QueryResult, D14)
class Score(BaseModel):
    value: float | None        # 0..1; None when the scorer does not apply to this case
    passed: bool | None
    detail: dict[str, Any] = {}

class CaseRun(BaseModel):      # what a scorer is given
    case: EvalCase
    role: str
    result: QueryResult        # with the run's trace when one was kept
    rows: list[list[Any]]      # the first terminal result, read from the results store

class Scorer(Protocol):
    interface_version: ClassVar[int]     # 1
    config_model: ClassVar[type[BaseModel]]
    modes: ClassVar[frozenset[str]]      # the modes it can score
    @classmethod
    def from_config(cls, cfg: BaseModel) -> "Scorer": ...
    def score(self, run: CaseRun) -> Score: ...

Built: evals core (V1)

nl2sql/evaluation/ is reorganised as the section says:

• The case format. suite.py holds EvalCase, EvalSuite, load_suite, select_cases and write_suite. outcome also takes clarify, which runs only with engine.evals.run(clarify=True); Wave B's clarification produces it.
• The Chinook suite moved. suites/chinook.yaml holds the gold data unchanged, pinned by test_chinook_suite_holds_the_gold_data_unchanged. Rows stay {column: value} maps, as the old file wrote them.
• The run and the report. run.py (run_evals) runs the three modes. report.py holds EvalReport and the boards, from tier 2's scoreboard.
• The scorers. scorers/ holds Score, CaseRun, Scorer and the five built-ins, declared as nl2sql.scorers entry points.
• Rows through the adapter. regenerate.py returns a RegenerateReport.
• The gold-plan fake. gold_llm.py answers from one table, RESPONSES, keyed by structured-output schema name. Wave B's PlannerAnswer is one more entry there.
• The gate's statistics. gate.py holds tier 2's question_flips, mcnemar_exact, compare_with_baseline and check_baseline, unchanged except that the deprecated max_accuracy_drop is gone. V2 builds engine.evals.gate on them.

engine.evals has load, run, regenerate and scorers. run takes clarify: bool = False. With curated on in mode full, a curated hit's record carries generated, and each board carries accuracy.generated. CaseRun also carries mode, columns, retrieval (what the retriever sent) and routed (mode retrieval's ranking). In mode full, retrieval is read from the run's schema retriever node.

Deleted: engine.benchmark, BenchmarkAPI, BenchmarkConfig, benchmark_runner.py, tier1.py, tier2.py, types.py, gold.py, retrieval_recall.py (now scorers/retrieval_recall.py, with sent, routed and retrieval_settings), datasets/ and nl2sql benchmark. records.py keeps its page texts, which still name nl2sql benchmark, so the committed pages stay byte-identical until V2 rewires publishing.

After review:

• Each board also reports accuracy.execution. It counts only result_match and outcome, the 0.6 tier 2 measure, so the published series stays comparable.
• Mode retrieval cases open a run scope too.
• regenerate refuses a packaged suite unless allow_packaged=True.
• feedback export refuses a configured suite and any file holding approved cases.

Built: the gate (V2)

engine.evals.gate(report, baseline) -> GateResult(ok, regressions, series_break, compared), with evals.yaml's gate settings, built on V1's statistics in nl2sql/evaluation/gate.py (gate, load_baseline). Where the code differs from the section above, after the owner's decision of 2026-10-15:

• It reads execution accuracy, not overall. A case passes the gate on result_match and outcome only (nl2sql.evaluation.report.execution_passed), the measure 0.6's tier 2 published, so a faithfulness miss is never a regression and a 0.7 run compares with the 0.6 history. "Both accuracies" are that measure as users see it and on the generated path (each case's generated run), the second only when the run had curated answers on. Mode retrieval has no execution and is read by status.
• A baseline is a report, a record or a benchmarks/ folder (the newest record of the run's mode, database and config). A record 0.6 wrote has no scores; its status was execution accuracy and is read so.
• A series break is not ok. The gate uses the history's own rule, records.series_key (mode, suite digest, schema fingerprint and, in mode full, roles and curated answers), plus the roles each run actually ran. A series break compares nothing for that config, and nl2sql eval run exits 2 after printing the other configs' regressions.
• The allowance is per mode. max_regressions and McNemar (two-sided, counted only when more cases were lost than gained) apply to mode full. Modes plans and retrieval run no model and allow no flip.
• A missing or unreadable baseline exits 2 with a one-line error.
• Routing has no suite yet (Phase R adds it), so CI gates mode retrieval on Chinook only. R3 added it: see the routing gate.

Built: the CLI and the records (V2)

nl2sql eval run, regenerate and publish are in cli/commands/eval.py. Where they differ from the table:

• run also takes --passes. --llm takes PRESET, FILE.yaml, NAME=PRESET or NAME=FILE. The report goes to --out, by default eval_<mode>.json next to nl2sql.yaml, with a run stamp (when, and which engine version and commit) for publish. Exit codes: 0, 1 on a regression, 2 on a usage error or a baseline that cannot be compared, 3 at the cost cap. Mode full refuses to start with CI set (the one CI read, in READS_THE_ENVIRONMENT).
• publish [REPORT ...] [--note] [--from DIR ...] writes one record per config of each report, then rebuilds docs/benchmarks.md and the README block from every record. With no report it only rebuilds the pages, as benchmark publish did.
• Records are keyed by mode and suite digest: benchmarks/<mode>/<database>/...json, kind the mode and dataset the suite's name and digest (several suites: their digests joined and hashed, report.suite_identity). Every mode is recorded; a record keeps the config's whole board.
• The headline accuracy is execution accuracy (metrics.accuracy is accuracy.execution), so the published series reads the same as 0.6's; accuracy_all_scorers and accuracy_generated sit beside it.
• 0.6's records stay readable. Kind tier2 is read as mode full (records.mode_of), so the history shows those runs in the mode full table, as a new series (the suite changed).
• Records are slim: one compact JSON line, each case's verdicts and cost only (records.slim_result). The full report is CI's eval-reports artifact. A retrieval record names its embedder.
• CI's baselines are recorded on CI (ubuntu-latest, Python 3.12): the integration job uploads eval_plans.json and eval_retrieval.json, and nl2sql eval publish records them from that artifact, the run's URL in the note. The steps are in benchmarks/README.md; a suite, schema or routing change re-records both the same way.

Built: seeding (V3)

engine.evals.seed(datasource_id, *, count=20, out, user_context, max_cost_usd=None) -> SeedReport and nl2sql eval seed, in evaluation/seed.py and evaluation/prompts.py (EVAL_SEEDER_PROMPT). The user guide is Benchmark your own database. Where the code differs from the steps above:

• The caller is required and checked first. No user_context, no role, or a role whose scope does not reach the datasource (or none of its tables) is a PermissionError before any call. The model is shown the tables through SchemaRetrieverNode.visible_tables(user_context, datasource_id), new and public: the planner's own rendering for that caller. outcome marks allowed each of the caller's roles whose own scope reads every table of the plan (after review of PR #328); other roles get no entry.
• Plans go through the pinned-plan entry. Each proposed plan runs as run_query(question, plan=SuppliedPlan(...), execute=False, user_context=...): plan.after, the validator and the generator under the caller's scope, with no model call, in a run scope of its own (run.started and run.finished). The draft keeps the plan that produced the SQL (result.sub_queries[0].plan), so a plan.after change wins. A question that reads as SQL (the decomposer's SubQuery check) is dropped as invalid.
• The SQL passes execute.before and execute.after (the session lead's decision on the review of PR #328): it is model-proposed, so an approval hook or Wave B's cost guard must see it. A deny, a hook failure or needs_approval drops the case (dropped_hooked). execute.after gets a summary of the first row (truncated: true). regenerate runs human-reviewed gold SQL and stays unhooked; both are documented in configuration/hooks.md. The rows are checked with execute_batches(request, 1) where the adapter streams; no row is kept.
• The structured output is SeedBatch (cases: SeedCase with question, difficulty and a PlanModel), the name a fake LLM rule answers. A call whose answer never parses counts its batch as invalid.
• The budget is Budget(max_cost_usd, max_llm_calls, setting="evals") with at most three calls per batch of five (CALLS_PER_BATCH), so a model answering nothing usable stops; the report says so (stopped_at_call_limit), and the CLI exits 3 as at the cost cap. Each call is shown the count of questions already written and the newest ten (RECENT_QUESTIONS), not all of them. max_cost_usd is required (the argument, else evals.yaml's), and an unpriced eval_seeder model is an UnknownPriceError before any call.
• Drafts never go into reviewed cases: out may not be a packaged suite, a suite evals.yaml names, or a file holding approved cases. The guard is nl2sql.evaluation.suite.check_target(path, configured), which feedback export calls too. SeedReport also carries path, dropped_hooked and stopped_at_call_limit.
• The CLI asks as --role, else policies.default_role, and exits 2 without one; it exits 2 on any refusal and 3 at the cost cap or the call limit, and refuses to run with CI set, as mode full does. It takes no --llm: llm.agents.eval_seeder picks the model.

score(ds) = sum over signals s:  1 / (RRF_K + rank_s(ds))

Built: entries, votes, fusion and the retrieval settings (R1)

• Entries. DatasourceExampleChunk (schema.datasource_example) in indexing/models.py; build_example_chunks(datasource_id, questions, schema_version) in indexing/chunk_builder.py takes the question list from its caller (SchemaChunkBuilder passes the orchestrator's sample_questions, or enrichment's), so Wave B adds the glossary's examples to that list without changing it. Blank and repeated questions are skipped; ids hash the question. DatasourceChunk lost examples. ENTRY_TYPES lists the new type.
• Routing. Retriever.route(question, datasource_ids, explain=None, *, limit=3, floor_ratio=0.5, allows=None) -> list[RouteCandidate] replaces retrieve_datasource_candidates, which is deleted. limit, floor_ratio and allows (an entry check) are keyword arguments the resolver fills from settings.routing.candidates, floor_ratio and the caller's scope (the retriever reads no settings and no policy). RouteCandidate (datasource_id, score, ranks, similarity) is in indexing/models.py; ROUTE_SIGNALS, fuse(rankings, rrf_k, datasource_ids, similarity=None) (an unranked datasource gets rank len(ranked) + 1) and shortlist(fused, limit, floor_ratio) (the floor on best similarity) are module functions in indexing/retrieval.py; VOTE_K = 20 and RRF_K = 60 are on Retriever. The trace gets one route.<signal> record per signal and a routing record (signals, fused with each similarity, a pool of the fused datasources, picks, dropped with floor or limit, rrf_k, vote_k, candidates, floor_ratio).
• Table scope (after the review of PR #316). The resolver's _route_filter passes allows: a table or column entry votes only when its (datasource, table) is visible in the live snapshot, a verified query only when every table its SQL reads is. Refused entries are dropped before ranking or recording, over-fetching so VOTE_K counts allowed entries. routing.candidates now defaults to 3, as D28 says.
• The resolver. The scope filter runs before routing: no reachable datasource is a SECURITY_VIOLATION with no search. It forwards at most max_datasources of the accepted candidates, best by fused score; the forwarded list stays sorted by id, so the decomposer's prompt does not depend on the fused order. ResolvedDatasource.metadata is empty on every path now (it held the datasource entry's metadata, which nothing read). MAX_DECOMPOSER_TABLES is settings.retrieval.decomposer_tables.
• The schema retriever reads settings.retrieval.table_k and planning_k; TABLE_K and PLANNING_K are deleted.
• The retrieval scorers (evaluation/scorers/retrieval_recall.py, after V1): routed() ranks with route under settings.routing (then cut at the caller's limit, max_datasources in mode retrieval), and retrieval_settings() reads settings.retrieval.
• Baseline. The 0.6 retrieval baseline, recorded on main before any of this changed, is benchmarks/retrieval/chinook/2026-10-15_653fa73_retrieval.json: table recall 0.9767, column recall 0.7185 and datasource_top1_accuracy 0.9535 (41 of 43 Chinook questions; the two misses went to support), over the demo's three datasources with the local embeddings.
• Measured after the review (key-free, local embeddings). Routing top-1 over 79 questions across the demo's three datasources (the 43 Chinook gold questions, the 20 sample questions and 16 paraphrases written for this check): 0.6 75/79; R1 before the review 62/79 (the curated free vote sent 17 of the 20 support and webanalytics questions to chinook); R1 after it 77/79. The expected datasource was among the shortlisted candidates for all 79 in every version.

# settings.yaml
routing:
  candidates: 3          # datasources shown to the answerability check
  max_datasources: 3     # datasources one question may use
  max_sub_queries: 6     # sub-queries one question may have
  floor_ratio: 0.5       # drop a candidate less similar than this share of the most similar
retrieval:
  full_snapshot_max_tables: 15     # exists
  table_k: 8                       # was SchemaRetrieverNode.TABLE_K
  planning_k: 12                   # was SchemaRetrieverNode.PLANNING_K
  decomposer_tables: 30            # was MAX_DECOMPOSER_TABLES in the resolver

Built: the sub-query limit (R2)

DecomposerNode reads settings.routing.max_sub_queries once, in its constructor (self.max_sub_queries), and the system prompt states it as rule 10 ("Use at most 6 sub_queries"), a {max_sub_queries} template variable, so the system message stays the same for one engine. The count is of the model's own sub-queries, before any is dropped as unmapped. Over it, the node returns an empty DecomposerResponse, no DAG and one QUESTION_TOO_BROAD error (ERROR, in FATAL_ERRORS), with the message above; the model is called once.

Built: join-key coercion and the check (R2)

• One set of rules. nl2sql/aggregation/join_keys.py (plain Python) holds the categories, DTYPE_CATEGORY, coercible(a, b) and join_dtype(left, right), the type both sides are cast to: the shared dtype; text parsed as the other side's type; int and float meet as int (not float: a non-whole value cannot equal a whole one, and a float cannot tell ids above 2^53 apart), date and datetime as datetime, a number and bool as bool. The decomposer's schema, the aggregator and the validator all read it.
• The decomposer. DecomposerResponse.validate_join_key_types() resolves each key against its side's expected_schema with resolve_key (so right.customer_id is customer_id); the side is the input's place in engine order (input_order). ExpectedColumn.join_key, JoinKeyPair.left_dtype and JoinKeyPair.right_dtype are SkipJsonSchema, set only by mark_join_keys() in decomposer/node.py, which replaces whatever the model sent and leaves a key it cannot find untyped rather than crash. join_key is excluded from the stable sub-query id and from the planner's expected_schema. CombineGroup.validate_roles() refuses two inputs on one side, or a third input, in a join/compare. The DAG carries the pair's dtypes in the combine node's join_keys attributes (dag.py dumps the pair, so it is unchanged).
• The aggregator. PolarsDuckdbEngine.combine casts each side first to its own declared dtype and then to the joined one, so a value is read the way its sub-query declared it (a number goes straight to the joined number type, and two float keys over integer frames join as int). A cast nulls, never changes, a value: a float or decimal that is not whole is not an int. Timestamps of any unit or zone become naive UTC microseconds (exact dtypes are compared, not the bare pl.Datetime class); text is parsed only by the ISO formats in _DATE_FORMATS, _NAIVE_FORMATS and _OFFSET_FORMATS, never by inference; a Time key declared a date or timestamp is null. Every one of those nulls is counted, so none ends the run as AGGREGATOR_FAILED. Text is stripped on both sides, of a text key too; a text boolean is true/false/t/f/yes/no/1/0; a number cast to a boolean is 1 or 0, anything else null. PolarsDuckdbEngine.unconverted_join_keys(inputs, join_keys) counts the values that became null (it collects the key columns only), and AggregationService.execute(..., warnings=[...]) gathers them; the node turns each into a JOIN_KEY_UNCONVERTED warning on the state's warnings, naming the datasource: "1 of 3 support.customer_id values are not whole numbers and joined nothing." S1 made it a RunWarning whose detail holds column, failed, total and dtype.
• The validator. pipeline/nodes/validator/join_keys.py, join_key_type_errors(plan, sub_query, relevant_tables, dialect), called from _validate_static before the buildability check. The column type is the schema snapshot's Column.type. sqlglot's BOOLEAN is a boolean, its numeric types a number (BIT included), its text types text, its temporal types temporal; anything else, an unparsable type or no type is other and not checked. The message is "The join key order_key is declared a date, but Orders.OrderKey is a number. Select a column that holds a date for this key."
• The demo. test_a_key_declared_text_on_one_side_joins_after_coercion (tests/e2e/test_multi_sub_query_fake_llm.py) joins Chinook's integer CustomerId to support's customer_id declared as text and gets the SQLite answer. The demo databases store both keys as integers, so the text-against-integer case itself is the unit test above. The design lists no realdb test for R2.

Built: the routing suite and gate (R3)

• The suite. suites/routing.yaml (ROUTING_SUITE_PATH in evaluation/suite.py): 78 approved, hand-written cases asked as admin, tagged by origin and datasource. They are R1's 79-question measurement (the 43 Chinook gold questions, the demo's 20 sample questions, 12 of which are gold questions too, and 16 paraphrases) plus 8 more support and webanalytics questions and 3 cross-database ones: 47 name chinook, 14 support, 14 webanalytics, 3 two of them. Every expected datasource was read against the schema. No case names tables, so the routing scorer alone decides a case and a recall miss cannot hide a routing one (the Chinook suite's retrieval status needs both). Ten support and webanalytics cases carry reviewed SQL, with rows written by regenerate; the others are routing only, with no sql.
• The scorer, after the owner's extra requirement. In mode retrieval routing passes a case only when routing ranks one of its datasources first (detail.top1) and shortlists all of them (detail.shortlist, as the section above said). Mode retrieval allows no flip, so the gate fails when either drops. The board gains routing (cases, top1, shortlist), and a retrieval record's metrics gain routing_top1, routing_shortlist and routing_cases, shown on the history page and the README beside recall. result_match no longer scores a case without sql: such a case says where a question goes, not what it returns.
• Where it runs. The demo's evals.yaml still names only chinook.yaml: mode plans needs a plan for every case with SQL, and a combined run would also move the Chinook records to a new database id. nl2sql init --demo copies routing.yaml with the other suites, and it is run by name. CI's integration job adds an "Eval gate (routing)" step (eval run --mode retrieval --suite evals/routing.yaml --baseline benchmarks --out demo-ci/eval_routing.json) and uploads eval_routing.json in eval-reports; its records live under benchmarks/retrieval/chinook-support-webanalytics/.
• Tests. tests/integration/test_routing_gate.py: test_routing_suite_cases_are_approved_and_regenerated and test_routing_recall_does_not_regress (the gate against benchmarks/).
• The wide schema. Its cases are in tests/realdb/test_routing_wide.py (test_routing_on_the_wide_schema), not in routing.yaml: they name the test bed's datasources, which the demo does not have. Pagila and the whole wide schema are indexed side by side on PostgreSQL only, since indexing 1,200 tables takes about 17 minutes on CI's runner; four questions about films must go to Pagila and four about payroll runs, ledger entries, vendor contacts and campaign clicks to the wide schema. After review it runs only when the realdb workflow sets REALDB_WIDE_ROUTING (nightly, on demand, and on pull requests that change indexing/), and the realdb job's timeout is 60 minutes.
• Routing cases are a kind (after review). EvalCase.kind is answer (the default, left out of a suite file) or routing. A routing case has no sql, gets no result_match verdict, and is left out of execution accuracy in every mode: the board's accuracy.execution, the record's headline metrics and flips, and the gate's execution measures. Mode retrieval still gates it by status. The loader rejects an approved answer case with an allowed role and no sql, so a missing answer can no longer pass as "any answer". Records keep each case's kind. The Chinook suite and its digest are unchanged. The three cross-database cases are tagged with both their datasources; the suite's header names the cases with two defensible answers. Both changed the suite's digest, so the routing baseline was recorded again from CI (run 38788733052, 2026-10-15_baaae0c_retrieval.json: the same 97.4% top-1 and 98.7% shortlist).
• Measured (key-free, local embeddings). On the routing suite, routing top-1 is 76/78 and shortlist recall 77/78. The misses: "Which customers complained most about refunds?" and "How many purchases were there per month in 2013?" rank chinook first; "Which customers both raised support tickets and made a purchase on the website?" drops webanalytics at the floor. The Chinook suite's retrieval gate is unchanged by the stricter scorer (all 43 Chinook cases rank chinook first).
• Baselines, recorded on CI (run 38778183715, ubuntu-latest, Python 3.12, the same numbers as on Windows): benchmarks/retrieval/chinook-support-webanalytics/2026-10-15_a33c8ce_retrieval.json (routing top-1 97.4%, shortlist 98.7%) and a new Chinook retrieval record, benchmarks/retrieval/chinook/2026-10-15_a33c8ce_retrieval.json, which carries the routing metrics (100% / 100%) with no flip against the last. The realdb wide-schema test passed on that run too.

engine.index.check(datasource_ids=None) -> list[DriftReport]

class DriftReport(BaseModel):
    datasource_id: str
    status: Literal["unchanged", "drifted", "not_indexed", "failed"]
    live_version: str | None          # the build pointer's schema version
    checked_at: datetime
    diff: SchemaDiff | None           # live database against the live build
    impact: DriftImpact | None
    error: str | None

class ColumnChange(BaseModel):
    table: str; column: str
    changes: dict[str, tuple[Any, Any]]     # "type", "nullable", "pk": (old, new)

class SchemaDiff(BaseModel):
    datasource_id: str
    from_version: str
    to_version: str | None                  # None: the live database
    tables_added: list[str]; tables_dropped: list[str]
    columns_added: list[ColumnRef]; columns_dropped: list[ColumnRef]
    columns_changed: list[ColumnChange]
    foreign_keys_added: list[str]; foreign_keys_dropped: list[str]
    engine_type: tuple[str, str] | None     # (old, new) when the adapter type changed
    def is_empty(self) -> bool: ...

def put_schema_diff(self, diff: SchemaDiff) -> None: ...                       # keyed by datasource and to_version
def get_schema_diff(self, datasource_id: str, schema_version: str) -> SchemaDiff | None: ...

Built: schema drift (D1)

nl2sql_sdk.schema holds SchemaDiff (with counts() and is_empty()), ColumnChange and ForeignKeyRef, exported from nl2sql_sdk and SchemaDiff from nl2sql too. Three shapes differ from the sketch above, so a caller's scope can filter them without parsing names: tables are TableRefs (not str), ColumnChange.table is a TableRef, and a foreign key is a ForeignKeyRef (table, columns, referred_table, referred_columns) rather than a string. A table added or dropped is listed once, without its columns or foreign keys. nl2sql/schema/diff.py has diff_contracts(old, new, *, from_version, to_version) and visible_diff(diff, allows); a seeded random property test holds that the diff is empty exactly when the fingerprints match. The store kind is at interface version 2 (INTERFACE_VERSIONS), all four built-in stores declare it, and SqliteState adds migration 0004_schema_diffs (schema_diffs, keyed by datasource and to_version; put_schema_diff replaces and refuses a diff with no to_version; keep_snapshots deletes a pruned version's diff). Wave B takes version 3 and migration 5, Wave C version 4 and migration 6. IndexingOrchestrator._record_drift runs after the commit (a failure is logged, never fails the live build) and publishes through nl2sql/schema/drift.py's publish_drift(ctx, diff, detected_by) on ctx.bus with the engine's tenant_id. The caller's scope (binding extra): engine.index.check(datasource_ids=None, *, user_context), versions(datasource_id, *, user_context) and diff(datasource_id, schema_version=None, *, user_context) all require user_context, as datasources.schema does (M1). A caller sees only the tables its Scope.allows: drift only in denied tables is unchanged (and a stored diff reads None, has_diff false); a datasource it cannot reach is left out of check() and, named, is the same ValueError as an unknown id. schema.drifted carries the whole diff's counts and no table name; it is audited and off the stream (STREAM_EVENTS). DriftReport and SchemaVersionInfo live in nl2sql/schema/drift.py; SchemaVersionInfo has schema_version, fingerprint, live, has_diff and no created_at, since the store interface records no creation time (the sqlite version string starts with one). The check reflects with the datasource's IndexScope copied to profile: catalog, sample_values: false; a failure's error is redact_for(ctx, ...). Impact: nl2sql/schema/impact.py's drift_impact(ctx, diff, tables) runs every finder in FINDERS over the (caller-visible) diff, each filling its own field of DriftImpact (verified_queries: ImpactedQuery(datasource_id, query_id, reasons); policy_rules: ImpactedRule(role, kind, pattern); preferred_tables). Wave B adds glossary_terms as one more field and one more finder appended to FINDERS; no caller changes. Verified queries are matched from inspect_verified_sql, which now also returns columns. After review: SchemaVersionInfo.fingerprint is of visible_contract(contract, scope.allows) (the denied tables and the foreign keys to them removed), so a denied-only change looks like a statistics-only one; the policy-rule impact compares against the live tables the caller may read; a check publishes schema.drifted once per drift (_publish_check_drift remembers (from_version, the diff's sha256) per datasource in a WeakKeyDictionary keyed by the runtime, so per engine and in memory only); inspect_verified_sql's columns are (schema, table, column) with the table its qualifier's alias names (or the only table read, or ""), and an unqualified column in a multi-table query is a "may read" reason. nl2sql index --check exits 2 when nothing was checked. nl2sql.policies.patterns_matching(roles, datasource_id, tables) takes policies.roles too and returns (role, kind, pattern) triples; the boundary test now flags imports of the nl2sql.policies.patterns module only (its prefix also caught patterns_matching), and api/indexing_api.py joined its access-decision list. nl2sql index --check also takes --role (repeatable, default admin as nl2sql run) and --json; a config that fails to load exits 2.

class RunWarning(BaseModel):         # not Warning, which is a Python builtin
    code: str                     # an ErrorCode value: VALUE_NOT_FOUND, JOIN_FAN_OUT, ALL_NULL_COLUMN, ...
    node: str
    message: str                  # what the answer says
    sub_query_id: str | None = None
    detail: dict[str, Any] = {}

Built: sanity checks and typed warnings (S1)

• Where. pipeline/nodes/executor/sanity/: values.py (moved from executor/value_check.py), fanout.py, nulls.py, and __init__.py's run_checks(plan, *, adapter, allowed, tables, row_limit, row_count, frame, can_retry). ExecutorNode._sanity calls it after execute.after (H1's order is kept: a result a hook stopped is never checked), when settings.pipeline.sanity_checks is on and the execution succeeded. An empty result gets the value check; a result with rows the fan-out and null checks. Each check is wrapped: one that raises reports nothing.
• Retries. can_retry(code) is true while retries remain, the plan is not pinned, and no earlier attempt of the sub-query raised that code, so each check retries at most once. Findings carry details.attempt; sql_agent.check_execution retries a VALUE_NOT_FOUND or JOIN_FAN_OUT ERROR of the current attempt (_SANITY_RETRIES).
• Fan-out. A table is repeated when a join reaches, on its far side, columns that are not the far table's whole primary key (from relevant_tables[].primary_key; an unknown key counts as unique); every table on the near side of that join in the plan's join tree is repeated, which covers the many-to-many sum. A COUNT, SUM or AVG (not DISTINCT) of a plain column of a repeated table is confirmed by SELECT 1 FROM <the plan's FROM and joins> WHERE <the plan's WHERE> AND NOT alias.pk IS NULL GROUP BY <the plan's groups>, alias.pk HAVING COUNT(*) > 1 LIMIT 1 (after review: a COUNT(*) against COUNT(DISTINCT pk) over the whole join gave false findings on outer joins and grouped plans), built with the generator's GeneratorNode.attach_joins (now a static method) and rendered by the adapter; the key keeps its reflected case. No probe without a single-column primary key or with a table the scope denies.
• Probes and hooks (owner's decision after review). A probe, the value check's or the fan-out check's, runs outside execute.before and execute.after. So no probe runs when any hook is attached at execute.before (run_checks(..., probes=False)): only the null check does. settings.pipeline.sanity_checks: false stays the off switch. The check lives in the executor, per sub-query, not in the aggregator beside unconverted_join_keys: the repeated rows are inside one sub-query's SQL, which the aggregator never sees.
• All-null. check_nulls(frame) reads the saved result's null counts through ArtifactStore.scan_parquet, never its values.
• Sample values. check_values(..., samples=) takes a column's snapshot sample values (from relevant_tables[].columns[].stats) before the DISTINCT probe; the existence probe still runs, since samples are partial.
• RunWarning lives in nl2sql/common/errors.py, beside PipelineError, not in api/query_api.py as planned: the nodes that produce warnings must not import the API layer. It is exported as nl2sql.RunWarning. RunWarning.of(error, sub_query_id) turns an error into a warning, its details left behind; an ERROR or CRITICAL one (an attempt a retry replaced) gets detail={"recovered": True} in place of the old severity field, and the answer synthesizer skips those. GraphState.warnings and SubgraphExecutionState.warnings are List[RunWarning]; every producer builds one (the aggregator's RESULT_TRUNCATED, HOOK_DENIED and R2's JOIN_KEY_UNCONVERTED, the curated matcher's CURATED_FALLBACK, the schema retriever's fallback as SCHEMA_RETRIEVAL_FAILED, result_from_state's row-read failure as UNKNOWN_ERROR). extra="forbid", so a stray dict fails loudly.
• The answer. The synthesizer's Data Notes now carry VALUE_NOT_FOUND, JOIN_FAN_OUT and ALL_NULL_COLUMN warnings, and each is appended to the answer's warnings.
• sanity.flagged is published through nl2sql.events.publish for each finding: code, sub_query_id, retry (true for a retried ERROR).
• The playground. NodeInspector.jsx reads a warning's code only, with no fallback to error_code; the bundle is rebuilt.
• Breaking for REST clients. QueryResult.warnings[] items lost error_code and severity for code and detail.recovered. Runs are not stored across an upgrade, so no older warning is read back.
• Tests. Every e2e test that opened the session demo's shared state.db (the fake_llm fixture, the RBAC API test, both evals tests, the temperature test) now gets a private copy.

Built: the MCP server and the caller's schema (M1)

nl2sql/mcp/server.py has build_server(engine, identity) -> FastMCP (official mcp SDK, the mcp extra mcp>=1.10,<2) and PREVIEW_ROWS = 20; nl2sql mcp --role ... --user ... is cli/commands/mcp.py (caller(policies, roles, user), mcp_command). Shapes the design did not fix: every tool returns a JSON object (its structured content). ask returns status (the run's RunStatus, so Wave B's needs_clarification needs no new field; its clarification gets its own key), answer (summary, content), sub_queries (id, datasource_id, sql, columns, rows, total_rows, truncated), truncated, source, warnings (code, message), trace_id, and errors (code, node, message), approvals_needed and stop when set. An error's or warning's message is kept only for the codes written for the asker (refusals, hooks, limits, providers, the answer checks); any other message is replaced by its code and the trace id, since it may carry an exception's text. An exception out of the engine is INTERNAL_ERROR, logged; ask with a datasource_id the caller cannot reach is DATASOURCE_NOT_FOUND without running. describe_table answers status: ok with the schema() table, or not_found (a denied table, a missing one and a table in an unreachable datasource alike) with the tables the caller may read, or ambiguous when a bare name matches tables in two schemas. nl2sql mcp also starts without a role under allow_all (with a warning), and refuses a role the policies do not define, under allow_all too. After the attack review of PR #325: the identity is resolved inside each tool's guard (a resolver that raises gives INTERNAL_ERROR); SECURITY_VIOLATION always carries the generic refusal (GENERIC_REFUSAL); FastMCP is built with explicit settings; and /api/schema without ?role= is a 400. datasources.schema(datasource_id, *, user_context) filters through policy_engine.scope(...).allows (schema_view(..., allows=)), drops a foreign key to a denied table, and is in the boundary test's ACCESS_DECISIONS. The playground's /api/schema requires ?role= (400 without it) and the page sends its Ask as role. mcp is in FORBIDDEN_AT_IMPORT (which is what test_import_nl2sql_does_not_load_mcp was to check), and test_mcp_uses_only_the_public_api holds nl2sql/mcp/ to the top-level nl2sql and mcp to nl2sql/mcp/.

Built: the three kinds (A0)

observer, hook and scorer are in nl2sql/registry.py (GROUPS, INTERFACE_VERSIONS at 1, add_* and the decorators). No built-in is declared yet: E2 adds the observers and V1 the scorers. E2 declared log, audit and otel and made the registry check an observer's requires.

Built: the five scorers (V1)

result_match, outcome, faithfulness, retrieval_recall and routing are declared in packages/nl2sql/pyproject.toml under nl2sql.scorers.

flowchart TD
  D["0. This design"] --> A0["A0. Foundation"]
  A0 --> E1["E1. Run events, stream, log correlation"]
  A0 --> H1["H1. Hooks: runner, six points, registry"]
  A0 --> V1["V1. Evals: suites, Chinook moved, scorers, run, regenerate"]
  A0 --> R1["R1. Routing: phrasing entries, votes, RRF, settings"]
  E1 --> E2["E2. Observers: log, audit, otel extra; llm.call, llm.changed"]
  H1 --> H2["H2. Approval"]
  V1 --> V2["V2. nl2sql eval, the gate, records"]
  R1 --> R2["R2. Limits, join-key coercion and check"]
  E2 --> E3["E3. Price table, cost, caps"]
  E1 --> E4["E4. Index, datasource, parse-retry, policy, purge events"]
  V1 --> R3["R3. Routing suite and gate"]
  R1 --> R3
  V2 --> V3["V3. Seeding"]
  E3 --> V3
  E4 --> D1["D1. Schema drift"]
  H1 --> S1["S1. Sanity checks, typed warnings"]
  E1 --> M1["M1. MCP server"]
  E3 --> REL["Release 0.7"]
  H2 --> REL
  R2 --> REL
  R3 --> REL
  V3 --> REL
  D1 --> REL
  S1 --> REL
  M1 --> REL