From 8475826d2d5d81baee5f27e349e979cd706aeb3a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 02:14:45 -0600 Subject: [PATCH] fix(wake-queue): deliver user comments whose wake a restart lost or a stale continuation dropped (#19992) 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 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 --- .../heartbeat-held-comment-wake.test.ts | 113 +++++++++++++++++- .../ownership-reconciliation-report.test.ts | 30 +++++ server/src/services/heartbeat.ts | 7 ++ server/src/services/heartbeat/queue.ts | 6 +- .../heartbeat/stale-comment-continuation.ts | 46 +++++++ .../native-finalization-reconciler.ts | 39 +++++- .../recovery/ownership-invariant-report.ts | 100 +++++++++++++++- server/src/services/work-intent-outbox.ts | 78 +++++++++++- tests/restart-chaos/grader.test.ts | 21 ++++ tests/restart-chaos/grader.ts | 27 +++-- 10 files changed, 446 insertions(+), 21 deletions(-) create mode 100644 server/src/services/heartbeat/stale-comment-continuation.ts diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index 2ee971fd4fb..24856d54ef2 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/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([ diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 730729d996b..d584a990654 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/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", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index ad1a131686d..785f697d1fe 100644 --- a/server/src/services/heartbeat.ts +++ b/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", diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 73dec05433f..0f0ebdf8f7d 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/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} diff --git a/server/src/services/heartbeat/stale-comment-continuation.ts b/server/src/services/heartbeat/stale-comment-continuation.ts new file mode 100644 index 00000000000..bedd51564e8 --- /dev/null +++ b/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 { + const context = run.contextSnapshot && typeof run.contextSnapshot === "object" ? run.contextSnapshot as Record : {}; + 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 }); +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 45e15f3feb7..7e22e0c20bf 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/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; diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 7650e5ffdda..269bb7e153d 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/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 { + 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; diff --git a/server/src/services/work-intent-outbox.ts b/server/src/services/work-intent-outbox.ts index 22c5b763635..89034210e03 100644 --- a/server/src/services/work-intent-outbox.ts +++ b/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>'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; +}): Promise { + 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; + }); +} diff --git a/tests/restart-chaos/grader.test.ts b/tests/restart-chaos/grader.test.ts index ec65b2a3e5e..416fc214eb0 100644 --- a/tests/restart-chaos/grader.test.ts +++ b/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 = {}) => 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)]; diff --git a/tests/restart-chaos/grader.ts b/tests/restart-chaos/grader.ts index 4dd4d2cc109..aaead120c04 100644 --- a/tests/restart-chaos/grader.ts +++ b/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[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();