M5 W3: atomic quote store and guarded immutable artifacts

data/m5_baseline_defects.json

@@ -3,9 +3,7 @@
   "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-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-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

@@ -26,27 +26,47 @@
       }
     },
     "quote_store": {
-      "schema_sha256": "6d2175921c806746d4f4232e99b05cccd1aceac385656c25af3cc83f1d3d3b14",
+      "schema_sha256": "d26de5eaaf158fd47094f886b9b94a94c21578b7f5bdf80a0cdeb22f3410a6d8",
       "tables": [
+        "artifact_guard_events",
+        "artifact_jobs",
+        "capability_access_events",
+        "delivery_jobs",
+        "issuance_events",
+        "quote_artifacts",
+        "quote_capabilities",
         "quote_idempotency",
+        "quote_issue_keys",
+        "quote_number_allocations",
         "quote_queue",
         "quote_revision_seq",
+        "quote_schema",
         "quote_seq",
         "quotes"
       ],
       "row_counts": {
+        "artifact_guard_events": 0,
+        "artifact_jobs": 0,
+        "capability_access_events": 0,
+        "delivery_jobs": 0,
+        "issuance_events": 0,
+        "quote_artifacts": 0,
+        "quote_capabilities": 0,
         "quote_idempotency": 1,
+        "quote_issue_keys": 0,
+        "quote_number_allocations": 0,
         "quote_queue": 1,
         "quote_revision_seq": 0,
+        "quote_schema": 1,
         "quote_seq": 1,
         "quotes": 1
       },
       "known_gaps": [
-        "idempotency is not tenant-bound",
-        "reserve/save/enqueue are separate commits",
-        "queue rows have no lease or timestamps"
+        "pre-W3 compatibility tables remain until migration evidence is retained",
+        "delivery job leases and authenticated call completion land in W5",
+        "legacy rows without assent or artifact proof remain explicitly held"
       ]
     }
   },
-  "combined_sha256": "6471bcd46b9fc33be420adf67e7183cb997364d36d8d94ec87113c6a32612de0"
+  "combined_sha256": "1a6d06e44585df041fe2fdb6b4869c4446f1309eaa9ae512dfeb2bf90efc55db"
 }

docs/M5_PRODUCTION_BUILD_E2E_EVAL_PLAN.md

@@ -721,6 +721,17 @@ Build:
   depth/oldest age/lease/retry, render/guard/provider duration, parity/contact/
   provider failures, hangup-to-accept, terminal callback status, link refusal,
   DB busy/disk, worker heartbeat, backup age/restore result, and rate-limit events.
+- Use OpenTelemetry as the vendor-neutral trace contract and PostHog as the
+  single production destination for tracing/AI observability plus pseudonymous
+  product/voice journey events and funnels (call connected, clarification,
+  readback, assent, hangup, provider acceptance, download). PostHog is not the
+  source of truth for quote facts, deterministic safety evals, or transactional
+  state. A shared pseudonymous correlation ID must join events, traces, eval
+  runs, and the latency report without exporting transcript text,
+  account/contact values, quote capability tokens, or document contents. Arize
+  Phoenix is an optional local W7 bake-off only; it does not become a second
+  required production sink unless measured workflow value justifies the extra
+  retention, privacy, correlation, and operational surface.
 - Add alerts/runbooks for parity incident, queue SLO, provider outage, stuck lease,
   missing contact, invalid-reference attack, disk pressure, failed backup/restore,
   key rotation/revocation, active-call restart, tenant offboarding, and rollback.
@@ -763,6 +774,19 @@ DoD:
   audit, independent resolution eval, and runtime readiness remain green.
 - The latency harness drives the actual multi-turn readback/assent protocol and
   reports workload, sample count, p50/p95/p99/max, commit, schema, and environment.
+- The same run exports stage spans and corresponding pseudonymous journey events
+  to a local PostHog-compatible capture sink. Its evidence bundle proves
+  trace/event/eval/report correlation and includes
+  STT endpointing/finalization, network/tool, resolver, optional query rewrite
+  and retrieval, LLM first-token, TTS first-audio, and player/telephony timing
+  when each stage exists. Instrumentation overhead and export failure are tested;
+  telemetry remains non-blocking and cannot alter the spoken or issued result.
+- Query-rewrite model racing is evaluated only on a non-binding retrieval lane,
+  against a single-model and raw-query fallback baseline. The report includes
+  answer/retrieval quality, fallback rate, provider-outage behavior, cost, and
+  p50/p75/p95/p99—not latency alone. No fastest-response-wins policy may bind a
+  SKU, requested quantity, account, price, availability, quote number, or
+  document field.
 - Twin per-turn order operations meet p95 <=800 ms; transactional issuance meets
   p95 <=300 ms; no renderer/provider call appears in the synchronous trace.
 - At P12 steady plus 2× burst, hangup-to-provider-acceptance p95 <=60 s, no orphan

