M5 W2: fresh sources and validated quote contract

data/inventory.json

@@ -15,7 +15,10 @@
   },
   "obsolete_forced_oos": 0,
   "out_of_stock": 1413,
-  "seed": 20260606,
+ "seed": 20260606,
+  "snapshot_as_of": "2026-06-08T10:00:00-04:00",
+  "source_id": "synthetic-inventory",
+  "source_version": "seed-20260606",
   "total": 9487
  },
  "records": {
@@ -37968,4 +37971,4 @@
    "qty_on_hand": 13
   }
  }
-}
+}

data/m5_baseline_defects.json

@@ -3,11 +3,9 @@
   "baseline_commit": "8001643fe503254ed9f12f0996894a49efe52342",
   "contract": "A probe passes only when its positive control works and the named baseline defect is behaviorally reproduced.",
   "defects": [
-    {"id": "M5-DEF-05", "owner_wave": "W2", "test": "tests/test_m5_baseline_defects.py::test_defect_quote_availability_uses_default_quantity_one", "expected_reason": "fulfillment_called_with_quantity_one"},
     {"id": "M5-DEF-07", "owner_wave": "W3", "test": "tests/test_m5_baseline_defects.py::test_defect_retry_does_not_repair_missing_queue_row", "expected_reason": "idempotent_retry_leaves_orphan_quote"},
     {"id": "M5-DEF-08", "owner_wave": "W5", "test": "tests/test_m5_baseline_defects.py::test_defect_queue_claim_is_not_exclusive", "expected_reason": "same_pending_job_claimed_twice"},
     {"id": "M5-DEF-09", "owner_wave": "W3", "test": "tests/test_m5_baseline_defects.py::test_defect_document_guard_accepts_cross_line_quantity_swap", "expected_reason": "flat_fact_set_loses_line_association"},
-    {"id": "M5-DEF-10", "owner_wave": "W2", "test": "tests/test_m5_baseline_defects.py::test_defect_revision_lookup_ignores_tenant_when_account_ids_collide", "expected_reason": "cross_tenant_revision_opened"},
     {"id": "M5-DEF-11", "owner_wave": "W7", "test": "tests/test_m5_baseline_defects.py::test_defect_latency_harness_rejects_chunked_readback", "expected_reason": "latency_harness_expects_immediate_quote_number"}
   ]
 }

data/m5_legacy_schema_lock.json

@@ -3,22 +3,24 @@
   "adoption": "migrate_backfill_preserve_nonempty",
   "components": {
     "customer": {
-      "schema_sha256": "0194ba0765b1012184fde37fdc7c6a020bc41e8e958e343edb66af434bf84989",
+      "schema_sha256": "d925196576d2c375572f62e1b71340203c8daf7f6f92a4cbc30d051dd480e648",
       "tables": [
         "accounts"
       ],
       "row_counts": {
         "accounts": 2
       },
-      "known_gap": "accounts table has no email or tenant_id column"
+      "known_gap": "legacy phone/email/contact verification remains explicit unknown until sourced"
     },
     "pricebook": {
-      "schema_sha256": "0cc59931cad2b089968da0717c5f3b633cba61c00a3c775451fd9f931f4a830f",
+      "schema_sha256": "0362ad23826a75ac58cd09fa3e782185c81ad545164133ff66d05baf4ad8e461",
       "tables": [
+        "pricebook_meta",
         "prices",
         "tiers"
       ],
       "row_counts": {
+        "pricebook_meta": 1,
         "prices": 1,
         "tiers": 1
       }
@@ -46,5 +48,5 @@
       ]
     }
   },
-  "combined_sha256": "6f2c894fc66dceef4f54d3038950458830d1bbef60fdb4f68bb5d508db0b2b85"
+  "combined_sha256": "6471bcd46b9fc33be420adf67e7183cb997364d36d8d94ec87113c6a32612de0"
 }

scripts/m5_legacy_inventory.py

