diff --git a/docs/configuration.md b/docs/configuration.md
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -279,6 +279,27 @@
 credentials all produce the same 401. Submitted paths are canonicalized through
 symlinks and must resolve to regular files inside the configured directory.
 
+### archive key ring
+
+`--archive-api-keys-file=<path>` (`JETSTREAM_ARCHIVE_API_KEYS_FILE`) adds
+named, metered archive credentials beside the fleet key:
+
+```
+# name:key[:mbps]     mbps omitted or 0 = unmetered
+evelyn:some-long-random-secret:8
+ci:another-secret
+```
+
+- keys authorize the archive endpoints exactly like `--archive-api-key`
+  (which stays valid and unmetered)
+- `mbps` is a per-key token bucket (4s burst) charged on served bytes;
+  a key in deficit gets `429 RateLimitExceeded` until it refills — the
+  SDK's fetch retry already backs off on 429
+- **revocation = delete the line.** the file's mtime is rechecked at most
+  once per second on the request path; no restart, no signal
+- an unreadable/missing file revokes every ring key (fail closed) and
+  logs a warning; the fleet key is unaffected
+
 ## OpenTelemetry
 
 Stream installs the same process-wide tracing provider as pinned Jetstream V2.
diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md
--- a/docs/semantic-parity.md
+++ b/docs/semantic-parity.md
@@ -200,3 +200,20 @@
 
 The previous `4d163e8` receipt remains historical evidence for that artifact.
 It is not a current admission and must not be copied forward.
+
+## archive key ring (2026-08-17, extension)
+
+upstream OSS has no archive auth at all — its only bearer surface is the
+single timestamp-import token (xrpcapi/auth.go). Bluesky's hosted
+instances gate and meter archives in a proprietary gateway in front
+(2 MB/s per key by default; the gateway exists to adjust limits per key —
+alex.bsky.team, 2026-08-17). stream's single `--archive-api-key` was
+already an extension mirroring that hosted behavior; the key ring
+(`--archive-api-keys-file`, serve/api_keys.zig) extends it into an
+in-process approximation of the gateway: named keys (`name:key[:mbps]`
+lines), per-key token-bucket byte budgets (429 + retry semantics the SDK
+already honors), and revocation by deleting a line (mtime-based reload,
+no restart). the fleet key remains valid and unmetered. rationale:
+metering protects compaction/live-delivery from uncapped archive drains
+on a ~200 MB/s volume; revocation avoids rotating the fleet credential
+per consumer.
diff --git a/src/main.zig b/src/main.zig
--- a/src/main.zig
+++ b/src/main.zig
@@ -14,6 +14,7 @@
 const ingest = @import("internal/ingest/ingest.zig");
 const verify = @import("internal/ingest/verify.zig");
 const xrpcapi = @import("internal/serve/xrpcapi.zig");
+const api_keys_mod = @import("internal/serve/api_keys.zig");
 const metrics = @import("internal/runtime/metrics.zig");
 const meta_store = @import("internal/storage/meta_store.zig");
 const pipeline_mod = @import("internal/ingest/pipeline.zig");
@@ -721,6 +722,15 @@
     xrpc_api.import = &import_manager;
     xrpc_api.import_token = cfg.timestamp_import_token;
     xrpc_api.archive_api_key = cfg.archive_api_key;
+    var key_ring: ?api_keys_mod.KeyRing = if (cfg.archive_api_keys_file.len > 0)
+        api_keys_mod.KeyRing.init(allocator, cfg.archive_api_keys_file)
+    else
+        null;
+    defer if (key_ring) |*ring| ring.deinit();
+    if (key_ring) |*ring| {
+        xrpc_api.key_ring = ring;
+        log.info("archive key ring enabled: {s} (revocation = edit the file)", .{cfg.archive_api_keys_file});
+    }
 
     // Killed getRepo downloads can leave incomplete scratch files. Bootstrap
     // owns backfill/, while steady repair has a separate root so a legitimate
diff --git a/src/internal/runtime/cli.zig b/src/internal/runtime/cli.zig
--- a/src/internal/runtime/cli.zig
+++ b/src/internal/runtime/cli.zig
@@ -205,6 +205,7 @@
     /// hosted Bluesky instances' edge behavior (Jetstream v2, 2026-08-13:
     /// live tail and dictionary open, archive behind a token).
     archive_api_key: []const u8 = "",
