From 8eb11fa8aa949ee5ef74c2e4f6651a001e2ef9d5 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 05:57:46 -0600 Subject: [PATCH] fix(runner): fail closed with a recorded cause when runner event commits stall (#20235) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A Runner streams numbered events to the control plane, which commits each one and acknowledges the committed prefix > - In restart chaos wave `w-20261109T093417-s9807`, t01 (claude R1, local), the control plane never committed Runner seq 34. Every failure path in `#event` closed the connection without recording why. The Runner reconnected 2,848 times over 14 minutes and nothing detected it > - This pull request makes that loop a detected, named failure that carries its cause > - The benefit is that a wedged event stream becomes a visible run failure with a reason, instead of a silent "running" task ## Linked Issues or Issue Description Refs #19831. Restart chaos wave `w-20261109T093417-s9807` (master `df4562ab38`), task t01, run `d08dc899`. **What happened** (Phase 2's re-check of the kept instance): - The provider kept working. Claude's transcript runs to 09:47:03 and ends in `paperclip_finish`. - The Runner, its ACPX provider state, and the controller journal agree on acked seq 33. The controller never commits Runner seq 34 (Write progressed, 09:36:25.892). - Phase 2 replayed the kept outbox offline. Runner seq 34 fails `validatePrpEvent`: its `/payload/target` contains a newline from a malformed Claude Write call. `#event` closes without a reason, and the Runner redelivers the same bytes forever. (Phase 2's separate PR stops the producers from emitting such paths.) - Every `#event` failure path closed the connection without recording a reason, so the cause was invisible. An outbox that already holds such an event can only be cleared by failing closed. - The `::N` ids in `heartbeat_run_events` come from the TS driver's own projection numbering, so there was no Runner sequence rewind. ## What Changed - `packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts`: - `#eventCommitFailed` replaces the bare `connection.close()` on each `#event` commit-failure path. It records a bounded cause: `event_sequence_mismatch` for an out-of-order event, `event_schema_invalid:` for an event that fails `validatePrpEvent`, the error's code or name (never its message) for a commit callback error, `journal_sequence_not_contiguous`, `journal_commit_refused`, or `checkpoint_flush_failed:`. - If failures continue for the same run at the same acknowledged cursor for at least 20 attempts and 45 s, it fails closed with `NativeSessionProtocolIntegrityError("runner_event_commit_stalled", " at source seq N")`, where N is the failing event's own sequence. Only a moved acknowledged cursor counts as progress (a replayed, already acknowledged duplicate does not), and a different run starts a fresh counter. A single transient failure still just reconnects, as before. - A checkpoint flush that fails before any outbound frame also counts, as `checkpoint_flush_failed:`. That includes the authentication challenge and the welcome, so a reconnect that cannot even authenticate still reaches the bound. - A journal write that throws (`#store.commit`, `#store.commitEvent`, or the welcome save) leaves the store unwritable until reload, so it fails closed at once with `journal_write_failed:`. - `packages/paperclip-runner/src/contracts/native-session-backend.ts`: new reason `runner_event_commit_stalled`, with an optional `detail` in the message. ## Verification Red/green: - "bounds commit-stall reconnects for an out-of-order event": a schema-valid event with the wrong `sourceSeq` fails closed with `event_sequence_mismatch at source seq 5`. Before this it closed silently forever. "… for 20 failures inside 45 s" pins the time bound: 20 failures over 19 s do not fail closed. - "fails closed when checkpoint storage keeps failing before a reconnect authenticates": with checkpoint persistence refusing (`ECONNREFUSED`), 20 reconnect attempts fail before authentication, then the control plane fails closed with `checkpoint_flush_failed:ECONNREFUSED at source seq 1`. Without this, those attempts close silently. - "fails closed at once when the journal cannot record a runner event": a journal write throwing `ENOSPC` fails closed with `journal_write_failed:ENOSPC at source seq 1`, and the private error text is not carried. On master the connection only closes. - `durable-prp-control-plane.test.ts` "fails closed with the schema issue when the runner keeps redelivering an invalid event (restart chaos t01)": an event that fails `validatePrpEvent` is redelivered 20 times over 57 s of mocked clock. On master the control plane only closes; with this PR it fails closed with `runner_event_commit_stalled` and detail `event_schema_invalid:/… at source seq 1`. - `durable-prp-control-plane.test.ts` "fails closed when runner events keep failing to commit without progress": 20 reconnects over 57 s of mocked clock, each commit failing with `ECONNRESET`. On master nothing fails closed. With this PR the 20th failure reports `runner_event_commit_stalled` with detail `ECONNRESET at source seq 1`, and the private error message is not carried. File: 201 passed, including the existing "keeps a non-typed commit failure retryable" cases. Static checks: - `tsc --noEmit` passes for `packages/paperclip-runner` and `server`. The server is unchanged. ## Risks - Storage that keeps failing event commits for at least 45 s and 20 attempts now fails the run (operator recovery) instead of looping silently. Shorter outages are unchanged. - The recorded cause is a code or error name only, never error text, so no tool content or credentials reach the run error. - No schema change and no change to persisted state formats. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../src/contracts/native-session-backend.ts | 8 +- .../durable-prp-control-plane.test.ts | 194 ++++++++++++++++++ .../durable-prp-control-plane.ts | 93 +++++++-- 3 files changed, 281 insertions(+), 14 deletions(-) diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index d4e3a4effea..282d74a83bc 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -188,12 +188,16 @@ export class NativeSessionProtocolIntegrityError extends Error { constructor( readonly reason: | "semantic_input_digest_mismatch" - | "source_event_replay_conflict", + | "source_event_replay_conflict" + | "runner_event_commit_stalled", + readonly detail?: string, ) { super( reason === "semantic_input_digest_mismatch" ? "native_event_replay_conflict: authenticated runner semantic input failed integrity validation; automatic recovery is stopped." - : "native_event_replay_conflict: authenticated runner event conflicts with committed history; automatic recovery is stopped.", + : reason === "runner_event_commit_stalled" + ? `native_event_replay_conflict: runner_event_commit_stalled: runner events kept failing to commit without progress (last cause: ${detail ?? "unknown"}); automatic recovery is stopped.` + : "native_event_replay_conflict: authenticated runner event conflicts with committed history; automatic recovery is stopped.", ); this.name = "NativeSessionProtocolIntegrityError"; } diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index 1332df7ddf6..699d5bdda26 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -1345,6 +1345,200 @@ it.each(["success", "integrity_failure"] as const)("ignores a replaced connectio } }); +it("fails closed with the schema issue when the runner keeps redelivering an invalid event (restart chaos t01)", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-schema-stall-")); + const commits = vi.fn(async () => undefined); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits, onProtocolIntegrityError }); + const envelope = semanticInputEvent(1); + const event = envelope.payload as Record; + event.sourceEventId = "invalid-write-target-1"; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + delete event.schemaVersion; + event.priority = 7; + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + let lease: string | null = null; + for (let attempt = 1; attempt <= 20; attempt += 1) { + const client: AuthenticatedClient = (await authenticate(core, lease ?? core.issueBootstrapTicket()))!; + lease = client.leaseToken ?? lease; + sendSecure(client, envelope); + await vi.waitFor(() => expect(client.socket.destroyed || client.socket.readyState === "closed" + || onProtocolIntegrityError.mock.calls.length > 0).toBe(true)); + clock += 3_000; + client.socket.destroy(); + } + expect(commits).not.toHaveBeenCalled(); + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as { reason: string; detail: string }; + expect(failure.reason).toBe("runner_event_commit_stalled"); + expect(failure.detail).toMatch(/^event_schema_invalid:\/\S* at source seq 1$/); + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("fails closed at once when the journal cannot record a runner event", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-journal-failure-")); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: vi.fn(async () => undefined), onProtocolIntegrityError }); + const envelope = semanticInputEvent(1); + const event = envelope.payload as Record; + event.sourceEventId = "journal-failure-1"; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + try { + await core.start(); + const client: AuthenticatedClient = (await authenticate(core, core.issueBootstrapTicket()))!; + const journal = core.store as unknown as { commit: (...args: unknown[]) => void; commitEvent: (...args: unknown[]) => Promise }; + const diskFull = Object.assign(new Error("no space on /private/disk"), { code: "ENOSPC" }); + vi.spyOn(journal, "commit").mockImplementation(() => { throw diskFull; }); + vi.spyOn(journal, "commitEvent").mockRejectedValue(diskFull); + sendSecure(client, envelope); + await vi.waitFor(() => expect(onProtocolIntegrityError).toHaveBeenCalledOnce()); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as Error & { reason: string; detail: string }; + expect(failure).toMatchObject({ reason: "runner_event_commit_stalled", detail: "journal_write_failed:ENOSPC at source seq 1" }); + expect(failure.message).not.toContain("/private/disk"); + client.socket.destroy(); + } finally { + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("fails closed when checkpoint storage keeps failing before a reconnect authenticates", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-checkpoint-stall-")); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: vi.fn(async () => undefined), onProtocolIntegrityError }); + let checkpointUp = true; + core.setCheckpointPersistence(async () => { + if (!checkpointUp) throw Object.assign(new Error("connect to private-db-host refused"), { code: "ECONNREFUSED" }); + }); + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + const tickets = Array.from({ length: 20 }, () => core.issueBootstrapTicket(60_000)); + await core.flushCheckpoint(); + checkpointUp = false; + for (const ticket of tickets) { + const client = await authenticate(core, ticket); + client?.socket.destroy(); + clock += 3_000; + if (onProtocolIntegrityError.mock.calls.length > 0) break; + } + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as Error & { reason: string; detail: string }; + expect(failure.reason).toBe("runner_event_commit_stalled"); + expect(failure.detail).toMatch(/^checkpoint_flush_failed:ECONNREFUSED at source seq 1$/); + expect(failure.message).not.toContain("private-db-host"); + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it.each([ + { label: "an out-of-order event", seq: 5, failCommit: false, stepMs: 3_000, failsClosed: true, + detail: "event_sequence_mismatch at source seq 5" }, + { label: "20 failures inside 45 s", seq: 1, failCommit: true, stepMs: 1_000, failsClosed: false, detail: null }, +])("bounds commit-stall reconnects for $label", async ({ seq, failCommit, stepMs, failsClosed, detail }) => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-stall-bound-")); + const commits = vi.fn(async () => { + if (failCommit) throw Object.assign(new Error("transient"), { code: "ECONNRESET" }); + }); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits, onProtocolIntegrityError }); + const envelope = semanticInputEvent(seq); + const event = envelope.payload as Record; + event.sourceEventId = `stall-bound-${seq}`; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + let lease: string | null = null; + for (let attempt = 1; attempt <= 20; attempt += 1) { + const client: AuthenticatedClient = (await authenticate(core, lease ?? core.issueBootstrapTicket()))!; + lease = client.leaseToken ?? lease; + sendSecure(client, envelope); + if (failCommit) await vi.waitFor(() => expect(commits).toHaveBeenCalledTimes(attempt)); + await new Promise(resolveTurn => setTimeout(resolveTurn, 20)); + clock += stepMs; + client.socket.destroy(); + } + if (failsClosed) { + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + expect(onProtocolIntegrityError.mock.calls[0]![0]).toMatchObject({ reason: "runner_event_commit_stalled", detail }); + } else { + expect(onProtocolIntegrityError).not.toHaveBeenCalled(); + } + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("fails closed when runner events keep failing to commit without progress", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-delivery-stall-")); + const commits = vi.fn(async () => { + throw Object.assign(new Error("connection reset by private-host-fixture"), { code: "ECONNRESET" }); + }); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits, onProtocolIntegrityError }); + const envelope = semanticInputEvent(1); + const event = envelope.payload as Record; + event.sourceEventId = "stalled-delivery-1"; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + let lease: string | null = null; + for (let attempt = 1; attempt <= 20; attempt += 1) { + const client: AuthenticatedClient = (await authenticate(core, lease ?? core.issueBootstrapTicket()))!; + lease = client.leaseToken ?? lease; + sendSecure(client, envelope); + await vi.waitFor(() => expect(commits).toHaveBeenCalledTimes(attempt)); + await new Promise(resolveTurn => setImmediate(resolveTurn)); + if (attempt < 20) expect(onProtocolIntegrityError).not.toHaveBeenCalled(); + clock += 3_000; + client.socket.destroy(); + } + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as Error; + expect(failure).toMatchObject({ reason: "runner_event_commit_stalled", detail: "ECONNRESET at source seq 1" }); + expect(failure.message).toContain("runner_event_commit_stalled"); + expect(failure.message).not.toContain("private-host-fixture"); + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + function renewalRequest(client: AuthenticatedClient, expiresAt: number): Record { return { protocol: "paperclip.runner", version: client.welcome.version, diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index e2e61d45b72..cf85f3ffbe6 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -73,6 +73,8 @@ class CommandJournalLimitError extends Error { // polling turn regains the event loop. Match the transport's explicit deferred // event bound so a valid burst is not compacted before it can be observed. const maxCommittedEventWindow = 4_096; +const EVENT_DELIVERY_STALL_FAILURES = 20; +const EVENT_DELIVERY_STALL_MS = 45_000; const maxStateBytes = 256 * 1024 * 1024; const authChallengeTtlMs = 5_000; const stableIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/; @@ -870,6 +872,19 @@ function validStoredWarmTransition(state: StoredCoreState): boolean { } } +function eventSchemaIssuePaths(issues: readonly { path?: unknown }[] | undefined): string { + const paths = [...new Set((issues ?? []).map(issue => typeof issue.path === "string" ? issue.path : "") + .map(path => path.replace(/[^A-Za-z0-9_./-]/g, "_").slice(0, 80)).filter(Boolean))]; + return paths.length ? paths.slice(0, 3).join(",") : "unknown"; +} + +function eventCommitFailureCause(error: unknown): string { + const code = isRecord(error) ? error.code : undefined; + if (typeof code === "string" && /^[A-Za-z0-9_.:-]{1,80}$/.test(code)) return code; + const name = error instanceof Error ? error.name : typeof error; + return /^[A-Za-z0-9_.:-]{1,80}$/.test(name) ? name : "unknown_error"; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -2040,6 +2055,7 @@ export class DurablePrpControlPlane { #onProtocolIntegrityError?: DurablePrpControlPlaneOptions["onProtocolIntegrityError"]; #onAuthenticatedConnection?: DurablePrpControlPlaneOptions["onAuthenticatedConnection"]; #protocolIntegrityError: NativeSessionProtocolIntegrityError | null = null; + #eventDeliveryStall: { runId: string; ackedSourceSeq: number; since: number; failures: number } | null = null; #connectionLeaseTtlMs: number; #persistCheckpoint: ((checkpoint: DurablePrpCheckpoint) => Promise) | null = null; #persistedRevision = -1; @@ -2211,10 +2227,15 @@ export class DurablePrpControlPlane { } const checkpoint = this.flushCheckpoint(); const frame = structuredClone(value); + let checkpointFailure: unknown = null; const sending = (this.#sendQueues.get(connection) ?? Promise.resolve()) - .then(() => checkpoint) + .then(() => checkpoint.catch(error => { checkpointFailure = error ?? new Error("checkpoint_failed"); throw error; })) .then(() => connection.sendJson(frame)) - .catch(() => connection.close()); + .catch(() => { + if (checkpointFailure !== null && this.#protocolIntegrityError === null) { + this.#eventCommitFailed(connection, `checkpoint_flush_failed:${eventCommitFailureCause(checkpointFailure)}`); + } else connection.close(); + }); this.#sendQueues.set(connection, sending); void sending.finally(() => { if (this.#sendQueues.get(connection) === sending) this.#sendQueues.delete(connection); @@ -3806,7 +3827,12 @@ export class DurablePrpControlPlane { this.#store.state.commandDeliveryCounts[command.commandId] = (this.#store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; } - this.#store.save(); + try { + this.#store.save(); + } catch (error) { + this.#failEventCommit(connection, `journal_write_failed:${eventCommitFailureCause(error)}`); + return; + } this.#sendJson(connection, { protocol, version: lease.protocolVersion, @@ -4124,6 +4150,27 @@ export class DurablePrpControlPlane { ); } + #eventCommitFailed(connection: AuthorityConnection, cause: string, sourceSeq?: number, now = Date.now()): void { + const acked = this.#store.state.ackedSourceSeq; + const runId = this.#identity.runId; + const stall = this.#eventDeliveryStall; + if (stall === null || stall.ackedSourceSeq !== acked || stall.runId !== runId) { + this.#eventDeliveryStall = { runId, ackedSourceSeq: acked, since: now, failures: 1 }; + } else { + stall.failures += 1; + if (stall.failures >= EVENT_DELIVERY_STALL_FAILURES && now - stall.since >= EVENT_DELIVERY_STALL_MS) { + this.#failEventCommit(connection, cause, sourceSeq); + return; + } + } + connection.close(); + } + + #failEventCommit(connection: AuthorityConnection, cause: string, sourceSeq?: number): void { + this.#failProtocolIntegrity(connection, new NativeSessionProtocolIntegrityError("runner_event_commit_stalled", + `${cause} at source seq ${sourceSeq ?? this.#store.state.ackedSourceSeq + 1}`)); + } + #failProtocolIntegrity( connection: AuthorityConnection, error: NativeSessionProtocolIntegrityError, @@ -4164,7 +4211,7 @@ export class DurablePrpControlPlane { } const validated = validatePrpEvent(envelope.payload); if (!validated.ok) { - connection.close(); + this.#eventCommitFailed(connection, `event_schema_invalid:${eventSchemaIssuePaths(validated.issues)}`); return; } const event = validated.event; @@ -4218,7 +4265,7 @@ export class DurablePrpControlPlane { ? sourceSeq !== this.#store.state.ackedSourceSeq + 1 : sourceSeq !== existing.sourceSeq ) { - connection.close(); + this.#eventCommitFailed(connection, "event_sequence_mismatch", sourceSeq); return; } if ( @@ -4277,7 +4324,7 @@ export class DurablePrpControlPlane { if (error instanceof NativeSessionProtocolIntegrityError) { this.#failProtocolIntegrity(connection, error); } else { - connection.close(); + this.#eventCommitFailed(connection, eventCommitFailureCause(error), sourceSeq); } return; } @@ -4324,13 +4371,28 @@ export class DurablePrpControlPlane { // turn before readers react or a replacement owner adopts their cursor. const candidate = buildCandidate(this.#store.state); if (candidate === null) { - connection.close(); + this.#eventCommitFailed(connection, "journal_sequence_not_contiguous", sourceSeq); + return; + } + try { + this.#store.commit(candidate, !isStreamingEvent(event)); + } catch (error) { + this.#failEventCommit(connection, `journal_write_failed:${eventCommitFailureCause(error)}`, sourceSeq); + return; + } + } else { + let committed: boolean; + try { + committed = await this.#store.commitEvent(buildCandidate, isCurrentOwner, !isStreamingEvent(event)); + } catch (error) { + if (isCurrentOwner()) this.#failEventCommit(connection, `journal_write_failed:${eventCommitFailureCause(error)}`, sourceSeq); + else connection.close(); + return; + } + if (!committed) { + if (isCurrentOwner()) this.#eventCommitFailed(connection, "journal_commit_refused", sourceSeq); return; } - this.#store.commit(candidate, !isStreamingEvent(event)); - } else if (!await this.#store.commitEvent(buildCandidate, isCurrentOwner, !isStreamingEvent(event))) { - if (isCurrentOwner()) connection.close(); - return; } if (!isCurrentOwner()) return; @@ -4350,7 +4412,14 @@ export class DurablePrpControlPlane { if (ownsLiveDispatch) this.#pendingSemanticCalls.add(liveCommandId!); let dispatched = false; try { - await this.flushCheckpoint(); + try { + await this.flushCheckpoint(); + } catch (error) { + if (this.#eventOwner === connection && this.#identity === identity) { + this.#eventCommitFailed(connection, `checkpoint_flush_failed:${eventCommitFailureCause(error)}`, sourceSeq); + } + return; + } if (this.#eventOwner !== connection || this.#identity !== identity) return; if (!this.#retainsSourceInputs() && ownsLiveDispatch && liveCall && liveCommandId && this.#onSemanticToolInput && !this.#getCommand(liveCommandId)) { const call = liveCall;