M5 W5: durable completion-gated delivery worker

data/m5_baseline_defects.json

@@ -3,7 +3,6 @@
   "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-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-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,13 +26,14 @@
       }
     },
     "quote_store": {
-      "schema_sha256": "d26de5eaaf158fd47094f886b9b94a94c21578b7f5bdf80a0cdeb22f3410a6d8",
+      "schema_sha256": "b78342c2f525a361338c7fb82102650af72229aeb2bff51e17c2ced7ec1fd4a8",
       "tables": [
         "artifact_guard_events",
         "artifact_jobs",
         "capability_access_events",
         "delivery_jobs",
         "issuance_events",
+        "provider_callback_events",
         "quote_artifacts",
         "quote_capabilities",
         "quote_idempotency",
@@ -42,14 +43,16 @@
         "quote_revision_seq",
         "quote_schema",
         "quote_seq",
-        "quotes"
+        "quotes",
+        "worker_heartbeats"
       ],
       "row_counts": {
         "artifact_guard_events": 0,
         "artifact_jobs": 0,
         "capability_access_events": 0,
         "delivery_jobs": 0,
         "issuance_events": 0,
+        "provider_callback_events": 0,
         "quote_artifacts": 0,
         "quote_capabilities": 0,
         "quote_idempotency": 1,
@@ -59,14 +62,15 @@
         "quote_revision_seq": 0,
         "quote_schema": 1,
         "quote_seq": 1,
-        "quotes": 1
+        "quotes": 1,
+        "worker_heartbeats": 0
       },
       "known_gaps": [
         "pre-W3 compatibility tables remain until migration evidence is retained",
-        "delivery job leases and authenticated call completion land in W5",
+        "credentialed provider receipts remain external to the schema fixture",
         "legacy rows without assent or artifact proof remain explicitly held"
       ]
     }
   },
-  "combined_sha256": "1a6d06e44585df041fe2fdb6b4869c4446f1309eaa9ae512dfeb2bf90efc55db"
+  "combined_sha256": "151432b988d1ace9cd6cd2af51039246f68d33f59a8977032cc678b808f592fb"
 }

docs/DECISION_LOG.md

@@ -1577,3 +1577,36 @@ migration tests pass offline. Production remains blocked on existing staging
 agent/tool/branch IDs, credentials, real hosted test IDs, an inbound call,
 credentialed webhook latency/loss evidence, and one exercised rollback receipt.
 **Status:** implementation locked; external staging proof outstanding.
+
+## 2026-07-11 — D21: delivery is at-least-once before send and manual after uncertainty
+
+**Decision.** Artifact and notification work are separate durable state
+machines. An artifact cannot leave `waiting_for_call_end` until an authenticated
+completion event or bounded reconciliation releases it. Rendering is leased and
+retryable; verified PDF/XLSX bytes are immutable and release per-channel
+delivery jobs only after document parity.
+
+A delivery lease has two recovery domains. A crash while merely `leased` is
+safe to retry. Crossing into `sending` marks provider acceptance as potentially
+ambiguous; a crash or connection loss from there enters `manual_review`, never
+an automatic duplicate send. Provider acceptance is not called “delivered.”
+Signed callbacks advance accepted work to delivered, bounced, undelivered, or
+permanent failure and are replay-safe and non-regressing.
+
+Frozen contact identity, version, channel, and destination fingerprint are
+revalidated immediately before provider I/O. Destination or unapproved metadata
+changes require fresh caller authorization. A quote may freeze multiple selected
+channel preflights; they share exactly one artifact job and retain one delivery
+job per channel.
+
+**Why.** Neither SQLite nor external providers can offer a meaningful global
+exactly-once transaction. The safe claim is explicit: retry only when no request
+was accepted, preserve stable provider keys, and surface every ambiguous outcome
+for human review.
+
+**Outcome.** The worker uses atomic `UPDATE ... RETURNING` leases, expiry
+recovery, parity gates, stable keys, credential-gated SendGrid/Twilio adapters,
+recipient allowlists, exact attachments/link-only SMS, signed callbacks,
+heartbeats, bounded polling, graceful drain, and operator inspect/retry/
+quarantine/revoke controls. **Status:** implemented and verified offline;
+credentialed staging sends and callbacks remain under P04.

docs/M5_BUILD_JOURNAL.md

@@ -735,3 +735,42 @@ environment/phone-or-SIP IDs and ElevenLabs credentials; a fetch-back receipt;
 an inbound staging call; actual hosted Agent Test IDs and raw results; a
 credentialed post-call latency/loss receipt; and one exercised rollback. No
 production or Vercel deployment was attempted.
