M5d: parity-gated off-call quote delivery

docs/M5_BUILD_JOURNAL.md

@@ -529,3 +529,49 @@ been measured. Multi-template theming stays out of scope for M5 regardless
 of renderer (spec §3.4) — noted in D15 as reportlab's one real cost versus
 WeasyPrint, since a future templating feature would fit WeasyPrint's
 HTML/CSS model better than reportlab's programmatic API.
+
+## 2026-07-10 — Session 6: M5d part 1 — off-call delivery worker
+
+Closed the render-to-delivery gap without moving document work onto the call
+path. Caller assent now saves the frozen quote and enqueues its revision in the
+same SQLite quote store. A deterministic `run_once` worker claims pending work,
+loads the verified account contact, renders PDF and XLSX, runs
+`verify_rendered_document` on both artifacts, and only then invokes the
+`Deliverer` protocol. CI uses the append-only `SimulatedDeliverer`; live
+email/SMS adapters remain credential-gated follow-up work.
+
+The parity gate is proven on the worker path, not merely imported: the frozen
+external oracle substitutes bytes rendered from an altered-price document,
+counts one real guard invocation, and observes zero delivery payloads, a
+blocked queue row, and a `QUOTE_DELIVERY_BLOCKED` journal event. The positive
+control drives the real `Gateway.converse` assent path, proves no renderer runs
+in-call, then processes one pending revision and observes one delivery for the
+verified account. Missing contact-of-record also fails closed and is journaled.
+
+Signed retrieval links use HMAC-SHA256 over quote number, revision, and the
+quote's own `valid_until`. Runtime configuration now separates the secret
+(`SKU_QUOTE_LINK_SECRET`) from the public HTTPS origin
+(`SKU_QUOTE_PUBLIC_BASE_URL`) and delivers a complete `/quote/{token}` URL, not
+a bare token. The external oracle mounts the runtime route against the same
+store and secret: the delivered URL returns HTTP 200, `application/pdf`, and
+the exact expected bytes; tampered tokens and the exact expiry boundary return
+404. The reserved example origin and insecure link secret both emit warnings
+until deployment configuration supplies real values.
+
+**Measured issuance baseline (synthetic twin, this reference machine):** 50
+issuances spanning one through six captured lines. The repository harness
+reported p50 0.643 ms and p95 1.432 ms; the separately hash-pinned oracle
+reported p95 2.186 ms. These are observed fixture numbers, not a production SLO
+or LTE claim, and the harness deliberately does not assert the provisional
+300 ms target.
+
+**Verification:** 754 tests collected; 729 passed and 25 skipped. Ruff and mypy
+are clean (78 source files). The planted-tamper and live-route external oracles
+pass. Round-trip identity is 9487/9487 with 96.96% full round-trip; the seeded
+noise audit records zero inventions; independent resolution eval is 14/14.
+
+**Accurate remaining:** real email/SMS provider adapters and credentials are
+not present. Quote revisions (same number, revision+1, fresh reads, prior
+revision superseded, account-match gate, and full re-assent) remain the second
+M5d dispatch. The table-backed worker is an explicit `run_once` consumer for
+M5; production scheduling/operations for that consumer are not claimed here.

pyproject.toml

@@ -129,9 +129,12 @@ files = [
     "src/model_provider", "src/observability", "src/gateway", "src/quoting",
 ]
 python_version = "3.10"
-# Don't report errors in dependencies outside the listed packages (third-party
-# SDKs, the runtime edge).
-follow_imports = "silent"
+# Do not parse dependencies outside the explicitly listed first-party packages
+# (third-party SDKs, the runtime edge). `silent` still parses dependency stubs;
+# NumPy 2.5 uses Python 3.12-only PEP 695 syntax that cannot be parsed under
+# this project's deliberate Python 3.10 target. Every first-party package in
+# the gate is named in `files`, so `skip` does not reduce their coverage.
+follow_imports = "skip"
 ignore_missing_imports = true
 warn_unused_ignores = true
 no_implicit_optional = true