@@ -16,7 +16,15 @@
 
 from gateway import Account  # noqa: E402
 from gateway.db_adapters import SqliteCustomerDB, SqlitePriceBook  # noqa: E402
-from quoting.models import QuoteDocument, QuoteHeader, QuoteLine, QuoteTerms  # noqa: E402
+from quoting.models import (  # noqa: E402
+    DeliveryPreflight,
+    QuoteDocument,
+    QuoteHeader,
+    QuoteLine,
+    QuotePresentation,
+    QuoteProvenance,
+    QuoteTerms,
+)
 from quoting.store import SqliteQuoteStore, order_content_hash  # noqa: E402
 from quoting.totals import build_totals, extended_price  # noqa: E402
 
@@ -42,6 +50,7 @@ def _sample_document(number: str) -> QuoteDocument:
     line = QuoteLine(
         line_no=1,
         sku="K5-24SBC",
+        tenant_id="tenant_001",
         catalog_description="chrome stack",
         requested_qty=10,
         uom="each",
@@ -50,6 +59,7 @@ def _sample_document(number: str) -> QuoteDocument:
         account_id="1001",
         extended_price=extended_price(Decimal("24.99"), 10),
         availability_note="in stock",
+        availability_status="available",
         ship_date=now + timedelta(days=1),
         ship_date_as_of=now,
         resolution_source="translator:parser",
@@ -70,6 +80,14 @@ def _sample_document(number: str) -> QuoteDocument:
         lines=(line,),
         totals=build_totals((line,)),
         terms=QuoteTerms(30, "Prices subject to change.", "Net 30."),
+        presentation=QuotePresentation(
+            "Legacy Seller", "", "legacy", "Legacy Customer", "1001"),
+        provenance=QuoteProvenance(
+            "legacy-build", "renderer-v1", "guard-v1", "agent-v1",
+            "legacy-catalog", ("legacy-price",), ("legacy-inventory",)),
+        delivery_preflight=DeliveryPreflight(
+            "email", "legacy-contact", "v1", "fingerprint",
+            "l***@example.test", True),
     )
 
 
@@ -112,7 +130,10 @@ def compute() -> dict:
                 "schema_sha256": customer_hash,
                 "tables": customer_tables,
                 "row_counts": _row_counts(customers._conn, customer_tables),
