diff --git a/src/runs/stream/append-guard.test.ts b/src/runs/stream/append-guard.test.ts new file mode 100644 index 000000000..4a7c77bd3 --- /dev/null +++ b/src/runs/stream/append-guard.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, test } from "bun:test"; +import type { AppendEventInput } from "../../db/repos/events.ts"; +import type { EventRow } from "../../db/schema.ts"; +import { appendOrDrop, MAX_CONSECUTIVE_APPEND_FAILURES, newAppendGuard } from "./append-guard.ts"; + +const input: AppendEventInput = { + runId: "run_test", + sandboxEventSeq: 1, + ts: "2026-05-08T12:00:01.000Z", + kind: "text", + payload: { seq: 1 }, +}; + +function events(plan: readonly boolean[]): { append: (i: AppendEventInput) => Promise } { + let i = 0; + return { + append: async (_i) => { + const ok = plan[i] ?? true; + i += 1; + if (!ok) throw new Error("unsupported Unicode escape sequence"); + return { id: i } as unknown as EventRow; + }, + }; +} + +describe("appendOrDrop", () => { + test("returns the row and resets the streak on success", async () => { + const guard = newAppendGuard(); + guard.consecutiveFailures = 2; + const row = await appendOrDrop(guard, events([true]), input); + expect(row).not.toBeNull(); + expect(guard.consecutiveFailures).toBe(0); + }); + + test("drops an isolated failure and returns null", async () => { + const guard = newAppendGuard(); + const logged: string[] = []; + const logger = { error: (_o: unknown, msg?: string) => logged.push(msg ?? "") }; + const row = await appendOrDrop(guard, events([false]), input, logger); + expect(row).toBeNull(); + expect(guard.consecutiveFailures).toBe(1); + expect(logged).toEqual(["bridge dropped event after append failure"]); + }); + + test("rethrows once failures reach the consecutive ceiling", async () => { + const guard = newAppendGuard(); + const repo = events(Array(MAX_CONSECUTIVE_APPEND_FAILURES).fill(false)); + for (let n = 1; n < MAX_CONSECUTIVE_APPEND_FAILURES; n += 1) { + expect(await appendOrDrop(guard, repo, input)).toBeNull(); + } + await expect(appendOrDrop(guard, repo, input)).rejects.toThrow("unsupported Unicode"); + }); + + test("a success between failures keeps the streak from reaching the ceiling", async () => { + const guard = newAppendGuard(); + const repo = events([false, false, true, false, false]); + for (const _ of [1, 2]) expect(await appendOrDrop(guard, repo, input)).toBeNull(); + expect(await appendOrDrop(guard, repo, input)).not.toBeNull(); + for (const _ of [1, 2]) expect(await appendOrDrop(guard, repo, input)).toBeNull(); + expect(guard.consecutiveFailures).toBe(2); + }); +}); diff --git a/src/runs/stream/append-guard.ts b/src/runs/stream/append-guard.ts new file mode 100644 index 000000000..d617d57c6 --- /dev/null +++ b/src/runs/stream/append-guard.ts @@ -0,0 +1,61 @@ +/** + * Bounded drop policy for a failed `events.append` (warren-fb5e). + * + * A poison event (Postgres jsonb rejecting U+0000) used to throw out of the + * bridge loop, flip `errored`, and make the reconnect loop replay the same + * event until `bridge_lost` killed a healthy run. Swallowing every failure + * is the opposite hazard: a DB outage would silently drop the whole stream + * and report success. This guard drops a failure and returns `null`, but + * rethrows once `MAX_CONSECUTIVE_APPEND_FAILURES` failures land in a row so + * a persistent store fault still reaches the reconnect path. Any success + * resets the streak. + */ + +import type { AppendEventInput, EventsRepo } from "../../db/repos/events.ts"; +import type { EventRow } from "../../db/schema.ts"; +import type { BridgeLogger } from "./types.ts"; + +export const MAX_CONSECUTIVE_APPEND_FAILURES = 3; + +export interface AppendGuard { + consecutiveFailures: number; +} + +export function newAppendGuard(): AppendGuard { + return { consecutiveFailures: 0 }; +} + +/** + * Append `input`, returning the row on success or `null` when the failure + * was dropped. Throws the append error itself once the streak reaches + * `MAX_CONSECUTIVE_APPEND_FAILURES`. + */ +export async function appendOrDrop( + guard: AppendGuard, + events: Pick, + input: AppendEventInput, + logger?: BridgeLogger, +): Promise { + try { + const row = await events.append(input); + guard.consecutiveFailures = 0; + return row; + } catch (err) { + guard.consecutiveFailures += 1; + const persistent = guard.consecutiveFailures >= MAX_CONSECUTIVE_APPEND_FAILURES; + logger?.error?.( + { + runId: input.runId, + kind: input.kind, + sandboxEventSeq: input.sandboxEventSeq, + consecutiveFailures: guard.consecutiveFailures, + err, + }, + persistent + ? "bridge append failed repeatedly; surfacing as stream error" + : "bridge dropped event after append failure", + ); + if (persistent) throw err; + return null; + } +} diff --git a/src/runs/stream/bridge.test.ts b/src/runs/stream/bridge.test.ts index 6cb9fbec4..e1b020350 100644 --- a/src/runs/stream/bridge.test.ts +++ b/src/runs/stream/bridge.test.ts @@ -105,6 +105,50 @@ describe("bridgeRunStream — event flow", () => { expect(rows).toEqual([1, 3]); }); + test("rethrows into errored once append fails three times in a row (warren-fb5e)", async () => { + repos.events.append = async () => { + throw new Error("connection refused"); + }; + const result = await bridgeRunStream({ + runId, + sandboxRunId, + repos, + broker, + sandboxId: "bur_aaaaaaaaaaaa", + runtimeProvider: makeProvider(), + source: source([evt(sandboxRunId, 1), evt(sandboxRunId, 2), evt(sandboxRunId, 3)]), + }); + expect(result.errored).toBe(true); + expect(result.written).toBe(0); + expect(result.dropped).toBe(2); + }); + + test("a dropped terminal event still sets terminalDetected (warren-fb5e)", async () => { + const origAppend = repos.events.append.bind(repos.events); + repos.events.append = async (input) => { + if (input.sandboxEventSeq === 2) throw new Error("unsupported Unicode escape sequence"); + return origAppend(input); + }; + const terminal = evt(sandboxRunId, 2, { + kind: "state_change", + stream: "system", + payload: { type: "result", subtype: "result", is_error: false, terminal_reason: "completed" }, + }); + const result = await bridgeRunStream({ + runId, + sandboxRunId, + repos, + broker, + sandboxId: "bur_aaaaaaaaaaaa", + runtimeProvider: makeProvider(), + source: source([evt(sandboxRunId, 1), terminal, evt(sandboxRunId, 3)]), + }); + expect(result.errored).toBe(false); + expect(result.terminalDetected).toEqual({ outcome: "succeeded" }); + expect(result.written).toBe(1); + expect(result.dropped).toBe(1); + }); + test("publishes each event to the broker after persisting", async () => { const sub = broker.subscribe(runId); const consumed: number[] = []; diff --git a/src/runs/stream/bridge.ts b/src/runs/stream/bridge.ts index f10f10073..b80ffc9b7 100644 --- a/src/runs/stream/bridge.ts +++ b/src/runs/stream/bridge.ts @@ -1,20 +1,13 @@ /** * `bridgeRunStream` — the main event-bridge pump (docs/design/agent-composition.md step 5; - * event durability per docs/design/runtime-and-supervisor.md). Splits out of the legacy - * monolithic `src/runs/stream.ts` (warren-041e / pl-9088 step 5): - * terminal-detection lives in `./terminal-detect.ts`, the run-state - * fallback in `./run-state-poller.ts`, cost-stats persistence in - * `./stats.ts`, and active-stream recovery in `./recover.ts`. + * event durability per docs/design/runtime-and-supervisor.md). Split out of the legacy + * `src/runs/stream.ts` (warren-041e / pl-9088 step 5): terminal-detection lives in + * `./terminal-detect.ts`, the run-state fallback in `./run-state-poller.ts`, cost-stats + * persistence in `./stats.ts`, and active-stream recovery in `./recover.ts`. * - * The bridge is the only writer to `events` (rows always land via - * `bridgeRunStream` → `EventsRepo.append`); the broker is published - * to immediately after each row commits so live tailers see fresh - * events without waiting on a polling interval. - * - * See the module-level commentary in `./index.ts` for the full - * resume / claim / terminal-detection / recovery semantics — keeping - * docs there so the doctored stream of consciousness stays in one - * place rather than fanning out across the split. + * The bridge is the only writer to `events` (`bridgeRunStream` → `EventsRepo.append`); the + * broker is published right after each row commits so live tailers never wait on a poll. + * `./index.ts` carries the full resume / claim / terminal-detection / recovery semantics. */ import type { EventStream, RunTerminalState } from "../../db/schema.ts"; @@ -29,6 +22,7 @@ import { newSessionStatsAccumulator, type SessionStatsAccumulator, } from "../usage-aggregate.ts"; +import { appendOrDrop, newAppendGuard } from "./append-guard.ts"; import { type CancelBurrowRunFn, enforceBudgetCap } from "./budget.ts"; import { providerStreamSource } from "./provider-source.ts"; import { defaultRunStateProbe, runStatePoller } from "./run-state-poller.ts"; @@ -152,6 +146,7 @@ export async function bridgeRunStream(input: BridgeRunStreamInput): Promise>; - try { - row = await repos.events.append({ + // warren-fb5e: a dropped append never flips `errored` by itself (a poison event + // would replay until bridge_lost); the guard rethrows after a bounded streak. + const row = await appendOrDrop( + appendGuard, + repos.events, + { runId, sandboxEventSeq: event.seq, ts: toIsoString(event.ts), @@ -214,13 +209,17 @@ export async function bridgeRunStream(input: BridgeRunStreamInput): Promise