diff --git a/.tangled/workflows/publish-docs.yml b/.tangled/workflows/publish-docs.yml
index 29d0ce2..a68002c 100644
--- a/.tangled/workflows/publish-docs.yml
+++ b/.tangled/workflows/publish-docs.yml
@@ -33,7 +33,7 @@ steps:
         publish=0
         for f in $(git diff-tree --no-commit-id --name-only -r HEAD); do
           case "$f" in
-            README.md|CHANGELOG.md|docs/*|devlog/*|scripts/publish-docs.zig|.tangled/workflows/publish-docs.yml)
+            README.md|CHANGELOG.md|docs/*|devlog/*|examples/*.md|scripts/publish-docs.zig|.tangled/workflows/publish-docs.yml)
               publish=1 ;;
           esac
         done
diff --git a/build.zig b/build.zig
index 828cddf..e5ff85c 100644
--- a/build.zig
+++ b/build.zig
@@ -189,6 +189,23 @@ 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",
         .root_module = b.createModule(.{
diff --git a/examples/07_streamplace_chat.md b/examples/07_streamplace_chat.md
new file mode 100644
index 0000000..bba3b98
--- /dev/null
+++ b/examples/07_streamplace_chat.md
@@ -0,0 +1,67 @@
+# 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/3ms6zywt7fb2e'
+```
+
+The second form reads that exact `place.stream.livestream` record from the
+streamer's PDS. Its TID supplies the lower replay boundary and its `endedAt`
+field supplies the upper chat boundary. The consumer still scans Jetstream up
+to the time the command began so records that reached the relay late are not
+lost. It then reports the count and exits:
+
+```text
+Watching @toni.bsky.team's interview at 11am!
+at://did:plc:.../place.stream.livestream/3ms6zywt7fb2e
+replaying this historical broadcast's chat...
+
+[2026-08-03T17:37:53.900Z] did:plc:...: heloo hello
+[2026-08-03T17:40:04.633Z] did:plc:...: oo this is groovy
+...
+replay complete: 26 messages
+```
+
+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`.
+
+Jetstream's replay window is finite. Exact historical selection prevents chat
+from adjacent broadcasts from leaking into the result, but it cannot recover a
+broadcast older than the relay's retained event history. A durable archival
+tool should run continuously and persist 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
new file mode 100644
index 0000000..387807d
--- /dev/null
+++ b/examples/07_streamplace_chat.zig
@@ -0,0 +1,193 @@
+//! replay a specific Streamplace broadcast's chat, or follow the latest one.
+//!
+//! 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 ChatPrinter = struct {
+    streamer_did: []const u8,
+    stream_rkey: []const u8,
+    ended_at: ?[]const u8,
+    count: usize = 0,
+
+    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 a broadcast
+        // before the selected one, even if they arrive late during replay.
+        if (std.mem.order(u8, commit.rkey, self.stream_rkey) == .lt) return;
+
+        const created_at = zat.json.getString(record, "createdAt") orelse return;
+        if (self.ended_at) |end| {
+            // Streamplace writes canonical UTC datetimes, so their ISO-8601
+            // spelling has the same order as the instants they represent.
+            if (std.mem.order(u8, created_at, end) == .gt) 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} 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 stream_rkey = at_uri.rkey() orelse return error.MissingLivestreamRkey;
+    const stream_tid = zat.Tid.parse(stream_rkey) orelse return error.InvalidLivestreamRkey;
+
+    // Read the selected record from its authoritative PDS. Historical records
+    // carry endedAt, which gives this replay an exact upper time boundary.
+    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.0",
+    );
+    defer pds.deinit();
+    const params = [_]zat.XrpcClient.QueryParam{
+        .{ .name = "repo", .value = streamer_did },
+        .{ .name = "collection", .value = "place.stream.livestream" },
+        .{ .name = "rkey", .value = stream_rkey },
+    };
+    const get_record = zat.Nsid.parse("com.atproto.repo.getRecord").?;
+    var response = try pds.queryParams(get_record, &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 title = zat.json.getString(parsed.value, "value.title") orelse "untitled stream";
+    const ended_at = zat.json.getString(parsed.value, "value.endedAt");
+
+    std.debug.print("{s}\n{s}\n{s}\n\n", .{
+        title,
+        livestream_uri,
+        if (ended_at == null)
+            "replaying chat, then following live..."
+        else
+            "replaying this historical broadcast's chat...",
+    });
+
+    var printer = ChatPrinter{
+        .streamer_did = streamer_did,
+        .stream_rkey = stream_rkey,
+        .ended_at = ended_at,
+    };
+    var client = zat.JetstreamClient.init(init.io, allocator, .{
+        .wanted_collections = &.{"place.stream.chat.message"},
+        // The livestream record's TID is its creation time in microseconds.
+        // Rewind one second so events at the boundary cannot be skipped.
+        .cursor = @as(i64, @intCast(stream_tid.timestamp())) - 1_000_000,
+        // A historical query replays through the moment this process began,
+        // catching records that reached Jetstream late, and then exits.
+        .end_cursor = if (ended_at != null)
+            @intCast(@divFloor(
+                std.Io.Timestamp.now(init.io, .real).nanoseconds,
+                std.time.ns_per_us,
+            ))
+        else
+            null,
+    });
+    defer client.deinit();
+
+    try client.subscribe(&printer);
+    std.debug.print("\nreplay complete: {d} messages\n", .{printer.count});
+}
+
+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.0",
+    );
+    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);
+}
diff --git a/src/internal/streaming/jetstream.zig b/src/internal/streaming/jetstream.zig
index eba2007..604e4db 100644
--- a/src/internal/streaming/jetstream.zig
+++ b/src/internal/streaming/jetstream.zig
@@ -40,6 +40,10 @@ pub const Options = struct {
     wanted_collections: []const []const u8 = &.{},
     wanted_dids: []const []const u8 = &.{},
     cursor: ?i64 = null,
+    /// Stop after replaying up to this Jetstream timestamp. Useful for finite
+    /// historical queries; unlike cursor, this is enforced by the client and
+    /// is not sent to the server.
+    end_cursor: ?i64 = null,
     max_message_size: usize = 1024 * 1024,
 };
 
@@ -174,6 +178,7 @@ pub const JetstreamClient = struct {
             log.info("connecting to host {d}/{d}: {s}", .{ effective_index + 1, self.options.hosts.len, host });
 
             self.connectAndRead(host, handler) catch |err| {
+                if (err == error.EndCursorReached) return;
                 if (comptime @hasDecl(@TypeOf(handler.*), "onError")) {
                     handler.onError(err);
                 } else {
@@ -269,6 +274,10 @@ fn WsHandler(comptime H: type) type {
                 return;
             };
 
+            if (self.client_state.options.end_cursor) |end| {
+                if (event.timeUs() >= end) return error.EndCursorReached;
+            }
+
             self.client_state.last_time_us = event.timeUs();
             self.handler.onEvent(event);
         }
@@ -444,6 +453,31 @@ test "cursor tracking via time_us" {
     try std.testing.expect(e2.timeUs() > e1.timeUs());
 }
 
+test "end cursor stops before dispatching the boundary event" {
+    const Handler = struct {
+        count: usize = 0,
+
+        pub fn onEvent(self: *@This(), _: Event) void {
+            self.count += 1;
+        }
+    };
+
+    var client = JetstreamClient.init(std.Options.debug_io, std.testing.allocator, .{
+        .end_cursor = 200,
+    });
+    var handler = Handler{};
+    var ws_handler = WsHandler(Handler){
+        .allocator = std.testing.allocator,
+        .handler = &handler,
+        .client_state = &client,
+    };
+
+    try std.testing.expectError(error.EndCursorReached, ws_handler.serverMessage(
+        \\{"did":"did:plc:a","time_us":200,"kind":"commit","commit":{"operation":"create","collection":"x","rkey":"1"}}
+    ));
+    try std.testing.expectEqual(@as(usize, 0), handler.count);
+}
+
 test "Event.timeUs works for all variants" {
     var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
     defer arena.deinit();