From abe6de0d9322f3d02de8b47a565332e9a98ad5e5 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 6 Jul 2026 04:40:32 +0300 Subject: [PATCH 1/5] =?UTF-8?q?fix(g2-s4):=20m13=20=E2=80=94=20bound=20the?= =?UTF-8?q?=20ops-surfaces=20orders.status=20journal=20scan?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fetch_pipeline_events(topic="orders.status") in the stuck-orders worklist (ops.py) and R1 reconciliation (reconciliation.py) had no limit — an unbounded full-table scan of the pipeline_events journal, an OOM/DoS risk at production scale (fine at demo scale today). Both now pass newest_first=True + a bounded journal_scan_limit() (env-tunable via AGENTFLOW_OPS_JOURNAL_SCAN_LIMIT, default 20k rows), and the "latest row per key" dict-building logic is flipped to "first row seen per key wins" to match the new descending scan order. A safety net, not a functional change — demo-scale data is far under the default cap either way. G2 audit m13. --- src/serving/api/routers/ops.py | 21 ++++++++-- src/serving/semantic_layer/query/engine.py | 13 +++++++ src/serving/semantic_layer/reconciliation.py | 41 ++++++++++++++++++-- 3 files changed, 68 insertions(+), 7 deletions(-) diff --git a/src/serving/api/routers/ops.py b/src/serving/api/routers/ops.py index fb6d461..de7153f 100644 --- a/src/serving/api/routers/ops.py +++ b/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( diff --git a/src/serving/semantic_layer/query/engine.py b/src/serving/semantic_layer/query/engine.py index 801711e..10efc3f 100644 --- a/src/serving/semantic_layer/query/engine.py +++ b/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 diff --git a/src/serving/semantic_layer/reconciliation.py b/src/serving/semantic_layer/reconciliation.py index ffc21b0..44eb2fe 100644 --- a/src/serving/semantic_layer/reconciliation.py +++ b/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 [] From 07ecf1cc0e3292b4bbca785ce2acf5ed02da5e7c Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 6 Jul 2026 04:41:05 +0300 Subject: [PATCH 2/5] =?UTF-8?q?fix(g2-s4):=20n4=20hardening=20=E2=80=94=20?= =?UTF-8?q?node-token=20guard,=20idempotency/branch-column=20notes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four small hardening items from the G2 audit (n4), none blockers: 1. node/config.py: resolve_node_config now fails fast if AGENTFLOW_NODE_TOKEN equals the public demo API key (DEMO_API_KEY, default "demo-key") — that key is published in the demo docs, so reusing it as the node-federation bearer token would let any public demo caller also authenticate as node-to-node federation. Covered by new tests in test_node_config.py. 2. node/ingest.py: documented-as-deferred. Idempotency is a Python check-then-act filter (_existing_event_ids), not a DB constraint — safe today because _INGEST_LOCK already serializes every write in the single center process (no concurrent-writer scenario exists). A DB backstop is also not a clean fit: the idempotency scope is narrower than the whole table (event_id unique only within topic IN ('events.validated', 'events.deadletter')), and DuckDB has no partial/filtered unique index support, so a table-wide UNIQUE(event_id) would over-constrain the journal. Revisit if a multi-center topology is ever built. 3. query/engine.py (committed alongside m13 in the previous commit): documented the tenant_id=None invariant on fetch_pipeline_events explicitly, and cross-referenced it in ops.py's _resolve_tenant_id — confirmed the only reachable path is AUTH_DISABLED dev/demo mode, not an oversight. 4. clickhouse_backend.py: documented-as-deferred. No `branch` column on ClickHouse's pipeline_events (unlike the DuckDB embedded schema, ADR 0012 N4) because the three-node demo's node-ingest path is DuckDB-only today (clickhouse_sink=None) — no ClickHouse caller needs it yet. Comment notes the additive ALTER-loop pattern to use if that ever changes. G2 audit n4. --- src/serving/backends/clickhouse_backend.py | 9 +++++ src/serving/node/config.py | 15 ++++++++ src/serving/node/ingest.py | 14 ++++++++ tests/unit/test_node_config.py | 40 ++++++++++++++++++++++ 4 files changed, 78 insertions(+) diff --git a/src/serving/backends/clickhouse_backend.py b/src/serving/backends/clickhouse_backend.py index 750a05a..c13b281 100644 --- a/src/serving/backends/clickhouse_backend.py +++ b/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 ( diff --git a/src/serving/node/config.py b/src/serving/node/config.py index 698a385..d1bd7f5 100644 --- a/src/serving/node/config.py +++ b/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. diff --git a/src/serving/node/ingest.py b/src/serving/node/ingest.py index 33c2ab0..6fa30f5 100644 --- a/src/serving/node/ingest.py +++ b/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 diff --git a/tests/unit/test_node_config.py b/tests/unit/test_node_config.py index efc5f8d..9a5fa62 100644 --- a/tests/unit/test_node_config.py +++ b/tests/unit/test_node_config.py @@ -195,3 +195,43 @@ def test_config_is_frozen() -> None: cfg = NodeConfig(role="standalone") with pytest.raises((AttributeError, TypeError)): cfg.role = "center" # type: ignore[misc] + + +def test_center_token_equal_to_default_demo_key_fails_fast() -> None: + # n4 (G2 audit): the well-known public demo API key must never double as + # the node-federation bearer token. + with pytest.raises(NodeConfigError, match="demo API key"): + resolve_node_config({"AGENTFLOW_NODE_ROLE": "center", "AGENTFLOW_NODE_TOKEN": "demo-key"}) + + +def test_edge_token_equal_to_default_demo_key_fails_fast() -> None: + with pytest.raises(NodeConfigError, match="demo API key"): + resolve_node_config( + { + "AGENTFLOW_NODE_ROLE": "edge", + "AGENTFLOW_NODE_BRANCH": "spb", + "AGENTFLOW_NODE_CENTER_URL": "https://example.test", + "AGENTFLOW_NODE_TOKEN": "demo-key", + } + ) + + +def test_token_equal_to_custom_demo_api_key_env_fails_fast() -> None: + # DEMO_API_KEY overrides the default "demo-key" (src/serving/api/main.py) — + # the guard must compare against whatever value is actually configured, + # not just the hardcoded default. + with pytest.raises(NodeConfigError, match="demo API key"): + resolve_node_config( + { + "AGENTFLOW_NODE_ROLE": "center", + "AGENTFLOW_NODE_TOKEN": "custom-demo-value", + "DEMO_API_KEY": "custom-demo-value", + } + ) + + +def test_token_distinct_from_demo_key_is_fine() -> None: + cfg = resolve_node_config( + {"AGENTFLOW_NODE_ROLE": "center", "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN} + ) + assert cfg.token == _NODE_TOKEN From 7c991e089f7e4abb0fce715149faf6445d94f8d0 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 6 Jul 2026 04:41:14 +0300 Subject: [PATCH 3/5] =?UTF-8?q?docs(g2-s4):=20B1=20=E2=80=94=20scope=20gen?= =?UTF-8?q?erator-spec=20=C2=A710's=20=E2=82=BD-only=20claim,=20carve=20ou?= =?UTF-8?q?t=20fanout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit §10 (edited in the recently-merged S3/n2 fix) stated verbatim that no generator or seed performs FX conversion at runtime and every branch seeds in ₽ — contradicted by postgres_oltp/fanout/02_seed.sql, which seeds the dxb branch's orders with currency='AED' on purpose (a CDC/multi-currency replication fixture proving the fan-out carries a real per-row currency column through CDC). No runtime bug: 04_ch_bridge.sql just replicates rows per-branch preserving the currency tag, nothing sums AED+RUB into one figure. Reworded to scope the ₽-only claim to the main vault seeds (synthetic_seed.sql, postgres_oltp/seed.sql) and explicitly carve out the fanout fixture as an intentional exception, converted to ₽ only in doc/comment prose via the §10 FX constants, never at runtime. G2 audit B1 (follow-up to the S3/n2 fix). --- docs/generator-spec.md | 33 +++++++++++++++++++++++---------- 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/docs/generator-spec.md b/docs/generator-spec.md index 4bd2fe0..369aed4 100644 --- a/docs/generator-spec.md +++ b/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). From 96de7da7ab1e7fc65b0a6c5216d50e2c51f652b3 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 6 Jul 2026 04:41:22 +0300 Subject: [PATCH 4/5] =?UTF-8?q?test(g2-s4):=20B2=20=E2=80=94=20restore=20b?= =?UTF-8?q?rand-neutral=20record=5Fsource=20prefix=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two DDL tests used to assert "x5__" not in raw as a negative guard against the old retail-dataset record_source prefix reappearing in header comments; removed during X5 removal (S2b) because keeping the literal string "x5__" in test source would itself violate the no-X5-anywhere rule. That left the invariant "the DDL's record_source example prefixes only use known-good conventions" untested. Restored brand-neutrally: extract the prefix tokens (e.g. 1c__, mp__, pg_ops__) named in each file's "source convention"/"source-agnostic" header-comment example, and assert they're all within an explicit allowlist {1c__, mp__, pg_ops__, wms__, ref__} — the conventions actually used across warehouse/agentflow/dv2/business_vault/ and the postgres DDL, confirmed by inspection. Catches a reintroduced bad prefix without the test ever naming it. G2 audit B2. --- tests/unit/test_dv2_business_vault_ddl.py | 35 +++++++++++++++++++++++ tests/unit/test_dv2_postgres_ddl.py | 35 +++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/tests/unit/test_dv2_business_vault_ddl.py b/tests/unit/test_dv2_business_vault_ddl.py index 0b1ab30..d98a7f4 100644 --- a/tests/unit/test_dv2_business_vault_ddl.py +++ b/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" + ) diff --git a/tests/unit/test_dv2_postgres_ddl.py b/tests/unit/test_dv2_postgres_ddl.py index 9b00d6c..33c168b 100644 --- a/tests/unit/test_dv2_postgres_ddl.py +++ b/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(): From 83f9ed4ef1e2394bc4c0ac488a9dae456f60fa38 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 6 Jul 2026 04:41:31 +0300 Subject: [PATCH 5/5] =?UTF-8?q?docs(g2-s4):=20B3=20=E2=80=94=20clarify=20v?= =?UTF-8?q?ault=5Frows.py=20is=20test-only,=20not=20a=20live=20write=20pat?= =?UTF-8?q?h?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit vault_rows.py is a production module (warehouse/agentflow/dv2/loaders/) that nothing in the shipped code imports — only tests/unit/test_dv2_postgres_ingestion.py does. It was relocated here during X5 removal (S2b) to carry generic vault-row Pydantic models formerly in the deleted X5 loader's schemas.py. Confirmed no other importer exists (grep across the repo); left in place rather than moved to tests/ (less invasive — warehouse/agentflow/dv2/* is already the established convention other test files import DDL/generation helpers from, e.g. dialects.py, generate_satellites.py; mypy's src/-only scope doesn't touch warehouse/ either way). Reworded vault_rows.py's module docstring and the test file's module docstring / ORDER_FEED_TABLES comment, which described a Python Postgres per-branch-order-feed write path that no longer exists — the real feed today is the SQL script promote_to_raw_vault_pg.sql (covered separately, further down the same test file). vault_rows's models exist only for (a) DDL-coverage/column-shape pinning and (b) generic fixture shapes for PostgresVaultWriter's own unit tests, not because the order feed calls the writer. G2 audit B3. --- tests/unit/test_dv2_postgres_ingestion.py | 30 +++++++++++++++---- warehouse/agentflow/dv2/loaders/vault_rows.py | 25 +++++++++++----- 2 files changed, 42 insertions(+), 13 deletions(-) diff --git a/tests/unit/test_dv2_postgres_ingestion.py b/tests/unit/test_dv2_postgres_ingestion.py index 742e437..f483afe 100644 --- a/tests/unit/test_dv2_postgres_ingestion.py +++ b/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), diff --git a/warehouse/agentflow/dv2/loaders/vault_rows.py b/warehouse/agentflow/dv2/loaders/vault_rows.py index d5e2511..cc00c63 100644 --- a/warehouse/agentflow/dv2/loaders/vault_rows.py +++ b/warehouse/agentflow/dv2/loaders/vault_rows.py @@ -1,11 +1,22 @@ -"""Vault-generic raw-vault row models shared by DV2 PostgreSQL ingestion. - -These pydantic models describe the hub / link / order-satellite shape used by -every per-branch order feed (``sat_order_header__1c__*`` / -``sat_order_pricing__1c__*``) and by ``PostgresVaultWriter``'s column-order / -DDL-coverage tests. They are entity-generic (not tied to any one source +"""Vault-generic raw-vault row models — test fixtures, not a live write path. + +These pydantic models describe the hub / link / order-satellite shape of the +per-branch order feed (``sat_order_header__1c__*`` / ``sat_order_pricing__1c__*``). +They were relocated here from the deleted X5 Retail Hero loader during its +removal (G2 audit S2b) because ``PostgresVaultWriter``'s own tests +(``tests/unit/test_dv2_postgres_ingestion.py``) needed generic row shapes to +keep running; that test file is their only importer today. + +**Not a live write path**: the per-branch order feed's actual production +writes go through the SQL script ``promote_to_raw_vault_pg.sql``, not Python. +These models exist purely as (a) a DDL-coverage/column-shape pin — their +fields must be a subset of the committed PostgreSQL DDL columns — and (b) a +generic fixture shape for ``PostgresVaultWriter``'s own unit tests (column +order, batching, hash-key handling), reusing the order-feed shape as a +convenient stand-in. They are entity-generic (not tied to any one source system or dataset) — see ``reference/vault_mapping.py`` for the analogous -models used by the supplier/product reference feed. +models used by the supplier/product reference feed, which *is* a live +Python write path (``reference/load_postgres.py``). """ from __future__ import annotations