scripts/latency_baseline.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Measure M5 in-call quote issuance with the offline production components.
+
+This reports observed local p50/p95 only. It intentionally does not enforce the
+spec's provisional 300 ms target.
+"""
+from __future__ import annotations
+
+import math
+import re
+import sys
+import tempfile
+from datetime import datetime
+from pathlib import Path
+from time import perf_counter_ns
+from zoneinfo import ZoneInfo
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path[:0] = [str(ROOT / 'src'), str(ROOT)]
+
+from fulfillment import load_inventory  # noqa: E402
+from gateway import (  # noqa: E402
+    Account,
+    ConversationJournal,
+    Gateway,
+    InMemoryCustomerDB,
+    SessionManager,
+    SyntheticPriceBook,
+)
+from quoting.store import SqliteQuoteStore  # noqa: E402
+from resolution import ResolutionService, catalog_content_version  # noqa: E402
+from sku_translator import FixtureCatalogIndex, InMemoryStore  # noqa: E402
+
+NOW = datetime(2026, 6, 8, 10, 0, tzinfo=ZoneInfo('America/New_York'))
+
+
+def _build_gateway(state_dir: Path) -> tuple[Gateway, SessionManager]:
+    catalog_path = ROOT / 'data' / 'catalog.csv'
+    catalog = FixtureCatalogIndex(str(catalog_path), tenant_id='tenant_001')
+    version = catalog_content_version(catalog_path)
+    service = ResolutionService(
+        catalog, InMemoryStore(), catalog_version=version)
+    accounts = [
+        Account('1001', 'DEMO TRUCK CENTER', '5550100100',
+                'quotes@example.test'),
+    ]
+    sessions = SessionManager(
+        secret=b'latency-baseline-session-secret',
+        customer_db=InMemoryCustomerDB(accounts), now_fn=lambda: 0.0)
+    gateway = Gateway(
+        service=service,
+        catalog=catalog,
+        inventory=load_inventory(ROOT / 'data' / 'inventory.json'),
+        catalog_version=version,
+        sessions=sessions,
+        journal=ConversationJournal(
+            path=state_dir / 'journal.jsonl', now_fn=lambda: NOW.isoformat()),
+        pricebook=SyntheticPriceBook.seeded(catalog.all_skus()),
+        quote_store=SqliteQuoteStore(':memory:'),
+        account_tier_of=lambda _account_id: 'preferred',
+        now_fn=lambda: NOW,
+    )
+    return gateway, sessions
+
+
+def percentile(values: list[float], fraction: float) -> float:
+    ordered = sorted(values)
+    return ordered[math.ceil(fraction * len(ordered)) - 1]
+
+
+def main() -> None:
+    durations_ms: list[float] = []
+    line_counts: list[int] = []
+    with tempfile.TemporaryDirectory(prefix='m5d-latency-') as td:
+        for run in range(50):
+            gateway, sessions = _build_gateway(Path(td) / str(run))
+            caller = f'latency-{run}'
+            token = sessions.open(caller, caller)
+            verified = gateway.converse(
+                caller, token, 'my account number is 1001')
+            if verified.kind != 'verify' or verified.refused is not None:
+                raise RuntimeError('synthetic account verification failed')
+            count = 1 + (run % 6)
+            eligible = tuple(
+                sku for sku in sorted(gateway.catalog.all_skus())
+                if re.match(r'^[A-Za-z]{1,4}\d', sku))
+            skus = eligible[:count]
+            for quantity, sku in enumerate(skus, start=1):
+                response = gateway.converse(
+                    caller, token, f'{quantity} of the {sku}')
+                if response.kind not in ('order', 'identify'):
+                    raise RuntimeError(f'line capture failed for {sku}')
+            if len(gateway._orders[caller].active_lines) != count:
+                raise RuntimeError(f'expected {count} active lines')
+
+            started = perf_counter_ns()
+            issued = gateway.converse(caller, token, 'place the order')
+            elapsed_ns = perf_counter_ns() - started
+            quote_number = issued.meta.get('quote_number')
+            if not quote_number:
+                raise RuntimeError('issuance returned no quote number')
+            queued = gateway.quote_store.queue_item(quote_number, 1)
+            if queued is None or queued.status != 'pending':
+                raise RuntimeError('issuance did not enqueue pending work')
+            durations_ms.append(max(elapsed_ns, 1) / 1_000_000)
+            line_counts.append(count)
+
+    p50 = percentile(durations_ms, 0.50)
+    p95 = percentile(durations_ms, 0.95)
+    print('M5d synthetic issuance latency baseline')
+    print(f'samples=50 line_counts={min(line_counts)}..{max(line_counts)}')
+    print(f'p50 {p50:.3f} ms')
+    print(f'p95 {p95:.3f} ms')
+
+
+if __name__ == '__main__':
+    main()

src/gateway/journal.py

@@ -39,6 +39,8 @@ class EventType(Enum):
     # M5 Stage B (docs/M5_SPOKEN_ORDER_TO_QUOTE_SPEC.md §1.1, §1.5):
     ORDER_LINE_CHANGE = 'order_line_change'      # a line was proposed/amended/withdrawn
     ORDER_ASSENT = 'order_assent'                 # full-order readback + explicit assent
+    QUOTE_DELIVERED = 'quote_delivered'
+    QUOTE_DELIVERY_BLOCKED = 'quote_delivery_blocked'
 
 
 # Identifier fields can encode customer identity even when they are not prose.

src/gateway/models.py

@@ -24,6 +24,7 @@ class Account:
     account_id: str
     name: str
     phone: str | None = None
+    email: str | None = None
 
 
 @dataclass(frozen=True)

src/gateway/orchestrator.py

@@ -761,6 +761,8 @@ def _issue_quote(self, sid, token, order) -> QuoteDocument:
         document = QuoteDocument(header=header, lines=tuple(lines),
                                  totals=build_totals(tuple(lines)), terms=terms)
         self.quote_store.save(document)
+        self.quote_store.enqueue(document.header.quote_number,
+                                 document.header.revision)
         return document
 
     def _fact_reader(self, sid, token, captured):

src/quoting/__init__.py

@@ -7,6 +7,13 @@
 """
 from __future__ import annotations
 