scripts/m5_legacy_inventory.py

@@ -145,9 +145,9 @@ def compute() -> dict:
                 "tables": quote_tables,
                 "row_counts": _row_counts(store._conn, quote_tables),
                 "known_gaps": [
-                    "idempotency is not tenant-bound",
-                    "reserve/save/enqueue are separate commits",
-                    "queue rows have no lease or timestamps",
+                    "pre-W3 compatibility tables remain until migration evidence is retained",
+                    "delivery job leases and authenticated call completion land in W5",
+                    "legacy rows without assent or artifact proof remain explicitly held",
                 ],
             },
         }

src/gateway/conversation_snapshot.py

@@ -5,8 +5,6 @@
 """
 from __future__ import annotations
 
-import hashlib
-import json
 from dataclasses import asdict
 from typing import Any
 
@@ -24,6 +22,7 @@
 from gateway.order_parser import OrderClause
 from gateway.session import Session
 from quoting.models import DeliveryPreflight
+from quoting.store import order_version_digest
 from sku_translator.quantity import RequestedQty
 
 SNAPSHOT_VERSION = 1
@@ -74,7 +73,9 @@ def dump_call(gateway, caller_id: str) -> dict[str, Any]:
         'order': None if order is None else {
             'lines': {key: _line(line) for key, line in order.state.lines.items()},
             'sequence': list(order.state.sequence), 'assented': order.state.assented,
-            'assent_turn': order.state.assent_turn, 'next_id': order._next_id},
+            'assent_turn': order.state.assent_turn,
+            'assent_occurred_at': order.state.assent_occurred_at,
+            'next_id': order._next_id},
         'session': None if session is None else {
             'session_id': session.session_id, 'channel_id': session.channel_id,
             'created_at': session.created_at, 'last_active': session.last_active,
@@ -133,6 +134,7 @@ def load_call(gateway, caller_id: str, value: dict[str, Any]) -> str | None:
         order.state.sequence = list(raw_order['sequence'])
         order.state.assented = raw_order['assented']
         order.state.assent_turn = raw_order['assent_turn']
+        order.state.assent_occurred_at = raw_order.get('assent_occurred_at')
         order._next_id = raw_order['next_id']
         gateway._orders[caller_id] = order
     raw_session = value.get('session')
@@ -204,11 +206,10 @@ def authoritative_events(before: dict[str, Any], after: dict[str, Any],
                     else 'line_resolution')
             events.append((kind, {'line_id': line_id, 'state': state,
                                   'sku': line.get('sku')}))
-    order_digest = hashlib.sha256(json.dumps(
-        [(line_id, new_lines[line_id].get('sku'),
-          new_lines[line_id].get('requested_qty'))
-         for line_id in new_order['sequence']], separators=(',', ':')
-    ).encode()).hexdigest()
+    order_digest = order_version_digest(tuple(
+        (line_id, new_lines[line_id].get('sku'),
+         new_lines[line_id].get('requested_qty'))
+        for line_id in new_order['sequence']))
     readback_receipt_id = f'readback:{order_digest[:24]}'
     readback_completed = (not before.get('completed_order_readback')
                           and after.get('completed_order_readback'))
@@ -230,14 +231,20 @@ def authoritative_events(before: dict[str, Any], after: dict[str, Any],
             'order_digest': order_digest,
             'line_versions': list(new_order['sequence']),
             'chunk_coverage': chunks, 'running_counts': running,
-            'occurred_at': (after.get('session') or {}).get('last_active')}))
+            'occurred_at': (
+                new_order.get('assent_occurred_at')
+                if assent_granted and new_order.get('assent_occurred_at') is not None
+                else (after.get('session') or {}).get('last_active'))}))
     if assent_granted:
         events.append(('assent_receipt', {
             'turn_id': turn_id, 'order_digest': order_digest,
             'readback_receipt_id': readback_receipt_id,
             'utterance_digest': text_digest,
             'line_versions': list(new_order['sequence']),
-            'occurred_at': (after.get('session') or {}).get('last_active')}))
+            'occurred_at': (
+                new_order.get('assent_occurred_at')
+                if new_order.get('assent_occurred_at') is not None
+                else (after.get('session') or {}).get('last_active'))}))
     return events
 
 

src/gateway/conversation_store.py

@@ -16,6 +16,10 @@
 })
 _LEGACY_QUOTE_TABLES = frozenset({
     'quote_idempotency', 'quote_queue', 'quote_revision_seq', 'quote_seq', 'quotes',
+    'quote_number_allocations', 'quote_issue_keys', 'issuance_events',
+    'artifact_jobs', 'delivery_jobs', 'quote_artifacts',
+    'quote_capabilities', 'artifact_guard_events', 'quote_schema',
+    'capability_access_events',
 })