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). 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/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/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 [] 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(): 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/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 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