diff --git a/.changelog/unreleased/fixed-631-review-trigger-own-feed.md b/.changelog/unreleased/fixed-631-review-trigger-own-feed.md new file mode 100644 index 00000000..953910f4 --- /dev/null +++ b/.changelog/unreleased/fixed-631-review-trigger-own-feed.md @@ -0,0 +1 @@ +- **The codex runtime's review-trigger loop reads the agent's own catch-up feed (Closes #631).** It uses a different cursor filename from the task loop and accepts loop IDs through 79 characters, including a 64-character agent ID plus `-review-trigger`. diff --git a/packages/cli/src/utils/codex-runtime.ts b/packages/cli/src/utils/codex-runtime.ts index f956c234..ce2d42f2 100644 --- a/packages/cli/src/utils/codex-runtime.ts +++ b/packages/cli/src/utils/codex-runtime.ts @@ -26,7 +26,7 @@ import { } from "./agent-lifecycle.js"; import { FlairClient, type PresenceActivity } from "./flair-client.js"; import { startTaskLoop } from "./flair-task-loop.js"; -import { handlePrOpened } from "./pr-review-trigger.js"; +import { startReviewTriggerLoop } from "./pr-review-trigger.js"; import { completeRuntimeMail, pollRuntimeMail, type RuntimeMailConfig, runtimeBootPreflight, sendRuntimeMail } from "./runtime-mail.js"; import { externalDispatchRefusal } from "./mail-tier.js"; import type { Envelope } from "@tpsdev-ai/agent"; @@ -853,16 +853,9 @@ export async function runCodexRuntime(config: CodexRuntimeConfig): Promise } }); - // Review trigger — listens for pr.opened and auto-requests K&S review + // Review trigger — polls the agent's own feed using a different cursor filename from the task loop. if (config.reviewTrigger) { - const triggerConfig = { - reviewers: config.reviewTrigger.reviewers, - agentId, - repo: config.reviewTrigger.repo, - }; - startTaskLoop(flair, `${agentId}-review-trigger`, async (event) => { - await handlePrOpened(event, triggerConfig); - }, { kinds: ["pr.opened"] }); + startReviewTriggerLoop(flair, agentId, config.reviewTrigger); console.log(`[${agentId}] Review trigger active → ${config.reviewTrigger.reviewers.join(", ")}`); } diff --git a/packages/cli/src/utils/flair-task-loop.ts b/packages/cli/src/utils/flair-task-loop.ts index ab5b5bc5..aa00febf 100644 --- a/packages/cli/src/utils/flair-task-loop.ts +++ b/packages/cli/src/utils/flair-task-loop.ts @@ -2,13 +2,16 @@ * flair-task-loop.ts — Polls OrgEventCatchup for task.assigned events * targeting this agent. Runtime-agnostic: works without OpenClaw. * - * CURSOR. The loop reads the catch-up feed through `FlairClient.getEventsSince` - * (the validated reader in org-event-read.ts, cli#619), always sending its - * position as Flair's `after` cursor, so Flair's own watermark for the agent - * never chooses where reading starts. The position is persisted in - * `~/.tps/cursors/-task-loop.json` as `{ position, updatedAt }` and - * replaced atomically (a temporary file in the same directory, fsynced, then - * renamed over it). With no usable cursor file, the loop sets the position to + * CURSOR. The loop reads `agentId`'s catch-up feed through + * `FlairClient.getEventsSince` (the validated reader in org-event-read.ts, + * cli#619), always sending its position as Flair's `after` cursor, so Flair's + * own watermark for the agent never chooses where reading starts. The position + * is persisted in `~/.tps/cursors/-task-loop.json` as + * `{ position, updatedAt }` and replaced atomically (a temporary file in the + * same directory, fsynced, then renamed over it). The task and review loops use + * different cursor filenames; callers that omit `loopId` use the `agentId` cursor. It must + * be a single safe filename component (see `LOOP_ID_PATTERN`); `startTaskLoop` refuses any other with + * `InvalidLoopIdError` before reading or writing a cursor. With no usable cursor file, the loop sets the position to * FIRST_BOOT_LOOKBACK_MS before now, once, and does not read the feed until * that position is saved: a failed save is logged by name and retried with * backoff, with the same position, and `stop()` ends the retrying. A legacy @@ -95,9 +98,31 @@ function cursorDir(): string { return join(homeDir(), ".tps", "cursors"); } -/** The task loop's cursor file for `agentId`. */ -export function taskLoopCursorPath(agentId: string): string { - return join(cursorDir(), `${agentId}-task-loop.json`); +/** + * The allowed set for `loopId`: a single filename component of 1..79 characters + * from `A-Z a-z 0-9 . _ -`, excluding the two dot components `.` and `..`. It + * names the cursor file `-task-loop.json` inside the cursors directory, + * so a path separator, a NUL, an empty or over-long string, or `.`/`..` is + * refused rather than allowed to resolve a cursor outside that directory. + */ +const LOOP_ID_PATTERN = /^[A-Za-z0-9._-]{1,79}$/; + +/** Whether `loopId` is a single safe filename component (see `LOOP_ID_PATTERN`). */ +export function isSafeLoopId(loopId: string): boolean { + return LOOP_ID_PATTERN.test(loopId) && loopId !== "." && loopId !== ".."; +} + +/** Thrown by `startTaskLoop` for a `loopId` that is not a single safe filename component. */ +export class InvalidLoopIdError extends Error { + constructor(loopId: string) { + super(`Invalid loopId ${JSON.stringify(loopId)}: it must be a single filename component matching ${LOOP_ID_PATTERN} and not "." or ".."`); + this.name = "InvalidLoopIdError"; + } +} + +/** The loop's cursor file for `loopId`. */ +export function taskLoopCursorPath(loopId: string): string { + return join(cursorDir(), `${loopId}-task-loop.json`); } /** @@ -125,10 +150,10 @@ export function cursorFilePosition(data: unknown): string | undefined { } /** The persisted feed position (Flair's `after` cursor), or undefined when there is none to resume from. */ -function loadCursor(agentId: string): string | undefined { - const path = taskLoopCursorPath(agentId); +function loadCursor(loopId: string): string | undefined { + const path = taskLoopCursorPath(loopId); if (!existsSync(path)) { - slog(`[${agentId}] No task loop cursor at ${path}; ${LOOKBACK_NOTE}`); + slog(`[${loopId}] No task loop cursor at ${path}; ${LOOKBACK_NOTE}`); return undefined; } let position: string | undefined; @@ -139,7 +164,7 @@ function loadCursor(agentId: string): string | undefined { problem = e instanceof SyntaxError ? "it is not JSON" : `it cannot be read (${(e as NodeJS.ErrnoException).code ?? (e as Error).name})`; } if (position === undefined) { - swarn(`[${agentId}] Task loop cursor at ${path} is unusable (${problem}); treating it as no cursor, so ${LOOKBACK_NOTE}`); + swarn(`[${loopId}] Task loop cursor at ${path} is unusable (${problem}); treating it as no cursor, so ${LOOKBACK_NOTE}`); } return position; } @@ -149,8 +174,8 @@ function loadCursor(agentId: string): string | undefined { * it, rename it over the cursor. Returns undefined on success, or the failure's * name (its code, such as EACCES, or its class) for the caller to log. */ -function saveCursor(agentId: string, position: string): string | undefined { - const path = taskLoopCursorPath(agentId); +function saveCursor(loopId: string, position: string): string | undefined { + const path = taskLoopCursorPath(loopId); const tmp = `${path}.${process.pid}.${Date.now()}.tmp`; try { mkdirSync(cursorDir(), { recursive: true }); @@ -190,11 +215,17 @@ export function startTaskLoop( flair: FlairClient, agentId: string, handler: TaskHandler, - opts: { pollIntervalMs?: number; kinds?: string[] } = {}, + opts: { pollIntervalMs?: number; kinds?: string[]; loopId?: string } = {}, ): { stop: () => Promise } { const basePollMs = opts.pollIntervalMs ?? DEFAULT_POLL_MS; const kinds = new Set(opts.kinds ?? ["task.assigned"]); - const loaded = loadCursor(agentId); + // Omitted loop IDs use the agentId cursor. + const loopId = opts.loopId ?? agentId; + // Refuse an unsafe loopId before any cursor is read or written: it names the + // cursor file, so one with a path separator or a traversal segment could + // resolve a cursor outside the cursors directory. + if (!isSafeLoopId(loopId)) throw new InvalidLoopIdError(loopId); + const loaded = loadCursor(loopId); // A new initial position is computed once, here; saving it is retried with // this same value until it succeeds, and the feed is not read before then. let position: string = loaded ?? initialPosition(); @@ -216,11 +247,11 @@ export function startTaskLoop( /** Save the initial position, retrying with backoff; false when stopped first. */ async function saveInitialPosition(): Promise { for (let attempt = 1; running; attempt++) { - const failure = saveCursor(agentId, position); + const failure = saveCursor(loopId, position); if (failure === undefined) return true; const delay = Math.min(basePollMs * 2 ** (attempt - 1), MAX_BACKOFF_MS); swarn( - `[${agentId}] Task loop cannot save its initial position to ${taskLoopCursorPath(agentId)} (${failure}); ` + + `[${loopId}] Task loop cannot save its initial position to ${taskLoopCursorPath(loopId)} (${failure}); ` + `the feed is not read until it is saved; retrying in ${delay} ms`, ); await sleep(delay); @@ -290,8 +321,8 @@ export function startTaskLoop( // A failed save is logged and the loop goes on from the position in // memory. The saved position is then behind it, so a restart replays // events rather than skipping them. - const failure = saveCursor(agentId, position); - if (failure !== undefined) swarn(`[${agentId}] Failed to persist cursor to ${taskLoopCursorPath(agentId)} (${failure})`); + const failure = saveCursor(loopId, position); + if (failure !== undefined) swarn(`[${loopId}] Failed to persist cursor to ${taskLoopCursorPath(loopId)} (${failure})`); } // Cap seen set diff --git a/packages/cli/src/utils/pr-review-trigger.ts b/packages/cli/src/utils/pr-review-trigger.ts index 54012a27..4552b647 100644 --- a/packages/cli/src/utils/pr-review-trigger.ts +++ b/packages/cli/src/utils/pr-review-trigger.ts @@ -10,6 +10,8 @@ import { spawnSync } from "node:child_process"; import snooplogg from "snooplogg"; +import type { FlairClient } from "./flair-client.js"; +import { startTaskLoop } from "./flair-task-loop.js"; const { log: slog, warn: swarn, error: serror } = snooplogg("tps:github"); @@ -74,6 +76,7 @@ export function reRequestReviewer( export async function handlePrOpened( event: { detail?: string; summary: string; refId?: string }, config: ReviewTriggerConfig, + deps: ReviewRequestDeps = {}, ): Promise { const detail = event.detail ?? ""; const prNumber = extractPrNumber(detail); @@ -83,5 +86,24 @@ export async function handlePrOpened( return; } - requestReviews(prNumber, config); + requestReviews(prNumber, config, deps); +} + +/** + * Poll the agent's own feed for `pr.opened`, using a different cursor filename from the task loop. + * The loop ID allows the CLI's 64-character agent ID plus `-review-trigger`. + */ +export function startReviewTriggerLoop( + flair: FlairClient, + agentId: string, + config: { reviewers: string[]; repo: string }, + opts: { pollIntervalMs?: number; deps?: ReviewRequestDeps } = {}, +): { stop: () => Promise } { + const triggerConfig: ReviewTriggerConfig = { reviewers: config.reviewers, agentId, repo: config.repo }; + return startTaskLoop( + flair, + agentId, + async (event) => { await handlePrOpened(event, triggerConfig, opts.deps); }, + { kinds: ["pr.opened"], loopId: `${agentId}-review-trigger`, pollIntervalMs: opts.pollIntervalMs }, + ); } diff --git a/packages/cli/test/flair-task-loop.test.ts b/packages/cli/test/flair-task-loop.test.ts index 7daae2bc..a122ee37 100644 --- a/packages/cli/test/flair-task-loop.test.ts +++ b/packages/cli/test/flair-task-loop.test.ts @@ -10,13 +10,13 @@ * the signing and parsing seams are exercised; only the HTTP endpoint is faked. */ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { chmodSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Writable } from "node:stream"; import { SnoopLogg } from "snooplogg"; -import { createFlairClient } from "../src/utils/flair-client.js"; -import { MAX_POSITION_SKEW_MS, startTaskLoop } from "../src/utils/flair-task-loop.js"; +import { createFlairClient, type FlairClient } from "../src/utils/flair-client.js"; +import { InvalidLoopIdError, MAX_POSITION_SKEW_MS, startTaskLoop } from "../src/utils/flair-task-loop.js"; import { createPatchShared } from "./helpers/patch-shared.js"; import { type CatchupFeed, @@ -507,16 +507,16 @@ describe("cli#624 — the initial position is set once and always sent", () => { }); describe("cli#624 — the initial position is saved before the first read", () => { - // A real write failure: the cursor directory is made unwritable, then restored. + // A regular file blocks creation of the cursor directory, including for root. const cursors = () => join(home, ".tps", "cursors"); const lockCursors = () => { - mkdirSync(cursors(), { recursive: true }); - chmodSync(cursors(), 0o500); + mkdirSync(join(home, ".tps"), { recursive: true }); + writeFileSync(cursors(), "blocked"); }; - const unlockCursors = () => chmodSync(cursors(), 0o700); + const unlockCursors = () => rmSync(cursors()); const saveFailures = () => warnings.split("cannot save its initial position").length - 1; - test.skipIf(process.getuid?.() === 0)( + test( "while the initial position cannot be saved nothing is read; after recovery the saved position is the one fixed at start, and a restart retries the first task", async () => { const startNow = Date.now(); @@ -535,9 +535,9 @@ describe("cli#624 — the initial position is saved before the first read", () = }, { pollIntervalMs: 20 }); try { expect(await waitFor(() => saveFailures() >= 3)).toBe(true); - expect(warnings).toContain(`cannot save its initial position to ${cursorFile()} (EACCES)`); + expect(warnings).toContain(`cannot save its initial position to ${cursorFile()} (EEXIST)`); expect(catchup.afters).toEqual([]); - // Half an hour passes before the directory is writable again: an initial + // Half an hour passes before the directory can be created: an initial // position recomputed now would start after e1. offset = 30 * 60_000; unlockCursors(); @@ -567,7 +567,7 @@ describe("cli#624 — the initial position is saved before the first read", () = }, ); - test.skipIf(process.getuid?.() === 0)("stop ends the retrying of the initial save promptly, and nothing is read", async () => { + test("stop ends the retrying of the initial save promptly, and nothing is read", async () => { const catchup = serveFeed({ events: [assigned("e1", 5)] }); lockCursors(); try { @@ -709,3 +709,27 @@ describe("cli#624 — the task loop's cursor file", () => { expect(readdirSync(join(home, ".tps", "cursors"))).toEqual([`${AGENT}-task-loop.json`]); }); }); + +describe("cli#631 — loopId is a single safe filename component", () => { + test("refuses a traversal, separator, dot or over-long loopId before reading or writing a cursor", () => { + // A client the loop could not use even if it got past the check: the refusal must come first. + const untouched = {} as FlairClient; + for (const bad of ["../../outside", "a/b", "..", ".", "a\\b", "x".repeat(80), "a\0b"]) { + expect(() => startTaskLoop(untouched, AGENT, async () => {}, { loopId: bad })).toThrow(InvalidLoopIdError); + } + // Nothing was read or written: the cursors directory does not exist, and the + // path a traversal loopId would reach is untouched. + expect(existsSync(join(home, ".tps", "cursors"))).toBe(false); + expect(existsSync(join(home, "outside-task-loop.json"))).toBe(false); + }); + + test("accepts a valid loopId and names its cursor file with it", async () => { + serveFeed({ events: [] }); + const loop = startTaskLoop(client(), AGENT, async () => {}, { loopId: "agent-a-review-trigger", pollIntervalMs: 60_000 }); + try { + expect(await waitFor(() => existsSync(join(home, ".tps", "cursors", "agent-a-review-trigger-task-loop.json")))).toBe(true); + } finally { + await loop.stop(); + } + }); +}); diff --git a/packages/cli/test/review-trigger-loop.test.ts b/packages/cli/test/review-trigger-loop.test.ts new file mode 100644 index 00000000..eceeec36 --- /dev/null +++ b/packages/cli/test/review-trigger-loop.test.ts @@ -0,0 +1,183 @@ +/** + * cli#631 — the review-trigger loop reads only feeds the agent may read: it + * reads the agent's OWN catch-up feed (`agentId`), the feed Flair serves an + * event addressed to that agent to. Flair refuses a non-admin agent's read of + * `-review-trigger` with 403. + * + * The catch-up route is `orgEventCatchupRoute`, the shared signed stub modelled + * on Flair's coded catch-up contract; it refuses a non-admin read of another + * participant's feed with 403, so a non-admin agent gets nothing from a foreign feed. + * The loop runs through the real FlairClient, signer and envelope reader; the + * HTTP catch-up endpoint is faked, and the case that drives the review request + * fakes the `gh-as` process call too (`spawnSyncImpl`). + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Writable } from "node:stream"; +import { SnoopLogg } from "snooplogg"; +import { createFlairClient } from "../src/utils/flair-client.js"; +import { InvalidLoopIdError } from "../src/utils/flair-task-loop.js"; +import { startReviewTriggerLoop } from "../src/utils/pr-review-trigger.js"; +import { + type CatchupFeed, + type CatchupRoute, + installStubFlairFetch, + orgEventCatchupRoute, + type StubFlair, + writeKeyFile, +} from "./helpers/stub-flair.js"; + +const AGENT = "agent-a"; +const AUTHOR = "agent-b"; +const SEED = Buffer.alloc(32, 0x31); +const REPO = "example/repo"; + +let stub: StubFlair | undefined; +let home: string; +let keyPath: string; +let warnings = ""; +let output: Writable; +let logger: SnoopLogg; +const savedHome = process.env.HOME; + +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "tps-review-trigger-")); + process.env.HOME = home; + keyPath = writeKeyFile(join(home, ".tps", "identity"), AGENT, SEED); + warnings = ""; + output = new Writable({ write(chunk, _encoding, done) { warnings += chunk.toString(); done(); } }); + logger = new SnoopLogg().enable("tps:flair").snoop().pipe(output, { colors: false }); +}); + +afterEach(() => { + logger.unsnoop().unpipe(output); + stub?.stop(); + stub = undefined; + if (savedHome === undefined) delete process.env.HOME; + else process.env.HOME = savedHome; + rmSync(home, { recursive: true, force: true }); +}); + +/** The review-trigger loop's cursor file under the test HOME (its own, not the task loop's). */ +const cursorFile = (agentId = AGENT) => join(home, ".tps", "cursors", `${agentId}-review-trigger-task-loop.json`); + +/** The cursor position the loop persisted, or undefined when it wrote none. */ +function cursorPosition(agentId = AGENT): string | undefined { + try { + return JSON.parse(readFileSync(cursorFile(agentId), "utf-8")).position; + } catch { + return undefined; + } +} + +const client = (agentId = AGENT) => createFlairClient(agentId, stub!.url, keyPath); + +/** A pr.opened event `handlePrOpened` can request reviews for. */ +const prOpened = (id: string, minutesAgo: number, targetIds: string[] = [AGENT]) => ({ + id, kind: "pr.opened", authorId: AUTHOR, summary: `PR ${id}`, + detail: `https://github.com/${REPO}/pull/1`, + createdAt: new Date(Date.now() - minutesAgo * 60_000).toISOString(), targetIds, +}); + +/** The feed position Flair gives an event: `\t`. */ +const positionOf = (e: { createdAt: string; id: string }) => `${e.createdAt}\t${e.id}`; + +/** Serve `feed` through the verifying stub, recording every participant a catch-up request names. */ +function serve(feed: CatchupFeed, agentId = AGENT): { route: CatchupRoute; asked: string[] } { + stub?.stop(); + const route = orgEventCatchupRoute(feed); + const asked: string[] = []; + stub = installStubFlairFetch({ [agentId]: SEED }, (req, url, caller) => { + if (!url.pathname.startsWith("/OrgEventCatchup/")) return undefined; + asked.push(decodeURIComponent(url.pathname.slice("/OrgEventCatchup/".length))); + return route(req, url, caller); + }); + return { route, asked }; +} + +/** Serve `answer` for every catch-up request, recording the `after` query. */ +function serveAnswer(answer: (req: Request, url: URL) => Response): CatchupRoute { + stub?.stop(); + const route = ((req: Request, url: URL) => { + route.afters.push(url.searchParams.get("after") ?? ""); + return answer(req, url); + }) as CatchupRoute; + route.afters = []; + route.served = []; + stub = installStubFlairFetch({ [AGENT]: SEED }, (req, url) => url.pathname.startsWith("/OrgEventCatchup/") ? route(req, url, "") : undefined); + return route; +} + +/** + * Poll `cond` until it holds or `timeoutMs` elapses; return whether it held. The + * default stays under bun's 5 s test timeout, so a failing test still stops its + * loop in `finally`. + */ +async function waitFor(cond: () => boolean, timeoutMs = 3_000): Promise { + const deadline = performance.now() + timeoutMs; + while (performance.now() < deadline) { + if (cond()) return true; + await Bun.sleep(10); + } + return cond(); +} + +const LOOP_OPTS = { pollIntervalMs: 30 }; + +describe("cli#631 — the review-trigger loop reads only feeds the agent may read", () => { + test.each([AGENT, "a".repeat(64)])("agent %s: with the HTTP catch-up endpoint faked and the gh-as process call faked (spawnSyncImpl), it reads the agent's own feed and requests the configured reviewers for a trigger addressed to the agent", async (agentId) => { + keyPath = writeKeyFile(join(home, ".tps", "identity"), agentId, SEED); + const e1 = prOpened("e1", 5, [agentId]); + const { route, asked } = serve({ events: [e1] }, agentId); + const ghCalls: Array<{ cmd: string; args: string[]; input?: string }> = []; + const deps = { + spawnSyncImpl: ((cmd: string, args: string[], opts?: { input?: string }) => { + ghCalls.push({ cmd, args, input: opts?.input }); + return { status: 0, stdout: "", stderr: "" }; + }) as any, + }; + const loop = startReviewTriggerLoop(client(agentId), agentId, { reviewers: ["reviewer-1"], repo: REPO }, { ...LOOP_OPTS, deps }); + try { + // The cursor advances only after the handler succeeds; a refused read advances nothing. + expect(await waitFor(() => cursorPosition(agentId) === positionOf(e1))).toBe(true); + expect(ghCalls).toHaveLength(1); + } finally { + await loop.stop(); + } + // The handler issued the review request the review-trigger loop exists for: gh-as, as the agent, + // on the PR in the event's /pull/ URL, with the configured reviewers as the payload. + expect(ghCalls[0]?.cmd).toBe("gh-as"); + expect(ghCalls[0]?.args).toEqual([ + agentId, "api", "--method", "POST", `repos/${REPO}/pulls/1/requested_reviewers`, "--input", "-", + ]); + expect(ghCalls[0]?.input).toBe(JSON.stringify({ reviewers: ["reviewer-1"] })); + // Every request named the agent's own feed. A non-admin agent's foreign-feed read would get 403. + expect(asked.length).toBeGreaterThan(0); + expect(new Set(asked)).toEqual(new Set([agentId])); + expect(route.served).toEqual(["e1"]); + expect(warnings).not.toContain("unavailable"); + }); + + test("refuses the 80-character review loop ID derived from a 65-character agent ID before cursor access", () => { + serve({ events: [] }); + expect(() => startReviewTriggerLoop(client(), "a".repeat(65), { reviewers: [], repo: REPO })) + .toThrow(InvalidLoopIdError); + expect(existsSync(join(home, ".tps", "cursors"))).toBe(false); + }); + + test("a refused read is reported unavailable, not treated as an empty feed", async () => { + const route = serveAnswer(() => Response.json({ error: "forbidden" }, { status: 403 })); + const loop = startReviewTriggerLoop(client(), AGENT, { reviewers: [], repo: REPO }, LOOP_OPTS); + try { + expect(await waitFor(() => route.afters.length >= 1)).toBe(true); + expect(await waitFor(() => warnings.includes("catch-up read unavailable"))).toBe(true); + } finally { + await loop.stop(); + } + expect(warnings).toContain("(HTTP 403)"); + // The initial position was saved; the refused read advanced the cursor no further. + expect(cursorPosition()).toBe(route.afters[0]); + }); +});