diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2ad4af9..abcfe94 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,20 @@
 # changelog
 
+## 0.1.1
+
+- the plan request now carries `kinds` (upstream WithKinds parity): the
+  planner prunes marker/sentinel blocks server-side instead of the client
+  fetching and discarding them. measured against the production archive:
+  a 3-collection account slice went 108.5s -> 5.3s (the network-wide
+  statusphere plan shrinks 83x, 140,705 -> 1,684 blocks). previously
+  `kinds` filtered client-side only.
+- `LiveClient`/`subscribe` send `Authorization: Bearer` on the websocket
+  handshake when `api_key` is set (upstream Go client parity; hosted
+  instances can gate the tail at their edge).
+- new example: `slice_sync` — one account's whole history from the
+  archive slice (the local-first pattern), plus `cutover_smoke` from the
+  seam-verification work.
+
 ## 0.1.0
 
 first tagged release of the zig jetstream service SDK — the service half
diff --git a/build.zig b/build.zig
index 0c63c32..003160d 100644
--- a/build.zig
+++ b/build.zig
@@ -115,4 +115,23 @@ pub fn build(b: *std.Build) void {
     if (b.args) |args| run_cutover.addArgs(args);
     const cutover_step = b.step("example-cutover-smoke", "live smoke: archive sweep -> gapless cutover -> live tail");
     cutover_step.dependOn(&run_cutover.step);
+
+    const slice = b.addExecutable(.{
+        .name = "example-slice-sync",
+        .root_module = b.createModule(.{
+            .root_source_file = b.path("examples/slice_sync.zig"),
+            .target = target,
+            .optimize = optimize,
+            .link_libc = true,
+            .imports = &.{
+                .{ .name = "zat", .module = zat.module("zat") },
+                .{ .name = "jetstream", .module = mod },
+            },
+        }),
+    });
+    b.installArtifact(slice);
+    const run_slice = b.addRunArtifact(slice);
+    if (b.args) |args| run_slice.addArgs(args);
+    const slice_step = b.step("example-slice-sync", "replay one account's whole history from the archive slice");
+    slice_step.dependOn(&run_slice.step);
 }
