diff --git a/README.md b/README.md
index aa69f1b..e0eb511 100644
--- a/README.md
+++ b/README.md
@@ -17,7 +17,9 @@ the live, phone-sized field report is at
   and crash-safe cutover into steady-state ingestion
 - Sync 1.1 commit verification, including PLC key rotation, MST inversion,
   op-CID checks, replay protection, durable repository-chain state, and
-  transparent authenticated whole-repo repair with ordered pending replay
+  transparent authenticated whole-repo repair with ordered pending replay;
+  the live scheduler matches Atmos's 32-worker per-DID FIFO, bounded
+  drop-oldest, completion-order batching, and safe cursor watermark behavior
 - durable metadata and archive publication through RocksDB transactions,
   coupled to segment fsync boundaries
 - tombstone-aware compaction and continuing repair of repositories that were
diff --git a/docs/live-scheduler.md b/docs/live-scheduler.md
new file mode 100644
index 0000000..6b313e5
--- /dev/null
+++ b/docs/live-scheduler.md
@@ -0,0 +1,50 @@
+# Live scheduler parity
+
+Stream's live path follows the scheduler and cursor behavior in pinned Atmos
+v0.2.14, which Jetstream V2 configures with its defaults.
+
+## Contract
+
+- Thirty-two workers execute at most thirty-two active DID chains. A new DID
+  backpressures the websocket dispatch path until a worker is available.
+- A worker owns one DID chain until it is empty. Events for that DID execute
+  in arrival order; other DIDs can complete and reach the writer first.
+- Each active DID may retain 64 pending events. The 65th pending arrival
+  removes the oldest pending event and appends the newest. The displaced seq
+  leaves the inflight set and increments the canonical
+  `jetstream_livestream_verify_queue_dropped_events_total` counter.
+- Results are delivered in completion order in batches of 50 or after 500 ms.
+  Completed-result buffering is bounded to Atmos's 4,096 entries; a full
+  result channel backpressures the worker that owns that DID chain.
+  A malformed frame, verifier error, replay, or repair-withheld event may be
+  collected silently; it does not manufacture a non-empty delivery batch.
+- After a successful non-empty batch, the relay cursor becomes the greater of
+  its prior value and `min(inflight)-1`. If nothing remains inflight, it uses
+  the largest seq in that batch. The pipeline is seeded from the durable
+  cursor at startup, so a replayed lower batch cannot regress it. The cursor
+  callback runs only after all event callbacks in the batch succeed.
+- Archive rows, verifier state, replay ratchets, and relay cursor retain their
+  existing fsync coupling. Cursor staging or durable metadata failure is
+  lifecycle-fatal.
+- Graceful cutover stops admission, drains every accepted frame and ordered
+  repair completion, then tears down worker tasks. Abrupt cancellation owns
+  and frees worker-local and writer-local jobs rather than abandoning them.
+
+## Offline receipts
+
+`zig build test` uses no public network. Scheduler tests encode real firehose
+control frames and use the production verifier with a temporary RocksDB
+store. Holding the verifier's real hosting-state lane supplies deterministic
+pressure without substituting a fake verifier:
+
+1. a second DID completes while the first is blocked;
+2. a second event for the blocked DID cannot pass the first;
+3. all 32 active DID chains occupy the pool and the 33rd submit remains
+   blocked until one completes;
+4. 65 pending events retain seqs 3 through 66, proving seq 2 was the exact
+   drop-oldest victim and left the inflight set; and
+5. cursor calculation returns the smallest inflight seq minus one, then the
+   batch maximum once the set is empty.
+
+These receipts complement the loopback HTTP and signed-CAR repair tests in
+`live-repair.md`; neither suite contacts a public PDS, relay, or PLC service.
diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md
index 3ae6129..6ca7d08 100644
--- a/docs/semantic-parity.md
+++ b/docs/semantic-parity.md
@@ -16,7 +16,7 @@ this document or `bootstrap-semantic-parity.md` is open.
 | JSS v1 storage | Upstream-produced fixture, reciprocal segment parsing, checksum/index/gloom vectors | closed for sealed format; rerun reciprocal corpus before experiment |
 | Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | implementation present; fresh full suite required |
 | Subscribe v1/v2 | Wire/filter/cursor/zstd unit and local e2e coverage | implementation present; official-client replay and below-floor/adversarial suite required |
-| Sync 1.1 live verification | Upstream requires durable per-DID chain/hosting state, MST inversion, op-CID consistency, rev replay/future guards, default acceptance of legacy-shaped commits, and transparent whole-repo repair | The verifier core is closed at the implementation boundary: diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof, signed inner/outer consistency, exact decimal size/future-rev/replay gates, exact legacy-shape detection with default `LegacyAccept`, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures route to repair; signature failures and outer/inner producer-integrity mismatches bypass it, and no failed event is archived. `prepareRepo` retains and authenticates the canonical signed complete head without a second multi-gigabyte parse/walk. Offline receipts use real signed CARs and real verifier state, not fixture verdicts. **Open:** replace the global ticket/backpressure scheduler with upstream's 32-worker per-DID FIFO, 64-event per-DID drop-oldest overflow, cross-DID completion delivery, and smallest-inflight cursor watermark. |
+| Sync 1.1 live verification | Upstream requires durable per-DID chain/hosting state, MST inversion, op-CID consistency, rev replay/future guards, default acceptance of legacy-shaped commits, transparent whole-repo repair, and Atmos delivery scheduling | **closed at the implementation boundary.** Diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof, signed inner/outer consistency, exact decimal size/future-rev/replay gates, exact legacy-shape detection with default `LegacyAccept`, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures route to repair; signature failures and outer/inner producer-integrity mismatches bypass it, and no failed event is archived. `prepareRepo` retains and authenticates the canonical signed complete head without a second multi-gigabyte parse/walk. The live path now uses Atmos's 32 worker slots, one worker-held FIFO chain per DID, a 64-pending drop-oldest boundary, completion-order batches of 50/500 ms across DIDs, and `min(inflight)-1` cursor watermarks. Dropped and silent events leave the inflight set without inventing a delivery batch; cursor advancement remains coupled to durable archive boundaries. Offline receipts use real encoded frames, real RocksDB verifier state, and actual worker blocking rather than fixture verdicts or scheduler mocks. See `live-scheduler.md`. |
 | Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. The writer emits `sync` then bounded 1,024-row `create_resync` batches and stages the fetched/pending chain checkpoints under the archive lock before any matching fsync. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. |
 | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **open — inventory and implementation receipt required** |
 | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | durable repo fields implemented; **operator repo/host views open** |
diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig
index 3cc6d84..710f998 100644
--- a/src/internal/bootstrap/lifecycle.zig
+++ b/src/internal/bootstrap/lifecycle.zig
@@ -324,6 +324,8 @@ fn runBootstrap(
         .emit_ctx = &consumer,
         .emit = ingest.Consumer.emitVerified,
         .emit_repair = ingest.Consumer.emitRepair,
+        .emit_cursor = ingest.Consumer.emitCursor,
+        .cursor_watermark = cursor_store.load(io) orelse 0,
     };
     var repair_scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined;
     const repair_scratch = try std.fmt.bufPrint(&repair_scratch_buf, "{s}/repair-scratch", .{live_data_dir});
diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig
index 2b7cf6e..775eff6 100644
--- a/src/internal/ingest.zig
+++ b/src/internal/ingest.zig
@@ -63,15 +63,14 @@ pub const Consumer = struct {
         if (self.getFatal()) |err| return err;
     }
 