+
+## 2026-07-11 — Session 11: W5 completion-gated durable delivery
+
+Replaced the legacy single queue with separate artifact and per-channel delivery
+machines. Authenticated post-call events atomically release only the matching
+conversation's outboxes; the worker also runs bounded ElevenLabs status
+reconciliation when credentials exist. Two SQLite connections cannot claim the
+same live lease. Lease attempts, expiry, timestamps, errors, stable provider
+keys, provider IDs, callbacks, and worker heartbeats survive restart.
+
+The renderer starts only after effective hangup and persists both guarded
+formats atomically. A verified artifact releases all selected channel jobs. A
+crash before the provider-send boundary retries; a crash or connection loss
+after that boundary moves to manual review, making duplicate risk explicit.
+Poison/quarantined artifacts do not block later work.
+
+Added credential-gated Twilio SMS and SendGrid email adapters. SMS is link-only;
+email attachments are the exact persisted guarded bytes. Both use recipient
+allowlists and stable delivery keys, classify 429/5xx/hard-4xx/ambiguous
+failures, and sanitize results. Twilio and SendGrid callback paths authenticate,
+deduplicate, tolerate out-of-order states, and never retain callback contact
+payloads. Production refuses `SimulatedDeliverer`; missing credentials fail
+adapter construction.
+
+The long-running worker has bounded jittered backoff, SIGTERM drain, heartbeat,
+and inspect/retry/quarantine/revoke-link commands. P16 was regenerated for quote
+schema v4, including delivery callback events and worker heartbeats. The W5
+double-claim baseline defect is removed; only the W7 latency defect remains.
+
+**Verification:** 923 tests collected and full pytest exits zero; ruff is clean;
+configured mypy reports no issues across 87 source files. Signed callback,
+lease/crash, contact-change, provider classification, exact attachment, multi-
+channel cardinality, hangup-release, reconciliation, migration, and production-
+simulation refusal tests pass.
+
+**External W5 proof still required:** tenant P04 channel rule, sender identities,
+Twilio/SendGrid credentials, staging recipient allowlists, one allowlisted email
+and SMS, provider IDs, signed delivered/undelivered callbacks, and downloaded
+artifact hash receipts. No production or Vercel deployment was attempted.

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     # because the parity gate must run by default, not behind an extra.
     "reportlab>=4.0",
     "pypdf>=4.0",
+    "cryptography>=42.0",    # SendGrid Event Webhook ECDSA verification
 ]
 
 [project.optional-dependencies]

scripts/m5_legacy_inventory.py

@@ -146,7 +146,7 @@ def compute() -> dict:
                 "row_counts": _row_counts(store._conn, quote_tables),
                 "known_gaps": [
                     "pre-W3 compatibility tables remain until migration evidence is retained",
-                    "delivery job leases and authenticated call completion land in W5",
+                    "credentialed provider receipts remain external to the schema fixture",
                     "legacy rows without assent or artifact proof remain explicitly held",
                 ],
             },

