diff --git a/build.zig b/build.zig
index e0d60f2..2ba3e3f 100644
--- a/build.zig
+++ b/build.zig
@@ -1,40 +1,5 @@
 const std = @import("std");
 
-// vendored libzstd, decode-only subset, pinned by url+hash in build.zig.zon
-// (same v1.5.7 pin as stream). no system library, cross-compiles with
-// -Dtarget. zig std's zstd decompressor measured ~20x slower than libzstd
-// on jss archive blocks and cannot verify frame content checksums.
-fn linkVendoredZstd(mod: *std.Build.Module, b: *std.Build, target: std.Build.ResolvedTarget, optimize: std.builtin.OptimizeMode) void {
-    const zstd_dep = b.dependency("zstd", .{});
-    const clib_mod = b.createModule(.{
-        .target = target,
-        .optimize = optimize,
-        .link_libc = true,
-    });
-    clib_mod.addCSourceFiles(.{
-        .root = zstd_dep.path("lib"),
-        .files = &.{
-            "common/debug.c",
-            "common/entropy_common.c",
-            "common/error_private.c",
-            "common/fse_decompress.c",
-            "common/xxhash.c",
-            "common/zstd_common.c",
-            "decompress/huf_decompress.c",
-            "decompress/zstd_ddict.c",
-            "decompress/zstd_decompress_block.c",
-            "decompress/zstd_decompress.c",
-        },
-        .flags = &.{
-            "-DZSTD_DISABLE_ASM",
-            "-DZSTD_MULTITHREAD=0",
-            "-DXXH_NAMESPACE=ZSTD_",
-        },
-    });
-    const clib = b.addLibrary(.{ .name = "zat-vendored-zstd", .root_module = clib_mod });
-    mod.linkLibrary(clib);
-}
-
 pub fn build(b: *std.Build) void {
     const target = b.standardTargetOptions(.{});
     const optimize = b.standardOptimizeOption(.{});
@@ -53,7 +18,6 @@ pub fn build(b: *std.Build) void {
             .{ .name = "websocket", .module = websocket.module("websocket") },
         },
     });
-    linkVendoredZstd(mod, b, target, optimize);
 
     const tests = b.addTest(.{ .root_module = mod });
 