-    /// raw frames from zat's client go through the verify pipeline; the
-    /// pipeline's writer thread calls emitVerified in ticket order.
+    /// Raw frames go through the Atmos-compatible scheduler. Its writer calls
+    /// emitVerified in completion order across DIDs and FIFO order per DID.
     pub fn onRawFrame(self: *Consumer, data: []const u8) void {
         const pipe = self.pipeline orelse return;
         pipe.submit(data) catch |err| {
-            // An Io future cancellation interrupts a producer blocked on the
-            // bounded ring during intentional shutdown. The pipeline retains
-            // every ticket accepted before that cancellation and drains them
-            // at the archive cutover boundary.
+            // Cancellation can interrupt a producer waiting for one of the
+            // 32 active-DID worker slots during intentional shutdown. Every
+            // frame accepted before that cancellation remains drain-owned.
             if (err == error.Canceled or err == error.Closed) return;
             log.err("pipeline submit failed: {s}", .{@errorName(err)});
             self.setFatal(err);
@@ -188,6 +187,16 @@ pub const Consumer = struct {
         try archive.flushBlock();
     }
 
+    /// Atmos advances its cursor only after a non-empty delivery batch
+    /// returns. The pipeline supplies that smallest-inflight watermark here;
+    /// it is staged against the latest local archive row, never an individual
+    /// event's possibly-out-of-order relay seq.
+    pub fn emitCursor(ctx: *anyopaque, watermark: i64) !void {
+        const self: *Consumer = @ptrCast(@alignCast(ctx));
+        const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds();
+        try self.noteWatermark(watermark, now_us, if (self.archive) |archive| archive.lastSeq() else 0);
+    }
+
     fn getFatal(self: *Consumer) ?anyerror {
         self.fatal_mu.lockUncancelable(self.io);
         defer self.fatal_mu.unlock(self.io);
@@ -241,7 +250,7 @@ pub const Consumer = struct {
             .identity => |identity| {
                 if (identity.seq <= try self.appliedSeq(.identity, identity.did)) {
                     if (self.stats) |stats| _ = stats.replayed_identity_events_dropped_total.fetchAdd(1, .monotonic);
-                    if (!replayed_after_repair) self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0);
+                    if (!replayed_after_repair) try self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0);
                     return;
                 }
                 // rotation signal: next commit re-resolves the key
@@ -249,13 +258,14 @@ pub const Consumer = struct {
             },
             .account => |account| if (account.seq <= try self.appliedSeq(.account, account.did)) {
                 if (self.stats) |stats| _ = stats.replayed_account_events_dropped_total.fetchAdd(1, .monotonic);
-                if (!replayed_after_repair) self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0);
+                if (!replayed_after_repair) try self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0);
                 return;
             },
             else => {},
         }
         var persisted: Persisted = .{};
