feat(reporting): add transactional notification outbox

docs/reporting-ledger-migration.md

@@ -50,21 +50,23 @@ obligation IDs, the named obligations must exist in the requested account.
 2. With the upgraded SDK, run `await store.create_schema()` before starting
    reporting work. It creates missing tables and applies the bundled
    `reporting_ledger_account_generations.sql`,
-   `reporting_ledger_obligation_currency.sql`, and
-   `reporting_ledger_reconciliation.sql` migrations in one transaction.
+   `reporting_ledger_obligation_currency.sql`,
+   `reporting_ledger_reconciliation.sql`, and
+   `reporting_notification_outbox.sql` migrations in one transaction.
 3. Restart reporting work with the upgraded SDK on every instance.
 
 For deployments managed by a migration tool, the standalone migration is
 [`reporting_ledger_account_generations.sql`](../src/adcp/reporting/ledger/reporting_ledger_account_generations.sql).
 It upgrades an existing beta.15 ledger by itself, including in autocommit mode.
-For a combined bootstrap and upgrade, run all four bundled files in one transaction:
+For a combined bootstrap and upgrade, run all five bundled files in one transaction:
 
 ```sh
 psql "$REPORTING_DATABASE_URL" --set=ON_ERROR_STOP=1 --single-transaction \
   -f src/adcp/reporting/ledger/reporting_ledger.sql \
   -f src/adcp/reporting/ledger/reporting_ledger_account_generations.sql \
   -f src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql \
-  -f src/adcp/reporting/ledger/reporting_ledger_reconciliation.sql
+  -f src/adcp/reporting/ledger/reporting_ledger_reconciliation.sql \
+  -f src/adcp/reporting/ledger/reporting_notification_outbox.sql
 ```
 
 Use the ledger's existing `search_path` and a role that owns its tables. All
@@ -74,9 +76,22 @@ leaves the old primary key in place and does not perform this upgrade.
 
 The reconciliation migration adds empty optional evidence tables and nullable,
 immutable managed digest/total evidence on revisions and adjustments. It does not infer historical
-evidence or enable a delivery tier. See the [storage contract](reporting-reconciliation-storage.md)
+evidence or enable a delivery tier. Re-running it also replaces the
+`reporting_reconciliation_reference_immutable()` guard function in place, so
+that a referenced configuration generation keeps its published content
+immutable while still accepting the rc.3 lifecycle columns (`activated_at`,
+`deactivated_at`, `automated_recovery_seconds`, `status_retention_days`). It
+rewrites no rows. See the [storage contract](reporting-reconciliation-storage.md)
 for the records, migration invariants, and deferred writer/handler work.
 
+The outbox migration adds empty event, delivery, and ordered status-dirty tables.
+It preserves existing ledger rows and does not backfill historical events.
+Notification enqueue remains disabled until the store is constructed with
+`notifications=True`. Opted-in schema readiness checks the complete installed chain,
+including constraints, indexes, and immutable-record guards. See the
+[notification outbox contract](reporting-notification-outbox.md) for trusted
+subscription wiring, retention, and the deferred status projector.
+
 The migration inspects the primary-key columns under an exclusive table lock
 and replaces the beta.15 global primary key with the account-qualified key.
 It preserves the constraint's name, including a renamed beta.15 key. Existing
@@ -167,7 +182,7 @@ large tables; production-sized lock time has not been benchmarked. Unexpected
 adopter column types/defaults fail and roll back rather than silently adapting.
 
 Stop and drain **all** beta.15 and #1169-only reporting writers before upgrading;
-they do not supply frozen currency. Run `create_schema()` or the three-file SQL
+they do not supply frozen currency. Run `create_schema()` or the bundled SQL
 command above, then start upgraded writers. If #1169 is already installed, its
 primary-key migration recognizes the account-qualified key without rebuilding
 it. The standalone currency SQL also runs atomically on an installed #1169

docs/reporting-notification-outbox.md

