fix(bridge): bound the append-drop policy and finalize dropped terminal events (warren-fb5e)

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<EventRow> } {
+	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);
+	});
+});

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<EventsRepo, "append">,
+	input: AppendEventInput,
+	logger?: BridgeLogger,
+): Promise<EventRow | null> {
+	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;
+	}
+}

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[] = [];

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<Brid
 	// Shape-sniffed in `extractClaudeUsage`; persisted on terminal only when
 	// no pi usage was observed (pi path wins for parity).
 	const claudeUsage: SessionStatsAccumulator = newSessionStatsAccumulator();
+	const appendGuard = newAppendGuard();
 
 	try {
 		for await (const event of source(ctrl.signal)) {
@@ -201,26 +196,30 @@ export async function bridgeRunStream(input: BridgeRunStreamInput): Promise<Brid
 				dropped += 1;
 				continue;
 			}
-			// warren-fb5e: a single failed append must not flip `errored` — that
-			// reclassifies a per-event failure as sandbox_unreachable and the
-			// reconnect loop replays the poison event until bridge_lost kills a healthy run.
-			let row: Awaited<ReturnType<typeof repos.events.append>>;
-			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),
 					kind: event.kind,
 					stream: normalizeStream(event.stream),
 					origin: event.origin ?? null, // warren-5a07: parse-boundary provenance
 					payload: event.payload,
-				});
-			} catch (err) {
+				},
+				input.logger,
+			);
+			if (row === null) {
 				dropped += 1;
-				input.logger?.error?.(
-					{ runId, kind: event.kind, sandboxEventSeq: event.seq, err },
-					"bridge dropped event after append failure",
-				);
+				// A dropped terminal event must still finalize the run (see warren-2206).
+				const droppedOutcome = detectRuntimeTerminal(event);
+				if (droppedOutcome !== null) {
+					terminalDetected = { outcome: droppedOutcome };
+					break;
+				}
 				continue;
 			}
 			written += 1;