-        if (self.archive) |a| persisted = self.persist(a, event, verdict, !replayed_after_repair) catch |err| {
+        const owns_cursor = !replayed_after_repair and self.pipeline == null;
+        if (self.archive) |a| persisted = self.persist(a, event, verdict, owns_cursor) catch |err| {
             // our own storage failing is fatal-worthy, not drop-worthy
             log.err("archive append failed: {s}", .{@errorName(err)});
             return err;
@@ -277,13 +287,13 @@ pub const Consumer = struct {
         if (persisted.last == 0) {
             if (self.archive) |a| {
                 const barrier = a.lastSeq();
-                try self.stageEventBoundary(event, verdict, barrier, !replayed_after_repair);
+                try self.stageEventBoundary(event, verdict, barrier, owns_cursor);
                 try self.commitDurableMetadata(
                     a.committed_seq.load(.acquire),
                     a.committed_seq.load(.acquire) + 1,
-                    if (!replayed_after_repair) event.seq() else null,
+                    if (owns_cursor) event.seq() else null,
                 );
-            } else if (!replayed_after_repair) self.noteSeq(event, time_us, 0);
+            } else if (owns_cursor) try self.noteSeq(event, time_us, 0);
         }
 
         if (self.to_stdout and frames.items.len > 0) {
@@ -296,15 +306,18 @@ pub const Consumer = struct {
         }
     }
 
-    fn noteSeq(self: *Consumer, event: zat.FirehoseEvent, time_us: i64, local_seq: u64) void {
+    fn noteSeq(self: *Consumer, event: zat.FirehoseEvent, time_us: i64, local_seq: u64) !void {
+        if (self.pipeline != null) return;
         const seq = event.seq() orelse return;
+        try self.noteWatermark(seq, time_us, local_seq);
+    }
+
+    fn noteWatermark(self: *Consumer, seq: i64, time_us: i64, local_seq: u64) !void {
         if (self.stats) |st| st.upstream_seq.store(seq, .monotonic);
         if (self.archive) |a| {
             if (a.committed_seq.load(.acquire) >= local_seq) {
-                self.commitDurableMetadata(a.committed_seq.load(.acquire), a.committed_seq.load(.acquire) + 1, seq) catch |err|
-                    log.err("metadata commit after archive sync failed: {s}", .{@errorName(err)});
-            } else self.pending_cursors.append(self.allocator, .{ .local_seq = local_seq, .upstream_seq = seq }) catch |err|
-                log.err("cursor staging failed: {s}", .{@errorName(err)});
+                try self.commitDurableMetadata(a.committed_seq.load(.acquire), a.committed_seq.load(.acquire) + 1, seq);
+            } else try self.pending_cursors.append(self.allocator, .{ .local_seq = local_seq, .upstream_seq = seq });
             return;
         }
         if (self.cursor_store) |cs| cs.update(self.io, seq, time_us);
diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig
index 1f27ccf..9dddb79 100644
--- a/src/internal/metrics.zig
+++ b/src/internal/metrics.zig
@@ -41,6 +41,7 @@ pub const Stats = struct {
     resync_queue_depth: std.atomic.Value(usize) = .init(0),
     resync_pending: std.atomic.Value(usize) = .init(0),
     livestream_decode_errors_total: Counter = .init(0),
+    verify_queue_dropped_total: Counter = .init(0),
     replayed_account_events_dropped_total: Counter = .init(0),
     replayed_identity_events_dropped_total: Counter = .init(0),
     /// highest physically-compacted seq (lag = stream_upstream_seq - this)
@@ -131,6 +132,8 @@ pub fn format(
         \\stream_resync_pending {d}
         \\# TYPE jetstream_livestream_stale_resyncs_dropped_total counter
         \\jetstream_livestream_stale_resyncs_dropped_total {d}
+        \\# TYPE jetstream_livestream_verify_queue_dropped_events_total counter
+        \\jetstream_livestream_verify_queue_dropped_events_total {d}
         \\
     , .{
         stats.resync_completed_total.load(.monotonic),
@@ -141,6 +144,7 @@ pub fn format(
         stats.resync_queue_depth.load(.monotonic),
         stats.resync_pending.load(.monotonic),
         stats.resync_stale_total.load(.monotonic),
+        stats.verify_queue_dropped_total.load(.monotonic),
     }) catch {};
 
     // Canonical upstream names. These are aliases only where Stream owns the
@@ -285,6 +289,7 @@ test "format renders counters" {
     _ = stats.ingest_append_errors_total.fetchAdd(8, .monotonic);
     stats.ingest_active_segment_bytes.store(9000, .monotonic);
     _ = stats.livestream_decode_errors_total.fetchAdd(2, .monotonic);
+    _ = stats.verify_queue_dropped_total.fetchAdd(9, .monotonic);
     _ = stats.replayed_account_events_dropped_total.fetchAdd(3, .monotonic);
     _ = stats.replayed_identity_events_dropped_total.fetchAdd(4, .monotonic);
     _ = stats.retry_attempts_total.fetchAdd(5, .monotonic);
@@ -312,6 +317,7 @@ test "format renders counters" {
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_append_errors_total 8") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_active_segment_bytes 9000") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_decode_errors_total 2") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_verify_queue_dropped_events_total 9") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_replayed_account_events_dropped_total 3") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_replayed_identity_events_dropped_total 4") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_attempts_total 5") != null);
diff --git a/src/internal/pipeline.zig b/src/internal/pipeline.zig
index 899b130..1c8b4d7 100644
--- a/src/internal/pipeline.zig
+++ b/src/internal/pipeline.zig
@@ -1,13 +1,6 @@
-//! ingest pipeline — parallel verification, ordered emission
-//!
-//! upstream-faithful: no event is served or archived unverified. the ws
-//! reader submits raw frames into a bounded queue (a full queue blocks the
-//! reader — TCP backpressure to the relay, like upstream); K verify workers
-//! decode and verify in parallel (blocking on DID resolution is fine here);
-//! a single writer drains a reorder buffer IN TICKET ORDER, so downstream
-//! (archive append order, tail order, cursor gating) is byte-identical to
-//! the old single-threaded path. cross-context signaling is Io.Mutex/
-//! Condition from one Io backend only (docs/lessons-from-zlay.md #2).
+//! Atmos-compatible live scheduler: 32 workers, FIFO per DID, 64 pending
+//! events per DID with drop-oldest overflow, completion-order delivery across
+//! DIDs, and a smallest-in-flight relay cursor watermark.
 
 const std = @import("std");
 const zat = @import("zat");
@@ -21,24 +14,30 @@ const Allocator = std.mem.Allocator;
 const log = std.log.scoped(.stream);
 
 const worker_count = 32;
-/// Per-DID verification is serialized through ordered hash lanes. Tickets
-/// register in relay order, then different lanes may verify concurrently.
-/// Hash collisions only reduce parallelism; they cannot reorder a repo.
-const verification_lane_count = 64;
-/// reorder-buffer capacity == max in-flight frames; bounds memory and
-/// gives the head-of-line window workers can verify ahead into
-const capacity = 256;
-
-const Slot = struct {
+const key_queue_capacity = worker_count * 2;
+const result_capacity = 4096;
+const batch_size = 50;
+const batch_timeout: Io.Clock.Duration = .{ .raw = Io.Duration.fromMilliseconds(500), .clock = .awake };
+
+const Job = struct {
     ticket: u64,
-    frame: []u8, // owned raw ws frame bytes
-    /// decode-once: the worker decodes into this arena; the writer consumes
-    /// the event and destroys the arena (null event = frame didn't decode)
+    frame: []u8,
     arena: ?*std.heap.ArenaAllocator = null,
     event: ?zat.FirehoseEvent = null,
-    verdict: verify.Result = .{}, // meaning "no objection" when no verifier
+    key: []const u8 = "",
+    seq: ?i64 = null,
+    verdict: verify.Result = .{},
     failure: ?anyerror = null,
-    done: bool = false,
+};
+
+const KeyChain = struct {
+    key: []u8,
+    pending: std.ArrayList(*Job) = .empty,
+};
+
+const Delivery = union(enum) {
+    job: *Job,
+    repair: *repair.Completion,
 };
 
 pub const Pipeline = struct {
@@ -47,31 +46,37 @@ pub const Pipeline = struct {
     verifier: ?*verify.Verifier,
     repair: ?*repair.Coordinator = null,
     stats: ?*metrics.Stats = null,
-    /// emit(ctx, frame_bytes, verdict) — runs on the writer thread, in
-    /// ticket order; owns nothing (pipeline frees the frame after).
+    /// Normal events are delivered in cross-DID verification completion
+    /// order, while each DID remains FIFO.
     emit_ctx: *anyopaque,
     emit: *const fn (*anyopaque, zat.FirehoseEvent, verify.Result) anyerror!void,
     emit_repair: ?*const fn (*anyopaque, *repair.Completion) anyerror!void = null,
+    emit_cursor: ?*const fn (*anyopaque, i64) anyerror!void = null,
 
     mutex: Io.Mutex = .init,
-    /// workers wait for work; reader waits for space; writer waits for done
     work_cond: Io.Condition = .init,
     space_cond: Io.Condition = .init,
-    done_cond: Io.Condition = .init,
-    verify_cond: Io.Condition = .init,
-    slots: [capacity]?Slot = @splat(null),
-    next_submit: u64 = 0, // ticket for the next submitted frame
-    next_claim: u64 = 0, // next ticket a worker should verify
-    next_emit: u64 = 0, // next ticket the writer will emit
+    wake: Io.Event = .unset,
+    chains: std.StringHashMapUnmanaged(*KeyChain) = .empty,
+    ready: std.ArrayList(*Job) = .empty,
+    ready_read: usize = 0,
+    completed: std.ArrayList(*Job) = .empty,
+    completed_read: usize = 0,
+    inflight: std.AutoHashMapUnmanaged(i64, void) = .empty,
+    processed_sparse: std.AutoHashMapUnmanaged(u64, void) = .empty,
+    processed_prefix: u64 = 0,
+    next_submit: u64 = 0,
+    claimed: u64 = 0,
+    processed: u64 = 0,
+    active_chains: usize = 0,
+    writer_jobs_buffered: usize = 0,
+    cursor_watermark: i64 = 0,
     accepting: bool = true,
     alive: bool = true,
     stopped: bool = false,
     fatal_error: ?anyerror = null,
-    next_verify_register: u64 = 0,
-    verify_lane_tail: [verification_lane_count]?u64 = @splat(null),
-    verify_lane_completed: [verification_lane_count]?u64 = @splat(null),
     repair_queue: std.ArrayList(*repair.Completion) = .empty,
-    repair_inflight: bool = false,
+    repair_buffered: usize = 0,
 
     // workers + writer run as io.concurrent tasks, never raw std.Threads:
     // they call io-backed I/O (DID resolution, archive writes), and Io
@@ -100,8 +105,7 @@ pub const Pipeline = struct {
         self.alive = false;
         self.work_cond.broadcast(self.io);
         self.space_cond.broadcast(self.io);
-        self.done_cond.broadcast(self.io);
-        self.verify_cond.broadcast(self.io);
+        self.wake.set(self.io);
         self.mutex.unlock(self.io);
     }
 
@@ -114,10 +118,15 @@ pub const Pipeline = struct {
             if (f.*) |*future| _ = future.cancel(self.io);
             f.* = null;
         }
-        for (&self.slots) |*slot| {
-            if (slot.*) |s| self.freeSlot(s);
-            slot.* = null;
-        }
+        for (self.ready.items[self.ready_read..]) |job| self.freeJob(job);
+        for (self.completed.items[self.completed_read..]) |job| self.freeJob(job);
+        self.ready.deinit(self.allocator);
+        self.completed.deinit(self.allocator);
+        var chains = self.chains.iterator();
+        while (chains.next()) |entry| self.freeChain(entry.value_ptr.*);
+        self.chains.deinit(self.allocator);
+        self.inflight.deinit(self.allocator);
+        self.processed_sparse.deinit(self.allocator);
         for (self.repair_queue.items) |completion| completion.deinit();
         self.repair_queue.deinit(self.allocator);
     }
@@ -131,22 +140,22 @@ pub const Pipeline = struct {
         self.accepting = false;
         // Signals are intentionally not counted by a condition variable. A
         // burst submitted immediately before capture cancellation can leave
-        // workers asleep even though next_claim trails next_submit, so the
+        // workers asleep even though claimed trails next_submit, so the
         // cutover must wake every worker to consume the fixed backlog.
         self.work_cond.broadcast(self.io);
         self.space_cond.broadcast(self.io);
-        log.debug("pipeline drain boundary: {d} submitted, {d} claimed, {d} emitted", .{
+        log.debug("pipeline drain boundary: {d} submitted, {d} claimed, {d} processed", .{
             self.next_submit,
-            self.next_claim,
-            self.next_emit,
+            self.claimed,
+            self.processed,
         });
-        while (self.alive and self.next_emit != self.next_submit) {
+        while (self.alive and (self.processed != self.next_submit or self.writer_jobs_buffered != 0)) {
             self.space_cond.wait(self.io, &self.mutex) catch {};
         }
         self.mutex.unlock(self.io);
         if (self.repair) |coordinator| try coordinator.drainAndStop();
         self.mutex.lockUncancelable(self.io);
-        while (self.alive and (self.repair_queue.items.len != 0 or self.repair_inflight))
+        while (self.alive and (self.repair_queue.items.len != 0 or self.repair_buffered != 0))
             self.space_cond.wait(self.io, &self.mutex) catch {};
         const failure = self.fatal_error;
         self.mutex.unlock(self.io);
@@ -154,12 +163,20 @@ pub const Pipeline = struct {
         if (failure) |err| return err;
     }
 
-    fn freeSlot(self: *Pipeline, slot: Slot) void {
-        if (slot.arena) |a| {
+    fn freeJob(self: *Pipeline, job: *Job) void {
+        if (job.arena) |a| {
             a.deinit();
             self.allocator.destroy(a);
         }
-        self.allocator.free(slot.frame);
+        self.allocator.free(job.frame);
+        self.allocator.destroy(job);
+    }
+
+    fn freeChain(self: *Pipeline, chain: *KeyChain) void {
+        for (chain.pending.items) |job| self.freeJob(job);
+        chain.pending.deinit(self.allocator);
+        self.allocator.free(chain.key);
+        self.allocator.destroy(chain);
     }
 
     pub fn fatalError(self: *Pipeline) ?anyerror {
@@ -168,75 +185,116 @@ pub const Pipeline = struct {
         return self.fatal_error orelse if (self.repair) |coordinator| coordinator.fatalError() else null;
     }
 
-    /// called by the ws reader: takes ownership of nothing — dupes `data`.
-    /// blocks when the buffer is full (backpressure).
+    /// called by the websocket reader. Decode happens on this dispatch path,
+    /// matching Atmos; only verification runs in the worker pool.
     pub const Gauges = struct { submitted: u64, claimed: u64, emitted: u64 };
     pub fn gauges(self: *Pipeline) Gauges {
         self.mutex.lockUncancelable(self.io);
         defer self.mutex.unlock(self.io);
-        return .{ .submitted = self.next_submit, .claimed = self.next_claim, .emitted = self.next_emit };
+        return .{ .submitted = self.next_submit, .claimed = self.claimed, .emitted = self.processed };
     }
 
     pub fn submit(self: *Pipeline, data: []const u8) !void {
+        const job = try self.allocator.create(Job);
+        errdefer self.allocator.destroy(job);
         const frame = try self.allocator.dupe(u8, data);
         errdefer self.allocator.free(frame);
+        job.* = .{ .ticket = 0, .frame = frame };
+        if (self.allocator.create(std.heap.ArenaAllocator)) |arena| {
+            arena.* = std.heap.ArenaAllocator.init(self.allocator);
+            if (zat.firehose.decodeFrame(arena.allocator(), frame)) |event| {
+                job.arena = arena;
+                job.event = event;
+                job.key = eventDid(event) orelse "";
+                job.seq = event.seq();
+            } else |err| {
+                log.debug("frame decode error: {s}", .{@errorName(err)});
+                if (self.stats) |stats| _ = stats.livestream_decode_errors_total.fetchAdd(1, .monotonic);
+                arena.deinit();
+                self.allocator.destroy(arena);
+            }
+        } else |_| return error.OutOfMemory;
+        errdefer if (job.arena) |arena| {
+            arena.deinit();
+            self.allocator.destroy(arena);
+        };
+
         self.mutex.lockUncancelable(self.io);
         defer self.mutex.unlock(self.io);
-        while (self.alive and self.accepting and self.next_submit - self.next_emit >= capacity) {
-            try self.space_cond.wait(self.io, &self.mutex);
-        }
         if (!self.alive or !self.accepting) return error.Closed;
-        std.debug.assert(self.next_submit - self.next_emit < capacity);
-        self.slots[self.next_submit % capacity] = .{ .ticket = self.next_submit, .frame = frame };
-        self.next_submit += 1;
-        self.work_cond.signal(self.io);
+        while (self.chains.get(job.key) == null and self.active_chains >= worker_count) {
+            self.space_cond.wait(self.io, &self.mutex) catch return error.Canceled;
+            if (!self.alive or !self.accepting) return error.Closed;
+        }
+
+        if (self.chains.get(job.key)) |chain| {
+            try chain.pending.ensureUnusedCapacity(self.allocator, 1);
+            if (job.seq) |seq| if (seq > 0) try self.inflight.ensureUnusedCapacity(self.allocator, 1);
+            if (chain.pending.items.len >= key_queue_capacity)
+                try self.processed_sparse.ensureUnusedCapacity(self.allocator, 1);
+            job.ticket = self.next_submit;
+            self.next_submit += 1;
+            if (job.seq) |seq| if (seq > 0) self.inflight.putAssumeCapacity(seq, {});
+            if (chain.pending.items.len >= key_queue_capacity) {
+                const dropped = chain.pending.orderedRemove(0);
+                self.dropJobLocked(dropped);
+            }
+            chain.pending.appendAssumeCapacity(job);
+        } else {
+            const chain = try self.allocator.create(KeyChain);
+            errdefer self.allocator.destroy(chain);
+            chain.* = .{ .key = try self.allocator.dupe(u8, job.key) };
+            errdefer self.allocator.free(chain.key);
+            try self.chains.ensureUnusedCapacity(self.allocator, 1);
+            try self.ready.ensureUnusedCapacity(self.allocator, 1);
+            if (job.seq) |seq| if (seq > 0) try self.inflight.ensureUnusedCapacity(self.allocator, 1);
+            job.ticket = self.next_submit;
+            self.next_submit += 1;
+            if (job.seq) |seq| if (seq > 0) self.inflight.putAssumeCapacity(seq, {});
+            self.chains.putAssumeCapacity(chain.key, chain);
+            self.ready.appendAssumeCapacity(job);
+            self.active_chains += 1;
+            self.work_cond.signal(self.io);
+        }
+    }
+
+    fn dropJobLocked(self: *Pipeline, job: *Job) void {
+        if (job.seq) |seq| _ = self.inflight.remove(seq);
+        self.markProcessedLocked(job.ticket);
+        if (self.stats) |stats| _ = stats.verify_queue_dropped_total.fetchAdd(1, .monotonic);
+        log.warn("verify queue overflow dropped oldest event: key={s} ticket={d}", .{ job.key, job.ticket });
+        self.freeJob(job);
     }
 
     fn workerLoop(self: *Pipeline) void {
         var resolver_storage = if (self.verifier) |v| v.makeResolver() else null;
         defer if (resolver_storage) |*r| r.deinit();
         while (true) {
-            // claim the next unverified ticket
             self.mutex.lockUncancelable(self.io);
-            while (self.alive and self.next_claim == self.next_submit) {
+            while (self.alive and self.ready_read == self.ready.items.len) {
+                self.compactReadyLocked();
                 self.work_cond.wait(self.io, &self.mutex) catch {};
             }
             if (!self.alive) {
                 self.mutex.unlock(self.io);
                 return;
             }
-            const ticket = self.next_claim;
-            self.next_claim += 1;
-            const frame = self.slots[ticket % capacity].?.frame;
+            var job = self.ready.items[self.ready_read];
+            self.ready_read += 1;
+            self.compactReadyLocked();
             self.mutex.unlock(self.io);
+            var owned_job: ?*Job = job;
+            defer if (owned_job) |owned| self.freeJob(owned);
+            while (true) {
+                self.mutex.lockUncancelable(self.io);
+                self.claimed += 1;
+                self.mutex.unlock(self.io);
 
-            // decode + verify outside the lock (this is the parallel part)
-            var arena: ?*std.heap.ArenaAllocator = null;
-            var event: ?zat.FirehoseEvent = null;
-            var verdict: verify.Result = .{};
-            var failure: ?anyerror = null;
-            if (self.allocator.create(std.heap.ArenaAllocator)) |a| {
-                a.* = std.heap.ArenaAllocator.init(self.allocator);
-                if (zat.firehose.decodeFrame(a.allocator(), frame)) |ev| {
-                    arena = a;
-                    event = ev;
-                } else |err| {
-                    log.debug("frame decode error: {s}", .{@errorName(err)});
-                    if (self.stats) |stats| _ = stats.livestream_decode_errors_total.fetchAdd(1, .monotonic);
-                    a.deinit();
-                    self.allocator.destroy(a);
-                }
-            } else |_| {}
-
-            // Every ticket must register, including malformed frames and
-            // allocation failures. Otherwise the first undecodable frame
-            // permanently blocks all later verification at next_register.
-            if (self.verifier) |v| {
-                const turn = self.beginVerification(ticket, if (event) |ev| eventDid(ev) else null);
-                if (turn.proceed) {
-                    defer self.endVerification(ticket, turn.lane);
-                    if (event) |ev| verdict = if (self.repair) |coordinator|
-                        coordinator.process(if (resolver_storage) |*r| r else undefined, frame, ev, ticket) catch |err| blk: {
+                var verdict: verify.Result = .{};
+                var failure: ?anyerror = null;
+                if (self.verifier) |v| {
+                    if (job.event) |ev| verdict = if (self.repair) |coordinator|
+                        coordinator.process(if (resolver_storage) |*r| r else undefined, job.frame, ev, job.ticket) catch |err| blk: {
                             failure = err;
                             break :blk .{};
                         }
@@ -262,56 +320,42 @@ pub const Pipeline = struct {
                         .info => .{},
                     };
                 }
-            }
-
-            self.mutex.lockUncancelable(self.io);
-            self.slots[ticket % capacity].?.arena = arena;
-            self.slots[ticket % capacity].?.event = event;
-            self.slots[ticket % capacity].?.verdict = verdict;
-            self.slots[ticket % capacity].?.failure = failure;
-            self.slots[ticket % capacity].?.done = true;
-            self.done_cond.broadcast(self.io);
-            self.mutex.unlock(self.io);
-        }
-    }
-
-    const VerificationTurn = struct { proceed: bool, lane: ?usize = null };
-
-    fn beginVerification(self: *Pipeline, ticket: u64, did: ?[]const u8) VerificationTurn {
-        self.mutex.lockUncancelable(self.io);
-        defer self.mutex.unlock(self.io);
 
-        // Registration itself is ticket ordered. That makes the predecessor
-        // relation deterministic even when decoding finishes out of order.
-        while (self.alive and ticket != self.next_verify_register) {
-            self.verify_cond.wait(self.io, &self.mutex) catch {};
-        }
-        if (!self.alive) return .{ .proceed = false };
-
-        const lane: ?usize = if (did) |value|
-            @intCast(std.hash.Wyhash.hash(0, value) % verification_lane_count)
-        else
-            null;
-        const predecessor = if (lane) |index| self.verify_lane_tail[index] else null;
-        if (lane) |index| self.verify_lane_tail[index] = ticket;
-        self.next_verify_register += 1;
-        self.verify_cond.broadcast(self.io);
-
-        if (lane) |index| {
-            while (self.alive and predecessor != null and self.verify_lane_completed[index] != predecessor) {
-                self.verify_cond.wait(self.io, &self.mutex) catch {};
+                self.mutex.lockUncancelable(self.io);
+                job.verdict = verdict;
+                job.failure = failure;
+                while (self.alive and self.completed.items.len - self.completed_read >= result_capacity)
+                    self.space_cond.wait(self.io, &self.mutex) catch {};
+                if (!self.alive) {
+                    self.mutex.unlock(self.io);
+                    return;
+                }
+                self.completed.append(self.allocator, job) catch |err| {
+                    self.mutex.unlock(self.io);
+                    owned_job = null;
+                    self.failJob(job, err);
+                    return;
+                };
+                owned_job = null;
+                const chain = self.chains.get(job.key).?;
+                if (chain.pending.items.len > 0) {
+                    job = chain.pending.orderedRemove(0);
+                    owned_job = job;
+                    self.wake.set(self.io);
+                    self.mutex.unlock(self.io);
+                    continue;
+                }
+                _ = self.chains.remove(chain.key);
+                chain.pending.deinit(self.allocator);
+                self.allocator.free(chain.key);
+                self.allocator.destroy(chain);
+                self.active_chains -= 1;
+                self.space_cond.broadcast(self.io);
+                self.wake.set(self.io);
+                self.mutex.unlock(self.io);
+                break;
             }
-            if (!self.alive) return .{ .proceed = false };
         }
-        return .{ .proceed = true, .lane = lane };
-    }
-
-    fn endVerification(self: *Pipeline, ticket: u64, lane: ?usize) void {
-        const index = lane orelse return;
-        self.mutex.lockUncancelable(self.io);
-        self.verify_lane_completed[index] = ticket;
-        self.verify_cond.broadcast(self.io);
-        self.mutex.unlock(self.io);
     }
 
     fn eventDid(event: zat.FirehoseEvent) ?[]const u8 {
@@ -325,76 +369,221 @@ pub const Pipeline = struct {
     }
 
     fn writerLoop(self: *Pipeline) void {
-        while (true) {
-            self.mutex.lockUncancelable(self.io);
-            while (self.alive) {
-                if (self.eligibleRepairIndex() != null) break;
-                const slot = self.slots[self.next_emit % capacity];
-                if (slot != null and slot.?.done and slot.?.ticket == self.next_emit) break;
-                self.done_cond.wait(self.io, &self.mutex) catch {};
-            }
-            if (!self.alive) {
-                self.mutex.unlock(self.io);
-                return;
-            }
-            if (self.eligibleRepairIndex()) |repair_index| {
-                const completion = self.repair_queue.orderedRemove(repair_index);
-                self.repair_inflight = true;
-                self.mutex.unlock(self.io);
-                const callback = self.emit_repair orelse {
-                    completion.deinit();
-                    self.failRepair(error.MissingRepairEmitter);
-                    return;
-                };
-                callback(self.emit_ctx, completion) catch |err| {
+        var batch: std.ArrayList(Delivery) = .empty;
+        var release_from: usize = 0;
+        defer {
+            for (batch.items[release_from..]) |delivery| switch (delivery) {
+                .job => |job| self.freeJob(job),
+                .repair => |completion| {
                     completion.deinit();
-                    self.failRepair(err);
-                    return;
-                };
-                completion.deinit();
+                    self.mutex.lockUncancelable(self.io);
+                    self.repair_buffered -= 1;
+                    self.mutex.unlock(self.io);
+                },
+            };
+            batch.deinit(self.allocator);
+        }
+        while (true) {
+            batch.clearRetainingCapacity();
+            release_from = 0;
+            var timed_out = false;
+            var deadline: ?Io.Timestamp = null;
+            while (batch.items.len < batch_size and !timed_out) {
+                self.wake.reset();
                 self.mutex.lockUncancelable(self.io);
-                self.repair_inflight = false;
-                self.space_cond.broadcast(self.io);
+                if (!self.alive) {
+                    self.mutex.unlock(self.io);
+                    return;
+                }
+                if (self.completed_read < self.completed.items.len) {
+                    const job = self.completed.items[self.completed_read];
+                    self.completed_read += 1;
+                    self.compactCompletedLocked();
+                    self.space_cond.signal(self.io);
+                    if (job.seq) |seq| _ = self.inflight.remove(seq);
+                    self.mutex.unlock(self.io);
+                    if (job.failure != null or isSilent(job)) {
+                        if (job.failure) |err| log.warn("verification worker rejected event: {s}", .{@errorName(err)});
+                        self.finishJob(job);
+                    } else {
+                        batch.append(self.allocator, .{ .job = job }) catch |err| {
+                            self.failJob(job, err);
+                            return;
+                        };
+                        self.mutex.lockUncancelable(self.io);
+                        self.writer_jobs_buffered += 1;
+                        self.mutex.unlock(self.io);
+                    }
+                    if (batch.items.len > 0 and deadline == null)
+                        deadline = Io.Timestamp.now(self.io, .awake).addDuration(batch_timeout.raw);
+                    if (!self.appendEligibleRepairs(&batch, &deadline)) return;
+                    continue;
+                }
                 self.mutex.unlock(self.io);
-                continue;
+                if (!self.appendEligibleRepairs(&batch, &deadline)) return;
+                if (batch.items.len >= batch_size) continue;
+                if (batch.items.len > 0) {
+                    const now = Io.Timestamp.now(self.io, .awake);
+                    if (now.toNanoseconds() >= deadline.?.toNanoseconds()) {
+                        timed_out = true;
+                        continue;
+                    }
+                    self.wake.waitTimeout(self.io, .{ .duration = .{
+                        .raw = now.durationTo(deadline.?),
+                        .clock = .awake,
+                    } }) catch |err| switch (err) {
+                        error.Timeout => timed_out = true,
+                        else => return,
+                    };
+                    continue;
+                }
+                self.wake.wait(self.io) catch return;
             }
-            const slot = self.slots[self.next_emit % capacity].?;
-            self.mutex.unlock(self.io);
 
-            if (slot.failure) |err| {
-                self.failSlot(slot, err);
-                return;
+            var max_seq: i64 = 0;
+            var normal_jobs: usize = 0;
+            for (batch.items, 0..) |delivery, i| switch (delivery) {
+                .job => |job| {
+                    if (job.event) |event| self.emit(self.emit_ctx, event, job.verdict) catch |err| {
+                        release_from = i + 1;
+                        self.failJob(job, err);
+                        return;
+                    };
+                    if (job.seq) |seq| max_seq = @max(max_seq, seq);
+                    self.finishJob(job);
+                    normal_jobs += 1;
+                    release_from = i + 1;
+                },
+                .repair => |completion| {
+                    const callback = self.emit_repair orelse {
+                        release_from = i + 1;
+                        completion.deinit();
+                        self.finishBufferedRepair();
+                        self.failPipeline(error.MissingRepairEmitter);
+                        return;
+                    };
+                    callback(self.emit_ctx, completion) catch |err| {
+                        release_from = i + 1;
+                        completion.deinit();
+                        self.finishBufferedRepair();
+                        self.failPipeline(err);
+                        return;
+                    };
+                    completion.deinit();
+                    self.finishBufferedRepair();
+                    release_from = i + 1;
+                },
+            };
+            if (max_seq > 0) {
+                const watermark = self.cursorAfterBatch(max_seq);
+                if (watermark > self.cursor_watermark) {
+                    if (self.emit_cursor) |callback| callback(self.emit_ctx, watermark) catch |err| {
+                        self.failPipeline(err);
+                        return;
+                    };
+                    self.cursor_watermark = watermark;
+                }
             }
+            self.finishWriterBatch(normal_jobs);
+        }
+    }
 
-            if (slot.event) |ev| self.emit(self.emit_ctx, ev, slot.verdict) catch |err| {
-                self.failSlot(slot, err);
-                return;
+    fn finishWriterBatch(self: *Pipeline, normal_jobs: usize) void {
+        if (normal_jobs == 0) return;
+        self.mutex.lockUncancelable(self.io);
+        self.writer_jobs_buffered -= normal_jobs;
+        self.space_cond.broadcast(self.io);
+        self.mutex.unlock(self.io);
+    }
+
+    fn isSilent(job: *const Job) bool {
+        const event = job.event orelse return true;
+        return switch (event) {
+            .commit, .sync => job.verdict.outcome != .valid,
+            else => false,
+        };
+    }
+
+    fn finishJob(self: *Pipeline, job: *Job) void {
+        self.mutex.lockUncancelable(self.io);
+        self.markProcessedLocked(job.ticket);
+        self.mutex.unlock(self.io);
+        self.freeJob(job);
+    }
+
+    fn markProcessedLocked(self: *Pipeline, ticket: u64) void {
+        self.processed += 1;
+        if (ticket == self.processed_prefix) {
+            self.processed_prefix += 1;
+            while (self.processed_sparse.remove(self.processed_prefix)) self.processed_prefix += 1;
+        } else if (ticket > self.processed_prefix) {
+            self.processed_sparse.put(self.allocator, ticket, {}) catch {
+                if (self.fatal_error == null) self.fatal_error = error.OutOfMemory;
+                self.alive = false;
             };
-            self.freeSlot(slot);
+        }
+        self.space_cond.broadcast(self.io);
+        self.wake.set(self.io);
+    }
+
+    fn cursorAfterBatch(self: *Pipeline, max_seq: i64) i64 {
+        self.mutex.lockUncancelable(self.io);
+        defer self.mutex.unlock(self.io);
+        var minimum: ?i64 = null;
+        var it = self.inflight.iterator();
+        while (it.next()) |entry| {
+            const seq = entry.key_ptr.*;
+            if (minimum == null or seq < minimum.?) minimum = seq;
+        }
+        return if (minimum) |seq| seq - 1 else max_seq;
+    }
 
-            // An item counts as emitted only after its callback succeeds.
-            // In Stream that callback includes durable archive persistence;
-            // advancing this ticket before success would lie about the
-            // pipeline boundary after a disk/database error.
+    fn appendEligibleRepairs(
+        self: *Pipeline,
+        batch: *std.ArrayList(Delivery),
+        deadline: *?Io.Timestamp,
+    ) bool {
+        while (batch.items.len < batch_size) {
             self.mutex.lockUncancelable(self.io);
-            self.slots[self.next_emit % capacity] = null;
-            self.next_emit += 1;
-            self.space_cond.signal(self.io);
+            if (!self.alive) {
+                self.mutex.unlock(self.io);
+                return false;
+            }
+            const repair_index = self.eligibleRepairIndex() orelse {
+                self.mutex.unlock(self.io);
+                return true;
+            };
+            const completion = self.repair_queue.orderedRemove(repair_index);
+            self.repair_buffered += 1;
             self.mutex.unlock(self.io);
+            batch.append(self.allocator, .{ .repair = completion }) catch |err| {
+                completion.deinit();
+                self.finishBufferedRepair();
+                self.failPipeline(err);
+                return false;
+            };
+            if (deadline.* == null)
+                deadline.* = Io.Timestamp.now(self.io, .awake).addDuration(batch_timeout.raw);
         }
+        return true;
+    }
+
+    fn finishBufferedRepair(self: *Pipeline) void {
+        self.mutex.lockUncancelable(self.io);
+        self.repair_buffered -= 1;
+        self.space_cond.broadcast(self.io);
+        self.mutex.unlock(self.io);
     }
 
     /// Caller holds `mutex`. Choose the oldest trigger whose ordinary ticket
     /// has already crossed the writer callback; a faster later repair cannot
     /// jump ahead of earlier relay input.
     fn eligibleRepairIndex(self: *Pipeline) ?usize {
-        var best: ?usize = null;
         for (self.repair_queue.items, 0..) |completion, i| {
-            if (completion.after_ticket >= self.next_emit) continue;
-            if (best == null or completion.after_ticket < self.repair_queue.items[best.?].after_ticket)
-                best = i;
+            if (completion.after_ticket < self.processed_prefix or self.processed_sparse.contains(completion.after_ticket))
+                return i;
         }
-        return best;
+        return null;
     }
 
     /// Repair workers call this while still holding their per-DID lane. The
@@ -406,32 +595,45 @@ pub const Pipeline = struct {
         defer self.mutex.unlock(self.io);
         if (!self.alive) return error.Closed;
         try self.repair_queue.append(self.allocator, completion);
-        self.done_cond.signal(self.io);
+        self.wake.set(self.io);
     }
 
-    fn failRepair(self: *Pipeline, err: anyerror) void {
-        self.mutex.lockUncancelable(self.io);
-        if (self.fatal_error == null) self.fatal_error = err;
-        self.repair_inflight = false;
-        self.alive = false;
-        self.work_cond.broadcast(self.io);
-        self.space_cond.broadcast(self.io);
-        self.done_cond.broadcast(self.io);
-        self.verify_cond.broadcast(self.io);
-        self.mutex.unlock(self.io);
+    fn failJob(self: *Pipeline, job: *Job, err: anyerror) void {
+        self.freeJob(job);
+        self.failPipeline(err);
     }
 
-    fn failSlot(self: *Pipeline, slot: Slot, err: anyerror) void {
+    fn failPipeline(self: *Pipeline, err: anyerror) void {
         self.mutex.lockUncancelable(self.io);
-        self.slots[self.next_emit % capacity] = null;
         if (self.fatal_error == null) self.fatal_error = err;
         self.alive = false;
         self.work_cond.broadcast(self.io);
         self.space_cond.broadcast(self.io);
-        self.done_cond.broadcast(self.io);
-        self.verify_cond.broadcast(self.io);
+        self.wake.set(self.io);
         self.mutex.unlock(self.io);
-        self.freeSlot(slot);
+    }
+
+    fn compactReadyLocked(self: *Pipeline) void {
+        if (self.ready_read == self.ready.items.len) {
+            self.ready.clearRetainingCapacity();
+            self.ready_read = 0;
+        }
+    }
+
+    fn compactCompletedLocked(self: *Pipeline) void {
+        if (self.completed_read == self.completed.items.len) {
+            self.completed.clearRetainingCapacity();
+            self.completed_read = 0;
+        } else if (self.completed_read >= result_capacity) {
+            const remaining = self.completed.items.len - self.completed_read;
+            std.mem.copyForwards(
+                *Job,
+                self.completed.items[0..remaining],
+                self.completed.items[self.completed_read..],
+            );
+            self.completed.items.len = remaining;
+            self.completed_read = 0;
+        }
     }
 };
 
@@ -453,7 +655,80 @@ const Collector = struct {
     }
 };
 
-test "pipeline preserves submission order under parallel workers" {
+fn controlFrame(allocator: Allocator, event: union(enum) {
+    identity: zat.firehose.IdentityEvent,
+    account: zat.firehose.AccountEvent,
+}) ![]u8 {
+    var out: Io.Writer.Allocating = .init(allocator);
+    errdefer out.deinit();
+    const tag: []const u8 = switch (event) {
+        .identity => "#identity",
+        .account => "#account",
+    };
+    try zat.cbor.encode(allocator, &out.writer, .{ .map = &.{
+        .{ .key = "op", .value = .{ .unsigned = 1 } },
+        .{ .key = "t", .value = .{ .text = tag } },
+    } });
+    switch (event) {
+        .identity => |identity| try zat.cbor.encode(allocator, &out.writer, .{ .map = &.{
+            .{ .key = "did", .value = .{ .text = identity.did } },
+            .{ .key = "seq", .value = .{ .unsigned = @intCast(identity.seq) } },
+            .{ .key = "time", .value = .{ .text = identity.time } },
+        } }),
+        .account => |account| try zat.cbor.encode(allocator, &out.writer, .{ .map = &.{
+            .{ .key = "active", .value = .{ .boolean = account.active } },
+            .{ .key = "did", .value = .{ .text = account.did } },
+            .{ .key = "seq", .value = .{ .unsigned = @intCast(account.seq) } },
+            .{ .key = "time", .value = .{ .text = account.time } },
+        } }),
+    }
+    return out.toOwnedSlice();
+}
+
+const SeqCollector = struct {
+    io: Io,
+    mutex: Io.Mutex = .init,
+    seqs: std.ArrayList(i64) = .empty,
+
+    fn emit(ctx: *anyopaque, event: zat.FirehoseEvent, _: verify.Result) !void {
+        const self: *SeqCollector = @ptrCast(@alignCast(ctx));
+        const seq = event.seq() orelse return;
+        self.mutex.lockUncancelable(self.io);
+        defer self.mutex.unlock(self.io);
+        try self.seqs.append(testing.allocator, seq);
+    }
+
+    fn len(self: *SeqCollector) usize {
+        self.mutex.lockUncancelable(self.io);
+        defer self.mutex.unlock(self.io);
+        return self.seqs.items.len;
+    }
+};
+
+const WakeCollector = struct {
+    io: Io,
+    emitted: Io.Event = .unset,
+
+    fn emit(ctx: *anyopaque, _: zat.FirehoseEvent, _: verify.Result) !void {
+        const self: *WakeCollector = @ptrCast(@alignCast(ctx));
+        self.emitted.set(self.io);
+    }
+};
+
+const CursorProbe = struct {
+    calls: usize = 0,
+    last: i64 = 0,
+
+    fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result) !void {}
+
+    fn cursor(ctx: *anyopaque, value: i64) !void {
+        const self: *CursorProbe = @ptrCast(@alignCast(ctx));
+        self.calls += 1;
+        self.last = value;
+    }
+};
+
+test "pipeline preserves same-key submission order under parallel workers" {
     var threaded: Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
 
@@ -496,6 +771,325 @@ test "pipeline preserves submission order under parallel workers" {
     }
 }
 
+test "offline batch timeout is measured from the first completed event" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var collector: WakeCollector = .{ .io = io };
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = io,
+        .verifier = null,
+        .emit_ctx = &collector,
+        .emit = WakeCollector.emit,
+    };
+    try pipeline.start();
+    defer pipeline.stop();
+    const first = &.{
+        0xa2, 0x61, 't', 0x65, '#', 'i', 'n',  'f', 'o', 0x62, 'o', 'p', 0x01,
+        0xa1, 0x64, 'n', 'a',  'm', 'e', 0x61, 'a',
+    };
+    const second = &.{
+        0xa2, 0x61, 't', 0x65, '#', 'i', 'n',  'f', 'o', 0x62, 'o', 'p', 0x01,
+        0xa1, 0x64, 'n', 'a',  'm', 'e', 0x61, 'b',
+    };
+    try pipeline.submit(first);
+    try io.sleep(Io.Duration.fromMilliseconds(350), .awake);
+    try pipeline.submit(second);
+    // A sliding timeout would fire 500 ms after the second completion. The
+    // upstream timer is armed by the first completion, so 300 ms is ample.
+    try collector.emitted.waitTimeout(io, .{
+        .duration = .{ .raw = Io.Duration.fromMilliseconds(300), .clock = .awake },
+    });
+    try pipeline.drainAndStop();
+}
+
+test "offline scheduler completes other DIDs while one verifier lane is blocked" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
+    const path_len = try tmp.dir.realPath(io, &path_buf);
+    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
+    defer meta.deinit();
+    var verifier = verify.Verifier.init(testing.allocator, io, "http://offline.invalid", &meta);
+    defer verifier.deinit();
+    var collector: SeqCollector = .{ .io = io };
+    defer collector.seqs.deinit(testing.allocator);
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = io,
+        .verifier = &verifier,
+        .emit_ctx = &collector,
+        .emit = SeqCollector.emit,
+    };
+    try pipeline.start();
+    defer pipeline.stop();
+
+    // Hold the real hosting-state lane so the first account event blocks in
+    // Verifier.onAccount. A different DID must still verify and reach the
+    // writer on another worker; no network or fake verifier is involved.
+    verifier.chain_mu.lockUncancelable(io);
+    var lane_held = true;
+    defer if (lane_held) verifier.chain_mu.unlock(io);
+    const slow = try controlFrame(testing.allocator, .{ .account = .{
+        .seq = 1,
+        .did = "did:plc:slow-lane",
+        .time = "2026-07-16T00:00:00Z",
+    } });
+    defer testing.allocator.free(slow);
+    const fast = try controlFrame(testing.allocator, .{ .identity = .{
+        .seq = 2,
+        .did = "did:plc:fast-lane",
+        .time = "2026-07-16T00:00:01Z",
+    } });
+    defer testing.allocator.free(fast);
+    try pipeline.submit(slow);
+    while (pipeline.gauges().claimed < 1) std.Thread.yield() catch {};
+    try pipeline.submit(fast);
+    const deadline = Io.Timestamp.now(io, .awake).addDuration(.fromSeconds(2));
+    while (collector.len() == 0 and Io.Timestamp.now(io, .awake).toNanoseconds() < deadline.toNanoseconds())
+        std.Thread.yield() catch {};
+    try testing.expectEqual(@as(usize, 1), collector.len());
+    try testing.expectEqual(@as(i64, 2), collector.seqs.items[0]);
+    verifier.chain_mu.unlock(io);
+    lane_held = false;
+    try pipeline.drainAndStop();
+    try testing.expectEqualSlices(i64, &.{ 2, 1 }, collector.seqs.items);
+}
+
+test "offline scheduler keeps one DID FIFO while its verifier lane is blocked" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
+    const path_len = try tmp.dir.realPath(io, &path_buf);
+    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
+    defer meta.deinit();
+    var verifier = verify.Verifier.init(testing.allocator, io, "http://offline.invalid", &meta);
+    defer verifier.deinit();
+    var collector: SeqCollector = .{ .io = io };
+    defer collector.seqs.deinit(testing.allocator);
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = io,
+        .verifier = &verifier,
+        .emit_ctx = &collector,
+        .emit = SeqCollector.emit,
+    };
+    try pipeline.start();
+    defer pipeline.stop();
+
+    const did = "did:plc:one-fifo";
+    verifier.chain_mu.lockUncancelable(io);
+    var lane_held = true;
+    defer if (lane_held) verifier.chain_mu.unlock(io);
+    const first = try controlFrame(testing.allocator, .{ .account = .{
+        .seq = 1,
+        .did = did,
+        .time = "2026-07-16T00:00:00Z",
+    } });
+    defer testing.allocator.free(first);
+    const second = try controlFrame(testing.allocator, .{ .identity = .{
+        .seq = 2,
+        .did = did,
+        .time = "2026-07-16T00:00:01Z",
+    } });
+    defer testing.allocator.free(second);
+    try pipeline.submit(first);
+    while (pipeline.gauges().claimed < 1) std.Thread.yield() catch {};
+    try pipeline.submit(second);
+    try io.sleep(Io.Duration.fromMilliseconds(600), .awake);
+    try testing.expectEqual(@as(usize, 0), collector.len());
+    verifier.chain_mu.unlock(io);
+    lane_held = false;
+    try pipeline.drainAndStop();
+    try testing.expectEqualSlices(i64, &.{ 1, 2 }, collector.seqs.items);
+}
+
+test "offline scheduler backpressures the thirty-third active DID" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
+    const path_len = try tmp.dir.realPath(io, &path_buf);
+    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
+    defer meta.deinit();
+    var verifier = verify.Verifier.init(testing.allocator, io, "http://offline.invalid", &meta);
+    defer verifier.deinit();
+    var collector: SeqCollector = .{ .io = io };
+    defer collector.seqs.deinit(testing.allocator);
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = io,
+        .verifier = &verifier,
+        .emit_ctx = &collector,
+        .emit = SeqCollector.emit,
+    };
+    try pipeline.start();
+    defer pipeline.stop();
+
+    verifier.chain_mu.lockUncancelable(io);
+    var lane_held = true;
+    defer if (lane_held) verifier.chain_mu.unlock(io);
+    for (0..worker_count) |i| {
+        var did_buf: [64]u8 = undefined;
+        const did = try std.fmt.bufPrint(&did_buf, "did:plc:active-{d}", .{i});
+        const frame = try controlFrame(testing.allocator, .{ .account = .{
+            .seq = @intCast(i + 1),
+            .did = did,
+            .time = "2026-07-16T00:00:00Z",
+        } });
+        defer testing.allocator.free(frame);
+        try pipeline.submit(frame);
+    }
+    while (pipeline.gauges().claimed < worker_count) std.Thread.yield() catch {};
+
+    const extra = try controlFrame(testing.allocator, .{ .identity = .{
+        .seq = worker_count + 1,
+        .did = "did:plc:thirty-third",
+        .time = "2026-07-16T00:00:01Z",
+    } });
+    defer testing.allocator.free(extra);
+    var started: Io.Event = .unset;
+    var finished: Io.Event = .unset;
+    var submit_error: ?anyerror = null;
+    const Submit = struct {
+        fn run(p: *Pipeline, frame: []const u8, began: *Io.Event, done: *Io.Event, result: *?anyerror) void {
+            began.set(p.io);
+            p.submit(frame) catch |err| {
+                result.* = err;
+            };
+            done.set(p.io);
+        }
+    };
+    var submit_future = try io.concurrent(Submit.run, .{ &pipeline, extra, &started, &finished, &submit_error });
+    defer _ = submit_future.cancel(io);
+    try started.wait(io);
+    try testing.expectError(error.Timeout, finished.waitTimeout(io, .{
+        .duration = .{ .raw = Io.Duration.fromMilliseconds(100), .clock = .awake },
+    }));
+    verifier.chain_mu.unlock(io);
+    lane_held = false;
+    try finished.wait(io);
+    _ = submit_future.await(io);
+    try testing.expect(submit_error == null);
+    try pipeline.drainAndStop();
+    try testing.expectEqual(@as(usize, worker_count + 1), collector.len());
+}
+
+test "offline per-DID overflow drops the oldest of 64 pending events" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var stats: metrics.Stats = .{};
+    var collector: SeqCollector = .{ .io = io };
+    defer collector.seqs.deinit(testing.allocator);
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = io,
+        .verifier = null,
+        .stats = &stats,
+        .emit_ctx = &collector,
+        .emit = SeqCollector.emit,
+    };
+    defer pipeline.stop();
+    const did = "did:plc:overflow";
+    for (1..67) |raw_seq| {
+        const frame = try controlFrame(testing.allocator, .{ .identity = .{
+            .seq = @intCast(raw_seq),
+            .did = did,
+            .time = "2026-07-16T00:00:00Z",
+        } });
+        defer testing.allocator.free(frame);
+        try pipeline.submit(frame);
+    }
+    const chain = pipeline.chains.get(did).?;
+    try testing.expectEqual(@as(usize, key_queue_capacity), chain.pending.items.len);
+    try testing.expectEqual(@as(?i64, 3), chain.pending.items[0].seq);
+    try testing.expectEqual(@as(?i64, 66), chain.pending.items[key_queue_capacity - 1].seq);
+    try testing.expect(!pipeline.inflight.contains(2));
+    try testing.expectEqual(@as(u64, 1), stats.verify_queue_dropped_total.load(.monotonic));
+}
+
+test "cursor watermark is smallest inflight minus one or batch maximum" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    var collector: SeqCollector = .{ .io = threaded.io() };
+    defer collector.seqs.deinit(testing.allocator);
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = threaded.io(),
+        .verifier = null,
+        .emit_ctx = &collector,
+        .emit = SeqCollector.emit,
+    };
+    defer pipeline.stop();
+    try pipeline.inflight.put(testing.allocator, 40, {});
+    try pipeline.inflight.put(testing.allocator, 7, {});
+    try testing.expectEqual(@as(i64, 6), pipeline.cursorAfterBatch(91));
+    pipeline.inflight.clearRetainingCapacity();
+    try testing.expectEqual(@as(i64, 91), pipeline.cursorAfterBatch(91));
+}
+
+test "loaded cursor baseline cannot regress on a replayed batch" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    var probe: CursorProbe = .{};
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = threaded.io(),
+        .verifier = null,
+        .emit_ctx = &probe,
+        .emit = CursorProbe.emit,
+        .emit_cursor = CursorProbe.cursor,
+        .cursor_watermark = 100,
+    };
+    try pipeline.start();
+    defer pipeline.stop();
+    const replay = try controlFrame(testing.allocator, .{ .identity = .{
+        .seq = 50,
+        .did = "did:plc:cursor-replay",
+        .time = "2026-07-16T00:00:00Z",
+    } });
+    defer testing.allocator.free(replay);
+    try pipeline.submit(replay);
+    try pipeline.drainAndStop();
+    try testing.expectEqual(@as(usize, 0), probe.calls);
+}
+
+test "graceful drain waits for the final cursor callback" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    var probe: CursorProbe = .{};
+    var pipeline: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = threaded.io(),
+        .verifier = null,
+        .emit_ctx = &probe,
+        .emit = CursorProbe.emit,
+        .emit_cursor = CursorProbe.cursor,
+    };
+    try pipeline.start();
+    defer pipeline.stop();
+    const event = try controlFrame(testing.allocator, .{ .identity = .{
+        .seq = 50,
+        .did = "did:plc:cursor-drain",
+        .time = "2026-07-16T00:00:00Z",
+    } });
+    defer testing.allocator.free(event);
+    try pipeline.submit(event);
+    try pipeline.drainAndStop();
+    try testing.expectEqual(@as(usize, 1), probe.calls);
+    try testing.expectEqual(@as(i64, 50), probe.last);
+}
+
 test "repair outbox waits for trigger ticket and selects relay order" {
     var threaded: Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
@@ -517,12 +1111,12 @@ test "repair outbox waits for trigger ticket and selects relay order" {
     try pipeline.repair_queue.append(testing.allocator, &earlier);
     defer pipeline.repair_queue.deinit(testing.allocator);
 
-    pipeline.next_emit = 4;
+    pipeline.processed_prefix = 4;
     try testing.expect(pipeline.eligibleRepairIndex() == null);
-    pipeline.next_emit = 5;
-    try testing.expectEqual(@as(?usize, 1), pipeline.eligibleRepairIndex());
-    pipeline.next_emit = 10;
+    pipeline.processed_prefix = 5;
     try testing.expectEqual(@as(?usize, 1), pipeline.eligibleRepairIndex());
+    pipeline.processed_prefix = 10;
+    try testing.expectEqual(@as(?usize, 0), pipeline.eligibleRepairIndex());
 }
 
 test "pipeline exposes emit failure without advancing emitted ticket" {
diff --git a/src/main.zig b/src/main.zig
index f3704d5..51e27cc 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -253,6 +253,8 @@ pub fn main(init: std.process.Init.Minimal) !void {
         .emit_ctx = &consumer,
         .emit = ingest.Consumer.emitVerified,
         .emit_repair = ingest.Consumer.emitRepair,
+        .emit_cursor = ingest.Consumer.emitCursor,
+        .cursor_watermark = cursor orelse 0,
     };
     var repair_coordinator: ?repair_mod.Coordinator = if (verifier) |*v| .{
         .allocator = allocator,