+    archive_api_keys_file: []const u8 = "",
     timestamp_import_dir: []const u8 = "",
 
     pub const overrides = .{
@@ -469,6 +470,7 @@
     /// hosted Bluesky instances' edge behavior (Jetstream v2, 2026-08-13:
     /// live tail and dictionary open, archive behind a token).
     archive_api_key: []const u8 = "",
+    archive_api_keys_file: []const u8 = "",
     timestamp_import_dir_arg: ?[]const u8 = null,
     repo_action_rate_limits: bool = true,
     store_fault_prefix: ?[]const u8 = null,
@@ -677,6 +679,8 @@
             cfg.plan_config.whole_segment_threshold = try std.fmt.parseFloat(f64, arg["--plan-whole-segment-threshold=".len..]);
         } else if (std.mem.startsWith(u8, arg, "--archive-api-key=")) {
             cfg.archive_api_key = arg["--archive-api-key=".len..];
+        } else if (std.mem.startsWith(u8, arg, "--archive-api-keys-file=")) {
+            cfg.archive_api_keys_file = arg["--archive-api-keys-file=".len..];
         } else if (std.mem.startsWith(u8, arg, "--timestamp-import-token=")) {
             cfg.timestamp_import_token = arg["--timestamp-import-token=".len..];
         } else if (std.mem.startsWith(u8, arg, "--timestamp-import-dir=")) {
diff --git a/src/internal/runtime/environment.zig b/src/internal/runtime/environment.zig
--- a/src/internal/runtime/environment.zig
+++ b/src/internal/runtime/environment.zig
@@ -50,6 +50,7 @@
     .{ .env = "JETSTREAM_TIMESTAMP_IMPORT_TOKEN", .flag = "timestamp-import-token" },
     .{ .env = "JETSTREAM_TIMESTAMP_IMPORT_DIR", .flag = "timestamp-import-dir" },
     .{ .env = "JETSTREAM_ARCHIVE_API_KEY", .flag = "archive-api-key" },
+    .{ .env = "JETSTREAM_ARCHIVE_API_KEYS_FILE", .flag = "archive-api-keys-file" },
 };
 
 pub fn processEntries(allocator: std.mem.Allocator) ![]const []const u8 {
@@ -133,7 +134,7 @@
 }
 
 test "unknown Jetstream variables are sorted, deduplicated, and scoped" {
-    try std.testing.expectEqual(@as(usize, 39), mappings.len);
+    try std.testing.expectEqual(@as(usize, 40), mappings.len);
     const entries = [_][]const u8{
         "JETSTREAM_ZZZ=1",
         "JETSTREAM_ADDR=127.0.0.1:0",
diff --git a/src/internal/serve/api_keys.zig b/src/internal/serve/api_keys.zig
new file mode 100644
--- /dev/null
+++ b/src/internal/serve/api_keys.zig
@@ -0,0 +1,260 @@
+//! named, revocable archive API keys with per-key byte-rate budgets.
+//!
+//! extension beyond upstream: the OSS server has NO archive auth at all
+//! (its only bearer surface is the single timestamp-import token,
+//! xrpcapi/auth.go); Bluesky gates and meters hosted archives in a
+//! proprietary gateway in front — 2 MB/s per key by default
+//! (alex.bsky.team, 2026-08-17), with the gateway existing precisely to
+//! adjust per-key limits. this ring approximates that gateway in-process:
+//! named keys, per-key byte budgets, and revocation without rotating the
+//! fleet credential. recorded in docs/semantic-parity.md.
+//!
+//! the key file is one entry per line:
+//!
+//!     name:key[:mbps]
+//!
+//! `#` comments and blank lines are ignored; a missing or 0 mbps means
+//! unmetered. revocation is deleting the line — the file's mtime is
+//! re-checked (at most once per second) on the request path, so no restart
+//! or signal is needed. the legacy single fleet key (--archive-api-key)
+//! keeps working beside the ring, unmetered.
+//!
+//! metering is a token bucket in bytes, charged AFTER each response (the
+//! bucket may go negative; the next request is refused with 429 until the
+//! deficit refills). burst capacity is 4 seconds of rate. presented keys
+//! are compared as sha256 digests in constant time, same as authorized().
+
+const std = @import("std");
+
+const Io = std.Io;
+const Allocator = std.mem.Allocator;
+const log = std.log.scoped(.api_keys);
+
+pub const burst_seconds = 4;
+/// stat the key file at most this often
+const reload_check_interval_us: i64 = std.time.us_per_s;
+
+pub const Entry = struct {
+    name: []const u8,
+    key_hash: [32]u8,
+    /// bytes per second; 0 = unmetered
+    rate_bps: u64,
+    /// token bucket level; negative = in deficit
+    bucket_bytes: i64,
+    last_refill_us: i64,
+};
+
+pub const KeyRing = struct {
+    allocator: Allocator,
+    path: []const u8,
+    mutex: Io.Mutex = .init,
+    entries: std.ArrayList(Entry) = .empty,
+    arena: std.heap.ArenaAllocator,
+    loaded_mtime_us: i64 = std.math.minInt(i64),
+    last_check_us: i64 = std.math.minInt(i64),
+
+    pub fn init(allocator: Allocator, path: []const u8) KeyRing {
+        return .{
+            .allocator = allocator,
+            .path = path,
+            .arena = std.heap.ArenaAllocator.init(allocator),
+        };
+    }
+
+    pub fn deinit(self: *KeyRing) void {
+        self.entries.deinit(self.allocator);
+        self.arena.deinit();
+        self.* = undefined;
+    }
+
+    /// resolve a bearer header to a live key entry. reloads the file when
+    /// its mtime changed (checked at most once per second), so adding or
+    /// deleting a line takes effect without a restart. returns null for
+    /// missing/unknown/revoked keys.
+    pub fn authorize(self: *KeyRing, io: Io, authorization: ?[]const u8, now_us: i64) ?*Entry {
+        const raw = authorization orelse return null;
+        const prefix = "Bearer ";
+        if (raw.len <= prefix.len or !std.ascii.eqlIgnoreCase(raw[0..prefix.len], prefix)) return null;
+        var presented_hash: [32]u8 = undefined;
+        std.crypto.hash.sha2.Sha256.hash(raw[prefix.len..], &presented_hash, .{});
+
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        self.maybeReload(io, now_us);
+        for (self.entries.items) |*entry| {
+            if (std.crypto.timing_safe.eql([32]u8, entry.key_hash, presented_hash)) return entry;
+        }
+        return null;
+    }
+
+    /// deficit check before serving. returns seconds to wait when the
+    /// key's bucket is negative (serve a 429), null when clear to serve.
+    pub fn precheck(self: *KeyRing, io: Io, entry: *Entry, now_us: i64) ?u64 {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        if (entry.rate_bps == 0) return null;
+        refill(entry, now_us);
+        if (entry.bucket_bytes >= 0) return null;
+        const deficit: u64 = @intCast(-entry.bucket_bytes);
+        return @max(1, deficit / entry.rate_bps);
+    }
+
+    /// charge served bytes to the key after the response. the bucket may
+    /// go negative — one oversized response is always allowed, and the
+    /// deficit gates the next one.
+    pub fn charge(self: *KeyRing, io: Io, entry: *Entry, bytes: u64, now_us: i64) void {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        if (entry.rate_bps == 0) return;
+        refill(entry, now_us);
+        entry.bucket_bytes -= @intCast(@min(bytes, std.math.maxInt(i64)));
+    }
+
+    fn refill(entry: *Entry, now_us: i64) void {
+        const elapsed_us = now_us - entry.last_refill_us;
+        entry.last_refill_us = now_us;
+        if (elapsed_us <= 0) return;
+        const cap: i64 = @intCast(entry.rate_bps * burst_seconds);
+        const gained: i64 = @intCast(@min(
+            (@as(u128, @intCast(elapsed_us)) * entry.rate_bps) / std.time.us_per_s,
+            @as(u128, @intCast(std.math.maxInt(i64))),
+        ));
+        entry.bucket_bytes = @min(entry.bucket_bytes +| gained, cap);
+    }
+
+    fn maybeReload(self: *KeyRing, io: Io, now_us: i64) void {
+        if (now_us -| self.last_check_us < reload_check_interval_us) return;
+        self.last_check_us = now_us;
+        const cwd = Io.Dir.cwd();
+        var file = cwd.openFile(io, self.path, .{}) catch |err| {
+            // a missing/unreadable file revokes everything — fail closed,
+            // loudly
+            if (self.entries.items.len > 0) {
+                log.warn("key file {s} unreadable ({s}); all ring keys revoked", .{ self.path, @errorName(err) });
+                self.entries.clearRetainingCapacity();
+                self.loaded_mtime_us = std.math.minInt(i64);
+            }
+            return;
+        };
+        defer file.close(io);
+        const stat = file.stat(io) catch return;
+        const mtime_us: i64 = stat.mtime.toMicroseconds();
+        if (mtime_us == self.loaded_mtime_us) return;
+
+        const bytes = cwd.readFileAlloc(io, self.path, self.allocator, .limited(1 << 20)) catch |err| {
+            log.warn("key file {s} read failed: {s}", .{ self.path, @errorName(err) });
+            return;
+        };
+        defer self.allocator.free(bytes);
+        self.loadFromSlice(bytes) catch |err| {
+            log.warn("key file {s} parse failed: {s}; keeping previous ring", .{ self.path, @errorName(err) });
+            return;
+        };
+        self.loaded_mtime_us = mtime_us;
+        log.info("archive key ring loaded: {d} keys from {s}", .{ self.entries.items.len, self.path });
+    }
+
+    /// parse `name:key[:mbps]` lines, replacing the ring. buckets of keys
+    /// that survive the reload keep their level (revoke-and-re-add resets).
+    pub fn loadFromSlice(self: *KeyRing, bytes: []const u8) !void {
+        var fresh: std.ArrayList(Entry) = .empty;
+        errdefer fresh.deinit(self.allocator);
+        var next_arena = std.heap.ArenaAllocator.init(self.allocator);
+        errdefer next_arena.deinit();
+
+        var lines = std.mem.tokenizeScalar(u8, bytes, '\n');
+        while (lines.next()) |raw_line| {
+            const line = std.mem.trim(u8, raw_line, " \t\r");
+            if (line.len == 0 or line[0] == '#') continue;
+            var parts = std.mem.splitScalar(u8, line, ':');
+            const name = parts.next() orelse return error.MalformedKeyLine;
+            const key = parts.next() orelse return error.MalformedKeyLine;
+            if (name.len == 0 or key.len == 0) return error.MalformedKeyLine;
+            const rate_bps: u64 = if (parts.next()) |mbps_str|
+                (std.fmt.parseInt(u64, mbps_str, 10) catch return error.MalformedKeyLine) * 1000 * 1000
+            else
+                0;
+            var hash: [32]u8 = undefined;
+            std.crypto.hash.sha2.Sha256.hash(key, &hash, .{});
+            var entry: Entry = .{
+                .name = try next_arena.allocator().dupe(u8, name),
+                .key_hash = hash,
+                .rate_bps = rate_bps,
+                .bucket_bytes = @intCast(rate_bps * burst_seconds),
+                .last_refill_us = 0,
+            };
+            for (self.entries.items) |old| {
+                if (std.crypto.timing_safe.eql([32]u8, old.key_hash, hash)) {
+                    entry.bucket_bytes = old.bucket_bytes;
+                    entry.last_refill_us = old.last_refill_us;
+                    break;
+                }
+            }
+            try fresh.append(self.allocator, entry);
+        }
+        self.entries.deinit(self.allocator);
+        self.entries = fresh;
+        self.arena.deinit();
+        self.arena = next_arena;
+    }
+};
+
+// === tests ===
+
+const testing = std.testing;
+
+test "loadFromSlice: names, rates, comments, revocation, bucket carry-over" {
+    var ring = KeyRing.init(testing.allocator, "/nonexistent");
+    defer ring.deinit();
+
+    try ring.loadFromSlice(
+        \\# fleet consumers
+        \\evelyn:sekrit-key-1:8
+        \\ci:other-key
+    );
+    try testing.expectEqual(@as(usize, 2), ring.entries.items.len);
+    try testing.expectEqualStrings("evelyn", ring.entries.items[0].name);
+    try testing.expectEqual(@as(u64, 8_000_000), ring.entries.items[0].rate_bps);
+    try testing.expectEqual(@as(u64, 0), ring.entries.items[1].rate_bps);
+
+    // drain some budget, then reload with the key surviving: level carries
+    ring.entries.items[0].bucket_bytes = -5_000_000;
+    try ring.loadFromSlice("evelyn:sekrit-key-1:8\n");
+    try testing.expectEqual(@as(usize, 1), ring.entries.items.len);
+    try testing.expectEqual(@as(i64, -5_000_000), ring.entries.items[0].bucket_bytes);
+
+    // revocation: the line is gone, the key is gone
+    try ring.loadFromSlice("ci:other-key\n");
+    try testing.expectEqual(@as(usize, 1), ring.entries.items.len);
+    try testing.expectEqualStrings("ci", ring.entries.items[0].name);
+
+    try testing.expectError(error.MalformedKeyLine, ring.loadFromSlice("no-key-here\n"));
+}
+
+test "bucket: burst allowed, deficit 429s, refill clears it" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var ring = KeyRing.init(testing.allocator, "/nonexistent");
+    defer ring.deinit();
+    try ring.loadFromSlice("e:k:2\n"); // 2 MB/s, 8 MB burst
+    const entry = &ring.entries.items[0];
+    entry.last_refill_us = 0;
+
+    // fresh bucket: clear to serve, then a 300 MB drain lands us in deficit
+    try testing.expectEqual(@as(?u64, null), ring.precheck(io, entry, 1_000_000));
+    ring.charge(io, entry, 300_000_000, 1_000_000);
+    const wait = ring.precheck(io, entry, 1_000_001) orelse return error.ExpectedDeficit;
+    // ~292 MB deficit at 2 MB/s ≈ 146s
+    try testing.expect(wait > 100 and wait < 200);
+
+    // after enough wall clock the bucket refills to burst and serves again
+    try testing.expectEqual(@as(?u64, null), ring.precheck(io, entry, 1_000_000 + 200 * std.time.us_per_s));
+
+    // unmetered key never blocks
+    try ring.loadFromSlice("open:k2\n");
+    const open = &ring.entries.items[0];
+    ring.charge(io, open, 1 << 40, 5);
+    try testing.expectEqual(@as(?u64, null), ring.precheck(io, open, 6));
+}
diff --git a/src/internal/serve/server.zig b/src/internal/serve/server.zig
--- a/src/internal/serve/server.zig
+++ b/src/internal/serve/server.zig
@@ -1281,6 +1281,12 @@
     modified_s: ?i64 = null,
     cache_max_age_s: u64 = 0,
 
+    /// bytes actually served on this request (0 until a 200 body went out);
+    /// the api key ring charges per-key budgets from this
+    pub fn servedBytes(self: *const XrpcResponder) u64 {
+        return self.observation.served_bytes;
+    }
+
     pub fn archiveModified(self: *XrpcResponder, modified_s: i64) void {
         // net/http treats the Unix epoch as an unspecified modification time.
         self.modified_s = if (modified_s == 0) null else modified_s;
diff --git a/src/internal/serve/xrpcapi.zig b/src/internal/serve/xrpcapi.zig
--- a/src/internal/serve/xrpcapi.zig
+++ b/src/internal/serve/xrpcapi.zig
@@ -11,6 +11,7 @@
 //! contract, including multipart/byteranges and mixed overlap handling.
 
 const std = @import("std");
+const api_keys = @import("api_keys.zig");
 const archive_mod = @import("../storage/archive.zig");
 const segment = @import("../storage/segment.zig");
 const import_manager = @import("../timestamp/manager.zig");
@@ -55,6 +56,10 @@
     /// Empty = open archive — the operator's opt-in, since gating an
     /// already-public surface must be a deliberate flip, not a default.
     archive_api_key: []const u8 = "",
+    /// named, revocable, per-key-metered archive keys (api_keys.zig).
+    /// checked when the fleet key above does not match; null = ring
+    /// disabled and the single key is the only credential.
+    key_ring: ?*api_keys.KeyRing = null,
     /// Keep one manifest planning pass in flight so public callers cannot
     /// multiply CPU and response allocation with concurrent requests.
     plan_mutex: Io.Mutex = .init,
@@ -95,9 +100,26 @@
             std.mem.eql(u8, method, "listSegments") or
             std.mem.eql(u8, method, "getSegment") or
             std.mem.eql(u8, method, "getBlock");
-        if (archive_method and self.archive_api_key.len > 0 and !authorized(self.archive_api_key, authorization)) {
-            respond.unauthorized();
-            return true;
+        var metered: ?*api_keys.Entry = null;
+        if (archive_method) {
+            const fleet_ok = self.archive_api_key.len > 0 and authorized(self.archive_api_key, authorization);
+            const gated = self.archive_api_key.len > 0 or self.key_ring != null;
+            if (gated and !fleet_ok) {
+                const ring = self.key_ring orelse {
+                    respond.unauthorized();
+                    return true;
+                };
+                const now_us = nowUs(self.io);
+                metered = ring.authorize(self.io, authorization, now_us) orelse {
+                    respond.unauthorized();
+                    return true;
+                };
+                if (ring.precheck(self.io, metered.?, now_us)) |_| {
+                    // the SDK's fetchWithRetry backs off on 429 (1s/2s/4s)
+                    respond.err(429, "RateLimitExceeded", "per-key byte budget exhausted; retry shortly");
+                    return true;
+                }
+            }
         }
         if (std.mem.eql(u8, method, "planSnapshot")) {
             if (!is_post) {
@@ -105,6 +127,7 @@
                 return true;
             }
             planSnapshot(self, respond, body);
+            self.chargeMetered(respond, metered);
             return true;
         }
         if (std.mem.eql(u8, method, "planBackfill")) {
@@ -128,7 +151,22 @@
         } else {
             respond.err(501, "MethodNotImplemented", method);
         }
+        self.chargeMetered(respond, metered);
         return true;
+    }
+
+    /// charge a metered key for the bytes the responder actually served.
+    /// post-hoc by design: the bucket may go negative and gates the NEXT
+    /// request, so one oversized response is always allowed through.
+    fn chargeMetered(self: *Api, respond: anytype, metered: ?*api_keys.Entry) void {
+        const entry = metered orelse return;
+        const ring = self.key_ring orelse return;
+        const bytes: u64 = if (comptime @hasDecl(@TypeOf(respond.*), "servedBytes")) respond.servedBytes() else 0;
+        ring.charge(self.io, entry, bytes, nowUs(self.io));
+    }
+
+    fn nowUs(io: Io) i64 {
+        return @intCast(@divTrunc(Io.Timestamp.now(io, .awake).nanoseconds, std.time.ns_per_us));
     }
 
     fn getZstdDictionary(_: *Api, respond: anytype, query: []const u8, if_none_match: ?[]const u8) void {
@@ -1651,6 +1689,67 @@
     var open: PlanResponder = .{};
     _ = open_api.handle(&open, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", null, .{});
     try testing.expect(open.code != 401);
+}
+
+test "key ring: named keys authorize, meter with 429, and revoke on reload" {
+    const testing = std.testing;
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buffer: [Io.Dir.max_path_bytes]u8 = undefined;
+    const path_len = try tmp.dir.realPath(io, &path_buffer);
+    var archive = try archive_mod.Archive.init(testing.allocator, io, path_buffer[0..path_len]);
+    defer archive.deinit();
+
+    // a real key file: the ring reloads on mtime, so revocation below is
+    // exercised the way an operator does it — by editing the file
+    var key_path_buf: [Io.Dir.max_path_bytes + 32]u8 = undefined;
+    const key_path = try std.fmt.bufPrint(&key_path_buf, "{s}/keys.conf", .{path_buffer[0..path_len]});
+    {
+        var f = try tmp.dir.createFile(io, "keys.conf", .{});
+        defer f.close(io);
+        try f.writeStreamingAll(io, "consumer:ring-key:1\n");
+    }
+    var ring = api_keys.KeyRing.init(testing.allocator, key_path);
+    defer ring.deinit();
+
+    var api: Api = .{ .allocator = testing.allocator, .io = io, .archive = &archive, .archive_api_key = "fleet", .key_ring = &ring };
+
+    // ring key authorizes where the fleet key would too
+    var ok: PlanResponder = .{};
+    _ = api.handle(&ok, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer ring-key", .{});
+    try testing.expect(ok.code != 401 and ok.code != 429);
+
+    // unknown key still 401s
+    var bad: PlanResponder = .{};
+    _ = api.handle(&bad, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer nope", .{});
+    try testing.expectEqual(@as(u16, 401), bad.code);
+
+    // drive the key into deficit: the next request is a 429; the fleet
+    // key is unmetered and keeps working
+    ring.charge(io, &ring.entries.items[0], 500_000_000, Api.nowUs(io));
+    var throttled: PlanResponder = .{};
+    _ = api.handle(&throttled, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer ring-key", .{});
+    try testing.expectEqual(@as(u16, 429), throttled.code);
+    var fleet: PlanResponder = .{};
+    _ = api.handle(&fleet, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer fleet", .{});
+    try testing.expect(fleet.code != 429 and fleet.code != 401);
+
+    // revocation: rewrite the file without the key; the next authorize
+    // reloads (mtime changed, check throttle bypassed via last_check)
+    {
+        var f = try tmp.dir.createFile(io, "keys.conf", .{ .truncate = true });
+        defer f.close(io);
+        try f.writeStreamingAll(io, "other:different-key\n");
+    }
+    ring.last_check_us = std.math.minInt(i64);
+    ring.loaded_mtime_us = std.math.minInt(i64);
+    var revoked: PlanResponder = .{};
+    _ = api.handle(&revoked, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer ring-key", .{});
+    try testing.expectEqual(@as(u16, 401), revoked.code);
 }
 
 test "planBackfill never answers 200 with a short plan under allocation failure" {