@@ -8,6 +8,12 @@ v2 put retries, idempotency, dead-lettering, encryption and telemetry *inside* e
 
 ## 2. Transport SPI
 
+> **As built ([ADR-033](ADRs/ADR-033-final-transport-spi-pull-receivers-and-the-in-memory-reference-broker.md)).** The SPI is pull-based: `ITransport` has `SendAsync`, `PublishAsync`, `CreateReceiverAsync`, `CreateDeadLetterReceiverAsync` and `ProvisionAsync`. `ITransportReceiver.ReceiveAsync(max, wait)` feeds the core's receive loop, and `ReceivedMessage` settles with `CompleteAsync`, `AbandonAsync(delay)`, `RenewAsync` or `DeadLetterAsync(reason)`.
+>
+> `TransportCapabilities` adds `OrderedDelivery`, `NativeDeadLetter`, `SubscriptionFilters`, `MaxBatchSize`, `Deduplication` and `MaxLockDuration`. Errors are typed: `TransportException.IsTransient`, a per-message `TransportSendException`, and `TopologyValidationException.Problems`.
+>
+> The callback-based sketch below is the original design input.
+
 ```csharp
 public interface ITransport
 {
 
@@ -0,0 +1,68 @@
+# ADR-033: The final transport SPI: pull receivers, CESQL filters and the in-memory reference broker
+
+| | |
+|---|---|
+| **Status** | Accepted |
+| **Date** | 2026-11-11 |
+| **Deciders** | v3 architecture review; WP-3.1 implementation |
+| **Amends** | [doc 07](../07-Cloud-Transports.md) §2, §7; [ADR-014](ADR-014-transport-spi-and-conformance-kit-parity-required.md) |
+
+## Context
+
+Doc 07 §2 sketched a callback-based SPI with a separate sender and provisioner. The core built in M1 uses a pull-based `ITransport`. Here the receive loop, per-key workers, bounded channels and settlement live in the core (`ReceiveEndpoint`, RX-R1..R6).
+
+WP-3.1 has to freeze the SPI before three provider workstreams start in parallel. It also has to turn the 12-case kit into the planned TCK-01..45.
+
+## Decision
+
+1. **Pull, not callbacks.** `ITransportReceiver.ReceiveAsync(max, wait)` stays.
+   - Every target broker has a native pull: SQS `ReceiveMessage`, Pub/Sub pull, Service Bus `ReceiveMessagesAsync`.
+   - Keeping the loop in the core keeps backpressure, per-key ordering and shutdown identical on every broker.
+   - The SPI stays at five types: `ITransport`, `ITransportReceiver`, `ReceivedMessage`, `TransportCapabilities` and `TransportTopology`.
+2. **Settlement is complete.** `ReceivedMessage` has `CompleteAsync`, `AbandonAsync(delay)` (native delay), `RenewAsync` (lock, visibility or lease extension) and `DeadLetterAsync(reason)` (the native backstop).
+   - A late settlement after a lost lock is ignored.
+   - The receive endpoint renews at half `ReceiveLockDuration` while a handler runs.
+   - Disposing a receiver releases what it holds unsettled.
+3. **Capabilities drive the core.** `TransportCapabilities` keeps the size, attribute, retention and delay limits. It gains:
+   - `OrderedDelivery`, `NativeDeadLetter`, `SubscriptionFilters`;
+   - `MaxBatchSize`, `Deduplication` (with its window), `MaxLockDuration`.
+
+   The core uses them to decide:
+   - client-side filtering when a broker has no native filters (UT-RX-22);
+   - whether to run the dead-letter ingestor;
+   - how the TCK judges deduplication.
+4. **Errors are typed.**
+   - `TransportException.IsTransient` (and `RetryAfter` for throttling) drives the error classifier.
+   - `TransportSendException` reports a partial batch per message: oversized messages are refused before anything is sent for them, and the rest are sent.
+   - `DestinationNotFoundException` names a missing endpoint or topic.
+   - `TopologyValidationException.Problems` is the full diff under `Validate`, typed by `TopologyProblemKind`.
+5. **The topology carries what brokers need to configure.** `TransportTopology` gains:
+   - `OrderedEndpoint` and `OrderedTopics`;
+   - `MaxDeliveries`, the dead-letter threshold: the retry budget (1 + delayed retries) plus two;
+   - per-topic `Filters`.
+
+   The full topology model and naming conventions remain WP-3.5. `SourceFlowRuntime.Topology()` builds this topology for the host and the CLI.
+6. **Subscription filters are CloudEvents SQL.**
+   - `CloudEventFilter` parses and evaluates CESQL 1.0 over the CloudEvents attributes: operators, `LIKE`, `IN`, `EXISTS` and the common functions. Errors never match.
+   - Routing compiles `Only<T>()` to `type IN (…)` and ANDs it with `Where(…)`.
+   - Transports translate `ExactTypes` and `TypePrefixes` natively. Anything else is evaluated client-side with the same evaluator, so results are identical (TCK-28).
+7. **The in-memory broker is the reference transport.** It sends messages through the real CloudEvents binary mapping with a 10-attribute budget, the strictest broker's, so the `sfext` overflow runs on every test. It also implements:
+   - ordered endpoints (one in-flight message per key, head-of-line within a key);
+   - locks with expiry and renewal;
+   - a maximum delivery count and a dead-letter queue;
+   - filtered fan-out, batches and partial failures;
+   - a provisioning diff;
+   - faults: outage, throttling, credential expiry, duplicates and dropped acks.
+
+   `StrictDestinations` gives real-broker semantics for unknown destinations. It is off by default, so tests can send to services that start later.
+8. **The TCK runs the real core.**
+   - `SourceFlow.Testing` carries a small generated domain (`TckModule`). Cases 18, 24, 25 and 45 start real services on the transport under test.
+   - The fixture contract adds `Faults` (null on real cloud; those cases are skipped, never weakened), `InjectDefectAsync` for the `Validate` cases, `QuietPeriod` and `DelayTolerance`.
+   - Payload protection cases 41–43 cover the transport side: ciphertext and marker are carried byte for byte and never inspected. Decryption and `DecryptionFailed` parking arrive with WP-5.1.
+
+## Consequences
+
+- Provider packages (WP-3.2..3.4) implement one small SPI, and must pass the same 45 cases as the in-memory broker.
+- Every in-memory test now exercises the CloudEvents wire mapping, so envelope regressions surface everywhere, not only in provider suites.
+- Brokers that cannot filter natively still filter correctly, at the cost of receiving and acknowledging unwanted events.
+- The broker dead-letter queue is a backstop. The ingestor moves its messages to the store, where `dlq replay` re-sends them.
@@ -45,3 +45,4 @@ Every ADR is **Proposed** until the v3 architecture review. "Traces" lists the [
 | [ADR-030](ADR-030-domain-queries-bind-at-startup.md) | Domain queries are checked in two stages: shape at build time, binding at startup | Accepted |
 | [ADR-031](ADR-031-cli-runs-store-commands-globally-and-model-commands-in-the-service.md) | The CLI runs store commands as a global tool and model commands inside the service | Accepted |
 | [ADR-032](ADR-032-blue-green-read-model-generations-behind-forward-only-aliases.md) | Blue/green read-model generations behind forward-only aliases | Accepted |
+| [ADR-033](ADR-033-final-transport-spi-pull-receivers-and-the-in-memory-reference-broker.md) | The final transport SPI: pull receivers, CESQL filters and the in-memory reference broker | Accepted |
@@ -224,6 +224,10 @@ These are additional to the SCK, which the in-memory store also runs.
 | UT-RX-14..16 | Shutdown: stop pulling; in-flight complete within timeout; remainder released | RX-R6 | new |
 | UT-RX-17..19 | Trace: consumer span linked to the producer `traceparent`; consumed and process-duration metrics | ADR-019 | → `CloudTelemetryTests.ExtractTraceParent_*` |
 | UT-RX-20..22 | Tenant restored from the envelope; actor restored; CESQL subscription filter applied client-side when not native | ADR-022, doc 07 | → `SnsMessageFilteringIntegrationTests` (concept) |
+| UT-RX-23 | Transport errors classified by `IsTransient` (outage and throttling transient; missing destination and oversized body permanent) | ADR-033 | → `AwsServiceThrottlingAndFailureTests` (concept) |
+| UT-RX-24 | A handler longer than the lock renews it, so the message is not redelivered | doc 07 §4, ADR-033 | new |
+| UT-RT-21..22 | Subscription filter from routing: `Only<T>()` becomes `type IN (…)` and is ANDed with `Where(…)`; a malformed filter fails at configuration | doc 07 §7 | new |
+| UT-CE-01..08 | CloudEvents SQL filter: operators and precedence, `LIKE` with escapes, `IN`/`EXISTS`, errors never match, functions, parse errors with positions, native translation (`ExactTypes`, `TypePrefixes`), envelope attributes | doc 07 §7, ADR-033 | new |
 | UT-DLQ-01..10 | Dead-letter manager: list and filter (consumer, reason, type, date), show, replay keeps the message id (the inbox still protects), discard with an audited reason, purge older than a cutoff, metrics | GAP-11 | → `InMemoryDeadLetterStoreTests` (11), `AwsDeadLetterQueuePropertyTests` (replay concepts) |
 
 ### 3.7 Subscriptions, projections, process managers (UT-SUB, UT-PRJ, UT-PM, WP-1.9)
 
@@ -64,7 +64,14 @@ Isolation: every test uses a unique prefix (`t{testId}{random}`) for schemas, qu
 
 ## 3. Transport Conformance Kit (TCK)
 
-`TransportConformanceTests` lives in `SourceFlow.Testing` and runs against in-memory, AWS (LocalStack, then real), Google Cloud (emulator, then real) and Azure (emulator, then real). Cases marked † may be `RealCloudOnly` with a written reason when an emulator diverges (as was already the case for two Azure session tests on `azure-cloud`).
+`TransportConformanceTests` lives in `SourceFlow.Testing` and runs against in-memory, AWS (LocalStack, then real), Google Cloud (emulator, then real) and Azure (emulator, then real).
+
+As built ([ADR-033](../ADRs/ADR-033-final-transport-spi-pull-receivers-and-the-in-memory-reference-broker.md)):
+- TCK-18, 24, 25 and 45 start real services on the transport under test, using the kit's generated `TckModule`.
+- TCK-32..34 need fault injection (`ITransportFixture.Faults`) and are skipped on real cloud.
+- TCK-37..40 use `InjectDefectAsync`.
+- TCK-41..43 cover the transport side of payload protection; decryption and `DecryptionFailed` parking come with WP-5.1.
+- TCK-00 checks the case list. Cases marked † may be `RealCloudOnly` with a written reason when an emulator diverges (as was already the case for two Azure session tests on `azure-cloud`).
 
 | ID | Case | v2 scenario sources |
 |----|------|---------------------|
 
@@ -4,14 +4,14 @@ Updated in the PR that completes a work package (CONTRIBUTING.md). Status values
 
 ## Progress summary
 
-Updated 2026-11-11. **34 of 63 work packages are done (54%)**, 0 in progress and 29 not started. By estimated effort, 66 of 138.5 engineer-weeks are done (48%).
+Updated 2026-11-11. **35 of 63 work packages are done (56%)**, 0 in progress and 28 not started. By estimated effort, 69 of 138.5 engineer-weeks are done (50%).
 
 | Milestone | Done | In progress | Todo | Effort done (ew) |
 |-----------|------|-------------|------|------------------|
 | M0 Foundations | 5 | 0 | 0 | 4 of 4 |
 | M1 Core | 15 | 0 | 0 | 37 of 37 |
 | M2 PostgreSQL | 11 | 0 | 0 | 19 of 19 |
-| M3 Cloud transports | 0 | 0 | 7 | 0 of 19 |
+| M3 Cloud transports | 1 | 0 | 6 | 3 of 19 |
 | M4 Tooling and operations | 2 | 0 | 5 | 5.5 of 16.5 |
 | M5 Security and migration | 1 | 0 | 4 | 0.5 of 11 |
 | Track S: Workflow DSL and Studio | 0 | 0 | 13 | 0 of 32 |
@@ -29,10 +29,11 @@ Completed in this round:
 - WP-5.3: redaction wiring (taxonomy, Erasing/HMAC redactors, `LogPayloads`).
 - WP-4.1: the `dotnet sourceflow` CLI (ADR-031).
 - WP-2.7: blue/green read-model generations behind forward-only aliases, including EF Core schema per generation (ADR-032, E2E-08).
+- WP-3.1: the final transport SPI and the 45-case transport conformance kit, passed by the in-memory reference broker (ADR-033).
 
-Nothing is in progress. M2 (PostgreSQL) is complete.
+Nothing is in progress. M2 (PostgreSQL) is complete; M3 has started (WP-3.1 done, the provider packages are next).
 
-Not started: the v2 cloud retirement (3.R), every cloud transport (M3), health and dashboards, SQL Server, event partitioning, chaos and load testing, payload protection, crypto-shredding (redaction is done), the v2 migration tool, and the whole Studio track.
+Not started: the cloud transport packages and provisioning (WP-3.2..3.6), the v2 cloud retirement (3.R), health and dashboards, SQL Server, event partitioning, chaos and load testing, payload protection, crypto-shredding (redaction is done), the v2 migration tool, and the whole Studio track.
 
 | WP | Title | Estimate (ew) | Status | Notes |
 |----|-------|---------------|--------|-------|
@@ -67,7 +68,7 @@ Not started: the v2 cloud retirement (3.R), every cloud transport (M3), health a
 | WP-2.9 | Storage conformance kit complete | 2 | done | SCK-01..58 in SourceFlow.Testing, run on in-memory, PostgreSQL and PostgreSQL with RLS. New in this package: SCK-23 (in-flight writes invisible), SCK-37 (batched retention via IStoreRetention, plus the host retention loop), SCK-54 (CsCheck property: snapshot + tail = full fold), PostgreSQL SCK-28 (LISTEN loss and reconnect), and SCK-L01 CsCheck linearisability of appends and the outbox. SCK-36 is covered by SCK-34. SCK-20 (one round trip) moves to WP-2.10, and SCK-65..72 (partitioned events) to WP-4.5 |
 | WP-2.10 | Performance baseline | 1 | done | benchmarks/SourceFlow.Benchmarks (BenchmarkDotNet PERF-01..04, 06; measurement harness PERF-05, 08, 10); baselines in benchmarks/baselines and the report in implementation/perf-baseline.md. One-round-trip PostgreSQL commit (SCK-20), batch-aware document projections and adaptive subscription batches. Deviations logged with owners (PERF-01/02 allocations, PERF-10 borderline). PERF-07 waits for M3 and PERF-09 for WP-4.7 |
 | WP-2.R | Retire v2 EF stores | 0.5 | done | Every `SourceFlow.Stores.EntityFramework.Tests` row is satisfied (SCK on three adapters, UT-EF-01..10). `src/SourceFlow.Stores.EntityFramework` and its tests are deleted |
-| WP-3.1 | Transport SPI final + Transport Conformance Kit | 3 | todo | |
+| WP-3.1 | Transport SPI final + Transport Conformance Kit | 3 | done | Final pull-based SPI (ADR-033): `RenewAsync`, `DeadLetterAsync`, `CreateDeadLetterReceiverAsync`, capabilities (ordered delivery, native DLQ, native filters, batch size, deduplication, lock), typed errors (`TransportException.IsTransient`, per-message `TransportSendException`, `DestinationNotFoundException`, `TopologyValidationException.Problems`), topology with ordering, max deliveries and filters (`SourceFlowRuntime.Topology()`). CloudEvents SQL filters (`CloudEventFilter`; `Only<T>()` and `Where(…)` in routing; native translation or client-side evaluation). Core: lock renewal while handling, client-side filtering, broker dead-letter ingestor. In-memory broker as the reference: CloudEvents binary wire with a 10-attribute budget, ordered endpoints, locks, max deliveries and DLQ, filtered fan-out, batches, provisioning diff, faults. TCK-01..45 in `SourceFlow.Testing` with a generated end-to-end domain and the fixture contract (faults, defects); all 45 pass on in-memory. UT-CE-01..08, UT-RT-21..22, UT-RX-22..24 |
 | WP-3.2 | `SourceFlow.Cloud.Aws` | 4 | todo | |
 | WP-3.3 | `SourceFlow.Cloud.Gcp` | 3.5 | todo | |
 | WP-3.4 | `SourceFlow.Cloud.Azure` | 4 | todo | |
 
@@ -47,6 +47,9 @@ internal static partial class Log
     [LoggerMessage(1203, LogLevel.Error, "Receive loop for {Endpoint} failed; continuing")]
     public static partial void ReceiveLoopFailed(ILogger logger, Exception exception, string endpoint);
 
+    [LoggerMessage(1210, LogLevel.Warning, "Message {MessageId} was dead-lettered by the broker on {Endpoint} ({Reason}); parked for replay")]
+    public static partial void BrokerDeadLetterIngested(ILogger logger, string messageId, string endpoint, string reason);
+
     [LoggerMessage(1301, LogLevel.Error, "Subscription {Subscription} partition {Partition} failed on event {EventId}; {Policy}")]
     public static partial void SubscriptionFailed(ILogger logger, Exception exception, string subscription, int partition, string eventId, string policy);
 
 
@@ -27,6 +27,7 @@ public sealed class SourceFlowHost : IHostedService, IHealthCheck, IAsyncDisposa
     private volatile bool _ready;
 
     private readonly Query.QueryService? _queries;
+    private DeadLetterIngestor? _ingestor;
 
     internal SourceFlowHost(SourceFlowRuntime runtime, OutboxRelay relay, ReceiveEndpoint receiver, SubscriptionEngine subscriptions, HostSettings settings, Query.QueryService? queries = null)
     {
@@ -57,12 +58,7 @@ public ValueTask<bool> SubscriptionsCaughtUpAsync(CancellationToken cancellation
     /// <inheritdoc />
     public async Task StartAsync(CancellationToken cancellationToken)
     {
-        var topology = new TransportTopology(
-            _runtime.Routing.Endpoint,
-            [.. _runtime.Routing.PublishedTopics],
-            [.. _runtime.Routing.Subscriptions.Select(s => s.Topic)],
-            [.. _runtime.Routing.Commands.Values.Select(c => c.Endpoint).Distinct(StringComparer.Ordinal)]);
-        await _runtime.Transport.ProvisionAsync(topology, _settings.Provisioning, cancellationToken).ConfigureAwait(false);
+        await _runtime.Transport.ProvisionAsync(_runtime.Topology(), _settings.Provisioning, cancellationToken).ConfigureAwait(false);
 
         await RegisterSchemasAsync(cancellationToken).ConfigureAwait(false);
         CheckDomainQueries();
@@ -75,6 +71,11 @@ [.. _runtime.Routing.Subscriptions.Select(s => s.Topic)],
         {
             await _receiver.StartAsync(stopping).ConfigureAwait(false);
             _loops.Add(Task.Run(() => _receiver.RunAsync(stopping), CancellationToken.None));
+            if (_runtime.Options.IngestBrokerDeadLetters && _runtime.Transport.Capabilities.NativeDeadLetter)
+            {
+                _ingestor = new DeadLetterIngestor(_runtime);
+                _loops.Add(Task.Run(() => LoopAsync(ct => _ingestor.RunOnceAsync(ct), stopping), CancellationToken.None));
+            }
         }
 
         if (_settings.Roles.HasFlag(SourceFlowRoles.Relay))
@@ -159,6 +160,10 @@ public async ValueTask DisposeAsync()
     {
         _stopping?.Dispose();
         await _receiver.DisposeAsync().ConfigureAwait(false);
+        if (_ingestor is not null)
+        {
+            await _ingestor.DisposeAsync().ConfigureAwait(false);
+        }
     }
 
     // Batched retention cleanup (doc 05 §9): cutoffs on the business clock, the interval on the system clock.
 
@@ -177,6 +177,10 @@ public ErrorClassifier()
         Treat<ArgumentException>(HandlingOutcome.Permanent);
         Treat<InvalidCastException>(HandlingOutcome.Permanent);
         Treat<PermanentException>(HandlingOutcome.Permanent);
+
+        // Broker errors say whether they are transient (outage, throttling) or not (missing destination, oversized body).
+        _rules.Insert(0, (e => e is Transport.TransportException { IsTransient: false }, HandlingOutcome.Permanent));
+        _rules.Insert(0, (e => e is Transport.TransportException { IsTransient: true }, HandlingOutcome.Transient));
     }
 
     /// <summary>Classifies <typeparamref name="TException"/> (and subclasses); later rules win.</summary>