fix(wake-queue): deliver user comments whose wake a restart lost or a stale continuation dropped (#19992)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A user's follow-up comment on a task must reach the agent exactly
once, through restarts (invariant 4)
> - The final unmasked local chaos storm (96 tasks, 31 restarts) passed
94 tasks. The 2 failures were both lost user comments
> - In t69, a restart killed the server between saving the comment and
writing its wake. In t72, a comment's own run was cancelled as stale,
because the earlier run completed the task first without reading the
comment
> - This pull request adds a reconciler rule that gives a lost comment
its wake, and saves the comment before a stale cancellation
> - The benefit is that a user's message is never silently dropped when
a restart or a completion race hits at the wrong moment

## Linked Issues or Issue Description

Refs #19956 (work-intent outbox), #19944, #19831 (plan, invariant 4).

**What happened?**
In chaos wave `w-20261108T055444-s9401` (local, unmasked, codex and
claude, 30-min K13 storm, master `46bf913fb9`), 94 of 96 tasks passed.
Two failed:
- **t69, R1 `rested_not_done`.** The user's comment was saved at
06:14:59.358 and reopened the task to `todo`. A K1 SIGKILL fired 28 ms
later, before the wake was written. No wake row exists. Recovery skips a
`todo` task whose last run succeeded, so the task sat with no run until
the wave ended.
- **t72, R4 `concurrent_runs`.** A comment posted during a run queued
its own run. That run was claimed at 06:16:04.874. The first run's
completion marked the task `done` 2 ms later. The comment's run was then
cancelled with `continuation_task_ownership_changed` before dispatch,
and the comment was lost.

**Expected behavior**
Each accepted user comment is delivered once, unless the task is
cancelled or reassigned, or a person closes it.

**Steps to reproduce**
`node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts
--replay w-20261108T055444-s9401/t69` (or `.../t72`).

## What Changed

- `recovery/ownership-invariant-report.ts`: a new reconciler-owned I4
rule, `user_comment_without_wake`. It finds a user comment that meets
all of these conditions:
  - it is newer than every run of the task;
- the task is assigned, is not a conversation, has no live run or lock,
and has at least one earlier run;
  - no wake request carries the comment.

A comment younger than 30 s is not reported yet, because it may still
get its normal wake. Its end-of-grace time is returned as `nextDueAt`.
- `native-runtime/native-finalization-reconciler.ts`: the backstop step
is now exported as `applyOwnershipReportBackstops`. For each such
comment it saves the missing outbox receipt (`executionWait: wake_lost`)
and promotes it.
- `work-intent-outbox.ts`: new `saveUndeliveredUserComments`, which the
reconciler rule and the stale-run path both use. It saves only live user
comments that no completed run received. It writes nothing for a task
that is reassigned, is a conversation, or has a status the caller does
not allow.
- `heartbeat/queue.ts`: the saved-receipt pass that already sends
pause-saved messages also sends `wake_lost` receipts. It uses the same
per-comment consumption check and the same chat-input exclusion.
- `heartbeat/stale-comment-continuation.ts` and `heartbeat.ts`: before a
stale continuation is cancelled, its undelivered user comments are saved
as a deferred receipt. The release drain then reopens the `done` task
and delivers them once. This applies only when an agent run that started
before the comment completed the task after it. A task that a person
marks done is left closed.
- `tests/restart-chaos/grader.ts`: R4 ignores runs cancelled before
dispatch (`continuation_task_ownership_changed`) when it looks for
overlapping execution. Those runs never executed. Comment consumption
still counts them.

## Verification

Each focused file ran alone in one process (`--maxWorkers=1`):
- `heartbeat-held-comment-wake.test.ts` (22 pass). This includes two new
cases:
- t69 shape: the reconciler rule fires, the backstop delivers the
comment once, and a second pass does nothing;
- t72 shape: the stale run is cancelled, the task reopens, and the
comment is delivered once.

  Both fail with the rule or the save disabled, and pass with them.
- `ownership-reconciliation-report.test.ts` (23 pass). It includes the
rule's boundaries: a carried comment, a too-young comment (deadline
only) and an already-read comment.
- `tests/restart-chaos/grader.test.ts` (17 pass). The new R4 case fails
with the old grader. A cancelled run that is not a pre-dispatch
cancellation still fails R4.
- Unchanged and passing: comment wake batching (27), refused assignment
wake (3), wake-queue postgres adapter (103), stale queue invalidation
(68), deferred receipt work (45). The first version also reopened a task
that a person had marked done, which broke one deferred-receipt test.
That case is now excluded.
- The server package typecheck and the restart-chaos harness typecheck
pass.
- Same-session chaos context on master: K6 crash loop
`w-20261108T063326-s9402`: 6 of 6 pass. The K4 long-hold wave is still
running. After merge, I'll replay only the targeted fault shapes (t69
and t72).

## Risks

- The reconciler rule runs on each report pass. It is bounded by the
1-hour lookback and the per-kind row cap, and it excludes conversation
tasks, so chat input keeps its own authorization.
- A comment younger than 30 s waits up to the end of its grace period
before the backstop acts.
- The stale-run save reopens a `done` task only for a comment that the
completing run could not have read.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code
execution in Claude Code.

## 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>

server/src/__tests__/heartbeat-held-comment-wake.test.ts