-                "known_gap": "accounts table has no email or tenant_id column",
+                "known_gap": (
+                    "legacy phone/email/contact verification remains explicit "
+                    "unknown until sourced"
+                ),
             },
             "pricebook": {
                 "schema_sha256": price_hash,

src/fulfillment/engine.py

@@ -45,6 +45,9 @@ class InventoryRecord:
     sku: str
     qty_on_hand: int
     lead_time_days: int | None  # None iff qty_on_hand > 0
+    source_id: str = ''
+    source_version: str = ''
+    as_of: datetime | None = None
 
     def __post_init__(self) -> None:
         if self.qty_on_hand < 0:
@@ -58,6 +61,9 @@ def __post_init__(self) -> None:
             raise ValueError(
                 f'{self.sku}: out-of-stock record requires lead_time_days >= 1'
             )
+        if self.as_of is not None and (
+                self.as_of.tzinfo is None or self.as_of.utcoffset() is None):
+            raise ValueError(f'{self.sku}: inventory as_of must be timezone-aware')
 
 
 @dataclass(frozen=True)
@@ -133,8 +139,14 @@ def load_inventory(path: str | Path) -> dict[str, InventoryRecord]:
     fails loudly here at startup rather than at quote time.
     """
     raw = json.loads(Path(path).read_text())
+    meta = raw.get('_meta', {})
+    as_of_raw = meta.get('snapshot_as_of')
+    as_of = datetime.fromisoformat(as_of_raw) if as_of_raw else None
     return {
         sku: InventoryRecord(sku=sku, qty_on_hand=rec['qty_on_hand'],
-                             lead_time_days=rec['lead_time_days'])
+                             lead_time_days=rec['lead_time_days'],
+                             source_id=meta.get('source_id', ''),
+                             source_version=meta.get('source_version', ''),
+                             as_of=as_of)
         for sku, rec in raw['records'].items()
     }

src/gateway/answers.py

@@ -19,6 +19,7 @@
     PriceAnswer,
 )
 from gateway.pricebook import PriceBook
+from gateway.source_freshness import source_is_fresh
 
 _MONTHS = ('January', 'February', 'March', 'April', 'May', 'June', 'July',
            'August', 'September', 'October', 'November', 'December')
@@ -56,7 +57,8 @@ def availability(sku: str, *, inventory: dict[str, InventoryRecord],
                  received_at, catalog_version: str, qty: int = 1
                  ) -> AvailabilityAnswer | None:
     rec = inventory.get(sku)
-    if rec is None:
+    if rec is None or not source_is_fresh(
+            rec.as_of, received_at, rec.source_id):
         return None
     try:
         res = ship_date(rec, qty, received_at,
@@ -68,15 +70,19 @@ def availability(sku: str, *, inventory: dict[str, InventoryRecord],
             quantity_on_hand=rec.qty_on_hand, ship_by_iso='',
             basis='beyond_calendar_horizon',
             plain=f"I can't quote a ship date for {sku} that far out yet.",
-            catalog_version=catalog_version)
+            catalog_version=catalog_version, source_id=rec.source_id,
+            source_version=rec.source_version,
+            source_as_of=rec.as_of.isoformat() if rec.as_of else '')
     ship_iso = res.ship_by.isoformat()
     return AvailabilityAnswer(
         sku=sku, in_stock=rec.qty_on_hand > 0,
         quantity_on_hand=rec.qty_on_hand, ship_by_iso=ship_iso,
         basis=res.basis,
         plain=_plain_availability(sku, rec.qty_on_hand > 0,
                                   rec.qty_on_hand, ship_iso),
-        catalog_version=catalog_version)
+        catalog_version=catalog_version, source_id=rec.source_id,
+        source_version=rec.source_version,
+        source_as_of=rec.as_of.isoformat() if rec.as_of else '')
 
 
 class PricingRefused(Exception):
@@ -85,18 +91,23 @@ class PricingRefused(Exception):
 
 
 def pricing(sku: str, auth: AuthorizationDecision, *, pricebook: PriceBook,
-            account_tier_of) -> PriceAnswer:
+            account_tier_of, source_now=None) -> PriceAnswer:
     # Second, independent gate (#10 defense-in-depth): even if the session
     # gate were bypassed, the pricing service itself refuses without a grant.
     if not auth.granted or auth.source in ('unverified', 'cross_account_denied'):
         raise PricingRefused(
             f'pricing refused for {sku}: authorization not granted '
             f'(source={auth.source})')
     tier = account_tier_of(auth.account_id)
-    price = pricebook.price(sku, tier)
-    if price is None:
+    snapshot = pricebook.snapshot(sku, tier, auth.account_id)
+    if snapshot is None:
         raise PricingRefused(f'no price on file for {sku} at tier {tier}')
+    if source_now is not None and not source_is_fresh(
+            snapshot.as_of, source_now, snapshot.source_id):
+        raise PricingRefused(f'price source is stale or unreadable for {sku}')
     return PriceAnswer(
-        sku=sku, account_id=auth.account_id, unit_price=price,
+        sku=sku, account_id=auth.account_id, unit_price=snapshot.unit_price,
         source=auth.source,
-        plain=f"For your account, {sku} is ${price:.2f} each.")
+        plain=f"For your account, {sku} is ${snapshot.unit_price:.2f} each.",
+        source_id=snapshot.source_id, source_version=snapshot.source_version,
+        source_as_of=snapshot.as_of.isoformat())

src/gateway/conversation_snapshot.py

@@ -23,6 +23,7 @@
 from gateway.order import DraftLine, LineState, Order
 from gateway.order_parser import OrderClause
 from gateway.session import Session
+from quoting.models import DeliveryPreflight
 from sku_translator.quantity import RequestedQty
 
 SNAPSHOT_VERSION = 1
@@ -90,6 +91,10 @@ def dump_call(gateway, caller_id: str) -> dict[str, Any]:
             'chunks': [[_line(line) for line in chunk] for chunk in chunks[0]],
             'next_index': chunks[1]},
         'completed_order_readback': caller_id in gateway._completed_order_readback,
+        'delivery_preflight': (
+            asdict(gateway._delivery_preflight_by_call[caller_id])
+            if caller_id in gateway._delivery_preflight_by_call else None),
+        'pending_quote_reference': gateway._pending_quote_reference.get(caller_id),
         'pending_clauses': [
             {'raw_span': clause.raw_span, 'part_text': clause.part_text,
              'requested_qty': asdict(clause.requested_qty), 'uom': clause.uom,
@@ -157,6 +162,11 @@ def load_call(gateway, caller_id: str, value: dict[str, Any]) -> str | None:
         gateway._pending_order_readback.pop(caller_id, None)
     _restore_set(gateway._completed_order_readback, caller_id,
                  value['completed_order_readback'])
+    raw_preflight = value.get('delivery_preflight')
+    _restore_mapping(gateway._delivery_preflight_by_call, caller_id,
+                     DeliveryPreflight(**raw_preflight) if raw_preflight else None)
+    _restore_mapping(gateway._pending_quote_reference, caller_id,
+                     value.get('pending_quote_reference'))
     pending_clauses = tuple(
         OrderClause(**{**clause,
                        'requested_qty': RequestedQty(**clause['requested_qty'])})

src/gateway/db_adapters.py

@@ -15,10 +15,11 @@
 from __future__ import annotations
 
 import sqlite3
+from datetime import datetime, timezone
 from decimal import Decimal
 
 from gateway.models import Account
-from gateway.pricebook import to_cents
+from gateway.pricebook import PriceSnapshot, to_cents
 
 
 def _connect(source) -> sqlite3.Connection:
@@ -37,10 +38,26 @@ class SqliteCustomerDB:
     def __init__(self, source) -> None:
         self._conn = _connect(source)
         self._conn.row_factory = sqlite3.Row
+        self._migrate_contact_columns()
+
+    def _migrate_contact_columns(self) -> None:
+        columns = {row['name'] for row in self._conn.execute(
+            'PRAGMA table_info(accounts)').fetchall()}
+        additions = {
+            'email': 'TEXT', 'tenant_id': 'TEXT', 'contact_id': 'TEXT',
+            'contact_version': 'TEXT',
+            'phone_verified': 'INTEGER NOT NULL DEFAULT 0',
+            'email_verified': 'INTEGER NOT NULL DEFAULT 0',
+        }
+        for name, declaration in additions.items():
+            if name not in columns:
+                self._conn.execute(
+                    f'ALTER TABLE accounts ADD COLUMN {name} {declaration}')
+        self._conn.commit()
 
     def by_number(self, account_no: str) -> Account | None:
         row = self._conn.execute(
-            'SELECT account_id, name, phone FROM accounts WHERE account_id = ?',
+            'SELECT * FROM accounts WHERE account_id = ?',
             (account_no.strip(),)).fetchone()
         return self._account(row) if row else None
 
@@ -52,24 +69,34 @@ def by_name(self, name: str) -> list[Account]:
         # the 0/1/many rule. LIKE param is escaped against wildcard injection.
         like = '%' + q.replace('%', r'\%').replace('_', r'\_') + '%'
         rows = self._conn.execute(
-            "SELECT account_id, name, phone FROM accounts "
+            "SELECT * FROM accounts "
             "WHERE LOWER(name) LIKE ? ESCAPE '\\' ORDER BY account_id",
             (like,)).fetchall()
         return [self._account(r) for r in rows]
 
     @staticmethod
     def _account(row) -> Account:
         return Account(account_id=row['account_id'], name=row['name'],
-                       phone=row['phone'])
+                       phone=row['phone'], email=row['email'],
+                       tenant_id=row['tenant_id'] or '',
+                       contact_id=row['contact_id'] or '',
+                       contact_version=row['contact_version'] or '',
+                       phone_verified=bool(row['phone_verified']),
+                       email_verified=bool(row['email_verified']))
 
     @classmethod
     def build(cls, path, accounts: list[Account]) -> 'SqliteCustomerDB':
         conn = _connect(path)
         conn.execute('CREATE TABLE IF NOT EXISTS accounts '
-                     '(account_id TEXT PRIMARY KEY, name TEXT, phone TEXT)')
+                     '(account_id TEXT PRIMARY KEY, name TEXT, phone TEXT, '
+                     'email TEXT, tenant_id TEXT, contact_id TEXT, '
+                     'contact_version TEXT, phone_verified INTEGER NOT NULL DEFAULT 0, '
+                     'email_verified INTEGER NOT NULL DEFAULT 0)')
         conn.execute('DELETE FROM accounts')
-        conn.executemany('INSERT INTO accounts VALUES (?, ?, ?)',
-                         [(a.account_id, a.name, a.phone) for a in accounts])
+        conn.executemany('INSERT INTO accounts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
+                         [(a.account_id, a.name, a.phone, a.email, a.tenant_id,
+                           a.contact_id, a.contact_version, int(a.phone_verified),
+                           int(a.email_verified)) for a in accounts])
         conn.commit()
         return cls(conn)
 
@@ -99,19 +126,45 @@ def price(self, sku: str, tier: str) -> Decimal | None:
         m = Decimal(mult['multiplier']) if mult else Decimal('1.0')
         return to_cents(Decimal(base['base']) * m)
 
+    def snapshot(self, sku: str, tier: str,
+                 account_id: str) -> PriceSnapshot | None:
+        value = self.price(sku, tier)
+        try:
+            meta = self._conn.execute(
+                'SELECT source_id, source_version, as_of FROM pricebook_meta '
+                'WHERE singleton=1').fetchone()
+        except sqlite3.OperationalError:
+            return None
+        if value is None or meta is None or not meta['as_of']:
+            return None
+        return PriceSnapshot(
+            sku, account_id, tier, value, meta['source_id'],
+            meta['source_version'], datetime.fromisoformat(meta['as_of']))
+
     @classmethod
     def build(cls, path, base_by_sku: dict[str, Decimal],
-              tier_multiplier: dict[str, Decimal]) -> 'SqlitePriceBook':
+              tier_multiplier: dict[str, Decimal], *,
+              source_id: str = 'sqlite-pricebook',
+              source_version: str = 'fixture-v1',
+              as_of: datetime | None = None) -> 'SqlitePriceBook':
         conn = _connect(path)
         conn.execute('CREATE TABLE IF NOT EXISTS prices '
                      '(sku TEXT PRIMARY KEY, base TEXT)')
         conn.execute('CREATE TABLE IF NOT EXISTS tiers '
                      '(tier TEXT PRIMARY KEY, multiplier TEXT)')
+        conn.execute('CREATE TABLE IF NOT EXISTS pricebook_meta '
+                     '(singleton INTEGER PRIMARY KEY CHECK(singleton=1), '
+                     'source_id TEXT, source_version TEXT, as_of TEXT)')
         conn.execute('DELETE FROM prices')
         conn.execute('DELETE FROM tiers')
         conn.executemany('INSERT INTO prices VALUES (?, ?)',
                          [(sku, str(base)) for sku, base in base_by_sku.items()])
         conn.executemany('INSERT INTO tiers VALUES (?, ?)',
                          [(tier, str(mult)) for tier, mult in tier_multiplier.items()])
+        snapshot_time = as_of or datetime.now(timezone.utc)
+        conn.execute('INSERT INTO pricebook_meta VALUES(1,?,?,?) '
+                     'ON CONFLICT(singleton) DO UPDATE SET source_id=excluded.source_id,'
+                     'source_version=excluded.source_version,as_of=excluded.as_of',
+                     (source_id, source_version, snapshot_time.isoformat()))
         conn.commit()
         return cls(conn)