scripts/quote_worker.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""Durable quote worker and bounded operator controls."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import random
+import signal
+import socket
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / 'src'))
+
+from quoting.store import SqliteQuoteStore  # noqa: E402
+from quoting.worker import run_artifact_once, run_delivery_once  # noqa: E402
+
+
+def _store() -> SqliteQuoteStore:
+    return SqliteQuoteStore(os.environ.get(
+        'SKU_QUOTE_STORE_DB', str(REPO / 'state' / 'quotes.db')))
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    sub = parser.add_subparsers(dest='command', required=True)
+    run = sub.add_parser('run')
+    run.add_argument('--max-poll-seconds', type=float, default=5.0)
+    run.add_argument('--lease-seconds', type=float, default=30.0)
+    run.add_argument('--once', action='store_true')
+    sub.add_parser('inspect')
+    retry = sub.add_parser('retry')
+    retry.add_argument('kind', choices=('artifact', 'delivery'))
+    retry.add_argument('tenant_id')
+    retry.add_argument('quote_number')
+    retry.add_argument('revision', type=int)
+    retry.add_argument('--channel')
+    quarantine = sub.add_parser('quarantine')
+    quarantine.add_argument('tenant_id')
+    quarantine.add_argument('quote_number')
+    quarantine.add_argument('revision', type=int)
+    quarantine.add_argument('reason')
+    revoke = sub.add_parser('revoke-link')
+    revoke.add_argument('token_hash')
+    return parser
+
+
+def _run(args: argparse.Namespace) -> int:
+    if not 0.05 <= args.max_poll_seconds <= 60 or args.lease_seconds <= 0:
+        raise SystemExit('poll and lease bounds are invalid')
+    from runtime.config import build_delivery_adapters, build_gateway
+
+    gateway, sessions = build_gateway()
+    store = gateway.quote_store
+    adapters = build_delivery_adapters()
+    reconciler = None
+    elevenlabs = {
+        'api_key': os.environ.get('ELEVENLABS_API_KEY', ''),
+        'secret': os.environ.get('ELEVENLABS_WEBHOOK_SECRET', ''),
+        'agent_id': os.environ.get('ELEVENLABS_AGENT_ID', ''),
+        'branch_id': os.environ.get('ELEVENLABS_STAGING_BRANCH_ID', ''),
+        'version_id': os.environ.get('ELEVENLABS_AGENT_VERSION_ID', ''),
+        'environment': os.environ.get('ELEVENLABS_ENVIRONMENT', ''),
+    }
+    if all(elevenlabs.values()):
+        from runtime.post_call import (
+            CallCompletionReconciler,
+            ElevenLabsConversationStatusSource,
+            PostCallPolicy,
+        )
+
+        def release(completion):
+            store.release_for_call_completion(
+                tenant_id=completion.tenant_id,
+                conversation_id=completion.conversation_id,
+                completion_event_id=completion.event_id,
+                effective_hangup_at=completion.effective_hangup_at,
+                received_at=completion.received_at)
+
+        reconciler = CallCompletionReconciler(
+            gateway.conversation_store,
+            ElevenLabsConversationStatusSource(elevenlabs['api_key']),
+            PostCallPolicy(
+                secret=elevenlabs['secret'], agent_id=elevenlabs['agent_id'],
+                branch_id=elevenlabs['branch_id'], version_id=elevenlabs['version_id'],
+                environment=elevenlabs['environment']),
+            completion_hook=release)
+    owner = f'{socket.gethostname()}:{os.getpid()}'
+    draining = False
+
+    def stop(_signum, _frame):
+        nonlocal draining
+        draining = True
+
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
+    idle = 0
+    while True:
+        now = datetime.now(timezone.utc)
+        store.heartbeat(worker_id=owner, now=now.timestamp(), draining=draining)
+        if draining:
+            return 0
+        if reconciler is not None:
+            for held in store.held_conversations():
+                reconciler.reconcile(
+                    tenant_id=str(held['tenant_id']),
+                    conversation_id=str(held['session_id']),
+                    held_since=float(held['held_since']), now=now.timestamp())
+        did_work = run_artifact_once(
+            store=store, owner=owner, now_fn=lambda: now,
+            lease_seconds=args.lease_seconds)
+        did_work = run_delivery_once(
+            store=store, owner=owner, adapters=adapters,
+            account_lookup=sessions.customer_db.by_number,
+            link_factory=gateway.quote_link_factory, now_fn=lambda: now,
+            lease_seconds=args.lease_seconds) or did_work
+        if args.once:
+            return 0
+        if did_work:
+            idle = 0
+            continue
+        idle = min(idle + 1, 8)
+        ceiling = min(args.max_poll_seconds, 0.05 * (2 ** idle))
+        time.sleep(random.uniform(0.05, max(0.05, ceiling)))
+
+
+def main() -> int:
+    args = _parser().parse_args()
+    store = _store()
+    if args.command == 'run':
+        return _run(args)
+    if args.command == 'inspect':
+        print(json.dumps(store.job_snapshot(), sort_keys=True, indent=2))
+        return 0
+    now = time.time()
+    if args.command == 'retry':
+        changed = store.operator_retry(
+            kind=args.kind, tenant_id=args.tenant_id,
+            quote_number=args.quote_number, revision=args.revision,
+            channel=args.channel, now=now)
+    elif args.command == 'quarantine':
+        changed = store.operator_quarantine(
+            tenant_id=args.tenant_id, quote_number=args.quote_number,
+            revision=args.revision, detail=args.reason, now=now)
+    else:
+        changed = store.revoke_capability(args.token_hash, at=datetime.now(timezone.utc))
+    print(json.dumps({'changed': changed}))
+    return 0 if changed else 2
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

src/gateway/conversation_store.py

@@ -19,7 +19,7 @@
     'quote_number_allocations', 'quote_issue_keys', 'issuance_events',
     'artifact_jobs', 'delivery_jobs', 'quote_artifacts',
     'quote_capabilities', 'artifact_guard_events', 'quote_schema',
-    'capability_access_events',
+    'capability_access_events', 'provider_callback_events', 'worker_heartbeats',
 })
 
 

src/quoting/__init__.py

@@ -8,9 +8,13 @@
 from __future__ import annotations
 
 from quoting.delivery import (
+    ChannelAdapter,
     Deliverer,
     DeliveryPayload,
+    ProviderResult,
+    SendGridEmailAdapter,
     SimulatedDeliverer,
+    TwilioSmsAdapter,
 )
 from quoting.document_guard import (
     DocumentParityViolation,
@@ -29,7 +33,14 @@
 )
 from quoting.render_pdf import render_pdf
 from quoting.render_xlsx import render_xlsx
-from quoting.store import QueueItem, QuoteStore, SqliteQuoteStore, order_content_hash
+from quoting.store import (
+    ArtifactJob,
+    DeliveryJob,
+    QueueItem,
+    QuoteStore,
+    SqliteQuoteStore,
+    order_content_hash,
+)
 from quoting.totals import build_totals, extended_price, subtotal
 
 __all__ = [
@@ -40,4 +51,6 @@
     'DocumentParityViolation', 'assert_document_parity', 'document_facts',
     'extract_pdf_facts', 'extract_xlsx_facts', 'verify_rendered_document',
     'Deliverer', 'DeliveryPayload', 'SimulatedDeliverer', 'QueueItem',
+    'ChannelAdapter', 'ProviderResult', 'TwilioSmsAdapter',
+    'SendGridEmailAdapter', 'ArtifactJob', 'DeliveryJob',
 ]