@@ -4,7 +4,7 @@ import { tmpdir } from "node:os";
 import path from "node:path";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { agents, agentWakeupRequests, authUsers, budgetPolicies, companies, companyMemberships, costEvents, createDb, heartbeatRuns,
+import { activityLog, agents, agentWakeupRequests, authUsers, budgetPolicies, companies, companyMemberships, costEvents, createDb, heartbeatRuns,
   issueComments, issues } from "@paperclipai/db";
 import { accountRunCost } from "../services/run-cost-accounting.js";
 import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
@@ -12,6 +12,8 @@ import { requestTaskRetry } from "../services/task-retry.js";
 import { queueIssueAssignmentWakeup } from "../services/issue-assignment-wakeup.js";
 import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js";
 import { heartbeatService } from "../services/heartbeat.js";
+import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
+import { applyOwnershipReportBackstops } from "../services/native-runtime/native-finalization-reconciler.js";
 import { promoteDeferredWorkIntents, TEMPORARY_REFUSAL_TRIGGERS, WORK_INTENT_TERMINAL_OUTCOMES } from "../services/work-intent-outbox.js";
 import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recovery-postgres.js";
 
@@ -116,6 +118,12 @@ describePostgres("saved intents while another run's accounting is pending", () =
     return woken;
   }
 
+  /** The agent's run closed the task, as a native status decision records it. */
+  const closeByAgent = (f: { companyId: string; issueId: string; agentId: string }, actorType = "agent") =>
+    db.insert(activityLog).values({ companyId: f.companyId, actorType, actorId: actorType === "agent" ? f.agentId : "person",
+      action: "issue.updated", entityType: "issue", entityId: f.issueId,
+      details: { source: "native_status_decision", fromStatus: "in_progress", toStatus: "done" } });
+
   async function settle(f: { companyId: string }) {
     await heartbeat.resumeQueuedRuns();
     await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId });
@@ -394,10 +402,111 @@ describePostgres("saved intents while another run's accounting is pending", () =
     expect(started).toHaveLength(1);
   }, 180_000);
 