+from quoting.delivery import (
+    Deliverer,
+    DeliveryPayload,
+    SimulatedDeliverer,
+    mint_link,
+    verify_link,
+)
 from quoting.document_guard import (
     DocumentParityViolation,
     assert_document_parity,
@@ -24,7 +31,7 @@
 )
 from quoting.render_pdf import render_pdf
 from quoting.render_xlsx import render_xlsx
-from quoting.store import QuoteStore, SqliteQuoteStore, order_content_hash
+from quoting.store import QueueItem, QuoteStore, SqliteQuoteStore, order_content_hash
 from quoting.totals import build_totals, extended_price, subtotal
 
 __all__ = [
@@ -34,4 +41,6 @@
     'render_pdf', 'render_xlsx',
     'DocumentParityViolation', 'assert_document_parity', 'document_facts',
     'extract_pdf_facts', 'extract_xlsx_facts', 'verify_rendered_document',
+    'Deliverer', 'DeliveryPayload', 'SimulatedDeliverer', 'mint_link',
+    'verify_link', 'QueueItem',
 ]

src/quoting/delivery.py

@@ -0,0 +1,84 @@
+"""M5d offline delivery seam and HMAC-signed quote links."""
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Protocol
+
+
+class AccountContact(Protocol):
+    account_id: str
+    name: str
+    email: str | None
+
+
+@dataclass(frozen=True)
+class DeliveryPayload:
+    account_id: str
+    destination: str
+    quote_number: str
+    revision: int
+    summary: str
+    pdf_bytes: bytes
+    xlsx_bytes: bytes
+    signed_link: str
+
+
+class Deliverer(Protocol):
+    def deliver(self, *, account: AccountContact, quote_number: str,
+                revision: int, summary: str, pdf_bytes: bytes,
+                xlsx_bytes: bytes, signed_link: str) -> None: ...
+
+
+@dataclass
+class SimulatedDeliverer:
+    """Offline deliverer used by CI; append-only for straightforward audit."""
+
+    payloads: list[DeliveryPayload] = field(default_factory=list)
+
+    def deliver(self, *, account: AccountContact, quote_number: str,
+                revision: int, summary: str, pdf_bytes: bytes,
+                xlsx_bytes: bytes, signed_link: str) -> None:
+        if account.email is None:
+            raise ValueError('contact_missing')
+        self.payloads.append(DeliveryPayload(
+            account_id=account.account_id, destination=account.email,
+            quote_number=quote_number, revision=revision, summary=summary,
+            pdf_bytes=pdf_bytes, xlsx_bytes=xlsx_bytes,
+            signed_link=signed_link))
+
+
+def _sign(payload: bytes, secret: bytes) -> str:
+    return hmac.new(secret, payload, hashlib.sha256).hexdigest()
+
+
+def mint_link(quote_number: str, revision: int, valid_until: datetime,
+              secret: bytes) -> str:
+    """Mint a token over quote identity and the quote's own validity horizon."""
+    payload = f'{quote_number}|{revision}|{valid_until.isoformat()}'.encode()
+    encoded = base64.urlsafe_b64encode(payload).rstrip(b'=').decode()
+    return f'{encoded}.{_sign(payload, secret)}'
+
+
+def verify_link(token: str, now: datetime,
+                secret: bytes) -> tuple[str, int] | None:
+    """Return quote identity only for a valid, unexpired token."""
+    try:
+        encoded, supplied_sig = token.split('.', 1)
+        padding = '=' * (-len(encoded) % 4)
+        payload = base64.b64decode(encoded + padding, altchars=b'-_', validate=True)
+        if not hmac.compare_digest(_sign(payload, secret), supplied_sig):
+            return None
+        quote_number, revision_text, expiry_text = payload.decode().split('|', 2)
+        revision = int(revision_text)
+        expiry = datetime.fromisoformat(expiry_text)
+        if now.tzinfo is None or expiry.tzinfo is None or now >= expiry:
+            return None
+        if not quote_number or revision < 1:
+            return None
+        return quote_number, revision
+    except (UnicodeDecodeError, ValueError):
+        return None