@@ -0,0 +1,221 @@
+# Transactional reporting notification outbox
+
+The optional reporting outbox commits a typed logical event in the ledger's
+transaction, then expands subscribers and delivers HTTP in separate phases.
+This implements the ledger and Managed readiness slice of [#1168](https://github.com/adcontextprotocol/adcp-client-python/issues/1168).
+It retains a durable status-dirty handoff for the later complete status projector.
+
+| Committed change | Retained notification work |
+| --- | --- |
+| New revision or adjustment | `reporting.ledger_changed` and ordered status-dirty evidence |
+| Configuration, obligation, readability, consumer statement or issue transition | Ordered status-dirty evidence |
+| Managed destination/reconciliation change | Consumer-scoped status-dirty evidence |
+| Verified materialization with its frozen Managed binding | `reporting.delivery_ready`, scoped to the reconciliation consumer |
+
+There is no `reporting.status_changed` emitter. Clock sweeps, complete status
+fingerprint deduplication, and webhook activity projection belong to #1168B.
+The capability helper omits `status_notification` and sets
+`supports_webhook_activity=false`.
+
+## Optional wiring
+
+Existing `ReportingLedgerStore` implementations and `ReportingProducer`
+constructors require no new methods, callback, or argument. Concrete memory
+and PostgreSQL stores accept the optional keyword `notifications=True`; the
+default is off. Core producer return values and exact-content reads keep their
+existing behavior. A memory outbox shares its ledger's rollback boundary but
+has no process-crash durability.
+
+```python
+from adcp.reporting.ledger import PgReportingReconciliationStore
+from adcp.reporting.outbox import (
+    PgReportingOutbox,
+    ReportingEnvelopeCipher,
+    ReportingNotificationWorker,
+)
+
+ledger = PgReportingReconciliationStore(pool=pool, notifications=True)
+await ledger.create_schema()
+outbox = PgReportingOutbox(pool=pool)
+worker = ReportingNotificationWorker(
+    outbox=outbox,
+    subscriptions=trusted_account_configurations,
+    signing=trusted_account_keyrings,
+    cipher=ReportingEnvelopeCipher(encryption_key, key_version="outbox-key-1"),
+)
+
+# At service startup, while arranging to schedule both worker phases:
+fields = await worker.advertised_notifications(ledger, account_id=account_id)
+# Merge fields into the producer's reporting_delivery capability block.
+
+# Each call performs at most one leased unit of work and returns a bool.
+await worker.expand_one(account_id=account_id)
+await worker.deliver_one(account_id=account_id)
+```
+
+Schedule both phases for the accounts the service is authorized to process.
+An empty turn is idle; use the service's normal scheduling/backoff policy.
+The SDK does not start a scheduler or contact subscribers during import or
+ledger construction. Supply a retained typed `ready_scope` to the capability
+helper to advertise readiness. It verifies the actual configuration,
+destination, and obligation binding. A profile label, capability string, or
+Core subscriber request cannot create a readiness event or capability.
+
+The helper verifies that the opted-in ledger and outbox use the same store or
+pool, that current registrations have usable authentication, and that the
+entire installed PostgreSQL chain matches its column, constraint, index,
+trigger, and guard-function contract. It does not infer operational readiness
+from the presence of objects. Continue scheduling the worker while advertising
+these fields. Custom stores can implement the additive outbox protocol;
+automatic capability verification conservatively covers the SDK reference stores.
+
+## Domain commit and replay
+
+Every domain mutation and its enqueue use the exact same connection inside
+an explicit `async with connection.transaction()`, including autocommit
+pools. An enqueue failure rolls back the mutation. Committed logical events
+are immutable and survive producer restarts. The causal key contains account,
+consumer namespace, closed cause kind, non-null cause ID, and cause generation.
+Readiness identifies the full `(account_id, consumer_id, materialization_id)`.
+An identical mutation replay creates no event; a conflicting identity cannot
+commit a ghost event.
+
+Each event owns one notification ID and `fired_at`. Every subscriber shares
+that ID. Each subscriber/emission generation gets a cryptographically random
+idempotency key and its own prepared body. Re-emitting an event with
+`outbox.reemit(...)` advances the emission generation and generates new keys;
+ordinary retry changes neither generation, key, nor body bytes.
+
+A configuration generation's *content* stays immutable, while the rc.3
+lifecycle state it carries -- activation, deactivation, and the recovery and
+retention windows -- keeps evolving on that same generation
+(`reporting-delivery-config-state.json` walks one generation from `ready` to
+`inactive`). A re-put that changes only those fields therefore applies to the
+retained generation and co-commits one status-dirty generation; an unchanged
+re-put stays a no-op and enqueues nothing; changed content is still a
+`CONFIGURATION_GENERATION_IMMUTABLE` conflict that writes nothing. The
+reconciliation reference guard exempts exactly this lifecycle state, so a
+Managed generation bound by reconciliation records can still be deactivated.
+
+A dirty record follows the retained evidence rather than a predicted
+transition: an idempotent re-acknowledge of an issue changes nothing and
+enqueues nothing, while a repeated waive does move `retired_at` and stays
+reconstructable. Existing issue return values are unchanged by opting in.
+
+Status-dirty records form an account-ordered journal, with a trusted optional
+generation/obligation/consumer/feed scope, cause generations, and immutable
+record references. Readability, configuration lifecycle, and issue lifecycle
+retain before-and-after evidence, so rapid reversals remain replayable.
+Consumer statements and reconciliation records remain readable in their
+original consumer namespace. Checkpoints use account/projector compare-and-set.
+They do not delete evidence or project a wire status. The clock-dirty method
+is a handoff seam, with no automatic clock sweep in this slice.
+
+The concrete issue methods accept optional `status_scope=ReportingStatusScope(...)`.
+The store validates supplied references in the issue's account and consumer
+namespace. It never parses `issue_key`. Without a typed scope, a dirty record
+invalidates the account and its specified consumer broadly. An occurrence's
+typed scope cannot be retargeted during its lifecycle.
+
+## Subscriber configuration and transport
+
+`ReportingSubscriptionResolver` must resolve from trusted account registration
+state. A `ReportingNotificationSubscription` combines account, principal,
+subscriber, normalized URL, exact event types, configuration revision,
+authorization and proof-of-control references, active/authorized/proof-valid
+flags, and one exclusive authentication mode. The registrar must actually
+verify principal authorization and the account/subscriber/URL proof of control;
+the references are evidence pointers, not substitutes for those checks.
+Never construct registrations from an unauthenticated request or notification.
+
+Fanout uses **active-at-expansion semantics**. It resolves one complete snapshot,
+validates uniqueness and fingerprints, then inserts all matching deliveries and
+the completed-expansion checkpoint in one transaction. A valid empty snapshot
+completes. A transient resolution error retries. A crash during insertion rolls
+back every subscriber row; a restart can resolve a new snapshot without mixing
+membership. An external resolver is not transactionally atomic with the ledger.
+
+AES-256-GCM encrypts the complete routing envelope, including the full query
+URL, credential mode/material, typed event and exact prepared body. Canonical
+AAD binds every routing column: account, consumer, subscriber/principal,
+delivery/event/notification/idempotency IDs, cause and emission generations,
+URL digest, configuration fingerprint, auth mode/signing scope, body digest,
+envelope version and encryption-key version. Decryption/authentication runs
+before configuration resolution, signing, DNS, or HTTP. Old encryption keys
+must remain available while retained deliveries can use them; `previous_keys`
+supports rotation without rewriting an immutable delivery.
+
+Every attempt resolves the exact account/subscriber/event configuration again,
+including authorization, proof validity and the full credential fingerprint.
+Removed, deactivated, or replaced registrations are terminally suppressed.
+Bodies are never retargeted. RFC 9421 attempts resolve current trusted signing
+material by account/principal/signing scope, producing fresh signatures during
+key rotation. Legacy bearer and HMAC modes are exclusive with RFC 9421.
+The worker constructs its own concrete hardened sender; injected sender
+objects/subclasses, arbitrary HTTP clients and URL rewrite hooks are not accepted.
+
+Prepared bodies use deterministic canonical UTF-8 and the exact bundled
+`3.2.0-rc.3` named schema validator, including the conditionals omitted by the
+generated models. The mapping is an allowlist with recursive secret rejection;
+it copies no reporting rows, resource/object lists, signed or activation URLs,
+credentials, tokens, extensions, or adopter metadata.
+
+The SDK-owned transport permits public HTTPS on port 443, validates all DNS
+answers, pins the accepted IP, verifies TLS against the original host, ignores
+proxy environment settings, and refuses redirects. Rebinding to loopback,
+private, link-local, metadata or IPv6 ULA addresses is rejected.
+Outbox state retains only closed local error classifications. HTTP library logs
+are suppressed within this transport context so URL queries, auth headers,
+signatures and provider responses cannot spill into logs; unrelated traffic
+retains its logging behavior.
+
+## Leases, retries and retention
+
+PostgreSQL claims use `FOR UPDATE SKIP LOCKED` in short transactions with
+unguessable tokens and database-clock expiration. ACK, retry release,
+suppression and quarantine are fenced by account, consumer namespace, token
+and expiry. The prepared-request seam rechecks the lease after DNS/signing,
+immediately before HTTP. HTTP runs outside database transactions. A crashed claim does not
+consume an HTTP retry budget. Network errors, 408/425/429/5xx, and transient
+signing/DNS failures retry; permanent 4xx, invalid payload/routing evidence,
+permanent scopes and poison rows quarantine without blocking other subscribers.
+
+HTTP acceptance and worker ACK cannot be a single transaction. Receivers must
+deduplicate using a trusted publisher identity that survives key rotation and
+the idempotency key. A lost ACK can produce another authenticated request with
+identical body/key and a fresh signature. Polling remains the recovery path.
+
+This slice retains events, expansion checkpoints, prepared bindings and dirty
+evidence indefinitely. It has no purge API or finite advertised activity horizon.
+Do not delete parent events, keys, or prepared bindings while any delivery is
+nonterminal or within an adopter's promised retry/activity retention horizon.
+
+## Migration and conformance
+
+`reporting_notification_outbox.sql` follows the reviewed four-file foundation
+chain. It adds notification events, expansion/delivery leases, ordered dirty
+records, projector checkpoints, and typed issue scope storage. It rewrites and
+backfills no ledger evidence. `create_schema()` installs all five steps atomically;
+opted-in stores and `PgReportingOutbox.create_schema()` also validate the complete
+installed contract before committing. Default-off Core startup preserves its
+compatibility with adopter indexes. The conservative notification readiness check
+requires the SDK table definitions, including their indexes and guards, to match
+the bundled contract. Concurrent and repeated installations serialize on the schema
+advisory lock. The standalone outbox SQL is atomic even on an autocommit
+connection with the foundation already installed. See
+[reporting ledger migrations](reporting-ledger-migration.md).
+
+The private shared harness is `tests/conformance/reporting/_reliable_support.py`.
+It injects `ManualClock`, scripted sync/async adapters, barriers, failure plans,
+bounded drains and deterministic destination/receiver stores into the same
+memory/PostgreSQL vectors. Essential cases are not marked `integration`.
+
+`test_reporting_notification_process_matrix.py` runs separately pooled producer,
+fanout, HTTP worker, receiver and observer services. Named IPC and SQL barriers
+cover commit/fanout and real TLS HTTP acceptance/ACK. All child, pipe, receiver
+and barrier waits have hard watchdogs with sanitized role/PID/checkpoint
+diagnostics; no timing sleeps control an interleaving. The receiver fixture
+self-check verifies both rotation keys before the full process lane is run.
+The distribution tests build an sdist, build its wheel, import a real base
+installation with PostgreSQL absent, then install `[pg]` and exercise migration,
+commit and restart from that wheel.

pyproject.toml

@@ -116,6 +116,8 @@ adcp-keygen = "adcp.signing.keygen:main"
 
 [project.optional-dependencies]
 dev = [
+    # The reporting outbox gate builds and installs the real sdist/wheel.
+    "build>=1.0",
     "pytest>=7.0.0",
     "pytest-asyncio>=0.21.0",
     "pytest-cov>=4.0.0",

src/adcp/reporting/ledger/delivery.py

@@ -406,14 +406,32 @@ class InMemoryReportingReconciliationStore(InMemoryReportingLedgerStore, _Reconc
     async def _commit(self, record: RecordT) -> tuple[RecordT, bool]:
         candidate = decode_record(payload(record))
         who = principal(candidate)
-        async with self._lock:
+        async with self._mutation():
             records = tuple(item.record for item in self._caller_changes(who))
             existing = replay(candidate, records)
             if existing is not None:
                 return cast(RecordT, existing), False
             context = self._delivery_context(candidate)
             stored = validate_transition(candidate, records, context, self._clock())
             self._append_reconciliation_change(stored)
+            if self._notification_state is not None:
+                from adcp.reporting.ledger.notification_events import (
+                    delivery_dirty,
+                    materialization_event,
+                )
+
+                event = materialization_event(
+                    stored,
+                    records,
+                    context.obligation,
+                    context.revision,
+                    context.configuration,
+                    self._clock(),
+                )
+                if event is not None:
+                    self._record_notification(event)
+                scope, reason, evidence = delivery_dirty(stored, context.obligation)
+                self._dirty_status(scope, reason, after=evidence)
             return cast(RecordT, stored), True
 
     def _append_reconciliation_change(self, record: ReportingDeliveryRecord) -> None:
@@ -461,6 +479,7 @@ def _delivery_context(self, record: ReportingDeliveryRecord) -> DeliveryContext:
             revision_id = record.adjusts_reporting_revision_id
         revision = self._revisions.get(revision_id) if revision_id is not None else None
         return DeliveryContext(
+            configuration=self._configurations.get(record.scope.generation_key),
             obligation=obligation,
             revision=revision,
             adjustment=(