@@ -144,23 +108,6 @@ pub fn build(b: *std.Build) void {
     const firehose_smoke_step = b.step("firehose-smoke", "run firehose smoke test (CBOR/CAR/CID on live data)");
     firehose_smoke_step.dependOn(&run_firehose_smoke.step);
 
-    // archive backfill smoke test (plan/fetch/decode on live archive data)
-    const archive_smoke = b.addExecutable(.{
-        .name = "archive-backfill-smoke",
-        .root_module = b.createModule(.{
-            .root_source_file = b.path("scripts/archive_backfill_smoke.zig"),
-            .target = target,
-            .optimize = optimize,
-            .link_libc = true,
-            .imports = &.{.{ .name = "zat", .module = mod }},
-        }),
-    });
-    b.installArtifact(archive_smoke);
-
-    const run_archive_smoke = b.addRunArtifact(archive_smoke);
-    const archive_smoke_step = b.step("archive-smoke", "run archive backfill smoke test (live archive data)");
-    archive_smoke_step.dependOn(&run_archive_smoke.step);
-
     // runnable examples (compile-checked usage recipes)
     const example_parse_identifiers = b.addExecutable(.{
         .name = "example-parse-identifiers",
@@ -242,22 +189,6 @@ pub fn build(b: *std.Build) void {
     const example_search_bluesky_step = b.step("example-search-bluesky", "search Bluesky posts via searchPostsV2");
     example_search_bluesky_step.dependOn(&run_example_search_bluesky.step);
 
-    const example_streamplace_chat = b.addExecutable(.{
-        .name = "example-streamplace-chat",
-        .root_module = b.createModule(.{
-            .root_source_file = b.path("examples/07_streamplace_chat.zig"),
-            .target = target,
-            .optimize = optimize,
-            .link_libc = true,
-            .imports = &.{.{ .name = "zat", .module = mod }},
-        }),
-    });
-    b.installArtifact(example_streamplace_chat);
-
-    const run_example_streamplace_chat = b.addRunArtifact(example_streamplace_chat);
-    if (b.args) |args| run_example_streamplace_chat.addArgs(args);
-    const example_streamplace_chat_step = b.step("example-streamplace-chat", "replay and follow a Streamplace chat");
-    example_streamplace_chat_step.dependOn(&run_example_streamplace_chat.step);
 
     const example_resolve_identity = b.addExecutable(.{
         .name = "example-resolve-identity",
diff --git a/build.zig.zon b/build.zig.zon
index 505b818..75691bc 100644
--- a/build.zig.zon
+++ b/build.zig.zon
@@ -8,10 +8,6 @@
             .url = "https://tangled.org/zzstoatzz.io/websocket.zig/archive/v0.1.11.tar.gz",
             .hash = "websocket-0.1.10-ZPISdUowBQAXIl_X9VpZozIxBLVf_o5wYjOMMdobrXF5",
         },
-        .zstd = .{
-            .url = "https://github.com/facebook/zstd/releases/download/v1.5.7/zstd-1.5.7.tar.gz",
-            .hash = "N-V-__8AAPZ7fwBg4JoCzM_0o2A8wxH2hsUUeiU1iuZv53L5",
-        },
         .@"atproto-interop-tests" = .{
             .url = "https://github.com/bluesky-social/atproto-interop-tests/archive/35bb5638ab1e5ce71fb88a0c95953fc557ef1925.tar.gz",
             .hash = "N-V-__8AAIp5AQCe4JjGmPl8CplTkCis8PF1qvn7QX6GwDfu",
diff --git a/examples/07_streamplace_chat.md b/examples/07_streamplace_chat.md
deleted file mode 100644
index 184120c..0000000
--- a/examples/07_streamplace_chat.md
+++ /dev/null
@@ -1,94 +0,0 @@
-# replay a Streamplace chat
-
-Streamplace chat messages are ordinary ATProto records in each chatter's repo.
-Every `place.stream.chat.message` record names the streamer's DID in its
-`streamer` field, so a Jetstream consumer can reconstruct a broadcast's chat.
-
-## choose a broadcast
-
-Pass a handle to replay the account's latest broadcast and then follow it live:
-
-```bash
-zig build example-streamplace-chat -- iame.li
-```
-
-Pass the exact livestream AT-URI to select a historical broadcast:
-
-```bash
-zig build example-streamplace-chat -- \
-  'at://did:plc:2zmxikig2sj7gqaezl5gntae/place.stream.livestream/3mst2znymsb2e'
-```
-
-The second form starts at that exact `place.stream.livestream` record on the
-streamer's PDS. Streamplace can roll a single viewing session into another
-livestream record, so the example also walks newer records and joins an
-immediately contiguous segment for the same media URL (a rolled-over session
-prints each joined record). The first record's TID supplies the lower replay
-boundary and the final segment's `endedAt` supplies the upper chat boundary,
-with five minutes of room for delayed ingestion.
-
-```text
-logical Streamplace session:
-- Loop engineering with @brittanyellich.com!
-  2026-08-11T16:48:57Z -> 2026-08-11T16:58:53.645Z
-  at://did:plc:.../place.stream.livestream/3mst2znymsb2e
-
-replaying from the archive (https://stream.waow.tech)...
-
-[2026-08-11T16:49:21.783Z] did:plc:...: ooo yeha i setup ssh between macs ...
-[2026-08-11T16:49:39.355Z] did:plc:...: i know how to setup some secure ...
-...
-archive replay: 17 messages from 633 blocks
-finishing the session tail over the live socket...
-
-replay complete: 17 observed messages across 1 segment
-```
-
-The output uses author DIDs because Jetstream carries repo writes, not hydrated
-profiles. The implementation is in `examples/07_streamplace_chat.zig` and is
-compile-checked by `zig build`. Build it with `-Doptimize=ReleaseSafe` — the
-archive pass decompresses hundreds of zstd blocks, which is slow in Debug.
-
-Historical chat comes from `zat.ArchiveBackfill` against
-[stream.waow.tech](https://stream.waow.tech)'s full-network archive, so replay
-does not depend on Jetstream cursor retention. The session's time window maps
-to plan seq bounds (`fetchSeqBounds`), so a recent broadcast plans only the
-handful of blocks it touches. Sessions inside the archive's merged bootstrap
-region (before 2026-08-04) still replay, but their rows are scattered per-DID,
-so the plan spans more blocks — expect a bulk download, and watch the
-`archive backfill: N blocks decoded` progress lines. The newest events sit in
-the archive's unsealed tail, so the example always finishes over the live
-socket from the covered boundary; the two passes overlap at the seam and the
-printer dedups by rkey.
-
-Replay reflects the archive's contents: records whose authors later deleted
-them are folded out by the archive's delete-compaction (the `.delete` events
-remain), and an instance's own documented coverage bounds apply.
-
-For current-state discovery without an archive, use the same shape as the
-[`collection_creators` flow](https://github.com/zzstoatzz/prefect-pack/blob/main/flows/collection_creators/main.py): enumerate every repo containing
-`place.stream.chat.message`, resolve each repo's PDS, fetch its current
-records, and filter them by `streamer` and time. That cannot recover deleted
-records. A durable chat archive should still consume continuously and persist
-both matching records and its cursor.
-
-Streamplace's `/api/websocket/<streamer-did>` endpoint is useful when hydrated
-author profiles, moderation events, and viewer counts matter more than complete
-history. Its initial chat burst contains only the most recent 100 messages.
-
-The record shape is documented in Streamplace's
-[`place.stream.chat.message` lexicon](https://stream.place/docs/lex-reference/chat/place-stream-chat-message/).
-The [`com.atproto.repo.getRecord`](https://atproto.com/lexicons/com-atproto-repo#getRecord)
-query retrieves the selected livestream from its PDS, and the replay cursor is
-part of the [Jetstream subscription API](https://github.com/bluesky-social/jetstream).
-
-<details>
-<summary>wiring zat into your own <code>build.zig</code></summary>
-
-```zig
-const zat = b.dependency("zat", .{}).module("zat");
-exe.root_module.addImport("zat", zat);
-```
-
-after `zig fetch --save https://tangled.org/zat.dev/zat/archive/main`.
-</details>
diff --git a/examples/07_streamplace_chat.zig b/examples/07_streamplace_chat.zig
deleted file mode 100644
index 80714dc..0000000
--- a/examples/07_streamplace_chat.zig
+++ /dev/null
@@ -1,421 +0,0 @@
-//! replay a Streamplace broadcast's chat, including contiguous rollover records.
-//!
-//! run: zig build example-streamplace-chat -- iame.li
-//!      zig build example-streamplace-chat -- at://did:plc:.../place.stream.livestream/3...
-
-const std = @import("std");
-const zat = @import("zat");
-
-const rollover_tolerance_us: i64 = 2 * std.time.us_per_s;
-const ingestion_grace_us: i64 = 5 * std.time.us_per_min;
-const archive_host = "https://stream.waow.tech";
-
-const Segment = struct {
-    uri: []u8,
-    rkey: []u8,
-    title: []u8,
-    created_at: []u8,
-    ended_at: ?[]u8,
-    url: []u8,
-
-    fn deinit(self: *Segment, allocator: std.mem.Allocator) void {
-        allocator.free(self.uri);
-        allocator.free(self.rkey);
-        allocator.free(self.title);
-        allocator.free(self.created_at);
-        if (self.ended_at) |value| allocator.free(value);
-        allocator.free(self.url);
-    }
-};
-
-const ChatPrinter = struct {
-    allocator: std.mem.Allocator,
-    streamer_did: []const u8,
-    first_stream_rkey: []const u8,
-    ended_at_us: ?i64,
-    count: usize = 0,
-    /// the archive pass and the live replay overlap at the seam by design
-    /// (delivery is at-least-once); "did/rkey" keys dedup the seam
-    seen: std.StringHashMapUnmanaged(void) = .empty,
-
-    fn deinit(self: *ChatPrinter) void {
-        var keys = self.seen.keyIterator();
-        while (keys.next()) |key| self.allocator.free(key.*);
-        self.seen.deinit(self.allocator);
-    }
-
-    pub fn onEvent(self: *ChatPrinter, event: zat.JetstreamEvent) void {
-        const commit = switch (event) {
-            .commit => |c| c,
-            else => return,
-        };
-        if (commit.operation != .create) return;
-        const record = commit.record orelse return;
-
-        const streamer = zat.json.getString(record, "streamer") orelse return;
-        if (!std.mem.eql(u8, streamer, self.streamer_did)) return;
-
-        // Chat records use TID keys. This excludes messages from broadcasts
-        // before the selected logical session, even if they arrive late.
-        if (std.mem.order(u8, commit.rkey, self.first_stream_rkey) == .lt) return;
-
-        const created_at = zat.json.getString(record, "createdAt") orelse return;
-        if (self.ended_at_us) |end| {
-            const created_at_us = parseUtcMicros(created_at) catch return;
-            if (created_at_us > end) return;
-        }
-
-        var key_buf: [512]u8 = undefined;
-        const key = std.fmt.bufPrint(&key_buf, "{s}/{s}", .{ commit.did, commit.rkey }) catch return;
-        const entry = self.seen.getOrPut(self.allocator, key) catch return;
-        if (entry.found_existing) return;
-        entry.key_ptr.* = self.allocator.dupe(u8, key) catch {
-            _ = self.seen.remove(key);
-            return;
-        };
-
-        const text = zat.json.getString(record, "text") orelse "";
-        self.count += 1;
-        std.debug.print("[{s}] {s}: {s}\n", .{ created_at, commit.did, text });
-    }
-
-    pub fn onConnect(self: *ChatPrinter, host: []const u8) void {
-        std.debug.print("connected to {s} ({d} observed messages so far)\n", .{ host, self.count });
-    }
-
-    pub fn onError(_: *ChatPrinter, err: anyerror) void {
-        std.debug.print("stream error: {s}\n", .{@errorName(err)});
-    }
-};
-
-pub fn main(init: std.process.Init) !void {
-    const allocator = init.gpa;
-    const args = try init.minimal.args.toSlice(init.arena.allocator());
-    if (args.len != 2) {
-        std.debug.print("usage: zig build example-streamplace-chat -- <handle|livestream-at-uri>\n", .{});
-        return error.MissingStream;
-    }
-
-    var handle_resolver = zat.HandleResolver.init(init.io, allocator);
-    defer handle_resolver.deinit();
-
-    var streamer_did: []const u8 = undefined;
-    var livestream_uri: []const u8 = undefined;
-    if (zat.AtUri.parse(args[1])) |uri| {
-        const collection = uri.collection() orelse return error.MissingCollection;
-        if (!std.mem.eql(u8, collection, "place.stream.livestream"))
-            return error.NotLivestreamUri;
-        _ = zat.Tid.parse(uri.rkey() orelse return error.MissingLivestreamRkey) orelse
-            return error.InvalidLivestreamRkey;
-
-        const authority = uri.authority();
-        streamer_did = if (zat.Did.parse(authority) != null)
-            try allocator.dupe(u8, authority)
-        else
-            try handle_resolver.resolve(zat.Handle.parse(authority) orelse
-                return error.InvalidAuthority);
-        livestream_uri = try allocator.dupe(u8, args[1]);
-    } else {
-        const handle = zat.Handle.parse(args[1]) orelse {
-            std.debug.print("expected an ATProto handle or livestream AT-URI: {s}\n", .{args[1]});
-            return error.InvalidStream;
-        };
-        streamer_did = try handle_resolver.resolve(handle);
-        livestream_uri = try latestLivestreamUri(init.io, allocator, streamer_did);
-    }
-    defer allocator.free(streamer_did);
-    defer allocator.free(livestream_uri);
-
-    const at_uri = zat.AtUri.parse(livestream_uri) orelse return error.InvalidLivestreamUri;
-    const first_rkey = at_uri.rkey() orelse return error.MissingLivestreamRkey;
-    const first_tid = zat.Tid.parse(first_rkey) orelse return error.InvalidLivestreamRkey;
-
-    var did_resolver = zat.DidResolver.init(init.io, allocator);
-    defer did_resolver.deinit();
-    var doc = try did_resolver.resolve(zat.Did.parse(streamer_did).?);
-    defer doc.deinit();
-    const pds_endpoint = doc.pdsEndpoint() orelse return error.NoPdsEndpoint;
-
-    var pds = zat.XrpcClient.initWithUserAgent(
-        init.io,
-        allocator,
-        pds_endpoint,
-        "zat-streamplace-chat-example/1.1",
-    );
-    defer pds.deinit();
-
-    var selected = try getLivestream(&pds, allocator, streamer_did, first_rkey);
-    defer selected.deinit(allocator);
-
-    var newer: std.ArrayList(Segment) = .empty;
-    defer {
-        for (newer.items) |*segment| segment.deinit(allocator);
-        newer.deinit(allocator);
-    }
-    if (selected.ended_at != null) {
-        try collectNewerLivestreams(&pds, allocator, streamer_did, first_rkey, &newer);
-    }
-
-    var replay_end = selected.ended_at;
-    var previous_end_us = if (selected.ended_at) |end| try parseUtcMicros(end) else null;
-    var previous_url: []const u8 = selected.url;
-    var segment_count: usize = 1;
-
-    std.debug.print("logical Streamplace session:\n", .{});
-    printSegment(&selected);
-
-    // listRecords is newest-first. Walking the collected prefix backwards
-    // visits the record immediately after the selected one first.
-    var i = newer.items.len;
-    while (i > 0 and previous_end_us != null) {
-        i -= 1;
-        const candidate = &newer.items[i];
-        const candidate_start_us = parseUtcMicros(candidate.created_at) catch break;
-        const gap_us = candidate_start_us - previous_end_us.?;
-        const same_target = previous_url.len == 0 or candidate.url.len == 0 or
-            std.mem.eql(u8, previous_url, candidate.url);
-        if (!same_target or gap_us < -rollover_tolerance_us or gap_us > rollover_tolerance_us) break;
-
-        segment_count += 1;
-        printSegment(candidate);
-        replay_end = candidate.ended_at;
-        previous_end_us = if (candidate.ended_at) |end| try parseUtcMicros(end) else null;
-        previous_url = candidate.url;
-    }
-
-    std.debug.print("\n", .{});
-    if (segment_count > 1) {
-        std.debug.print("joined {d} contiguous livestream records; Streamplace rolled over the AT-URI mid-session.\n", .{segment_count});
-    }
-
-    var printer = ChatPrinter{
-        .allocator = allocator,
-        .streamer_did = streamer_did,
-        .first_stream_rkey = first_rkey,
-        .ended_at_us = if (replay_end) |end| try parseUtcMicros(end) else null,
-    };
-    defer printer.deinit();
-
-    // The livestream record's TID is its creation time in microseconds.
-    // Rewind one second so events at the boundary cannot be skipped.
-    const start_us = @as(i64, @intCast(first_tid.timestamp())) - std.time.us_per_s;
-    const end_cursor = if (replay_end) |end|
-        (try parseUtcMicros(end)) + ingestion_grace_us
-    else
-        null;
-
-    // Historical chat comes from the archive, not from live-replay retention:
-    // stream.waow.tech archives every event, so any session since its live
-    // capture began replays completely — including messages later deleted.
-    // The witnessed-time window maps to plan seq bounds, which keeps the
-    // fetch to the handful of blocks the session actually touches.
-    const bounds = try zat.ArchiveBackfill.fetchSeqBounds(init.io, allocator, archive_host, start_us, end_cursor, null);
-    var archive_boundary_us: ?i64 = null;
-    if (bounds.covered) {
-        std.debug.print("replaying from the archive ({s})...\n\n", .{archive_host});
-        const result = try zat.ArchiveBackfill.run(init.io, allocator, .{
-            .host = archive_host,
-            .collections = &.{"place.stream.chat.message"},
-            .after_seq = bounds.after_seq,
-            .before_seq = bounds.before_seq,
-            .concurrency = 6,
-        }, &printer);
-        archive_boundary_us = result.last_time_us;
-        std.debug.print("\narchive replay: {d} messages from {d} blocks\n", .{ printer.count, result.blocks_decoded });
-    } else {
-        std.debug.print("session predates the archive's live capture; only live replay is available.\n", .{});
-    }
-
-    // The newest events live in the archive's unsealed tail, so a complete
-    // replay always finishes over the live socket from the covered boundary
-    // (the seam overlaps; the printer dedups). For a live session this is
-    // also the follow.
-    std.debug.print("{s}\n\n", .{
-        if (replay_end == null)
-            "catching up over the live socket, then following live..."
-        else
-            "finishing the session tail over the live socket...",
-    });
-    var client = zat.JetstreamClient.init(init.io, allocator, .{
-        .wanted_collections = &.{"place.stream.chat.message"},
-        .cursor = if (archive_boundary_us) |t| t - std.time.us_per_s else start_us,
-        // Stop close to the broadcast end, with room for delayed ingestion.
-        // The idle timeout handles the case where no matching event lands
-        // exactly beyond that boundary.
-        .end_cursor = end_cursor,
-        .replay_idle_timeout_ms = if (end_cursor != null) 10_000 else null,
-    });
-    defer client.deinit();
-
-    try client.subscribe(&printer);
-    std.debug.print("\nreplay complete: {d} observed messages across {d} segment{s}\n", .{
-        printer.count,
-        segment_count,
-        if (segment_count == 1) "" else "s",
-    });
-}
-
-fn printSegment(segment: *const Segment) void {
-    std.debug.print("- {s}\n  {s} -> {s}\n  {s}\n", .{
-        segment.title,
-        segment.created_at,
-        segment.ended_at orelse "live",
-        segment.uri,
-    });
-}
-
-fn getLivestream(
-    client: *zat.XrpcClient,
-    allocator: std.mem.Allocator,
-    repo: []const u8,
-    rkey: []const u8,
-) !Segment {
-    const params = [_]zat.XrpcClient.QueryParam{
-        .{ .name = "repo", .value = repo },
-        .{ .name = "collection", .value = "place.stream.livestream" },
-        .{ .name = "rkey", .value = rkey },
-    };
-    const method = zat.Nsid.parse("com.atproto.repo.getRecord").?;
-    var response = try client.queryParams(method, &params);
-    defer response.deinit();
-    if (!response.ok()) {
-        std.debug.print("getRecord failed ({d}): {s}\n", .{ @intFromEnum(response.status), response.body });
-        return error.LivestreamLookupFailed;
-    }
-
-    var parsed = try response.json();
-    defer parsed.deinit();
-    const uri = zat.json.getString(parsed.value, "uri") orelse return error.MissingLivestreamUri;
-    const value = zat.json.getPath(parsed.value, "value") orelse return error.MissingLivestreamRecord;
-    return copySegment(allocator, uri, value);
-}
-
-fn collectNewerLivestreams(
-    client: *zat.XrpcClient,
-    allocator: std.mem.Allocator,
-    repo: []const u8,
-    selected_rkey: []const u8,
-    out: *std.ArrayList(Segment),
-) !void {
-    var cursor: ?[]u8 = null;
-    defer if (cursor) |value| allocator.free(value);
-
-    while (true) {
-        var params: std.ArrayList(zat.XrpcClient.QueryParam) = .empty;
-        defer params.deinit(allocator);
-        try params.appendSlice(allocator, &.{
-            .{ .name = "repo", .value = repo },
-            .{ .name = "collection", .value = "place.stream.livestream" },
-            .{ .name = "limit", .value = "100" },
-        });
-        if (cursor) |value| try params.append(allocator, .{ .name = "cursor", .value = value });
-
-        const method = zat.Nsid.parse("com.atproto.repo.listRecords").?;
-        var response = try client.queryParams(method, params.items);
-        defer response.deinit();
-        if (!response.ok()) return error.LivestreamListFailed;
-
-        var parsed = try response.json();
-        defer parsed.deinit();
-        const records = zat.json.getArray(parsed.value, "records") orelse return error.MissingRecords;
-        for (records) |record| {
-            const uri = zat.json.getString(record, "uri") orelse continue;
-            const parsed_uri = zat.AtUri.parse(uri) orelse continue;
-            const rkey = parsed_uri.rkey() orelse continue;
-            if (std.mem.eql(u8, rkey, selected_rkey)) return;
-            const value = zat.json.getPath(record, "value") orelse continue;
-            try out.append(allocator, try copySegment(allocator, uri, value));
-        }
-
-        const next = zat.json.getString(parsed.value, "cursor") orelse return;
-        const next_copy = try allocator.dupe(u8, next);
-        if (cursor) |value| allocator.free(value);
-        cursor = next_copy;
-    }
-}
-
-fn copySegment(allocator: std.mem.Allocator, uri: []const u8, value: std.json.Value) !Segment {
-    const parsed_uri = zat.AtUri.parse(uri) orelse return error.InvalidLivestreamUri;
-    const rkey = parsed_uri.rkey() orelse return error.MissingLivestreamRkey;
-    const created_at = zat.json.getString(value, "createdAt") orelse return error.MissingCreatedAt;
-    const ended_at = zat.json.getString(value, "endedAt");
-    return .{
-        .uri = try allocator.dupe(u8, uri),
-        .rkey = try allocator.dupe(u8, rkey),
-        .title = try allocator.dupe(u8, zat.json.getString(value, "title") orelse "untitled stream"),
-        .created_at = try allocator.dupe(u8, created_at),
-        .ended_at = if (ended_at) |end| try allocator.dupe(u8, end) else null,
-        .url = try allocator.dupe(u8, zat.json.getString(value, "url") orelse ""),
-    };
-}
-
-fn parseUtcMicros(value: []const u8) !i64 {
-    if (value.len < 20 or value[4] != '-' or value[7] != '-' or value[10] != 'T' or
-        value[13] != ':' or value[16] != ':' or value[value.len - 1] != 'Z')
-        return error.InvalidDatetime;
-
-    const year = try std.fmt.parseInt(u16, value[0..4], 10);
-    const month_int = try std.fmt.parseInt(u4, value[5..7], 10);
-    const day = try std.fmt.parseInt(u5, value[8..10], 10);
-    const hour = try std.fmt.parseInt(u5, value[11..13], 10);
-    const minute = try std.fmt.parseInt(u6, value[14..16], 10);
-    const second = try std.fmt.parseInt(u6, value[17..19], 10);
-    if (year < std.time.epoch.epoch_year or month_int < 1 or month_int > 12 or
-        hour > 23 or minute > 59 or second > 59)
-        return error.InvalidDatetime;
-
-    const month: std.time.epoch.Month = @enumFromInt(month_int);
-    if (day < 1 or day > std.time.epoch.getDaysInMonth(year, month)) return error.InvalidDatetime;
-
-    var days: i64 = 0;
-    var y: u16 = std.time.epoch.epoch_year;
-    while (y < year) : (y += 1) days += std.time.epoch.getDaysInYear(y);
-    var m: u4 = 1;
-    while (m < month_int) : (m += 1) {
-        days += std.time.epoch.getDaysInMonth(year, @enumFromInt(m));
-    }
-    days += @as(i64, day) - 1;
-
-    var micros = (days * std.time.s_per_day + @as(i64, hour) * std.time.s_per_hour +
-        @as(i64, minute) * std.time.s_per_min + @as(i64, second)) * std.time.us_per_s;
-    if (value.len > 20) {
-        if (value[19] != '.') return error.InvalidDatetime;
-        const fraction = value[20 .. value.len - 1];
-        if (fraction.len == 0) return error.InvalidDatetime;
-        var fractional_us: i64 = 0;
-        var digits: usize = 0;
-        while (digits < fraction.len and digits < 6) : (digits += 1) {
-            const c = fraction[digits];
-            if (c < '0' or c > '9') return error.InvalidDatetime;
-            fractional_us = fractional_us * 10 + (c - '0');
-        }
-        while (digits < 6) : (digits += 1) fractional_us *= 10;
-        micros += fractional_us;
-    }
-    return micros;
-}
-
-fn latestLivestreamUri(
-    io: std.Io,
-    allocator: std.mem.Allocator,
-    streamer_did: []const u8,
-) ![]u8 {
-    var url_buffer: [512]u8 = undefined;
-    const url = try std.fmt.bufPrint(&url_buffer, "https://stream.place/api/livestream/{s}", .{streamer_did});
-    var transport = zat.HttpTransport.initWithUserAgent(io, allocator, "zat-streamplace-chat-example/1.1");
-    defer transport.deinit();
-    var response = try transport.fetch(.{ .url = url, .max_response_size = 1024 * 1024 });
-    defer response.deinit(allocator);
-    if (response.status != .ok) return error.LivestreamLookupFailed;
-
-    var parsed = try std.json.parseFromSlice(std.json.Value, allocator, response.body, .{});
-    defer parsed.deinit();
-    const uri = zat.json.getString(parsed.value, "uri") orelse return error.MissingLivestreamUri;
-    return try allocator.dupe(u8, uri);
-}
-
-test "parse Streamplace UTC datetime to microseconds" {
-    try std.testing.expectEqual(@as(i64, 0), try parseUtcMicros("1970-01-01T00:00:00Z"));
-    try std.testing.expectEqual(@as(i64, 1785780238315000), try parseUtcMicros("2026-08-03T18:03:58.315Z"));
-    try std.testing.expectError(error.InvalidDatetime, parseUtcMicros("2026-02-30T00:00:00Z"));
-}
diff --git a/scripts/archive_backfill_smoke.zig b/scripts/archive_backfill_smoke.zig
deleted file mode 100644
index 19d1d64..0000000
--- a/scripts/archive_backfill_smoke.zig
+++ /dev/null
@@ -1,120 +0,0 @@
-const std = @import("std");
-const zat = @import("zat");
-
-pub fn main() !void {
-    var da: std.heap.DebugAllocator(.{}) = .init;
-    defer _ = da.deinit();
-    const allocator = da.allocator();
-
-    std.debug.print("archive backfill smoke: pollz collections from stream.waow.tech\n", .{});
-
-    var threaded = std.Io.Threaded.init(allocator, .{
-        .async_limit = .limited(8),
-        .concurrent_limit = .limited(8),
-    });
-    defer threaded.deinit();
-    const io = threaded.io();
-
-    var handler = Handler{};
-    const t0 = std.Io.Clock.awake.now(io);
-    const serial = try zat.ArchiveBackfill.run(io, allocator, .{
-        .host = "https://stream.waow.tech",
-        .collections = &.{ "tech.waow.pollz.poll", "tech.waow.pollz.vote" },
-        .max_blocks = 15,
-    }, &handler);
-    const t1 = std.Io.Clock.awake.now(io);
-    const serial_ms = @divTrunc(t0.durationTo(t1).toNanoseconds(), std.time.ns_per_ms);
-    handler = Handler{ .printed = 3 };
-    const result = try zat.ArchiveBackfill.run(io, allocator, .{
-        .host = "https://stream.waow.tech",
-        .collections = &.{ "tech.waow.pollz.poll", "tech.waow.pollz.vote" },
-        // a full plan spans tens of GB; the smoke only proves plan/fetch/decode
-        .max_blocks = 15,
-        .concurrency = 6,
-    }, &handler);
-    const t2 = std.Io.Clock.awake.now(io);
-    const conc_ms = @divTrunc(t1.durationTo(t2).toNanoseconds(), std.time.ns_per_ms);
-    std.debug.print("15 blocks serial: {d}ms, concurrency=6: {d}ms\n", .{ serial_ms, conc_ms });
-    if (serial.events_delivered != result.events_delivered) return error.ConcurrencyChangedResults;
-    if (serial.position.?.block_index != result.position.?.block_index) return error.ConcurrencyChangedResults;
-
-    std.debug.print(
-        "pass 1: {d} blocks (truncated={}), {d} events delivered ({d} polls, {d} votes, {d} deletes), position={?any}, last_time_us={?d}\n",
-        .{ result.blocks_decoded, result.truncated, result.events_delivered, handler.polls, handler.votes, handler.deletes, result.position, result.last_time_us },
-    );
-    if (result.events_delivered == 0) return error.NoEvents;
-    if (handler.missing_record != 0) return error.MissingRecords;
-
-    // resume from pass 1's position: must pick up strictly after it
-    const resumed = try zat.ArchiveBackfill.run(io, allocator, .{
-        .host = "https://stream.waow.tech",
-        .collections = &.{ "tech.waow.pollz.poll", "tech.waow.pollz.vote" },
-        .max_blocks = 10,
-        .start_after = result.position,
-    }, &handler);
-    std.debug.print("pass 2 (resumed): {d} blocks, {d} events, position={?any}\n", .{ resumed.blocks_decoded, resumed.events_delivered, resumed.position });
-    const p1 = result.position.?;
-    const p2 = resumed.position orelse return error.NoResumeProgress;
-    if (p2.segment_index < p1.segment_index or
-        (p2.segment_index == p1.segment_index and p2.block_index <= p1.block_index))
-        return error.ResumeWentBackwards;
-
-    // checksum verification on a real sealed segment (smallest in the archive)
-    try verifySmallestSegment(io, allocator);
-}
-
-fn verifySmallestSegment(io: std.Io, allocator: std.mem.Allocator) !void {
-    var transport = zat.HttpTransport.init(io, allocator);
-    defer transport.deinit();
-
-    var listed = try transport.fetch(.{ .url = "https://stream.waow.tech/xrpc/network.bsky.jetstream.listSegments?limit=1000" });
-    defer listed.deinit(allocator);
-    if (listed.status != .ok) return error.ListSegmentsFailed;
-
-    var arena_state = std.heap.ArenaAllocator.init(allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-    const parsed = try std.json.parseFromSliceLeaky(std.json.Value, arena, listed.body, .{});
-    const segments = parsed.object.get("segments").?.array.items;
-    var smallest = segments[0].object;
-    for (segments[1..]) |s| {
-        if (s.object.get("sizeBytes").?.integer < smallest.get("sizeBytes").?.integer) smallest = s.object;
-    }
-    const name = smallest.get("name").?.string;
-    const checksum = try std.fmt.parseInt(u64, smallest.get("checksum").?.string, 16);
-
-    const url = try std.fmt.allocPrint(arena, "https://stream.waow.tech/xrpc/network.bsky.jetstream.getSegment?name={s}", .{name});
-    var fetched = try transport.fetch(.{ .url = url });
-    defer fetched.deinit(allocator);
-    if (fetched.status != .ok) return error.SegmentFetchFailed;
-
-    try zat.ArchiveBackfill.verifySegment(fetched.body, checksum);
-    std.debug.print("checksum verified: {s} ({d} bytes, xxh3={x:0>16})\n", .{ name, fetched.body.len, checksum });
-}
-
-const Handler = struct {
-    polls: u64 = 0,
-    votes: u64 = 0,
-    deletes: u64 = 0,
-    missing_record: u64 = 0,
-    printed: u64 = 0,
-
-    pub fn onEvent(self: *Handler, event: zat.JetstreamEvent) void {
-        const commit = event.commit;
-        if (commit.operation == .delete) {
-            self.deletes += 1;
-            return;
-        }
-        if (commit.record == null) {
-            self.missing_record += 1;
-            return;
-        }
-        if (std.mem.endsWith(u8, commit.collection, ".poll")) self.polls += 1 else self.votes += 1;
-        if (self.printed < 3) {
-            self.printed += 1;
-            const question = zat.json.getString(commit.record.?, "question") orelse
-                zat.json.getString(commit.record.?, "answer") orelse "?";
-            std.debug.print("  sample: {s} {s}/{s} time_us={d} field={s}\n", .{ commit.did, commit.collection, commit.rkey, commit.time_us, question });
-        }
-    }
-};
diff --git a/src/internal/streaming/archive_backfill.zig b/src/internal/streaming/archive_backfill.zig
deleted file mode 100644
index 9cb0b2a..0000000
--- a/src/internal/streaming/archive_backfill.zig
+++ /dev/null
@@ -1,1187 +0,0 @@
-//! archive backfill — replay a jetstream archive through the live-tail handler
-//!
-//! consumes the network.bsky.jetstream archive API (planSnapshot / getBlock /
-//! getSegment; planSnapshot was named planBackfill before jetstream's
-//! proposal-0015 rename) served by archiving jetstream instances such as
-//! stream.waow.tech: plan → fetch sealed jss v1 segment blocks → decompress
-//! (plain zstd, no dictionary) → columnar decode → deliver each matching row
-//! as a jetstream.Event through the same onEvent handler the live client uses.
-//!
-//! after `run` returns, connect the live client with a cursor at or before
-//! the returned `last_time_us` (or the plan boundary) — overlap is fine for
-//! idempotent consumers; delivery is at-least-once. the plan covers SEALED
-//! segments only: the newest ~one segment of events is unsealed, so a
-//! complete drain is always archive pass + live replay from the boundary.
-//!
-//! scope (measured 2026-08-07): retrieval is block-granular, so an UNBOUNDED
-//! plan over a sparse collection reads ~1,300x more than it keeps
-//! (streamplace chat: 40 GB fetched for ~30 MB of rows; one block per ~4096
-//! events regardless of how few match). two ways that cost disappears:
-//! - bound the plan: `after_seq`/`before_seq` (see fetchSeqBounds for
-//!   mapping a witnessed-time window to bounds — a two-hour window planned
-//!   ~25 blocks when measured 2026-08-12)
-//! - or, for a sparse collection's CURRENT records only, use a collection
-//!   directory (lightrail.microcosm.blue) + per-repo listRecords — though
-//!   that cannot express deletes or event ordering.
-//! whole-segment mode buffers one full segment in memory (~277 MB);
-//! block mode buffers one block.
-//!
-//! jss v1 format: https://tangled.org/zat.dev/stream → docs/jss-format-v1.md
-
-const std = @import("std");
-const jetstream = @import("jetstream.zig");
-const sync = @import("sync.zig");
-const cbor = @import("../repo/cbor.zig");
-const zstd = @import("zstd.zig");
-const multibase = @import("../crypto/multibase.zig");
-const HttpTransport = @import("../xrpc/transport.zig").HttpTransport;
-
-const mem = std.mem;
-const json = std.json;
-const Allocator = mem.Allocator;
-const Io = std.Io;
-const log = std.log.scoped(.zat);
-
-pub const Event = jetstream.Event;
-
-pub const Options = struct {
-    /// base URL of an archiving jetstream instance, e.g. "https://stream.waow.tech"
-    host: []const u8,
-    /// collections to replay; empty = all
-    collections: []const []const u8 = &.{},
-    /// dids to replay; empty = all
-    dids: []const []const u8 = &.{},
-    /// bearer credential for token-gated archives (Bluesky's hosted
-    /// instances 401 archive endpoints without one; the live tail never
-    /// needs it). mirrors the Go client's WithAPIKey. sent as
-    /// "Authorization: Bearer <key>" on plan/list/segment/block requests.
-    api_key: ?[]const u8 = null,
-    /// exclusive lower seq bound for the plan; null = from the beginning.
-    /// derive from a time window with fetchSeqBounds.
-    after_seq: ?u64 = null,
-    /// inclusive upper seq bound for the plan; null = to the sealed tip
-    before_seq: ?u64 = null,
-    /// stop after decoding this many blocks. a full plan can span tens of GB;
-    /// use this to bound a partial or exploratory run. when the limit stops
-    /// the run early, the plan is not fully covered — do not treat
-    /// `planned_through_seq` as a completed-backfill boundary.
-    max_blocks: ?u64 = null,
-    /// resume a prior run: skip everything up to and including this position
-    /// (take it from the prior Result.position). segment indices are stable
-    /// for a given archive, so a re-plan covers at least the same range.
-    start_after: ?Position = null,
-    /// verify each whole-mode segment's jss self-checksum (xxh3 over header
-    /// and footer, cross-checked against the plan's checksum) before decoding.
-    /// off by default. per-block zstd content checksums are always verified
-    /// by libzstd regardless of this flag.
-    verify_checksums: bool = false,
-    /// block fetches kept in flight. serial fetch is round-trip-bound (~0.2
-    /// MB/s at 900ms RTT measured 2026-08-08; 8 in flight was 7.2x faster),
-    /// so WAN consumers want 4-8. delivery order is unaffected: events
-    /// always arrive in plan order. stay modest — each slot holds its own
-    /// connection, and politeness is the caller's job. clamped to [1, 16].
-    concurrency: usize = 1,
-};
-
-/// a block's place in the plan; Result.position reports the last decoded one
-pub const Position = struct {
-    segment_index: u32,
-    block_index: u32,
-};
-
-pub const Result = struct {
-    /// archive coverage boundary from the plan. resume the live client with a
-    /// cursor at or before the timestamp of the last delivered event.
-    planned_through_seq: u64,
-    sealed_tip_seq: u64,
-    events_delivered: u64 = 0,
-    blocks_decoded: u64 = 0,
-    /// true when Options.max_blocks stopped the run before the plan was exhausted
-    truncated: bool = false,
-    /// last decoded block; feed to Options.start_after to resume
-    position: ?Position = null,
-    /// witnessed_at of the last delivered event, in the same µs domain as the
-    /// live client's cursor. null if nothing matched.
-    last_time_us: ?i64 = null,
-};
-
-/// replay the archive through `handler` (same contract as JetstreamClient:
-/// `fn onEvent(*H, Event) void`). event slices are only valid during the
-/// onEvent call. blocks until the plan is exhausted.
-pub fn run(io: Io, allocator: Allocator, options: Options, handler: anytype) !Result {
-    var transport = HttpTransport.init(io, allocator);
-    defer transport.deinit();
-
-    const authorization = try bearerHeader(allocator, options.api_key);
-    defer if (authorization) |a| allocator.free(a);
-
-    var plan_arena = std.heap.ArenaAllocator.init(allocator);
-    defer plan_arena.deinit();
-
-    const plan = try fetchPlan(plan_arena.allocator(), &transport, options, authorization);
-
-    var result: Result = .{
-        .planned_through_seq = plan.planned_through_seq,
-        .sealed_tip_seq = plan.sealed_tip_seq,
-    };
-
-    const concurrency = @min(@max(options.concurrency, 1), max_concurrency);
-    var window = try FetchWindow.init(io, allocator, concurrency);
-    defer window.deinit();
-    window.authorization = authorization;
-
-    for (plan.segments) |segment| {
-        if (segment.blocks) |ranges| {
-            for (ranges) |range| {
-                var bi = range.first;
-                while (bi <= range.last) : (bi += 1) {
-                    if (skippedByResume(options.start_after, segment.index, bi)) continue;
-                    if (limitReached(options, &result)) {
-                        window.discardAll();
-                        return result;
-                    }
-                    if (window.isFull()) window.deliverOldest(allocator, options, handler, &result) catch |err| {
-                        window.discardAll();
-                        return err;
-                    };
-                    window.start(options, segment.name, .{
-                        .segment_index = segment.index,
-                        .block_index = bi,
-                    }) catch |err| {
-                        window.discardAll();
-                        return err;
-                    };
-                }
-            }
-        } else {
-            if (options.start_after) |sa| if (segment.index < sa.segment_index) continue;
-            // whole-segment fetches are serial; flush pending blocks first so
-            // delivery stays in plan order
-            window.deliverAll(allocator, options, handler, &result) catch |err| {
-                window.discardAll();
-                return err;
-            };
-            if (limitReached(options, &result)) return result;
-            const url = try std.fmt.allocPrint(
-                allocator,
-                "{s}/xrpc/network.bsky.jetstream.getSegment?name={s}",
-                .{ options.host, segment.name },
-            );
-            defer allocator.free(url);
-            var fetched = try fetchWithRetry(io, &transport, url, authorization);
-            defer fetched.deinit(allocator);
-            try deliverSegment(allocator, fetched.body, options, segment, handler, &result);
-        }
-    }
-
-    window.deliverAll(allocator, options, handler, &result) catch |err| {
-        window.discardAll();
-        return err;
-    };
-    return result;
-}
-
-const max_concurrency = 16;
-
-const FetchError = anyerror;
-
-fn fetchJob(io: Io, transport: *HttpTransport, url: []const u8, authorization: ?[]const u8) FetchError!HttpTransport.FetchResult {
-    return fetchWithRetry(io, transport, url, authorization);
-}
-
-/// a FIFO window of in-flight block fetches. starting is out-of-order I/O;
-/// delivery always awaits the oldest first, so handler order equals plan
-/// order regardless of concurrency. each slot owns a transport (one
-/// connection) reused as the window slides.
-const FetchWindow = struct {
-    io: Io,
-    allocator: Allocator,
-    transports: []HttpTransport,
-    slots: []Inflight,
-    /// borrowed from run(); outlives every in-flight fetch
-    authorization: ?[]const u8 = null,
-    head: usize = 0,
-    count: usize = 0,
-
-    const Inflight = struct {
-        future: std.Io.Future(FetchError!HttpTransport.FetchResult),
-        url: []u8,
-        position: Position,
-    };
-
-    fn init(io: Io, allocator: Allocator, concurrency: usize) !FetchWindow {
-        const transports = try allocator.alloc(HttpTransport, concurrency);
-        errdefer allocator.free(transports);
-        const slots = try allocator.alloc(Inflight, concurrency);
-        for (transports) |*t| t.* = HttpTransport.init(io, allocator);
-        return .{ .io = io, .allocator = allocator, .transports = transports, .slots = slots };
-    }
-
-    fn deinit(self: *FetchWindow) void {
-        self.discardAll();
-        for (self.transports) |*t| t.deinit();
-        self.allocator.free(self.transports);
-        self.allocator.free(self.slots);
-    }
-
-    fn isFull(self: *const FetchWindow) bool {
-        return self.count == self.slots.len;
-    }
-
-    fn start(self: *FetchWindow, options: Options, segment_name: []const u8, position: Position) !void {
-        std.debug.assert(!self.isFull());
-        const url = try std.fmt.allocPrint(
-            self.allocator,
-            "{s}/xrpc/network.bsky.jetstream.getBlock?segment={s}&blockIndex={d}",
-            .{ options.host, segment_name, position.block_index },
-        );
-        errdefer self.allocator.free(url);
-        const slot = (self.head + self.count) % self.slots.len;
-        self.slots[slot] = .{
-            .future = self.io.async(fetchJob, .{ self.io, &self.transports[slot], url, self.authorization }),
-            .url = url,
-            .position = position,
-        };
-        self.count += 1;
-    }
-
-    fn deliverOldest(self: *FetchWindow, allocator: Allocator, options: Options, handler: anytype, result: *Result) !void {
-        std.debug.assert(self.count > 0);
-        const slot = &self.slots[self.head];
-        self.head = (self.head + 1) % self.slots.len;
-        self.count -= 1;
-        defer self.allocator.free(slot.url);
-        var fetched = try slot.future.await(self.io);
-        defer fetched.deinit(allocator);
-        try deliverFrame(allocator, fetched.body, options, slot.position, handler, result);
-    }
-
-    /// drain in order, stopping (and discarding the rest) if max_blocks hits
-    fn deliverAll(self: *FetchWindow, allocator: Allocator, options: Options, handler: anytype, result: *Result) !void {
-        while (self.count > 0) {
-            if (limitReached(options, result)) {
-                self.discardAll();
-                return;
-            }
-            try self.deliverOldest(allocator, options, handler, result);
-        }
-    }
-
-    /// await and drop everything in flight without decoding
-    fn discardAll(self: *FetchWindow) void {
-        while (self.count > 0) {
-            const slot = &self.slots[self.head];
-            self.head = (self.head + 1) % self.slots.len;
-            self.count -= 1;
-            if (slot.future.await(self.io)) |fetched| {
-                var discarded = fetched;
-                discarded.deinit(self.allocator);
-            } else |_| {}
-            self.allocator.free(slot.url);
-        }
-    }
-};
-
-/// decode every block of a sealed in-memory jss segment through `handler` —
-/// the offline counterpart of `run` for a segment already on disk (local
-/// archive copies, getSegment bodies fetched out-of-band). same delivery
-/// contract; Options.host is unused. honors collections/dids/max_blocks/
-/// verify_checksums.
-pub fn replaySegment(allocator: Allocator, body: []const u8, options: Options, handler: anytype) !Result {
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    try deliverSegment(allocator, body, options, .{
-        .name = "",
-        .index = 0,
-        .checksum = null,
-        .blocks = null,
-    }, handler, &result);
-    return result;
-}
-
-/// resume skip: everything at or before start_after is already delivered
-fn skippedByResume(start_after: ?Position, segment_index: u32, block_index: u32) bool {
-    const sa = start_after orelse return false;
-    if (segment_index < sa.segment_index) return true;
-    return segment_index == sa.segment_index and block_index <= sa.block_index;
-}
-
-const fetch_attempts = 4;
-
-/// GET with bounded retries: transport errors and retryable statuses
-/// (5xx, 429) back off 1s/2s/4s; other non-200s fail immediately. a
-/// multi-GB run should not be discarded over one transient failure.
-fn fetchWithRetry(io: Io, transport: *HttpTransport, url: []const u8, authorization: ?[]const u8) !HttpTransport.FetchResult {
-    var attempt: u32 = 0;
-    var backoff_s: u64 = 1;
-    while (true) {
-        attempt += 1;
-        if (transport.fetch(.{ .url = url, .authorization = authorization })) |result| {
-            if (result.status == .ok) return result;
-            var discarded = result;
-            discarded.deinit(transport.allocator);
-            const code = @intFromEnum(result.status);
-            const retryable = code >= 500 or result.status == .too_many_requests;
-            if (!retryable) return error.FetchFailed;
-            if (attempt >= fetch_attempts) return error.FetchRetriesExhausted;
-            log.warn("fetch {d} (attempt {d}/{d}), retrying in {d}s: {s}", .{ code, attempt, fetch_attempts, backoff_s, url });
-        } else |err| {
-            if (err == error.Canceled) return err;
-            if (attempt >= fetch_attempts) return err;
-            log.warn("fetch error {s} (attempt {d}/{d}), retrying in {d}s: {s}", .{ @errorName(err), attempt, fetch_attempts, backoff_s, url });
-        }
-        try io.sleep(Io.Duration.fromSeconds(@intCast(backoff_s)), .awake);
-        backoff_s *= 2;
-    }
-}
-
-fn limitReached(options: Options, result: *Result) bool {
-    const max = options.max_blocks orelse return false;
-    if (result.blocks_decoded < max) return false;
-    result.truncated = true;
-    return true;
-}
-
-// === plan ===
-
-const BlockRange = struct { first: u32, last: u32 };
-
-const PlannedSegment = struct {
-    name: []const u8,
-    index: u32,
-    /// jss self-checksum from the plan (16 hex chars on the wire); null if absent
-    checksum: ?u64,
-    /// null = whole-segment mode
-    blocks: ?[]const BlockRange,
-};
-
-const Plan = struct {
-    planned_through_seq: u64,
-    sealed_tip_seq: u64,
-    segments: []const PlannedSegment,
-};
-
-fn planRequestBody(arena: Allocator, options: Options) ![]const u8 {
-    var body: std.Io.Writer.Allocating = .init(arena);
-    var stringify: json.Stringify = .{ .writer = &body.writer };
-    try stringify.beginObject();
-    try stringify.objectField("collections");
-    try stringify.beginArray();
-    for (options.collections) |c| try stringify.write(c);
-    try stringify.endArray();
-    try stringify.objectField("dids");
-    try stringify.beginArray();
-    for (options.dids) |d| try stringify.write(d);
-    try stringify.endArray();
-    if (options.after_seq) |seq| {
-        try stringify.objectField("afterSeq");
-        try stringify.write(seq);
-    }
-    if (options.before_seq) |seq| {
-        try stringify.objectField("beforeSeq");
-        try stringify.write(seq);
-    }
-    try stringify.endObject();
-    return body.written();
-}
-
-fn bearerHeader(allocator: Allocator, api_key: ?[]const u8) !?[]u8 {
-    const key = api_key orelse return null;
-    return try std.fmt.allocPrint(allocator, "Bearer {s}", .{key});
-}
-
-fn fetchPlan(arena: Allocator, transport: *HttpTransport, options: Options, authorization: ?[]const u8) !Plan {
-    const body = try planRequestBody(arena, options);
-    const url = try std.fmt.allocPrint(arena, "{s}/xrpc/network.bsky.jetstream.planSnapshot", .{options.host});
-    var fetched = try transport.fetch(.{
-        .url = url,
-        .method = .POST,
-        .payload = body,
-        .authorization = authorization,
-    });
-    defer fetched.deinit(transport.allocator);
-    if (fetched.status != .ok) {
-        log.err("planSnapshot failed: {d} {s}", .{ @intFromEnum(fetched.status), fetched.body });
-        return error.PlanFailed;
-    }
-    return parsePlan(arena, fetched.body);
-}
-
-// === time window → seq bounds ===
-
-/// one sealed segment's metadata row from listSegments, reduced to what
-/// bounds derivation needs
-pub const SegmentSpan = struct {
-    min_seq: u64,
-    max_seq: u64,
-    min_witnessed_us: i64,
-    max_witnessed_us: i64,
-};
-
-pub const SeqBounds = struct {
-    after_seq: ?u64,
-    before_seq: ?u64,
-    /// false when no sealed segment's witnessed window intersects the
-    /// requested window — the window predates the archive's live capture or
-    /// lies entirely in the unsealed tail. plan nothing; go straight to the
-    /// live client (or drop the bounds for a full historical scan).
-    covered: bool,
-};
-
-/// reduce a witnessed-time window to plan seq bounds: the bounds admit every
-/// segment whose witnessed span intersects [start_us, end_us]. sound for
-/// live-captured rows (an event is witnessed at or after it happened);
-/// bootstrap-merged rows carry bootstrap-era witnessed times, so windows
-/// older than the archive's live capture report covered=false rather than
-/// pretending.
-pub fn seqBoundsForWindow(spans: []const SegmentSpan, start_us: i64, end_us: ?i64) SeqBounds {
-    var after: ?u64 = null;
-    var before: ?u64 = null;
-    var covered = false;
-    for (spans) |span| {
-        const ends_before = span.max_witnessed_us < start_us;
-        const starts_after = if (end_us) |e| span.min_witnessed_us > e else false;
-        if (ends_before or starts_after) continue;
-        covered = true;
-        if (after == null or span.min_seq -| 1 < after.?) after = span.min_seq -| 1;
-        if (before == null or span.max_seq > before.?) before = span.max_seq;
-    }
-    // an unbounded end means "through the sealed tip": drop the upper bound
-    if (end_us == null) before = null;
-    return .{ .after_seq = after, .before_seq = before, .covered = covered };
-}
-
-/// fetch every sealed segment's metadata from `host` and derive the seq
-/// bounds covering the witnessed window [start_us, end_us]. end_us == null
-/// means "through the sealed tip".
-pub fn fetchSeqBounds(io: Io, allocator: Allocator, host: []const u8, start_us: i64, end_us: ?i64, api_key: ?[]const u8) !SeqBounds {
-    var transport = HttpTransport.init(io, allocator);
-    defer transport.deinit();
-    const authorization = try bearerHeader(allocator, api_key);
-    defer if (authorization) |a| allocator.free(a);
-    var arena_state = std.heap.ArenaAllocator.init(allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    var spans: std.ArrayList(SegmentSpan) = .empty;
-    var cursor: ?[]const u8 = null;
-    while (true) {
-        const url = if (cursor) |c|
-            try std.fmt.allocPrint(arena, "{s}/xrpc/network.bsky.jetstream.listSegments?limit=1000&cursor={s}", .{ host, c })
-        else
-            try std.fmt.allocPrint(arena, "{s}/xrpc/network.bsky.jetstream.listSegments?limit=1000", .{host});
-        var fetched = try fetchWithRetry(io, &transport, url, authorization);
-        defer fetched.deinit(allocator);
-        const parsed = try json.parseFromSliceLeaky(json.Value, arena, fetched.body, .{});
-        const root = switch (parsed) {
-            .object => |o| o,
-            else => return error.MalformedSegmentList,
-        };
-        const segments = switch (root.get("segments") orelse return error.MalformedSegmentList) {
-            .array => |a| a.items,
-            else => return error.MalformedSegmentList,
-        };
-        if (segments.len == 0) break;
-        for (segments) |seg| {
-            try spans.append(arena, .{
-                .min_seq = try getJsonU64(seg, "minSeq"),
-                .max_seq = try getJsonU64(seg, "maxSeq"),
-                .min_witnessed_us = @intCast(try getJsonU64(seg, "minWitnessedAt")),
-                .max_witnessed_us = @intCast(try getJsonU64(seg, "maxWitnessedAt")),
-            });
-        }
-        // a short page is the end even if a cursor is present
-        if (segments.len < 1000) break;
-        const next = root.get("cursor") orelse break;
-        cursor = switch (next) {
-            .string => |s| try arena.dupe(u8, s),
-            .integer => |i| try std.fmt.allocPrint(arena, "{d}", .{i}),
-            else => break,
-        };
-    }
-    return seqBoundsForWindow(spans.items, start_us, end_us);
-}
-
-fn parsePlan(arena: Allocator, body: []const u8) !Plan {
-    const parsed = try json.parseFromSliceLeaky(json.Value, arena, body, .{});
-    const root = switch (parsed) {
-        .object => |o| o,
-        else => return error.MalformedPlan,
-    };
-
-    const segments_val = root.get("segments") orelse return error.MalformedPlan;
-    const segments_json = switch (segments_val) {
-        .array => |a| a.items,
-        else => return error.MalformedPlan,
-    };
-
-    const segments = try arena.alloc(PlannedSegment, segments_json.len);
-    for (segments_json, segments) |seg_val, *out| {
-        const seg = switch (seg_val) {
-            .object => |o| o,
-            else => return error.MalformedPlan,
-        };
-        const name = switch (seg.get("name") orelse return error.MalformedPlan) {
-            .string => |s| s,
-            else => return error.MalformedPlan,
-        };
-        const mode = switch (seg.get("mode") orelse return error.MalformedPlan) {
-            .string => |s| s,
-            else => return error.MalformedPlan,
-        };
-        var blocks: ?[]const BlockRange = null;
-        if (mem.eql(u8, mode, "blocks")) {
-            const ranges_json = switch (seg.get("blocks") orelse return error.MalformedPlan) {
-                .array => |a| a.items,
-                else => return error.MalformedPlan,
-            };
-            const ranges = try arena.alloc(BlockRange, ranges_json.len);
-            for (ranges_json, ranges) |range_val, *r| {
-                r.* = .{
-                    .first = try getJsonU32(range_val, "first"),
-                    .last = try getJsonU32(range_val, "last"),
-                };
-            }
-            blocks = ranges;
-        }
-        const checksum: ?u64 = if (seg.get("checksum")) |v| switch (v) {
-            .string => |s| std.fmt.parseInt(u64, s, 16) catch null,
-            else => null,
-        } else null;
-        out.* = .{
-            .name = name,
-            .index = try getJsonU32(seg_val, "index"),
-            .checksum = checksum,
-            .blocks = blocks,
-        };
-    }
-
-    return .{
-        .planned_through_seq = try getJsonU64(.{ .object = root }, "plannedThroughSeq"),
-        .sealed_tip_seq = try getJsonU64(.{ .object = root }, "sealedTipSeq"),
-        .segments = segments,
-    };
-}
-
-fn getJsonU64(val: json.Value, key: []const u8) !u64 {
-    const obj = switch (val) {
-        .object => |o| o,
-        else => return error.MalformedPlan,
-    };
-    return switch (obj.get(key) orelse return error.MalformedPlan) {
-        .integer => |i| if (i < 0) error.MalformedPlan else @intCast(i),
-        else => error.MalformedPlan,
-    };
-}
-
-fn getJsonU32(val: json.Value, key: []const u8) !u32 {
-    return std.math.cast(u32, try getJsonU64(val, key)) orelse error.MalformedPlan;
-}
-
-// === jss v1 segment walk ===
-
-const header_size = 256;
-const block_index_entry_size = 52;
-const max_block_event_count = 1 << 18;
-const max_block_count = 1 << 20;
-
-fn deliverSegment(allocator: Allocator, body: []const u8, options: Options, segment: PlannedSegment, handler: anytype, result: *Result) !void {
-    if (body.len < header_size) return error.MalformedSegment;
-    if (!mem.eql(u8, body[0..4], "jss0")) return error.MalformedSegment;
-    const block_count = mem.readInt(u32, body[14..18], .little);
-    const block_index_offset = mem.readInt(u64, body[90..98], .little);
-    if (block_count > max_block_count) return error.MalformedSegment;
-
-    if (options.verify_checksums) try verifySegment(body, segment.checksum);
-
-    for (0..block_count) |bi| {
-        if (skippedByResume(options.start_after, segment.index, @intCast(bi))) continue;
-        if (limitReached(options, result)) return;
-        const entry_off = block_index_offset + bi * block_index_entry_size;
-        if (entry_off + block_index_entry_size > body.len) return error.MalformedSegment;
-        const offset = mem.readInt(u64, body[entry_off..][0..8], .little);
-        const compressed_size = mem.readInt(u32, body[entry_off + 8 ..][0..4], .little);
-        const frame_start = offset + 8;
-        if (frame_start + compressed_size > body.len) return error.MalformedSegment;
-        try deliverFrame(allocator, body[frame_start..][0..compressed_size], options, .{
-            .segment_index = segment.index,
-            .block_index = @intCast(bi),
-        }, handler, result);
-    }
-}
-
-/// verify a sealed segment's self-checksum: xxh3_64(seed 0) over
-/// header[12..256] ++ file[footer_offset..EOF], stored at header offset 4.
-/// also cross-checks the plan's checksum for the segment when present.
-pub fn verifySegment(body: []const u8, plan_checksum: ?u64) !void {
-    if (body.len < header_size) return error.MalformedSegment;
-    if (!mem.eql(u8, body[0..4], "jss0")) return error.MalformedSegment;
-    const stored = mem.readInt(u64, body[4..12], .little);
-    if (stored == 0) return error.SegmentUnsealed;
-    if (plan_checksum) |expected| {
-        if (expected != stored) return error.ChecksumMismatch;
-    }
-    const footer_offset = mem.readInt(u64, body[58..66], .little);
-    if (footer_offset < header_size or footer_offset > body.len) return error.MalformedSegment;
-    var hasher = std.hash.XxHash3.init(0);
-    hasher.update(body[12..header_size]);
-    hasher.update(body[footer_offset..]);
-    if (hasher.final() != stored) return error.ChecksumMismatch;
-}
-
-fn deliverFrame(allocator: Allocator, compressed: []const u8, options: Options, position: Position, handler: anytype, result: *Result) !void {
-    var arena_state = std.heap.ArenaAllocator.init(allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const raw = zstd.decompressAlloc(arena, compressed) catch |err| switch (err) {
-        error.OutOfMemory => return error.OutOfMemory,
-        else => return error.BlockDecompressFailed,
-    };
-
-    try deliverBlock(arena, raw, options, handler, result);
-    result.blocks_decoded += 1;
-    result.position = position;
-    // a bounded-but-large plan (a session inside the merged bootstrap
-    // region can touch thousands of blocks) must not look like a hang
-    if (result.blocks_decoded % 256 == 0)
-        log.info("archive backfill: {d} blocks decoded, {d} events delivered", .{ result.blocks_decoded, result.events_delivered });
-}
-
-// jss event kinds (1..7)
-const kind_create = 1;
-const kind_update = 2;
-const kind_delete = 3;
-const kind_create_resync = 7;
-
-/// decode one decompressed columnar block and deliver matching rows.
-/// `arena` backs per-row record decoding; the caller resets it per block.
-fn deliverBlock(arena: Allocator, raw: []const u8, options: Options, handler: anytype, result: *Result) !void {
-    if (raw.len < 4) return error.MalformedBlock;
-    const n: usize = mem.readInt(u32, raw[0..4], .little);
-    if (n == 0) {
-        if (raw.len != 4) return error.MalformedBlock;
-        return;
-    }
-    if (n > max_block_event_count) return error.MalformedBlock;
-
-    var o: usize = 4;
-    const seq_base = o;
-    o += 8 * n; // seq
-    const wit_base = o;
-    o += 8 * n; // witnessed_at
-    o += 8 * n; // indexed_at (unused)
-    const kind_base = o;
-    o += n;
-    const col_len_base = o;
-    o += n;
-    const did_len_base = o;
-    o += 2 * n;
-    const rkey_len_base = o;
-    o += n;
-    const rev_len_base = o;
-    o += n;
-    const pay_len_base = o;
-    o += 4 * n;
-    if (o > raw.len) return error.MalformedBlock;
-
-    // blob section bases from length totals
-    var col_total: usize = 0;
-    var did_total: usize = 0;
-    var rkey_total: usize = 0;
-    var rev_total: usize = 0;
-    var pay_total: usize = 0;
-    for (0..n) |i| {
-        col_total += raw[col_len_base + i];
-        did_total += mem.readInt(u16, raw[did_len_base + 2 * i ..][0..2], .little);
-        rkey_total += raw[rkey_len_base + i];
-        rev_total += raw[rev_len_base + i];
-        pay_total += mem.readInt(u32, raw[pay_len_base + 4 * i ..][0..4], .little);
-    }
-    if (o + col_total + did_total + rkey_total + rev_total + pay_total != raw.len)
-        return error.MalformedBlock;
-
-    var col_off = o;
-    var did_off = col_off + col_total;
-    var rkey_off = did_off + did_total;
-    var rev_off = rkey_off + rkey_total;
-    var pay_off = rev_off + rev_total;
-
-    for (0..n) |i| {
-        const col_len = raw[col_len_base + i];
-        const did_len = mem.readInt(u16, raw[did_len_base + 2 * i ..][0..2], .little);
-        const rkey_len = raw[rkey_len_base + i];
-        const rev_len = raw[rev_len_base + i];
-        const pay_len = mem.readInt(u32, raw[pay_len_base + 4 * i ..][0..4], .little);
-
-        const collection = raw[col_off..][0..col_len];
-        const did = raw[did_off..][0..did_len];
-        const rkey = raw[rkey_off..][0..rkey_len];
-        const rev = raw[rev_off..][0..rev_len];
-        const payload = raw[pay_off..][0..pay_len];
-        col_off += col_len;
-        did_off += did_len;
-        rkey_off += rkey_len;
-        rev_off += rev_len;
-        pay_off += pay_len;
-
-        const operation: sync.CommitAction = switch (raw[kind_base + i]) {
-            kind_create, kind_create_resync => .create,
-            kind_update => .update,
-            kind_delete => .delete,
-            // identity/account/sync marker rows carry no record; the live-tail
-            // pass re-covers them, so the archive pass skips them.
-            else => continue,
-        };
-
-        // planned blocks can contain other collections' rows — always filter client-side
-        if (options.collections.len > 0 and !containsString(options.collections, collection)) continue;
-        if (options.dids.len > 0 and !containsString(options.dids, did)) continue;
-
-        const record: ?json.Value = if (operation != .delete and payload.len > 0) blk: {
-            const decoded = cbor.decode(arena, payload) catch return error.MalformedRecord;
-            break :blk try cborToJson(arena, decoded.value);
-        } else null;
-
-        const witnessed_at = mem.readInt(i64, raw[wit_base + 8 * i ..][0..8], .little);
-        _ = mem.readInt(u64, raw[seq_base + 8 * i ..][0..8], .little);
-
-        handler.onEvent(.{ .commit = .{
-            .did = did,
-            .time_us = witnessed_at,
-            .rev = if (rev_len > 0) rev else null,
-            .operation = operation,
-            .collection = collection,
-            .rkey = rkey,
-            .record = record,
-        } });
-        result.events_delivered += 1;
-        result.last_time_us = witnessed_at;
-    }
-}
-
-fn containsString(haystack: []const []const u8, needle: []const u8) bool {
-    for (haystack) |s| if (mem.eql(u8, s, needle)) return true;
-    return false;
-}
-
-// === DAG-CBOR record → the json.Value shape the live client delivers ===
-
-/// bytes and links use the atproto lex-JSON conventions the live jetstream
-/// emits: {"$bytes": base64-no-pad} and {"$link": cid-base32-string}.
-fn cborToJson(arena: Allocator, value: cbor.Value) Allocator.Error!json.Value {
-    return switch (value) {
-        .null => .null,
-        .boolean => |b| .{ .bool = b },
-        .unsigned => |u| if (std.math.cast(i64, u)) |i|
-            .{ .integer = i }
-        else
-            .{ .float = @floatFromInt(u) },
-        .negative => |i| .{ .integer = i },
-        .text => |s| .{ .string = s },
-        .bytes => |b| blk: {
-            const b64 = std.base64.standard_no_pad.Encoder;
-            const encoded = try arena.alloc(u8, b64.calcSize(b.len));
-            _ = b64.encode(encoded, b);
-            var obj: json.ObjectMap = .empty;
-            try obj.put(arena, "$bytes", .{ .string = encoded });
-            break :blk .{ .object = obj };
-        },
-        .cid => |c| blk: {
-            const encoded = multibase.base32lower.encode(arena, c.raw) catch return error.OutOfMemory;
-            var obj: json.ObjectMap = .empty;
-            try obj.put(arena, "$link", .{ .string = encoded });
-            break :blk .{ .object = obj };
-        },
-        .array => |items| blk: {
-            var arr = json.Array.init(arena);
-            try arr.ensureTotalCapacityPrecise(items.len);
-            for (items) |item| arr.appendAssumeCapacity(try cborToJson(arena, item));
-            break :blk .{ .array = arr };
-        },
-        .map => |entries| blk: {
-            var obj: json.ObjectMap = .empty;
-            try obj.ensureTotalCapacity(arena, entries.len);
-            for (entries) |entry| obj.putAssumeCapacity(entry.key, try cborToJson(arena, entry.value));
-            break :blk .{ .object = obj };
-        },
-    };
-}
-
-// === tests ===
-
-const testing = std.testing;
-
-const CollectingHandler = struct {
-    arena: Allocator,
-    events: std.array_list.Managed(jetstream.CommitEvent),
-
-    fn init(arena: Allocator) CollectingHandler {
-        return .{ .arena = arena, .events = .init(arena) };
-    }
-
-    pub fn onEvent(self: *CollectingHandler, event: Event) void {
-        // copy string fields — block memory dies after deliverBlock returns
-        var commit = event.commit;
-        commit.did = self.arena.dupe(u8, commit.did) catch unreachable;
-        commit.collection = self.arena.dupe(u8, commit.collection) catch unreachable;
-        commit.rkey = self.arena.dupe(u8, commit.rkey) catch unreachable;
-        commit.record = null;
-        self.events.append(commit) catch unreachable;
-    }
-};
-
-/// build a columnar jss block from rows for tests
-fn buildTestBlock(arena: Allocator, rows: []const struct {
-    seq: u64,
-    witnessed_at: i64,
-    kind: u8,
-    collection: []const u8,
-    did: []const u8,
-    rkey: []const u8,
-    rev: []const u8,
-    payload: []const u8,
-}) ![]u8 {
-    var out: std.Io.Writer.Allocating = .init(arena);
-    const w = &out.writer;
-    try w.writeInt(u32, @intCast(rows.len), .little);
-    for (rows) |r| try w.writeInt(u64, r.seq, .little);
-    for (rows) |r| try w.writeInt(i64, r.witnessed_at, .little);
-    for (rows) |_| try w.writeInt(i64, 0, .little); // indexed_at
-    for (rows) |r| try w.writeByte(r.kind);
-    for (rows) |r| try w.writeByte(@intCast(r.collection.len));
-    for (rows) |r| try w.writeInt(u16, @intCast(r.did.len), .little);
-    for (rows) |r| try w.writeByte(@intCast(r.rkey.len));
-    for (rows) |r| try w.writeByte(@intCast(r.rev.len));
-    for (rows) |r| try w.writeInt(u32, @intCast(r.payload.len), .little);
-    for (rows) |r| try w.writeAll(r.collection);
-    for (rows) |r| try w.writeAll(r.did);
-    for (rows) |r| try w.writeAll(r.rkey);
-    for (rows) |r| try w.writeAll(r.rev);
-    for (rows) |r| try w.writeAll(r.payload);
-    return out.written();
-}
-
-// {"$type": "tech.waow.pollz.vote", "answer": 2}
-const test_record_cbor = [_]u8{
-    0xa2, // map(2)
-    0x65, '$', 't', 'y', 'p', 'e', // "$type"
-    0x74, 't', 'e', 'c', 'h', '.',
-    'w',  'a', 'o', 'w', '.', 'p',
-    'o',  'l', 'l', 'z', '.', 'v',
-    'o',  't', 'e',
-    0x66, 'a', 'n', 's', 'w', 'e', 'r', // "answer"
-    0x02,
-};
-
-test "deliverBlock decodes rows, filters collections, maps kinds" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const raw = try buildTestBlock(arena, &.{
-        .{ .seq = 10, .witnessed_at = 1000, .kind = kind_create, .collection = "tech.waow.pollz.vote", .did = "did:plc:alice", .rkey = "r1", .rev = "rev1", .payload = &test_record_cbor },
-        .{ .seq = 11, .witnessed_at = 1001, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:bob", .rkey = "r2", .rev = "rev2", .payload = &test_record_cbor },
-        .{ .seq = 12, .witnessed_at = 1002, .kind = kind_delete, .collection = "tech.waow.pollz.vote", .did = "did:plc:carol", .rkey = "r3", .rev = "", .payload = "" },
-        .{ .seq = 13, .witnessed_at = 1003, .kind = kind_create_resync, .collection = "tech.waow.pollz.vote", .did = "did:plc:dave", .rkey = "r4", .rev = "rev4", .payload = &test_record_cbor },
-        .{ .seq = 14, .witnessed_at = 1004, .kind = 5, .collection = "$account", .did = "did:plc:eve", .rkey = "", .rev = "", .payload = "" },
-    });
-
-    var handler = CollectingHandler.init(arena);
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    try deliverBlock(arena, raw, .{
-        .host = "https://example.test",
-        .collections = &.{"tech.waow.pollz.vote"},
-    }, &handler, &result);
-
-    try testing.expectEqual(@as(u64, 3), result.events_delivered);
-    try testing.expectEqual(@as(i64, 1003), result.last_time_us.?);
-
-    try testing.expectEqualStrings("did:plc:alice", handler.events.items[0].did);
-    try testing.expectEqual(sync.CommitAction.create, handler.events.items[0].operation);
-    try testing.expectEqual(@as(i64, 1000), handler.events.items[0].time_us);
-
-    try testing.expectEqual(sync.CommitAction.delete, handler.events.items[1].operation);
-    try testing.expectEqualStrings("did:plc:carol", handler.events.items[1].did);
-
-    // create_resync maps to create
-    try testing.expectEqual(sync.CommitAction.create, handler.events.items[2].operation);
-    try testing.expectEqualStrings("did:plc:dave", handler.events.items[2].did);
-}
-
-test "deliverBlock decodes DAG-CBOR record into json.Value" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const raw = try buildTestBlock(arena, &.{
-        .{ .seq = 1, .witnessed_at = 42, .kind = kind_create, .collection = "tech.waow.pollz.vote", .did = "did:plc:alice", .rkey = "r1", .rev = "rev1", .payload = &test_record_cbor },
-    });
-
-    const RecordHandler = struct {
-        answer: ?i64 = null,
-        type_ok: bool = false,
-
-        pub fn onEvent(self: *@This(), event: Event) void {
-            const record = event.commit.record orelse return;
-            const obj = record.object;
-            self.type_ok = mem.eql(u8, obj.get("$type").?.string, "tech.waow.pollz.vote");
-            self.answer = obj.get("answer").?.integer;
-        }
-    };
-
-    var handler = RecordHandler{};
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    try deliverBlock(arena, raw, .{ .host = "https://example.test" }, &handler, &result);
-
-    try testing.expect(handler.type_ok);
-    try testing.expectEqual(@as(i64, 2), handler.answer.?);
-}
-
-test "deliverBlock rejects trailing bytes and truncation" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const raw = try buildTestBlock(arena, &.{
-        .{ .seq = 1, .witnessed_at = 42, .kind = kind_create, .collection = "c", .did = "d", .rkey = "r", .rev = "", .payload = "" },
-    });
-
-    const NoopHandler = struct {
-        pub fn onEvent(_: *@This(), _: Event) void {}
-    };
-    var handler = NoopHandler{};
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    const opts = Options{ .host = "https://example.test" };
-
-    const with_trailing = try mem.concat(arena, u8, &.{ raw, "x" });
-    try testing.expectError(error.MalformedBlock, deliverBlock(arena, with_trailing, opts, &handler, &result));
-    try testing.expectError(error.MalformedBlock, deliverBlock(arena, raw[0 .. raw.len - 1], opts, &handler, &result));
-}
-
-test "deliverBlock accepts empty compacted block" {
-    const NoopHandler = struct {
-        pub fn onEvent(_: *@This(), _: Event) void {}
-    };
-    var handler = NoopHandler{};
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    try deliverBlock(testing.allocator, &.{ 0, 0, 0, 0 }, .{ .host = "https://example.test" }, &handler, &result);
-    try testing.expectEqual(@as(u64, 0), result.events_delivered);
-}
-
-test "deliverFrame decompresses a plain zstd frame" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    // empty block (event_count 0) wrapped in a hand-built zstd frame:
-    // magic, FHD (single-segment, 1-byte FCS), content size 4,
-    // block header (last, raw, size 4), payload
-    const frame = [_]u8{ 0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x04, 0x21, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
-
-    const NoopHandler = struct {
-        pub fn onEvent(_: *@This(), _: Event) void {}
-    };
-    var handler = NoopHandler{};
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    try deliverFrame(arena, &frame, .{ .host = "https://example.test" }, .{ .segment_index = 7, .block_index = 3 }, &handler, &result);
-    try testing.expectEqual(@as(u64, 0), result.events_delivered);
-    try testing.expectEqual(@as(u64, 1), result.blocks_decoded);
-    try testing.expectEqual(@as(u32, 7), result.position.?.segment_index);
-    try testing.expectEqual(@as(u32, 3), result.position.?.block_index);
-}
-
-test "cborToJson converts bytes and cids to lex-JSON conventions" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const bytes_json = try cborToJson(arena, .{ .bytes = "hi" });
-    try testing.expectEqualStrings("aGk", bytes_json.object.get("$bytes").?.string);
-
-    const cid_raw = [_]u8{ 0x01, 0x71, 0x12, 0x04, 0xde, 0xad, 0xbe, 0xef };
-    const cid_json = try cborToJson(arena, .{ .cid = .{ .raw = &cid_raw } });
-    const link = cid_json.object.get("$link").?.string;
-    try testing.expect(link.len > 1 and link[0] == 'b');
-
-    const nested = try cborToJson(arena, .{ .array = &.{ .{ .unsigned = 7 }, .null, .{ .boolean = true } } });
-    try testing.expectEqual(@as(i64, 7), nested.array.items[0].integer);
-    try testing.expectEqual(json.Value.null, nested.array.items[1]);
-    try testing.expectEqual(true, nested.array.items[2].bool);
-}
-
-test "plan request carries seq bounds only when set" {
-    var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const unbounded = try planRequestBody(arena, .{ .host = "", .collections = &.{"a.b.c"} });
-    try std.testing.expect(mem.indexOf(u8, unbounded, "afterSeq") == null);
-    try std.testing.expect(mem.indexOf(u8, unbounded, "beforeSeq") == null);
-
-    const bounded = try planRequestBody(arena, .{
-        .host = "",
-        .collections = &.{"a.b.c"},
-        .after_seq = 100,
-        .before_seq = 200,
-    });
-    try std.testing.expect(mem.indexOf(u8, bounded, "\"afterSeq\":100") != null);
-    try std.testing.expect(mem.indexOf(u8, bounded, "\"beforeSeq\":200") != null);
-}
-
-test "seqBoundsForWindow admits intersecting segments, reports uncovered windows" {
-    // three sealed segments; witnessed spans overlap at the edges the way
-    // real segments do (per-connection skew)
-    const spans = [_]SegmentSpan{
-        .{ .min_seq = 1, .max_seq = 100, .min_witnessed_us = 1000, .max_witnessed_us = 2000 },
-        .{ .min_seq = 101, .max_seq = 200, .min_witnessed_us = 1900, .max_witnessed_us = 3000 },
-        .{ .min_seq = 201, .max_seq = 300, .min_witnessed_us = 2900, .max_witnessed_us = 4000 },
-    };
-
-    // a window inside the middle segment admits it plus the overlapping edge
-    const mid = seqBoundsForWindow(&spans, 2500, 2600);
-    try std.testing.expect(mid.covered);
-    try std.testing.expectEqual(@as(?u64, 100), mid.after_seq);
-    try std.testing.expectEqual(@as(?u64, 200), mid.before_seq);
-
-    // an open end means "through the sealed tip": no upper bound
-    const open = seqBoundsForWindow(&spans, 3500, null);
-    try std.testing.expect(open.covered);
-    try std.testing.expectEqual(@as(?u64, 200), open.after_seq);
-    try std.testing.expectEqual(@as(?u64, null), open.before_seq);
-
-    // a window before all witnessed spans (bootstrap-era session) is honest
-    const before_archive = seqBoundsForWindow(&spans, 10, 20);
-    try std.testing.expect(!before_archive.covered);
-
-    // straddling windows admit everything they touch
-    const straddle = seqBoundsForWindow(&spans, 1500, 3500);
-    try std.testing.expect(straddle.covered);
-    try std.testing.expectEqual(@as(?u64, 0), straddle.after_seq);
-    try std.testing.expectEqual(@as(?u64, 300), straddle.before_seq);
-}
-
-test "parsePlan handles blocks and whole-segment modes" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    const plan = try parsePlan(arena,
-        \\{"plannedThroughSeq": 12345, "sealedTipSeq": 12000, "segments": [
-        \\  {"name": "seg_0000000001", "index": 0, "checksum": "abc", "minSeq": 1, "maxSeq": 100,
-        \\   "mode": "blocks", "blocks": [{"first": 0, "last": 2}, {"first": 5, "last": 5}]},
-        \\  {"name": "seg_0000000002", "index": 1, "checksum": "def", "minSeq": 101, "maxSeq": 200,
-        \\   "mode": "segment"}
-        \\]}
-    );
-
-    try testing.expectEqual(@as(u64, 12345), plan.planned_through_seq);
-    try testing.expectEqual(@as(u64, 12000), plan.sealed_tip_seq);
-    try testing.expectEqual(@as(usize, 2), plan.segments.len);
-    try testing.expectEqualStrings("seg_0000000001", plan.segments[0].name);
-    try testing.expectEqual(@as(u32, 0), plan.segments[0].index);
-    try testing.expectEqual(@as(u64, 0xabc), plan.segments[0].checksum.?);
-    try testing.expectEqual(@as(u32, 1), plan.segments[1].index);
-    const ranges = plan.segments[0].blocks.?;
-    try testing.expectEqual(@as(u32, 0), ranges[0].first);
-    try testing.expectEqual(@as(u32, 2), ranges[0].last);
-    try testing.expectEqual(@as(u32, 5), ranges[1].first);
-    try testing.expect(plan.segments[1].blocks == null);
-}
-
-test "deliverSegment walks header and block index" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    // one empty block, hand-built zstd frame as in the frame test
-    const frame = [_]u8{ 0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x04, 0x21, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
-
-    var out: std.Io.Writer.Allocating = .init(arena);
-    const w = &out.writer;
-    try w.writeAll("jss0");
-    try w.splatByteAll(0, 10); // checksum u64 + version u16
-    try w.writeInt(u32, 1, .little); // block_count @14
-    try w.splatByteAll(0, 72); // through offset 90
-    const frame_offset: u64 = header_size;
-    const block_index_offset: u64 = header_size + 8 + frame.len;
-    try w.writeInt(u64, block_index_offset, .little); // block_index_offset @90
-    try w.splatByteAll(0, 158); // reserved → header complete @256
-    try w.writeInt(u64, frame.len, .little); // block length prefix
-    try w.writeAll(&frame);
-    // block index entry
-    try w.writeInt(u64, frame_offset, .little);
-    try w.writeInt(u32, frame.len, .little);
-    try w.writeInt(u32, 4, .little); // uncompressed_size
-    try w.writeInt(u32, 0, .little); // event_count
-    try w.splatByteAll(0, 32); // seq/witnessed bounds
-
-    const NoopHandler = struct {
-        pub fn onEvent(_: *@This(), _: Event) void {}
-    };
-    var handler = NoopHandler{};
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    const seg = PlannedSegment{ .name = "seg_test", .index = 0, .checksum = null, .blocks = null };
-    try deliverSegment(arena, out.written(), .{ .host = "https://example.test" }, seg, &handler, &result);
-    try testing.expectEqual(@as(u64, 0), result.events_delivered);
-    try testing.expectEqual(@as(u64, 1), result.blocks_decoded);
-}
-
-test "skippedByResume skips through the resume position, nothing after" {
-    const sa = Position{ .segment_index = 5, .block_index = 10 };
-    try testing.expect(skippedByResume(sa, 4, 999));
-    try testing.expect(skippedByResume(sa, 5, 9));
-    try testing.expect(skippedByResume(sa, 5, 10));
-    try testing.expect(!skippedByResume(sa, 5, 11));
-    try testing.expect(!skippedByResume(sa, 6, 0));
-    try testing.expect(!skippedByResume(null, 0, 0));
-}
-
-test "verifySegment checks the jss self-checksum and the plan's" {
-    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
-    defer arena_state.deinit();
-    const arena = arena_state.allocator();
-
-    // minimal sealed segment: 256B header + 4-byte footer
-    var out: std.Io.Writer.Allocating = .init(arena);
-    const w = &out.writer;
-    try w.writeAll("jss0");
-    try w.splatByteAll(0, 54); // checksum + version..offset 58
-    try w.writeInt(u64, header_size, .little); // footer_offset @58
-    try w.splatByteAll(0, 190); // rest of header
-    try w.writeAll("foot");
-    const body = out.written();
-
-    var hasher = std.hash.XxHash3.init(0);
-    hasher.update(body[12..header_size]);
-    hasher.update(body[header_size..]);
-    const checksum = hasher.final();
-    mem.writeInt(u64, body[4..12], checksum, .little);
-
-    try verifySegment(body, checksum);
-    try verifySegment(body, null);
-    try testing.expectError(error.ChecksumMismatch, verifySegment(body, checksum + 1));
-
-    body[body.len - 1] ^= 1; // corrupt the footer
-    try testing.expectError(error.ChecksumMismatch, verifySegment(body, checksum));
-    body[body.len - 1] ^= 1;
-
-    mem.writeInt(u64, body[4..12], 0, .little); // unsealed
-    try testing.expectError(error.SegmentUnsealed, verifySegment(body, null));
-}
-
-test "deliverSegment rejects bad magic" {
-    const NoopHandler = struct {
-        pub fn onEvent(_: *@This(), _: Event) void {}
-    };
-    var handler = NoopHandler{};
-    var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 };
-    const bogus = [_]u8{0} ** header_size;
-    try testing.expectError(error.MalformedSegment, deliverSegment(
-        testing.allocator,
-        &bogus,
-        .{ .host = "https://example.test" },
-        .{ .name = "seg_test", .index = 0, .checksum = null, .blocks = null },
-        &handler,
-        &result,
-    ));
-}
diff --git a/src/internal/streaming/zstd.zig b/src/internal/streaming/zstd.zig
deleted file mode 100644
index fd7d57a..0000000
--- a/src/internal/streaming/zstd.zig
+++ /dev/null
@@ -1,83 +0,0 @@
-//! minimal libzstd decode bindings
-//!
-//! backed by the vendored facebook/zstd v1.5.7 source (pinned by url+hash in
-//! build.zig.zon, decode-only subset compiled in build.zig — no system
-//! library). two reasons this exists instead of std.compress.zstd:
-//! measured ~20x faster on jss archive blocks, and libzstd verifies the
-//! frame content checksum (present on every jss block), which std does not
-//! implement.
-
-const std = @import("std");
-const Allocator = std.mem.Allocator;
-
-const c = struct {
-    extern fn ZSTD_getFrameContentSize(src: [*]const u8, src_size: usize) c_ulonglong;
-    extern fn ZSTD_decompress(dst: [*]u8, dst_cap: usize, src: [*]const u8, src_len: usize) usize;
-    extern fn ZSTD_isError(code: usize) c_uint;
-
-    const contentsize_unknown: c_ulonglong = std.math.maxInt(c_ulonglong); // (ull)-1
-    const contentsize_error: c_ulonglong = std.math.maxInt(c_ulonglong) - 1; // (ull)-2
-};
-
-/// jss spec cap on a decompressed block; also the zstd-bomb guard
-pub const max_decompressed: usize = 1 << 30;
-
-pub const DecompressError = error{
-    /// frame header carries no content size — jss writers always set it
-    UnknownFrameSize,
-    /// malformed frame, size lie, or content checksum mismatch
-    DecompressFailed,
-    BlockOversize,
-    OutOfMemory,
-};
-
-/// decompress one frame whose header declares its content size (all jss
-/// blocks do). libzstd verifies the frame content checksum when present.
-pub fn decompressAlloc(allocator: Allocator, frame: []const u8) DecompressError![]u8 {
-    if (frame.len == 0) return error.DecompressFailed;
-    const size = c.ZSTD_getFrameContentSize(frame.ptr, frame.len);
-    if (size == c.contentsize_unknown) return error.UnknownFrameSize;
-    if (size == c.contentsize_error) return error.DecompressFailed;
-    if (size > max_decompressed) return error.BlockOversize;
-
-    const dst = try allocator.alloc(u8, @intCast(size));
-    errdefer allocator.free(dst);
-    const written = c.ZSTD_decompress(dst.ptr, dst.len, frame.ptr, frame.len);
-    if (c.ZSTD_isError(written) != 0 or written != dst.len) return error.DecompressFailed;
-    return dst;
-}
-
-// === tests ===
-
-const testing = std.testing;
-
-// hand-built frame: magic, FHD (single-segment | checksum), content size 4,
-// one raw last-block of "zat!", XXH64(seed 0) low 32 bits as content checksum
-fn testFrame(buf: *[17]u8, corrupt_checksum: bool) []const u8 {
-    const content = "zat!";
-    buf[0..4].* = .{ 0x28, 0xb5, 0x2f, 0xfd };
-    buf[4] = 0x20 | 0x04; // single_segment + content_checksum
-    buf[5] = content.len;
-    buf[6..9].* = .{ 0x21, 0x00, 0x00 }; // last block, raw, size 4
-    buf[9..13].* = content.*;
-    const check: u32 = @truncate(std.hash.XxHash64.hash(0, content));
-    std.mem.writeInt(u32, buf[13..17], if (corrupt_checksum) check ^ 1 else check, .little);
-    return buf;
-}
-
-test "decompressAlloc round-trips a checksummed frame" {
-    var buf: [17]u8 = undefined;
-    const out = try decompressAlloc(testing.allocator, testFrame(&buf, false));
-    defer testing.allocator.free(out);
-    try testing.expectEqualStrings("zat!", out);
-}
-
-test "decompressAlloc rejects a corrupted content checksum" {
-    var buf: [17]u8 = undefined;
-    try testing.expectError(error.DecompressFailed, decompressAlloc(testing.allocator, testFrame(&buf, true)));
-}
-
-test "decompressAlloc rejects garbage" {
-    try testing.expectError(error.DecompressFailed, decompressAlloc(testing.allocator, "not a zstd frame"));
-    try testing.expectError(error.DecompressFailed, decompressAlloc(testing.allocator, ""));
-}
diff --git a/src/root.zig b/src/root.zig
index 017920d..d46d509 100644
--- a/src/root.zig
+++ b/src/root.zig
@@ -61,13 +61,15 @@ pub const CommitAction = sync.CommitAction;
 pub const EventKind = sync.EventKind;
 pub const AccountStatus = sync.AccountStatus;
 
-// jetstream
+// jetstream (the v1 /subscribe client). DEPRECATED HOME: the jetstream
+// service SDK now lives at tangled.org/zat.dev/jetstream (ArchiveBackfill
+// moved there; the subscribeEvents v2 client lands there). this v1 client
+// stays functional here until its consumers migrate.
 pub const jetstream = @import("internal/streaming/jetstream.zig");
 pub const JetstreamClient = jetstream.JetstreamClient;
 pub const JetstreamEvent = jetstream.Event;
 
 // firehose (raw CBOR event stream)
-pub const ArchiveBackfill = @import("internal/streaming/archive_backfill.zig");
 
 pub const firehose = @import("internal/streaming/firehose.zig");
 pub const FirehoseClient = firehose.FirehoseClient;
@@ -106,8 +108,6 @@ comptime {
         _ = @import("internal/repo/mst.zig");
         _ = @import("internal/repo/mst_test.zig");
         _ = @import("internal/repo/repo.zig");
-        _ = @import("internal/streaming/archive_backfill.zig");
-        _ = @import("internal/streaming/zstd.zig");
         _ = @import("internal/streaming/firehose.zig");
         _ = @import("internal/streaming/jetstream.zig");
         _ = @import("internal/streaming/sync.zig");