+  // Chaos s9401/t69: a restart landed between the comment commit (which
+  // reopened the finished task to todo) and its wake write. Reconciler rule:
+  // an undelivered user comment newer than the task's last run gets a wake.
+  it("delivers once a user comment whose wake a restart lost after a succeeded run", async () => {
+    const f = await seedHeldComment();
+    await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId));
+    // The run that finished the task was native, as in the wave, and the
+    // comment came after it; the comment's wake was never written.
+    await db.update(heartbeatRuns).set({ runtimeMode: "native", startedAt: new Date(Date.now() - 120_000),
+      createdAt: new Date(Date.now() - 120_000) }).where(eq(heartbeatRuns.id, f.sourceRunId));
+    await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId));
+    await db.update(issueComments).set({ createdAt: new Date(Date.now() - 60_000), updatedAt: new Date(Date.now() - 60_000) })
+      .where(eq(issueComments.id, f.commentId));
+    await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.issueId));
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId))
+      .toEqual([f.commentId]);
+    await applyOwnershipReportBackstops(db, report);
+    await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) })
+      .where(eq(agentWakeupRequests.companyId, f.companyId));
+    await settle(f); await settle(f);
+    expect(deliveries(f.commentId)).toHaveLength(1);
+    // Delivered: the rule no longer fires, and nothing delivers it again.
+    const again = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(again.violations.filter(row => row.reason === "user_comment_without_wake")).toEqual([]);
+    await applyOwnershipReportBackstops(db, again);
+    await settle(f);
+    expect(deliveries(f.commentId)).toHaveLength(1);
+  }, 180_000);
+
+  // Chaos s9401/t72: a message queued its own run while the first run was
+  // finishing. The first run completed the task, so the queued run was
+  // cancelled as a stale continuation and the unread message was dropped.
+  it("reopens a finished task for a message whose queued run turned stale, and delivers it once", async () => {
+    const f = await seedHeldComment();
+    await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId));
+    await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId));
+    // The message's own queued run; the first run then completed the task.
+    const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation",
+      triggerDetail: "system", reason: "issue_commented", status: "queued", requestedByActorType: "user", requestedByActorId: f.userId,
+      payload: { issueId: f.issueId, commentId: f.commentId, mutation: "comment" } }).returning();
+    const [stale] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation",
+      triggerDetail: "system", status: "queued", wakeupRequestId: wake.id,
+      contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: f.commentId, wakeCommentId: f.commentId,
+        wakeCommentIds: [f.commentId], wakeReason: "issue_commented", source: "issue.comment" } }).returning();
+    await db.update(agentWakeupRequests).set({ runId: stale.id }).where(eq(agentWakeupRequests.id, wake.id));
+    // The first run, which started before the message, completed the task after it.
+    await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() + 1_000) }).where(eq(heartbeatRuns.id, f.sourceRunId));
+    await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId));
+    await closeByAgent(f);
+    await settle(f); await settle(f);
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, stale.id)))[0]).toMatchObject({
+      status: "cancelled", errorCode: "continuation_task_ownership_changed" });
+    await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) })
+      .where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution")));
+    await settle(f); await settle(f);
+    expect(deliveries(f.commentId)).toHaveLength(1);
+    await settle(f);
+    expect(deliveries(f.commentId)).toHaveLength(1);
+  }, 180_000);
+
+  // The durable half: the stale check ran before the closing run settled (or
+  // its save failed), so only the cancelled stale wake carries the message.
+  // Once the closing run settles, the reconciler saves it and the drain
+  // reopens the task and delivers it once.
+  it("recovers a message a stale continuation dropped once the closing run settles", async () => {
+    const f = await seedHeldComment();
+    await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId));
+    await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(),
+      error: "continuation_task_ownership_changed" }).where(eq(agentWakeupRequests.id, f.wakeId));
+    await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() + 1_000) }).where(eq(heartbeatRuns.id, f.sourceRunId));
+    await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId));
+    // A person closed it: that decision stands.
+    await closeByAgent(f, "user");
+    const byPerson = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(byPerson.violations.filter(row => row.reason === "user_comment_dropped_by_stale_continuation")).toEqual([]);
+    // The agent's run closed it later without reading the message.
+    await closeByAgent(f);
+    // Another message's own turn succeeded later; it did not receive this one.
+    await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation",
+      triggerDetail: "system", status: "succeeded", startedAt: new Date(Date.now() + 2_000), finishedAt: new Date(Date.now() + 3_000),
+      contextSnapshot: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentIds: [randomUUID()],
+        executionContinuation: { messages: [{ id: f.commentId }] } } });
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(report.violations.filter(row => row.reason === "user_comment_dropped_by_stale_continuation")
+      .map(row => row.recordId)).toEqual([f.commentId]);
+    expect(await applyOwnershipReportBackstops(db, report)).toEqual({ retryNeeded: false });
+    await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) })
+      .where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution")));
+    await settle(f); await settle(f);
+    expect(deliveries(f.commentId)).toHaveLength(1);
+    const again = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(again.violations.filter(row => row.recordId === f.commentId)).toEqual([]);
+  }, 180_000);
+
+  it("keeps a reconciler pass due when the backstop cannot save a message", async () => {
+    const report = { violations: [{ rule: "I4_exactly_once_intent", kind: "issue_comment", recordId: randomUUID(),
+      companyId: "not-a-company-id", runId: null, issueId: "not-an-issue-id", reason: "user_comment_without_wake" }] };
+    expect(await applyOwnershipReportBackstops(db, report as never)).toEqual({ retryNeeded: true });
+  });
+
   it("keeps the terminal outcomes closed and every temporary refusal mapped to a release trigger", () => {
     expect([...WORK_INTENT_TERMINAL_OUTCOMES]).toEqual(["delivered", "consumed", "task_cancelled", "hard_stop"]);
     for (const refusal of ["budget.blocked", "company.inactive", "agent.not_invokable", "issue_tree_hold_active",
-      "heartbeat.scheduling_suppressed"]) expect(TEMPORARY_REFUSAL_TRIGGERS[refusal]).toBeTruthy();
+      "heartbeat.scheduling_suppressed", "wake_lost"]) expect(TEMPORARY_REFUSAL_TRIGGERS[refusal]).toBeTruthy();
   });
 
   it.each([

server/src/__tests__/ownership-reconciliation-report.test.ts

@@ -13,6 +13,7 @@ import {
   createDb,
   environmentLeases,
   heartbeatRuns,
+  issueComments,
   issues,
   nativeRunFinalizations,
   sandboxCreationOperations,
@@ -406,6 +407,35 @@ describe("report-only ownership reconciliation", () => {
     expect(report.violations.filter(violation => violation.rule === "I4_exactly_once_intent")).toEqual([]);
   });
 
+  // Reconciler rule (chaos s9401/t69): an undelivered user comment newer than
+  // the task's last run gets a wake. A carried, a too-young, or an
+  // already-read comment does not fire it.
+  it("finds a user comment newer than the task's last run that no wake carries", async () => {
+    const issueId = randomUUID();
+    await db.insert(issues).values({ id: issueId, companyId, title: "Reopened", status: "todo", assigneeAgentId: agentId });
+    await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", invocationSource: "manual",
+      runtimeMode: "native", contextSnapshot: { issueId }, startedAt: minutesAgo(5), createdAt: minutesAgo(5), finishedAt: minutesAgo(4) });
+    const [lost, carried, young] = await db.insert(issueComments).values([
+      { companyId, issueId, authorUserId: "user-1", body: "lost wake", createdAt: minutesAgo(2) },
+      { companyId, issueId, authorUserId: "user-1", body: "carried", createdAt: minutesAgo(2) },
+      { companyId, issueId, authorUserId: "user-1", body: "young", createdAt: new Date(Date.now() - 5_000) },
+    ]).returning();
+    // An agent's own comment written under its user (local CLI) is not user intent.
+    await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", authorAgentId: agentId,
+      body: "self", createdAt: minutesAgo(2) });
+    await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "queued",
+      payload: { issueId, commentId: carried.id } });
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId)).toEqual([lost.id]);
+    // The young comment may still get its normal wake: a deadline, not a report.
+    expect(report.nextDueAt?.getTime()).toBe(young.createdAt.getTime() + 30_000);
+    // A run that started after the comment read it.
+    await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", invocationSource: "manual",
+      contextSnapshot: { issueId }, startedAt: minutesAgo(1), createdAt: minutesAgo(1), finishedAt: new Date() });
+    const later = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
+    expect(later.violations.filter(row => row.reason === "user_comment_without_wake")).toEqual([]);
+  });
+
   it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => {
     const updatedAt = minutesAgo(1);
     await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution",

server/src/services/heartbeat.ts

@@ -320,6 +320,7 @@ import {
 import { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js";
 export { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js";
 import { StaleExecutionContinuationError } from "./execution-continuation.js";
+import { deferUndeliveredCommentsOfStaleRun } from "./heartbeat/stale-comment-continuation.js";
 import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils";
 
 import {
@@ -3796,6 +3797,12 @@ export function heartbeatService(
         // The queued continuation became obsolete before adapter dispatch.
         // Use cancellation settlement so wakeup, issue ownership, agent state,
         // and notifications agree; do not retry work for the previous owner.
+        // A user's message it carried that no run read is saved first, so the
+        // release reopens the finished task for it (work-intent outbox). If
+        // this fast path fails or the closing run has not settled yet, the
+        // ownership reconciler saves it once that run settles.
+        await deferUndeliveredCommentsOfStaleRun(db, run).catch(err =>
+          logger.warn({ err, runId: run.id }, "could not save the messages of a stale continuation"));
         await cancelRunInternal(run.id, outerErr.code, {
           errorCode: outerErr.code,
           eventMessage: "stale execution continuation cancelled before dispatch",

server/src/services/heartbeat/queue.ts

@@ -1779,8 +1779,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
         // which resumes through its own live authorization), and only pauses
         // this query can see lifted. A refusal by any other gate rewrites the
         // receipt's reason, so it never retries in a loop here.
-        sql`${agentWakeupRequests.payload}->'executionWait'->>'trigger' = 'pause_lifted'`,
-        sql`${agentWakeupRequests.payload}->'executionWait'->>'reason' in ('company.inactive', 'agent.not_invokable')`,
+        // A message the reconciler saved after its wake was lost waits on no
+        // hold, so the same explicit wake sends it.
+        sql`${agentWakeupRequests.payload}->'executionWait'->>'trigger' in ('pause_lifted', 'reconciler_backstop')`,
+        sql`${agentWakeupRequests.payload}->'executionWait'->>'reason' in ('company.inactive', 'agent.not_invokable', 'wake_lost')`,
         eq(agentWakeupRequests.requestedByActorType, "user"),
         sql`coalesce(${agentWakeupRequests.idempotencyKey}, '') not like 'chat-inbound:%'`,
         sql`exists (select 1 from agents pause_agent where pause_agent.id = ${agentWakeupRequests.agentId}

server/src/services/heartbeat/stale-comment-continuation.ts

@@ -0,0 +1,46 @@
+import { agentWakeupRequests, heartbeatRuns, issueComments, type Db } from "@paperclipai/db";
+import { and, eq, inArray, sql } from "drizzle-orm";
+import { saveUndeliveredUserComments, taskClosedByAgentSql } from "../work-intent-outbox.js";
+
+const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+
+/**
+ * Work-intent outbox (invariant 4). A user's message can wake its own queued
+ * run while an earlier run of the task is still finishing. When that earlier
+ * run completes the task first, the queued run reaches a done task and is
+ * cancelled as a stale continuation before dispatch, but the earlier run never
+ * read the message. Cancelling it alone would drop the message (chaos
+ * s9401/t72).
+ *
+ * Before that cancellation, the message's undelivered comments are saved as a
+ * deferred receipt of the same user. The run's cancellation releases the task,
+ * and the release drain reopens the done task for the user's message and
+ * delivers it once. Only a task that an agent run completed without reading
+ * the message qualifies: that run started before the comment and succeeded
+ * after it. A task a person marked done ends the message's intent.
+ * Returns the saved receipt id, or null when nothing is owed.
+ */
+export async function deferUndeliveredCommentsOfStaleRun(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise<string | null> {
+  const context = run.contextSnapshot && typeof run.contextSnapshot === "object" ? run.contextSnapshot as Record<string, unknown> : {};
+  const issueId = typeof context.issueId === "string" ? context.issueId : null;
+  const commentIds = Array.isArray(context.wakeCommentIds)
+    ? context.wakeCommentIds.filter((id): id is string => typeof id === "string" && UUID.test(id)) : [];
+  if (!issueId || !UUID.test(issueId) || commentIds.length === 0) return null;
+  const unread = await db.select({ id: issueComments.id }).from(issueComments).where(and(
+    eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, commentIds),
+    sql`exists (select 1 from ${heartbeatRuns} completing where completing.company_id = ${run.companyId}
+      and completing.agent_id = ${run.agentId} and completing.id <> ${run.id} and completing.status = 'succeeded'
+      and (completing.context_snapshot->>'issueId' = ${issueId} or completing.native_issue_id = ${issueId})
+      and completing.started_at < ${issueComments.createdAt} and completing.finished_at >= ${issueComments.createdAt})`));
+  const owed = commentIds.filter(id => unread.some(row => row.id === id));
+  if (owed.length === 0) return null;
+  // An overlapping run is not proof the agent closed the task: a person can
+  // mark it done while that run finishes. Only an agent-made close reopens.
+  const [closed] = await db.execute<{ byAgent: boolean }>(sql`select ${taskClosedByAgentSql(run.companyId, issueId)} as "byAgent"`);
+  if (closed?.byAgent !== true) return null;
+  const [wake] = run.wakeupRequestId ? await db.select().from(agentWakeupRequests)
+    .where(and(eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.id, run.wakeupRequestId))) : [];
+  return saveUndeliveredUserComments(db, { companyId: run.companyId, issueId, agentId: run.agentId, commentIds: owed,
+    allowedStatuses: ["done"], requestedByActorId: wake?.requestedByActorType === "user" ? wake.requestedByActorId : null,
+    source: wake?.source, triggerDetail: wake?.triggerDetail });
+}

server/src/services/native-runtime/native-finalization-reconciler.ts

@@ -18,12 +18,14 @@ import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator
 import {
   createOwnershipReportState,
   DEFERRED_INTENT_NOT_PROMOTED,
+  USER_COMMENT_DROPPED_BY_STALE_RUN,
+  USER_COMMENT_WITHOUT_WAKE,
   ownershipReportFingerprint,
   recheckSettlingKinds,
   runOwnershipReconciliationPass,
   type OwnershipReport,
 } from "../recovery/ownership-invariant-report.js";
-import { promoteDeferredWorkIntents } from "../work-intent-outbox.js";
+import { deferredWorkIntentWait, promoteDeferredWorkIntents, saveUndeliveredUserComments } from "../work-intent-outbox.js";
 import { createHash, randomUUID } from "node:crypto";
 import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, not, notInArray, or, sql, type SQL } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -824,13 +826,37 @@ export const OWNERSHIP_REPORT_INTERVAL_MS = 30_000;
 /** Section E backstop: a deferred intent whose hold cleared without a
  * release event is promoted through the one promotion function. Promotion
  * re-runs normal admission, so a wrong report never bypasses a gate. */
-async function promoteMissedDeferredIntents(db: Db, report: OwnershipReport) {
+export async function applyOwnershipReportBackstops(db: Db, report: OwnershipReport): Promise<{ retryNeeded: boolean }> {
+  let retryNeeded = false;
+  // Rules: an undelivered user comment newer than the task's last run gets a
+  // wake, and a comment a stale continuation dropped reopens its done task.
+  // Save the missing outbox entry, then promote it.
+  for (const violation of report.violations.filter(row =>
+    row.reason === USER_COMMENT_WITHOUT_WAKE || row.reason === USER_COMMENT_DROPPED_BY_STALE_RUN)) {
+    if (!violation.companyId || !violation.issueId) continue;
+    const lost = violation.reason === USER_COMMENT_WITHOUT_WAKE;
+    try {
+      const [issue] = await db.select({ assigneeAgentId: issues.assigneeAgentId }).from(issues)
+        .where(and(eq(issues.companyId, violation.companyId), eq(issues.id, violation.issueId)));
+      if (!issue?.assigneeAgentId) continue;
+      const saved = await saveUndeliveredUserComments(db, { companyId: violation.companyId, issueId: violation.issueId,
+        agentId: issue.assigneeAgentId, commentIds: [violation.recordId], allowedStatuses: lost ? ["todo", "in_progress"] : ["done"],
+        ...(lost ? { executionWait: deferredWorkIntentWait("wake_lost", "This message's wake was lost; it is sent now.") } : {}) });
+      if (saved) await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: violation.companyId!,
+        trigger: "reconciler_backstop", issueId: violation.issueId }));
+    } catch (error) {
+      // The message still has no saved receipt: keep a pass due.
+      retryNeeded = true;
+      logger.warn({ err: error, commentId: violation.recordId }, "could not save a user comment without a wake");
+    }
+  }
   const companies = new Set(report.violations.filter(violation => violation.reason === DEFERRED_INTENT_NOT_PROMOTED)
     .flatMap(violation => violation.companyId ? [violation.companyId] : []));
   for (const companyId of companies) {
     await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId, trigger: "reconciler_backstop" }))
-      .catch(error => logger.warn({ err: error, companyId }, "deferred intent backstop promotion failed"));
+      .catch(error => { retryNeeded = true; logger.warn({ err: error, companyId }, "deferred intent backstop promotion failed"); });
   }
+  return { retryNeeded };
 }
 
 export function registerOwnershipReconciliationWork(
@@ -895,12 +921,13 @@ export function registerOwnershipReconciliationWork(
       lastPassAt = Date.now();
       pendingWake = false;
       const report = await runOwnershipReconciliationPass(db, { state });
-      await promoteMissedDeferredIntents(db, report);
+      const backstop = await applyOwnershipReportBackstops(db, report);
       const fingerprint = ownershipReportFingerprint(report);
       const changing = report.inProgress > 0 && fingerprint !== lastFingerprint;
       // A settling kind's live owner can die without writing a signal, so
-      // keep one bounded re-check due while such records exist.
-      const fullPassDue = changing || report.awaitingSecondProof > 0 || report.retryNeeded;
+      // keep one bounded re-check due while such records exist. A backstop
+      // that could not save or promote keeps one due as well.
+      const fullPassDue = changing || report.awaitingSecondProof > 0 || report.retryNeeded || backstop.retryNeeded;
       watchOnly = !fullPassDue && report.settleWatch > 0;
       const followUp = fullPassDue || report.settleWatch > 0
         ? Date.now() + OWNERSHIP_REPORT_INTERVAL_MS : null;

server/src/services/recovery/ownership-invariant-report.ts

@@ -6,7 +6,8 @@ import {
   issues,
   type Db,
 } from "@paperclipai/db";
-import { TEMPORARY_REFUSAL_TRIGGERS } from "../work-intent-outbox.js";
+import { taskClosedByAgentSql, TEMPORARY_REFUSAL_TRIGGERS } from "../work-intent-outbox.js";
+import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js";
 import { logger } from "../../middleware/logger.js";
 import { persistActivity } from "../activity-log.js";
 import { proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js";
@@ -366,6 +367,97 @@ async function checkDeferredIntentsPromoted(db: Db, now: Date, limit: number): P
 }
 
 export const DEFERRED_INTENT_NOT_PROMOTED = "deferred_after_hold_cleared";
+export const USER_COMMENT_WITHOUT_WAKE = "user_comment_without_wake";
+/** A user comment younger than this may still be getting its normal wake. */
+const COMMENT_WAKE_GRACE_MS = 30_000;
+
+export const USER_COMMENT_DROPPED_BY_STALE_RUN = "user_comment_dropped_by_stale_continuation";
+
+/** Invariant 4, the durable half of the stale-continuation rule (chaos
+ * s9401/t72). A user comment on a done task whose only carrying wakes were
+ * cancelled as stale continuations, that a settled agent run which started
+ * before it never read (it finished after the comment), and that no later
+ * run received (the shared per-comment delivery check). The decision waits for the closing run to settle, so it
+ * also covers a stale check that ran before that run reached succeeded, or a
+ * save that failed. The worker saves it so the release drain reopens the
+ * task and delivers it once. */
+async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Promise<InvariantViolation[]> {
+  const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString();
+  const rows = await db.execute<{ id: string; company_id: string; issue_id: string; agent_id: string }>(sql`
+    select c.id, c.company_id, c.issue_id, i.assignee_agent_id as agent_id from issue_comments c
+    join issues i on i.id = c.issue_id and i.company_id = c.company_id
+    where c.author_user_id is not null and c.deleted_at is null
+      and c.author_agent_id is null and c.created_by_run_id is null and c.created_at >= ${since}::timestamptz
+      and i.status = 'done' and i.assignee_agent_id is not null and i.conversation_agent_id is null
+      and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}
+      and exists (select 1 from agent_wakeup_requests stale where stale.company_id = c.company_id
+        and stale.payload->>'issueId' = c.issue_id::text and stale.status = 'cancelled'
+        and stale.error = 'continuation_task_ownership_changed'
+        and (stale.payload->>'commentId' = c.id::text
+          or coalesce(stale.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
+      and not exists (select 1 from agent_wakeup_requests open_wake where open_wake.company_id = c.company_id
+        and open_wake.payload->>'issueId' = c.issue_id::text
+        and open_wake.status not in ('cancelled', 'skipped', 'failed', 'coalesced')
+        and (open_wake.payload->>'commentId' = c.id::text
+          or coalesce(open_wake.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
+      and exists (select 1 from heartbeat_runs closing where closing.company_id = c.company_id
+        and closing.agent_id = i.assignee_agent_id and closing.status = 'succeeded'
+        and (closing.context_snapshot->>'issueId' = c.issue_id::text or closing.native_issue_id = c.issue_id)
+        and closing.started_at < c.created_at and closing.finished_at >= c.created_at)
+      and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id
+        and (r.context_snapshot->>'issueId' = c.issue_id::text or r.native_issue_id = c.issue_id)
+        and r.status in ('queued', 'running', 'scheduled_retry'))
+    order by c.created_at limit ${Math.min(limit, MAX_RECORDS_PER_KIND)}`);
+  // Consumption is per comment: a later run counts only if it received this
+  // comment (its own wake input, or a retry's task history), never ordinary
+  // task history of another message's turn.
+  const owed: typeof rows[number][] = [];
+  for (const row of rows) {
+    const delivered = await deliveredQueuedCommentIds(db, { companyId: row.company_id, issueId: row.issue_id,
+      agentId: row.agent_id, commentIds: [row.id] });
+    if (!delivered.has(row.id)) owed.push(row);
+  }
+  return owed.map(row => ({ rule: "I4_exactly_once_intent" as const, kind: "issue_comment", recordId: row.id,
+    companyId: row.company_id, runId: null, issueId: row.issue_id, reason: USER_COMMENT_DROPPED_BY_STALE_RUN }));
+}
+
+/** Invariant 4, reconciler-owned rule: an undelivered user comment newer than
+ * the task's last run gets a wake. A comment commits before its wake is
+ * written, so a restart in between loses the wake (chaos s9401/t69). This
+ * finds a user comment on an assigned, non-conversation task with no live run
+ * or lock, newer than every run of the task, that no wake request carries.
+ * The reconciliation worker saves it as a deferred receipt and promotes it. */
+async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
+  const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString();
+  const rows = await db.execute<{ id: string; company_id: string; issue_id: string; created_at: string }>(sql`
+    select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from issue_comments c
+    join issues i on i.id = c.issue_id and i.company_id = c.company_id
+    where c.author_user_id is not null and c.deleted_at is null
+      and c.author_agent_id is null and c.created_by_run_id is null
+      and c.created_at >= ${since}::timestamptz
+      and i.assignee_agent_id is not null and i.conversation_agent_id is null
+      and i.status in ('todo', 'in_progress') and i.execution_run_id is null
+      and exists (select 1 from heartbeat_runs prior where prior.company_id = c.company_id
+        and (prior.context_snapshot->>'issueId' = c.issue_id::text or prior.native_issue_id = c.issue_id))
+      and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id
+        and (r.context_snapshot->>'issueId' = c.issue_id::text or r.native_issue_id = c.issue_id)
+        and (r.status in ('queued', 'running', 'scheduled_retry') or coalesce(r.started_at, r.created_at) >= c.created_at))
+      and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id
+        and w.payload->>'issueId' = c.issue_id::text
+        and (w.payload->>'commentId' = c.id::text
+          or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
+    order by c.created_at limit ${Math.min(limit, MAX_RECORDS_PER_KIND)}`);
+  // A comment inside the grace may still get its normal wake: check it again
+  // when the grace ends (a saved deadline), and report only older ones.
+  const graceEnd = now.getTime() - COMMENT_WAKE_GRACE_MS;
+  const young = [...rows].map(row => new Date(row.created_at).getTime()).filter(at => at >= graceEnd);
+  return {
+    violations: [...rows].filter(row => new Date(row.created_at).getTime() < graceEnd)
+      .map(row => ({ rule: "I4_exactly_once_intent" as const, kind: "issue_comment", recordId: row.id,
+        companyId: row.company_id, runId: null, issueId: row.issue_id, reason: USER_COMMENT_WITHOUT_WAKE })),
+    nextDueAt: young.length ? new Date(Math.min(...young) + COMMENT_WAKE_GRACE_MS) : null,
+  };
+}
 
 /** Invariant 4 with section E's closed outcomes: an accepted intent (a user's
  * request or a message) ended terminally on a refusal that
@@ -536,6 +628,12 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
   for (const check of [
     () => checkAdmissionIndependence(db, state, now, limit),
     () => checkExactlyOnceIntents(db, state, now, limit),
+    () => checkStaleDroppedComments(db, now, limit),
+    async () => {
+      const result = await checkUserCommentsWithoutWake(db, now, limit);
+      if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt;
+      return result.violations;
+    },
     async () => {
       const result = await checkDeferredIntentsPromoted(db, now, limit);
       if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt;

server/src/services/work-intent-outbox.ts

@@ -1,6 +1,8 @@
-import { and, eq, sql } from "drizzle-orm";
-import { agentWakeupRequests, type Db } from "@paperclipai/db";
-import { notifyDeferredExecutionEligibility, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js";
+import { and, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm";
+import { agentWakeupRequests, issueComments, issues, type Db } from "@paperclipai/db";
+import { notifyDeferredExecutionEligibility, notifyDeferredExecutionWork, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js";
+import { withQueuedCommentIdsInWakePayload } from "./issue-queued-comment-queue.js";
+import { deliveredQueuedCommentIds } from "./queued-comment-delivery.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 
@@ -37,6 +39,8 @@ export const TEMPORARY_REFUSAL_TRIGGERS: Readonly<Record<string, WorkIntentHoldT
   "agent.not_invokable": "pause_lifted",
   "issue_tree_hold_active": "tree_hold_released",
   "heartbeat.scheduling_suppressed": "scheduling_resumed",
+  /** A user message whose wake was lost (the reconciler saved it). */
+  wake_lost: "reconciler_backstop",
 };
 
 export function temporaryRefusalTrigger(reason: string): WorkIntentHoldTrigger | null {
@@ -78,3 +82,71 @@ export async function promoteDeferredWorkIntents(tx: object, input: {
     input.issueId ? sql`${agentWakeupRequests.payload}->>'issueId' = ${input.issueId}` : undefined,
   ));
 }
+
+const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
+
+/** SQL: the latest recorded `done` transition of the task was not made by a
+ * person. A task a person closed (or with no recorded close) is never
+ * reopened for an earlier message. */
+export function taskClosedByAgentSql(companyId: unknown, issueId: unknown) {
+  return sql`coalesce((select closed.actor_type from activity_log closed
+    where closed.company_id = ${companyId} and closed.entity_type = 'issue' and closed.entity_id = ${issueId}::text
+      and closed.action = 'issue.updated'
+      and (closed.details->>'toStatus' = 'done' or closed.details->>'status' = 'done')
+    order by closed.created_at desc limit 1), 'user') <> 'user'`;
+}
+
+/**
+ * Record the missing outbox entry for user messages whose wake was lost or
+ * ended without delivering them: one deferred receipt of the user that
+ * carries the undelivered comments. Only the task's assigned agent, and only
+ * live user comments that no completed run received, are saved. Statuses the
+ * caller allows (for example `done`, which the release drain reopens for a
+ * user's message) gate it. Normal admission and the release drain deliver
+ * the receipt once. Returns the receipt id, or null when nothing is owed.
+ */
+export async function saveUndeliveredUserComments(db: Db, input: {
+  companyId: string; issueId: string; agentId: string; commentIds: string[];
+  allowedStatuses: readonly string[]; requestedByActorId?: string | null;
+  source?: string | null; triggerDetail?: string | null;
+  /** Names why the receipt waits and what promotes it. */
+  executionWait?: Record<string, unknown>;
+}): Promise<string | null> {
+  if (input.commentIds.length === 0) return null;
+  return db.transaction(async tx => {
+    const [issue] = await tx.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId,
+      conversationAgentId: issues.conversationAgentId }).from(issues)
+      .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update");
+    // A reassigned or cancelled task ends the intent; a conversation keeps
+    // its own inbound authorization path.
+    if (!issue || issue.assigneeAgentId !== input.agentId || issue.conversationAgentId ||
+        !input.allowedStatuses.includes(issue.status)) return null;
+    if (issue.status === "done") {
+      const [closed] = await tx.execute<{ byAgent: boolean }>(sql`select ${taskClosedByAgentSql(input.companyId, input.issueId)} as "byAgent"`);
+      if (closed?.byAgent !== true) return null;
+    }
+    const live = await tx.select({ id: issueComments.id, authorUserId: issueComments.authorUserId }).from(issueComments).where(and(
+      eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId),
+      inArray(issueComments.id, input.commentIds), isNotNull(issueComments.authorUserId), isNull(issueComments.deletedAt),
+      // A direct human comment: no agent author and no creating run.
+      isNull(issueComments.authorAgentId), isNull(issueComments.createdByRunId)));
+    const delivered = await deliveredQueuedCommentIds(tx as unknown as Db, { companyId: input.companyId,
+      issueId: input.issueId, agentId: input.agentId, commentIds: live.map(row => row.id) });
+    const remaining = input.commentIds.filter(id => live.some(row => row.id === id) && !delivered.has(id));
+    if (remaining.length === 0) return null;
+    const latest = remaining.at(-1)!;
+    await notifyDeferredExecutionWork(tx, { companyId: input.companyId });
+    await notifyDeferredWakesForIssue(tx, { companyId: input.companyId, issueId: input.issueId });
+    const [saved] = await tx.insert(agentWakeupRequests).values({
+      companyId: input.companyId, agentId: input.agentId, source: input.source ?? "automation",
+      triggerDetail: input.triggerDetail ?? "system", reason: "issue_commented", status: "deferred_issue_execution",
+      requestedByActorType: "user",
+      requestedByActorId: input.requestedByActorId ?? live.find(row => row.id === latest)?.authorUserId ?? null,
+      payload: withQueuedCommentIdsInWakePayload({ issueId: input.issueId, mutation: "comment", commentId: latest,
+        ...(input.executionWait ? { executionWait: input.executionWait } : {}),
+        [DEFERRED_WAKE_CONTEXT_KEY]: { issueId: input.issueId, taskId: input.issueId, wakeReason: "issue_commented",
+          source: "issue.comment", wakeCommentIds: remaining, wakeCommentId: latest } }, remaining),
+    }).returning({ id: agentWakeupRequests.id });
+    return saved?.id ?? null;
+  });
+}

tests/restart-chaos/grader.test.ts

@@ -127,6 +127,27 @@ describe("R1–R6 grading on fixture timelines", () => {
     expect(gradeR4(record(afterSuccess)).detail).toMatch(/comment_consumed_twice: c1 by r1,r2/);
   });
 
+  // s9401/t72: the message's run was claimed just before the first run ended,
+  // then cancelled as stale before dispatch; a later run delivered the message.
+  it("passes R4 for a stale continuation cancelled before dispatch whose message a later run delivered once", () => {
+    const samples = [sample(90_000, "done", [
+      run("r1", "succeeded", 1_000, 40_000),
+      run("r2", "cancelled", 30_000, 40_060, { startedAt: iso(39_960), wakeCommentIds: ["c1"],
+        errorCode: "continuation_task_ownership_changed" }),
+      run("r3", "succeeded", 45_000, 80_000, { wakeCommentIds: ["c1"] }),
+    ])];
+    expect(gradeR4(record(samples))).toMatchObject({ verdict: "pass" });
+    // A poll that saw both running (claiming marks running before dispatch)
+    // does not change that.
+    const polled = [sample(39_990, "in_progress", [run("r1", "running", 1_000), run("r2", "running", 30_000, null, { startedAt: iso(39_960) })]),
+      ...samples];
+    expect(gradeR4(record(polled))).toMatchObject({ verdict: "pass" });
+    // Any other cancelled run that overlapped still executed.
+    const executed = [sample(90_000, "done", [run("r1", "succeeded", 1_000, 40_000),
+      run("r2", "cancelled", 30_000, 40_060, { startedAt: iso(39_960), errorCode: "cancelled" })])];
+    expect(gradeR4(record(executed)).detail).toMatch(/concurrent_runs/);
+  });
+
   it("fails R5 on a silent stall longer than three minutes after ready, but not during outages or named waits", () => {
     const idle = (at: number, extra: Partial<TaskSample> = {}) => sample(at, "in_progress", [run("r1", "failed", 1_000, 30_000)], extra);
     const stalled = [idle(60_000), idle(60_000 + STALL_LIMIT_MS / 2), idle(60_000 + STALL_LIMIT_MS + 5_000)];

tests/restart-chaos/grader.ts

@@ -103,6 +103,9 @@ export function gradeR3(record: TaskRecord): RuleResult {
   return { verdict: "pass", detail: "one Retry started a run that completed the task" };
 }
 
+/** Run cancellations decided before adapter dispatch: the run never executed. */
+const PRE_DISPATCH_CANCELLATIONS = new Set(["continuation_task_ownership_changed"]);
+
 /**
  * R4 is about duplicate execution. A queued or scheduled wake waiting behind a
  * running run (for example a follow-up comment, cancelled when the running run
@@ -114,22 +117,32 @@ export function gradeR3(record: TaskRecord): RuleResult {
  * after a successful run is a replay.
  */
 export function gradeR4(record: TaskRecord): RuleResult {
-  for (const sample of reachable(record.samples)) {
-    const running = (sample.runs ?? []).filter(run => run.status === "running");
-    if (running.length > 1) return { verdict: "fail", detail: `concurrent_runs: ${running.map(run => `${run.id}(${run.status})`).join(",")}` };
-  }
   const finalRuns = new Map<string, NonNullable<TaskSample["runs"]>[number]>();
   for (const sample of reachable(record.samples)) for (const run of sample.runs ?? []) finalRuns.set(run.id, run);
+  // Claiming marks a run running before dispatch, so a run that ended as a
+  // pre-dispatch cancellation never executed, whatever a poll saw.
+  const neverExecuted = (id: string) => {
+    const final = finalRuns.get(id);
+    return final?.status === "cancelled" && PRE_DISPATCH_CANCELLATIONS.has(final.errorCode ?? "");
+  };
+  for (const sample of reachable(record.samples)) {
+    const running = (sample.runs ?? []).filter(run => run.status === "running" && !neverExecuted(run.id));
+    if (running.length > 1) return { verdict: "fail", detail: `concurrent_runs: ${running.map(run => `${run.id}(${run.status})`).join(",")}` };
+  }
   const begun = [...finalRuns.values()].filter(run => run.startedAt)
     .sort((a, b) => (ms(a.createdAt) || 0) - (ms(b.createdAt) || 0));
+  // A continuation cancelled as stale before adapter dispatch was claimed but
+  // never executed (chaos s9401/t72), so it cannot overlap another run. Its
+  // message is saved and delivered by a later run, counted once below.
+  const executed = begun.filter(run => !(run.status === "cancelled" && PRE_DISPATCH_CANCELLATIONS.has(run.errorCode ?? "")));
   // Execution windows seen between polls: started → finished (or still going at the last sample).
   const lastAt = Math.max(0, ...reachable(record.samples).map(sample => sample.at));
   const window = (run: (typeof begun)[number]) => ({ start: ms(run.startedAt) || 0,
     end: run.finishedAt ? ms(run.finishedAt) || 0 : ACTIVE_RUN_STATUSES.has(run.status) ? lastAt : ms(run.startedAt) || 0 });
-  for (let index = 0; index < begun.length; index += 1) for (let other = index + 1; other < begun.length; other += 1) {
-    const [a, b] = [window(begun[index]!), window(begun[other]!)];
+  for (let index = 0; index < executed.length; index += 1) for (let other = index + 1; other < executed.length; other += 1) {
+    const [a, b] = [window(executed[index]!), window(executed[other]!)];
     if (a.start < b.end && b.start < a.end) {
-      return { verdict: "fail", detail: `concurrent_runs: ${begun[index]!.id}(${begun[index]!.status}),${begun[other]!.id}(${begun[other]!.status}) executed together` };
+      return { verdict: "fail", detail: `concurrent_runs: ${executed[index]!.id}(${executed[index]!.status}),${executed[other]!.id}(${executed[other]!.status}) executed together` };
     }
   }
   const consumers = new Map<string, typeof begun>();