fix(runner): fail closed with a recorded cause when runner event commits stall (#20235)

## 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 `<instance>:<run>: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:<issue paths>` 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:<cause>`.
- 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",
"<cause> 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:<cause>`. 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:<code>`.
- `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 <noreply@anthropic.com>

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";
   }

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<string, unknown>;
+  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<string, unknown>;
+  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<boolean> };
+    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<string, unknown>;
+  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<void>(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<string, unknown>;
+  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<void>(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<string, unknown> {
   return {
     protocol: "paperclip.runner", version: client.welcome.version,

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<string, unknown> {
   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<void>) | 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;