G2 S4: hardening — bounded journal scan, node-token guard, doc/test follow-ups

docs/generator-spec.md

@@ -258,16 +258,29 @@ legend. Targets:
 
 ## 10. Currencies and determinism
 
-- **All seeded amounts are ₽, in every branch.** In the legend narrative dxb
-  invoices in AED and ala in KZT, but the v1 seeds and the vault store only
-  the ₽ figures of §1 — no generator or seed performs an FX conversion at
-  runtime, and cross-branch aggregates work directly in ₽. The **pinned demo
-  FX constants** (not live rates; internally consistent with a 90 ₽/USD
-  world): `AED = 24.50 ₽`, `KZT = 0.175 ₽`, `CNY = 12.40 ₽` — kept in
-  `reference/legend.py` solely as the fixed conversion basis for any
-  doc/evidence sentence that quotes a non-₽ figure (e.g. FOB in CNY). If a
-  future revision stores branch-local currencies, these are the constants it
-  must use.
+- **All seeded amounts in the main vault seeds are ₽, in every branch.**
+  `synthetic_seed.sql` and `postgres_oltp/seed.sql` (the seeds that back the
+  vault/serving demo and §1's rates) store only the ₽ figures — no generator
+  or seed in that path performs an FX conversion at runtime, and cross-branch
+  aggregates there work directly in ₽. In the legend narrative dxb invoices
+  in AED and ala in KZT, but nothing in the main seed path materializes that.
+  **Exception: `postgres_oltp/fanout/02_seed.sql`.** This is a separate,
+  intentional CDC/multi-currency replication fixture (the per-branch
+  Postgres→ClickHouse fan-out demo) — it seeds `orders.currency` as the local
+  tag per branch (msk = RUB, dxb = AED) on purpose, to prove the fan-out
+  carries a real per-row currency column through CDC.
+  `postgres_oltp/fanout/04_ch_bridge.sql` only replicates each branch's rows
+  into its own ClickHouse database, preserving whatever currency tag was
+  seeded — it never sums AED and RUB into one figure. The AED amounts there
+  are converted to ₽ only in this doc's/that file's comments, for illustrative
+  reference, using the FX constants below — never at runtime or in any
+  aggregation query. The **pinned demo FX constants** (not live rates;
+  internally consistent with a 90 ₽/USD world): `AED = 24.50 ₽`, `KZT = 0.175
+  ₽`, `CNY = 12.40 ₽` — kept in `reference/legend.py` solely as the fixed
+  conversion basis for any doc/evidence sentence that quotes a non-₽ figure
+  (e.g. FOB in CNY, or the fanout fixture's AED totals). If a future revision
+  stores branch-local currencies in the main seed path, these are the
+  constants it must use.
 - Generator seed constant stays `20260626`; everything derives
   deterministically from it. Timestamps keep today's mechanics (relative
   `NOW()` in serving demo, `load_ts = now64()` in vault seeds).

src/serving/api/routers/ops.py

@@ -29,6 +29,7 @@
     ReconciliationFinding,
     check_journal_vs_store,
     check_stuck_replay,
+    journal_scan_limit,
 )
 from src.serving.semantic_layer.stage_clock import (
     coerce_dt,
@@ -73,6 +74,11 @@ class StuckOrdersResponse(BaseModel):
 
 
 def _resolve_tenant_id(request: Request) -> str | None:
+    # n4 (G2 audit): None here means auth is disabled (dev/demo mode) — a
+    # real authenticated request always carries a concrete `tenant_key.tenant`
+    # (AuthMiddleware). Passed through to `fetch_pipeline_events`, where
+    # tenant_id=None is the documented cross-tenant-scan invariant, not an
+    # oversight — see QueryEngine.fetch_pipeline_events's docstring.
     tenant_key = getattr(request.state, "tenant_key", None)
     tenant_id = getattr(request.state, "tenant_id", None) or getattr(tenant_key, "tenant", None)
     return cast("str | None", tenant_id)
@@ -134,19 +140,28 @@ def _build_stuck_orders_payload(
 
     order_rows = engine.fetch_orders_by_status(ladder, tenant_id=tenant_id)
     stage_rows = engine.fetch_pipeline_events(
-        tenant_id=tenant_id, topic="orders.status", newest_first=False
+        tenant_id=tenant_id,
+        topic="orders.status",
+        newest_first=True,
+        limit=journal_scan_limit(),
     )
 
     # Latest journal row per (order, event_type), matching each order's
     # *current* status — not merely the most recent row overall (§1.4).
-    # Ascending iteration means the last write for a key wins.
+    # Descending iteration (`newest_first=True`, bounded per m13's
+    # `journal_scan_limit()` — a size safety net against an unbounded scan of
+    # the whole journal at production scale): the first row seen per key is
+    # its latest — skip once a key is already recorded.
     latest_by_key: dict[tuple[str, str], dict[str, Any]] = {}
     for row in stage_rows:
         entity_id = row.get("entity_id")
         event_type = row.get("event_type")
         if not entity_id or not event_type:
             continue
-        latest_by_key[(str(entity_id), str(event_type))] = row
+        key = (str(entity_id), str(event_type))
+        if key in latest_by_key:
+            continue
+        latest_by_key[key] = row
 
     all_items = [
         _build_stuck_order_item(

src/serving/backends/clickhouse_backend.py

@@ -358,6 +358,15 @@ def ensure_schema(self) -> None:
             expect_json=False,
             translate=False,
         )
+        # n4 (G2 audit): no `branch` column here, unlike the DuckDB embedded
+        # schema (ADR 0012 N4, `src/processing/local_pipeline.py`) — deferred,
+        # not an oversight. The three-node demo's node-ingest write path
+        # (`src/serving/node/ingest.py`) always applies through
+        # `_process_event(..., clickhouse_sink=None)`, i.e. it is DuckDB-only
+        # today; no ClickHouse caller reads or filters on a branch tag. Add
+        # `branch Nullable(String)` to the ALTER loop below (same
+        # additive/idempotent pattern) if/when node ingest is ever wired to
+        # write through ClickHouse.
         self._request(
             f"""
             CREATE TABLE IF NOT EXISTS {self._database}.pipeline_events (

src/serving/node/config.py

@@ -102,6 +102,21 @@ def resolve_node_config(env: Mapping[str, str] | None = None) -> NodeConfig:
             f"AGENTFLOW_NODE_TOKEN is required in role={role!r} — the center "
             "accepts ingest with it, an edge sends it — but is unset or empty."
         )
+    # n4 (G2 audit): the node token must not equal the public demo API key
+    # (``DEMO_API_KEY``, default ``demo-key`` — published in the demo docs).
+    # ``node/ingest.py``'s bearer check is already a distinct auth path from
+    # the API-key auth middleware, but nothing stopped an operator from
+    # *configuring* the same well-known value for both — which would let any
+    # public demo caller also authenticate as node-to-node federation. Fail
+    # fast at boot instead of silently accepting the collision.
+    demo_key = env.get("DEMO_API_KEY", "demo-key")
+    if demo_key and token == demo_key:
+        raise NodeConfigError(
+            "AGENTFLOW_NODE_TOKEN must not equal the public demo API key "
+            f"(DEMO_API_KEY={demo_key!r}) — that key is published in the demo "
+            "docs, so reusing it as the node token would let any public demo "
+            "caller authenticate as node-to-node federation."
+        )
 
     if role == "center":
         # The center does not emit, so AGENTFLOW_NODE_CENTER_URL is ignored.

src/serving/node/ingest.py

@@ -42,6 +42,20 @@
 # one process = one lock.
 _INGEST_LOCK = threading.Lock()
 
+# n4 (G2 audit): idempotency (`_existing_event_ids` below, N5) is a Python
+# check-then-act filter, not a DB-level constraint — deliberately deferred,
+# not an oversight. It is safe today because `_INGEST_LOCK` already
+# serializes every ingest write within the single center process (ADR 0012's
+# topology has exactly one center; there is no concurrent-writer scenario to
+# race). A DB-level backstop is not a clean fit either: the idempotency scope
+# is `event_id` unique *within* `topic IN ('events.validated',
+# 'events.deadletter')` only (derived rows such as the `orders.status`
+# journal entry intentionally reuse the same event_id with an `-status`
+# suffix), and DuckDB does not support partial/filtered unique indexes — a
+# table-wide `UNIQUE(event_id)` would over-constrain the journal beyond what
+# this check actually needs. Revisit if a multi-center topology (concurrent
+# writers) is ever built.
+
 
 class NodeEventBatch(BaseModel):
     """One edge->center push. ``origin_branch`` is constrained to the live edge

src/serving/semantic_layer/query/engine.py

@@ -112,6 +112,19 @@ def fetch_pipeline_events(
         stage clock, ops-surfaces-spec.md §1.2/§3.2) — orthogonal to
         ``event_type``/``validated_only``, usable without an ``entity_id``
         for a bulk scan across many entities in one query.
+
+        ``tenant_id=None`` is an explicit invariant, not incidental behaviour
+        (n4, G2 audit): it means "no tenant filter" — an unscoped, cross-
+        tenant scan. Two call shapes reach it deliberately: the webhook
+        dispatcher's background scan, which is intentionally tenant-agnostic
+        (it matches events against every registered webhook regardless of
+        tenant); and the ``/v1/ops/*`` and ``/v1/stream/*`` routers, whose
+        per-request ``tenant_id`` is resolved from ``request.state`` and is
+        ``None`` in exactly one situation — auth is disabled
+        (``AGENTFLOW_AUTH_DISABLED``/``app.state.auth_disabled``, dev/demo
+        mode only; ``AuthMiddleware`` always sets a concrete tenant on an
+        authenticated request). A genuinely multi-tenant deployment with auth
+        enabled never produces ``tenant_id=None`` from those routers.
         """
         # Deliberately uncached (unlike _table_columns): the journal is created
         # and widened by out-of-process writers, so a scan must see schema

src/serving/semantic_layer/reconciliation.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
@@ -21,6 +22,30 @@
 
 _STATUS_EVENT_PREFIX = "order.status."
 
+_DEFAULT_JOURNAL_SCAN_LIMIT = 20_000
+
+
+def journal_scan_limit() -> int:
+    """Row cap for the ops-surfaces ``orders.status`` journal scan (G2 audit
+    m13): ``fetch_pipeline_events`` with no ``limit`` is a full-table scan of
+    the whole ``pipeline_events`` journal — fine at demo scale (a few dozen
+    rows) but an unbounded-memory/DoS risk once a long-running deployment has
+    accumulated real history. Callers pair this with ``newest_first=True`` so
+    the bounded window is always the *most recent* N rows, and take the first
+    row seen per key as that key's latest state — a size safety net, not a
+    functional change, at demo scale the journal never gets close to the
+    default cap. Env-tunable via ``AGENTFLOW_OPS_JOURNAL_SCAN_LIMIT`` for
+    operators who want a different ceiling.
+    """
+    raw = (os.getenv("AGENTFLOW_OPS_JOURNAL_SCAN_LIMIT") or "").strip()
+    if not raw:
+        return _DEFAULT_JOURNAL_SCAN_LIMIT
+    try:
+        value = int(raw)
+    except ValueError:
+        return _DEFAULT_JOURNAL_SCAN_LIMIT
+    return value if value > 0 else _DEFAULT_JOURNAL_SCAN_LIMIT
+
 
 @dataclass(frozen=True)
 class ReconciliationFinding:
@@ -59,7 +84,10 @@ def check_journal_vs_store(
     ]
 
     stage_rows = engine.fetch_pipeline_events(
-        tenant_id=tenant_id, topic="orders.status", newest_first=False
+        tenant_id=tenant_id,
+        topic="orders.status",
+        newest_first=True,
+        limit=journal_scan_limit(),
     )
     latest_by_entity: dict[str, tuple[str, datetime | None]] = {}
     for row in stage_rows:
@@ -68,9 +96,14 @@ def check_journal_vs_store(
         if not entity_id or not event_type.startswith(_STATUS_EVENT_PREFIX):
             continue
         status = event_type[len(_STATUS_EVENT_PREFIX) :]
-        # Ascending iteration (`newest_first=False`): the last write per
-        # entity_id wins, matching the stuck-orders worklist's own scan.
-        latest_by_entity[str(entity_id)] = (status, coerce_dt(row.get("processed_at")))
+        # Descending iteration (`newest_first=True`, bounded per m13's
+        # `journal_scan_limit()`): the first row seen per entity_id is its
+        # latest status — skip once a key is already recorded, matching the
+        # stuck-orders worklist's own scan.
+        key = str(entity_id)
+        if key in latest_by_entity:
+            continue
+        latest_by_entity[key] = (status, coerce_dt(row.get("processed_at")))
     if not latest_by_entity:
         return []
 

tests/unit/test_dv2_business_vault_ddl.py

@@ -36,6 +36,30 @@ def _strip_comments(sql: str) -> str:
     return sql
 
 
+# B2 (brand-neutral restoration of a guard dropped during X5 removal, G2
+# audit): the header comment's illustrative "which source conventions does
+# this admit" example must only ever name known-good record_source prefixes.
+# Checked against an explicit allowlist rather than a literal bad string, so
+# this test never itself has to spell out the retired dataset's prefix to
+# guard against it reappearing.
+_RECORD_SOURCE_PREFIX_ALLOWLIST = {"1c__", "mp__", "pg_ops__", "wms__", "ref__"}
+_SOURCE_CONVENTION_PAREN_RE = re.compile(r"source convention\s*\(([^)]*)\)", re.IGNORECASE)
+_SOURCE_AGNOSTIC_RE = re.compile(r"source-agnostic:\s*([^;]*);", re.IGNORECASE)
+_PREFIX_TOKEN_RE = re.compile(r"([a-z0-9](?:[a-z0-9]|_(?!_))*)__")
+
+
+def _record_source_example_prefixes(raw: str) -> set[str]:
+    """Prefix tokens (``1c__``, ``mp__``, ...) named in a DDL header comment's
+    illustrative "source convention"/"source-agnostic" example list — not
+    every record_source value the file's views/tables actually use, just the
+    documented examples."""
+    prefixes: set[str] = set()
+    for pattern in (_SOURCE_CONVENTION_PAREN_RE, _SOURCE_AGNOSTIC_RE):
+        for match in pattern.finditer(raw):
+            prefixes.update(f"{token}__" for token in _PREFIX_TOKEN_RE.findall(match.group(1)))
+    return prefixes
+
+
 def _bv_sql_files() -> list[Path]:
     return [Path(p) for p in sorted(glob.glob(str(BV_DIR / "*.sql")))]
 
@@ -89,3 +113,14 @@ def test_customer_mdm_views_admit_all_source_conventions():
         assert "mp__" in raw, (
             f"bv_customer_mdm__{branch}.sql should document the mp__ marketplace-feed convention (domain.md §5.3)"
         )
+        example_prefixes = _record_source_example_prefixes(raw)
+        assert example_prefixes, (
+            f"bv_customer_mdm__{branch}.sql should document its record_source "
+            "source-convention examples in the header comment"
+        )
+        assert example_prefixes <= _RECORD_SOURCE_PREFIX_ALLOWLIST, (
+            f"bv_customer_mdm__{branch}.sql documents unexpected record_source "
+            f"prefix(es) {sorted(example_prefixes - _RECORD_SOURCE_PREFIX_ALLOWLIST)} "
+            f"outside the allowlist {sorted(_RECORD_SOURCE_PREFIX_ALLOWLIST)} — a "
+            "stale/regressed dataset prefix may have leaked back into the header comment"
+        )

tests/unit/test_dv2_postgres_ddl.py

@@ -44,6 +44,30 @@ def _strip_comments(sql: str) -> str:
     return sql
 
 
+# B2 (brand-neutral restoration of a guard dropped during X5 removal, G2
+# audit): the header comment's illustrative "which source conventions does
+# this admit" example must only ever name known-good record_source prefixes.
+# Checked against an explicit allowlist rather than a literal bad string, so
+# this test never itself has to spell out the retired dataset's prefix to
+# guard against it reappearing.
+_RECORD_SOURCE_PREFIX_ALLOWLIST = {"1c__", "mp__", "pg_ops__", "wms__", "ref__"}
+_SOURCE_CONVENTION_PAREN_RE = re.compile(r"source convention\s*\(([^)]*)\)", re.IGNORECASE)
+_SOURCE_AGNOSTIC_RE = re.compile(r"source-agnostic:\s*([^;]*);", re.IGNORECASE)
+_PREFIX_TOKEN_RE = re.compile(r"([a-z0-9](?:[a-z0-9]|_(?!_))*)__")
+
+
+def _record_source_example_prefixes(raw: str) -> set[str]:
+    """Prefix tokens (``1c__``, ``mp__``, ...) named in a DDL header comment's
+    illustrative "source convention"/"source-agnostic" example list — not
+    every record_source value the file's views/tables actually use, just the
+    documented examples."""
+    prefixes: set[str] = set()
+    for pattern in (_SOURCE_CONVENTION_PAREN_RE, _SOURCE_AGNOSTIC_RE):
+        for match in pattern.finditer(raw):
+            prefixes.update(f"{token}__" for token in _PREFIX_TOKEN_RE.findall(match.group(1)))
+    return prefixes
+
+
 def _pg_sql_files() -> list[Path]:
     files = [
         PG_DIR / "00_schema.sql",
@@ -169,6 +193,17 @@ def test_customer_mdm_views_admit_all_source_conventions():
     assert "mp__" in raw, (
         "03_business_vault.sql should document the mp__ marketplace-feed convention (domain.md §5.3)"
     )
+    example_prefixes = _record_source_example_prefixes(raw)
+    assert example_prefixes, (
+        "03_business_vault.sql should document its record_source source-convention "
+        "examples in the header comment"
+    )
+    assert example_prefixes <= _RECORD_SOURCE_PREFIX_ALLOWLIST, (
+        "03_business_vault.sql documents unexpected record_source prefix(es) "
+        f"{sorted(example_prefixes - _RECORD_SOURCE_PREFIX_ALLOWLIST)} outside the "
+        f"allowlist {sorted(_RECORD_SOURCE_PREFIX_ALLOWLIST)} — a stale/regressed "
+        "dataset prefix may have leaked back into the header comment"
+    )
 
 
 def test_hubs_and_links_have_bytea_primary_keys():

tests/unit/test_dv2_postgres_ingestion.py

@@ -1,11 +1,26 @@
 """Unit tests for the DV2 PostgreSQL ingestion repoint (no Docker).
 
-The raw vault moved off ClickHouse onto PostgreSQL; these tests pin that the
-per-branch order feed and the supplier reference can actually *feed* that
-PostgreSQL vault without a live database:
+The raw vault moved off ClickHouse onto PostgreSQL. Two genuinely different
+write paths are covered here, and it matters which is which (B3, G2 audit):
+
+* The **supplier/reference feed** is a real Python write path:
+  ``PostgresVaultWriter`` + ``reference/load_postgres.py`` actually connect
+  and insert (see ``test_reference_load_writes_all_tables``).
+* The **per-branch order feed** (``hub_customer``/``hub_order``/
+  ``sat_order_header__1c__*``/...) is *not* written by Python in production
+  — its live path is the SQL script ``promote_to_raw_vault_pg.sql`` (covered
+  separately below, "PostgreSQL-native OLTP -> vault promotion"). The
+  ``vault_rows`` Pydantic models used in ``ORDER_FEED_TABLES`` below exist
+  only as (a) a DDL-coverage/column-shape pin (model fields vs. the
+  committed DDL) and (b) a generic fixture shape for ``PostgresVaultWriter``'s
+  own unit tests, which exercise its column-order/batching mechanics against
+  a fake DB-API connection — reusing the order-feed shape as a stand-in, not
+  because the order feed itself calls the writer.
+
+This file otherwise pins:
 
 * ``build_insert_sql`` renders an idempotent, sqlglot-valid INSERT;
-* every column the feeds insert exists in the committed PostgreSQL DDL (the
+* every column a feed inserts exists in the committed PostgreSQL DDL (the
   guard that catches a model/DDL or generic-vs-entity-name drift);
 * ``PostgresVaultWriter`` streams rows through a fake DB-API connection in the
   right column order and batches, with hash keys preserved as ``bytes``.
@@ -167,8 +182,11 @@ def test_build_insert_sql_satellite_requires_single_hash_key():
 
 # --- inserted columns exist in the committed PostgreSQL DDL -------------------
 
-# Tables the per-branch order feed (1c) inserts into, with the row model whose
-# fields become the INSERT column list.
+# DDL-coverage pin, not a live write path (see module docstring): the tables
+# the per-branch order feed (1c) logically owns, with the vault_rows model
+# whose fields must be a subset of each table's committed DDL columns. The
+# feed's actual live writes go through promote_to_raw_vault_pg.sql, not these
+# models.
 ORDER_FEED_TABLES = [
     ("hub_customer", "01_hubs.sql", vault_rows.HubCustomer),
     ("hub_product", "01_hubs.sql", vault_rows.HubProduct),