diff --git a/build.zig.zon b/build.zig.zon
index b231805..5dfa04a 100644
--- a/build.zig.zon
+++ b/build.zig.zon
@@ -1,6 +1,6 @@
 .{
     .name = .jetstream,
-    .version = "0.1.0",
+    .version = "0.1.1",
     .minimum_zig_version = "0.16.0-dev.3070+b22eb176b",
     .fingerprint = 0xee435acce6d720e8,
     .dependencies = .{
diff --git a/examples/slice_sync.zig b/examples/slice_sync.zig
new file mode 100644
index 0000000..7344f5a
--- /dev/null
+++ b/examples/slice_sync.zig
@@ -0,0 +1,94 @@
+//! slice sync: everything one account (or one app's collections) ever did,
+//! replayed from the whole-network archive in seconds — the local-first
+//! pattern from the Jetstream v2 launch thread ("backfill all of my Semble
+//! data into the browser: 12 seconds, 383 requests, 190kb over the wire").
+//!
+//! the planSnapshot request carries the did/collection filters, so the
+//! server prunes to just the blocks containing matching rows — the client
+//! never downloads the archive, only the slice.
+//!
+//! run: zig build example-slice-sync -- <handle-or-did> [collection,collection,...]
+//!      zig build example-slice-sync -- zzstoatzz.io app.bsky.feed.post
+
+const std = @import("std");
+const zat = @import("zat");
+const jetstream_sdk = @import("jetstream");
+
+const SliceStats = struct {
+    rows: usize = 0,
+    first_time_us: i64 = 0,
+    last_time_us: i64 = 0,
+    by_collection: std.StringHashMapUnmanaged(usize) = .empty,
+    arena: std.mem.Allocator,
+
+    pub fn onEvent(self: *@This(), event: jetstream_sdk.Event) bool {
+        self.rows += 1;
+        if (self.first_time_us == 0) self.first_time_us = event.time_us;
+        self.last_time_us = event.time_us;
+        if (event.payload == .commit) {
+            const gop = self.by_collection.getOrPut(self.arena, event.payload.commit.collection) catch return true;
+            if (!gop.found_existing) {
+                gop.key_ptr.* = self.arena.dupe(u8, event.payload.commit.collection) catch return true;
+                gop.value_ptr.* = 0;
+            }
+            gop.value_ptr.* += 1;
+        }
+        return true;
+    }
+};
+
+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-slice-sync -- <handle-or-did> [collections,...]\n", .{});
+        return error.MissingActor;
+    }
+
+    var did: []const u8 = args[1];
+    if (!std.mem.startsWith(u8, did, "did:")) {
+        var resolver = zat.HandleResolver.init(init.io, allocator);
+        defer resolver.deinit();
+        const resolved = try resolver.resolve(zat.Handle.parse(args[1]) orelse return error.InvalidHandle);
+        defer allocator.free(resolved);
+        did = try init.arena.allocator().dupe(u8, resolved);
+    }
+
+    // token-gated archives (stream.waow.tech is one) need a bearer key
+    const api_key: ?[]const u8 = if (std.c.getenv("JETSTREAM_API_KEY")) |v| std.mem.span(v) else null;
+
+    var collections: std.ArrayList([]const u8) = .empty;
+    if (args.len > 2) {
+        var it = std.mem.tokenizeScalar(u8, args[2], ',');
+        while (it.next()) |c| try collections.append(init.arena.allocator(), c);
+    }
+
+    std.debug.print("slicing the archive for {s} ({d} collection filters)\n", .{ did, collections.items.len });
+    const started = std.Io.Timestamp.now(init.io, .awake).nanoseconds;
+
+    var stats = SliceStats{ .arena = init.arena.allocator() };
+    try jetstream_sdk.subscribe(init.io, allocator, .{
+        .hosts = &.{"https://stream.waow.tech"},
+        .after_seq = 0,
+        .snapshot_only = true,
+        .dids = &.{did},
+        .collections = collections.items,
+        // jim's sparse-backfill recipe: kind=commit changes the PLAN, not
+        // just client filtering — without it, marker/sentinel rows admit
+        // nearly every block (measured 83x more blocks for a network-wide
+        // collection slice)
+        .kinds = &.{.commit},
+        .api_key = api_key,
+    }, &stats);
+
+    const elapsed_ms = @divFloor(std.Io.Timestamp.now(init.io, .awake).nanoseconds - started, std.time.ns_per_ms);
+    std.debug.print("\n{d} rows in {d}ms\n", .{ stats.rows, elapsed_ms });
+    var it = stats.by_collection.iterator();
+    while (it.next()) |entry| {
+        std.debug.print("  {s}: {d}\n", .{ entry.key_ptr.*, entry.value_ptr.* });
+    }
+    if (stats.rows > 0) {
+        const span_days = @divFloor(stats.last_time_us - stats.first_time_us, std.time.us_per_s * 86_400);
+        std.debug.print("  spanning {d} days of history\n", .{span_days});
+    }
+}
diff --git a/src/archive_backfill.zig b/src/archive_backfill.zig
index 42cba62..bb12080 100644
--- a/src/archive_backfill.zig
+++ b/src/archive_backfill.zig
@@ -51,6 +51,11 @@ pub const Options = struct {
     host: []const u8,
     /// collections to replay; empty = all
     collections: []const []const u8 = &.{},
+    /// plan-level kind pruning (upstream WithKinds -> plan request
+    /// "kinds"). without it the plan admits marker/sentinel blocks —
+    /// measured 83x more blocks for a network-wide collection slice.
+    /// empty = all kinds.
+    kinds: []const livedecode.Kind = &.{},
     /// dids to replay; empty = all
     dids: []const []const u8 = &.{},
     /// bearer credential for token-gated archives (Bluesky's hosted
@@ -397,6 +402,12 @@ fn planRequestBody(arena: Allocator, options: Options) ![]const u8 {
         try stringify.objectField("beforeSeq");
         try stringify.write(seq);
     }
+    if (options.kinds.len > 0) {
+        try stringify.objectField("kinds");
+        try stringify.beginArray();
+        for (options.kinds) |k| try stringify.write(@tagName(k));
+        try stringify.endArray();
+    }
     try stringify.endObject();
     return body.written();
 }
@@ -1120,6 +1131,16 @@ test "cborToJson converts bytes and cids to lex-JSON conventions" {
     try testing.expectEqual(true, nested.array.items[2].bool);
 }
 
+test "plan request carries kinds only when set" {
+    var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
+    defer arena_state.deinit();
+    const arena = arena_state.allocator();
+    const with_kinds = try planRequestBody(arena, .{ .host = "h", .kinds = &.{ .commit, .sync } });
+    try testing.expect(std.mem.indexOf(u8, with_kinds, "\"kinds\":[\"commit\",\"sync\"]") != null);
+    const without = try planRequestBody(arena, .{ .host = "h" });
+    try testing.expect(std.mem.indexOf(u8, without, "kinds") == null);
+}
+
 test "plan request carries seq bounds only when set" {
     var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
     defer arena_state.deinit();
diff --git a/src/client.zig b/src/client.zig
index 83559b4..1041a57 100644
--- a/src/client.zig
+++ b/src/client.zig
@@ -97,8 +97,10 @@ pub const Options = struct {
     kinds: []const livedecode.Kind = &.{},
     collections: []const []const u8 = &.{},
     dids: []const []const u8 = &.{},
-    /// bearer credential for token-gated archives (planSnapshot/getSegment/
-    /// getBlock only; the live tail never needs it)
+    /// bearer credential for token-gated archives and, matching the
+    /// upstream Go client, sent on the live websocket handshake too
+    /// (hosted instances can gate the tail at their edge; the OSS server
+    /// does not)
     api_key: ?[]const u8 = null,
     /// block fetches in flight during the archive sweep (see ArchiveBackfill)
     concurrency: usize = 4,
@@ -257,6 +259,7 @@ fn runOnHost(
             .host = host,
             .collections = options.collections,
             .dids = options.dids,
+            .kinds = options.kinds,
             .api_key = options.api_key,
             .after_seq = cursor,
             .concurrency = options.concurrency,
@@ -322,6 +325,7 @@ fn liveOptions(options: Options, host: []const u8, cursor: ?u64, dedup_floor: u6
         .collections = options.collections,
         .dids = options.dids,
         .zstd_dict = dict,
+        .api_key = options.api_key,
         .stall_timeout_ns = options.failover_stall_ns,
     };
     if (options.backoff_min_ns > 0) out.backoff_min_ns = options.backoff_min_ns;
diff --git a/src/live.zig b/src/live.zig
index b71198b..43691b8 100644
--- a/src/live.zig
+++ b/src/live.zig
@@ -73,6 +73,12 @@ pub const Options = struct {
     /// then degrades to an uncompressed tail.
     refetch_dict: ?*const fn (ctx: ?*anyopaque, allocator: Allocator) ?[]u8 = null,
     refetch_dict_ctx: ?*anyopaque = null,
+    /// bearer credential sent as an Authorization header on the websocket
+    /// handshake, matching the upstream Go client (engine.go: the
+    /// negotiation transport carries the key on every request including
+    /// the live dial). the OSS server never gates the live tail, but
+    /// hosted instances can at their edge. null = no auth sent.
+    api_key: ?[]const u8 = null,
     /// declare the session stalled when no EVENT arrives for this long
     /// (surfaced as transient error.LiveStalled → reconnect/failover).
     /// event-based, not frame-based: a connected-but-silent server that
@@ -198,12 +204,19 @@ pub const LiveClient = struct {
         });
         defer client.deinit();
 
-        var headers_buf: [512]u8 = undefined;
-        const headers = try std.fmt.bufPrint(
-            &headers_buf,
-            "Host: {s}\r\nSec-WebSocket-Protocol: {s}\r\n",
-            .{ target.host, subprotocol },
-        );
+        var headers_buf: [1024]u8 = undefined;
+        const headers = if (self.options.api_key) |key|
+            try std.fmt.bufPrint(
+                &headers_buf,
+                "Host: {s}\r\nSec-WebSocket-Protocol: {s}\r\nAuthorization: Bearer {s}\r\n",
+                .{ target.host, subprotocol, key },
+            )
+        else
+            try std.fmt.bufPrint(
+                &headers_buf,
+                "Host: {s}\r\nSec-WebSocket-Protocol: {s}\r\n",
+                .{ target.host, subprotocol },
+            );
         client.handshake(path, .{ .headers = headers }) catch |err| {
             if (err == error.InvalidHandshakeResponse) {
                 if (client.handshake_failure) |failure| return classifyRejection(failure);