From 98e0f1fc0d972113cb14e299d54cd8ac6c2e7780 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 03:03:15 -0600 Subject: [PATCH] fix(native-runtime): a stopped run cannot close a task another run continues; deliver a pre-restart comment at once (#20216) 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 > - When the board stops a run and then posts a comment asking the agent to continue, that comment starts a new run, and only that new run should decide when the task is done > - In restart chaos wave `w-20261109T081456-s9901`, t16 (claude S1, K1) closed the task as done without the line the user asked for, and t10 (codex S1, K6) never woke a run for the continue comment, so it needed a Retry > - In t16, the stopped run's saved "done" result was accepted 10 s after the continue run started. In t10, a restart killed the server 21 ms after the comment committed, before its wake was written, and the comment backstop then waited out its 30 s grace > - This pull request stops a run from closing a task that another run now holds, and makes the backstop deliver at once a comment that predates the server process > - The benefit is that a user's latest direction after a Stop is neither overridden by the stopped run nor lost to a restart ## Linked Issues or Issue Description Refs #19831. Restart chaos wave `w-20261109T081456-s9901` (master `09e24dd12c`), tasks t16 and t10. **t16, from the kept database (CHA-20):** - 08:29:19: the board stopped run 1 (`6e33d123`). Its saved result reported done, but finalization failed on the stop (`finalization_failed_claim_preserved`, a retryable failure). - 08:29:41: the user's continue comment started run 2 (`2cb71c5b`). Run 2 held the task's execution and saw the comment. - 08:29:51: the finalization reconciler retried run 1's saved result and committed `completion_claim_policy_accepted` for run 1. The task closed as done. - From then on, every Paperclip tool call from run 2 was refused (`paperclip_runner_tool_binding_not_authorized`), so it could not add the `CONTINUE-…` line. Its result (blocked) was then preserved as `terminal_status_preserved`. - Run 2 ending `succeeded` with `native_session_lease_lost` follows from this: its result was accepted before the lease was lost. The comment was delivered; the model saw it. **t10 (CHA-14):** - The Stop moved the task to blocked. The user's comment reopened it to todo at 08:26:28.633, and K6 killed the server at 08:26:28.654, before the route wrote the comment's wake. No wake request exists for that comment. - This is not caused by the Stop, #20066 or #20180: the wake is written after the comment commits, and the kill landed in between. - The I4 backstop ("a user comment without a wake") matched the comment, but treated it as possibly still getting its normal wake for 30 s. The pass interval then pushed the check to 02:27:16, and the harness clicked Retry at 02:26:59. ## What Changed - `status-arbiter.ts`: - New `heldByAnotherRun(issue, runId)`. - `supersedeStaleOwnerCompletion` also keeps status, with reason `completion_run_superseded`, when the issue's `executionRunId` is a different run. A run that was reassigned away still gets `completion_owner_superseded`. - `native-run-finalizer.ts` and `native-status-reassessment.ts` pass the run id to that guard. - `status-decision-committer.ts` rechecks the same rule under the issue lock: a done decision for a non-conversation issue held by another run raises `NativeStatusRaceError`. The finalizer then re-arbitrates and keeps status. - `ownership-invariant-report.ts`: the comment-without-wake grace now only applies to comments created after this server process started (`getServerProcessStartedAt`). Separate live server replicas are not supported, and the wake is written by the process that accepted the comment, so an older comment is owed at once. The startup pass now delivers it. `runOwnershipReconciliationPass` takes an optional `processStartedAt`. ## Verification Red/green: - `native-stale-owner-completion.test.ts`: three new tests. - The unit policy: another run holding the task keeps status; the same run, or none, still closes it. - "does not let a stopped run's saved done result close the task a newer run is continuing" (the t16 shape). - "does not close the task when a newer run takes it after arbitration but before the status commit" (only the committer recheck covers this). - On master: 3 failed. With this PR: 13 passed. - `ownership-reconciliation-report.test.ts`: "finds at once a young user comment whose wake a restart lost after the board stopped the run" (the t10 shape). It fails on master and passes with this PR; the same comment in a process that accepted it still waits for its grace deadline. The file: 27 passed. Regression and static checks: - 16 nearby native finalization, status projection, workspace finalizer and arbiter suites: 439 passed. - `tsc --noEmit` passes for `server`, and for the two test files. ## Risks - A done result now waits when a different run holds the task. That run reaches its own decision; the earlier result stays as history. If the holder is stale, the existing execution-lock recovery clears it, and a later reassessment can then close the task. - After a restart, a user comment newer than the task's last run, with no wake, is delivered at the startup pass rather than 30 s later. A replica that is still live and writing the same wake is not a supported topology. - No schema change and no change to persisted run state. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- .../ownership-reconciliation-report.test.ts | 23 +++++- .../native-runtime/native-run-finalizer.ts | 2 +- .../native-stale-owner-completion.test.ts | 70 +++++++++++++++++++ .../native-status-reassessment.ts | 2 +- .../services/native-runtime/status-arbiter.ts | 14 +++- .../status-decision-committer.ts | 4 ++ .../recovery/ownership-invariant-report.ts | 13 ++-- 7 files changed, 117 insertions(+), 11 deletions(-) diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 309f0ab683f..f6ce47f45d5 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -457,17 +457,36 @@ describe("report-only ownership reconciliation", () => { 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 }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false, + processStartedAt: minutesAgo(10) }); 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 }); + const later = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false, + processStartedAt: minutesAgo(10) }); expect(later.violations.filter(row => row.reason === "user_comment_without_wake")).toEqual([]); }); + it("finds at once a young user comment whose wake a restart lost after the board stopped the run", async () => { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Stopped then continued", status: "todo", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "cancelled", invocationSource: "assignment", + runtimeMode: "native", contextSnapshot: { issueId }, startedAt: minutesAgo(4), createdAt: minutesAgo(4), finishedAt: minutesAgo(1) }); + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", + body: "I stopped your previous run on purpose. Please continue.", createdAt: new Date(Date.now() - 12_000) }).returning(); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false, + processStartedAt: new Date(Date.now() - 2_000) }); + expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId)) + .toEqual([comment!.id]); + const sameProcess = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false, + processStartedAt: minutesAgo(10) }); + expect(sameProcess.violations.filter(row => row.reason === "user_comment_without_wake")).toEqual([]); + expect(sameProcess.nextDueAt?.getTime()).toBe(comment!.createdAt.getTime() + 30_000); + }); + it("finds a user comment that a restart left without a wake on a task its agent then closed", async () => { const closedTask = async (actorType: "agent" | "user") => { const issueId = randomUUID(); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 069a415d072..52d3aa178fc 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1604,7 +1604,7 @@ export async function finalizeNativeRun(input: { // A reassigned-away run keeps its accepted result, but cannot close the // new owner's task. The status committer rechecks this under the issue lock. if (!isConversation(authoritativeIssue)) { - proposedDecision = supersedeStaleOwnerCompletion(proposedDecision, authoritativeIssue, run.agentId); + proposedDecision = supersedeStaleOwnerCompletion(proposedDecision, authoritativeIssue, run.agentId, run.id); } if (proposedDecision.statusAction === "done") { const continuation = record(run.contextSnapshot?.executionContinuation); diff --git a/server/src/services/native-runtime/native-stale-owner-completion.test.ts b/server/src/services/native-runtime/native-stale-owner-completion.test.ts index 3541caa00b3..3c1bb94b1a7 100644 --- a/server/src/services/native-runtime/native-stale-owner-completion.test.ts +++ b/server/src/services/native-runtime/native-stale-owner-completion.test.ts @@ -27,6 +27,12 @@ describe("stale owner completion policy", () => { expect(supersedeStaleOwnerCompletion(done, issue, "agent-a")).toEqual({ ...done, statusAction: "preserve", toStatus: "in_progress", reasonCode: "completion_owner_superseded", effects: [] }); }); + it("preserves status when another run of the same owner now holds the task", () => { + expect(supersedeStaleOwnerCompletion(done, { ...owner, executionRunId: "run-2" }, "agent-a", "run-1")).toEqual({ ...done, + statusAction: "preserve", toStatus: "in_progress", reasonCode: "completion_run_superseded", effects: [] }); + expect(supersedeStaleOwnerCompletion(done, { ...owner, executionRunId: "run-1" }, "agent-a", "run-1")).toBe(done); + expect(supersedeStaleOwnerCompletion(done, { ...owner, executionRunId: null }, "agent-a", "run-1")).toBe(done); + }); it("leaves non-terminal decisions to their existing owner checks", () => { const blocked: NativeStatusDecision = { ...done, statusAction: "blocked", toStatus: "blocked", reasonCode: "x" }; expect(supersedeStaleOwnerCompletion(blocked, { ...owner, assigneeAgentId: "agent-b" }, "agent-a")).toBe(blocked); @@ -92,6 +98,46 @@ describe.skipIf(!support.supported)("native completion after reassignment", () = .where(eq(heartbeatRuns.id, f.runId)); } + /** The board's Stop, then the user's continue comment starts a new run that holds the task. */ + async function stopAndContinue(f: Awaited>) { + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "cancelled", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + const continuationRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: continuationRunId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, status: "running", runtimeMode: "native", + contextSnapshot: { issueId: f.issueId, source: "issue.comment" } }); + await db.update(issues).set({ executionRunId: continuationRunId, checkoutRunId: continuationRunId }) + .where(eq(issues.id, f.issueId)); + return continuationRunId; + } + + function actBeforeStatusCommit(act: () => Promise) { + let acted = false; + const racing = new Proxy(db, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver); + if (property !== "update") return typeof value === "function" ? value.bind(target) : value; + return (table: unknown) => { + const builder = target.update(table as typeof nativeRunFinalizations); + if (table !== nativeRunFinalizations || acted) return builder; + const set = builder.set.bind(builder); + return Object.assign(builder, { set: (values: Record) => { + const next = set(values as never); + if (values.phase !== "arbitrating" || !("assessmentId" in values)) return next; + const where = next.where.bind(next); + return Object.assign(next, { where: (...args: Parameters) => { + const query = where(...args); + return { then: (resolve: (value: unknown) => unknown, reject: (error: unknown) => unknown) => + query.then(async (result) => { acted = true; await act(); return result; }).then(resolve, reject) }; + } }); + } }); + }; + }, + }) as Db; + return { racing, acted: () => acted }; + } + const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; const decision = async (runId: string) => { const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); @@ -154,6 +200,30 @@ describe.skipIf(!support.supported)("native completion after reassignment", () = expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", assigneeAgentId: f.successorId }); expect((await decision(f.runId)).row).toMatchObject({ reasonCode: "completion_owner_superseded" }); }); + it("does not let a stopped run's saved done result close the task a newer run is continuing", async () => { + const f = await acceptedDoneResult(); + const continuationRunId = await stopAndContinue(f); + + await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", executionRunId: continuationRunId }); + const { coordinator, row } = await decision(f.runId); + expect(coordinator.phase).toBe("committed"); + expect(row).toMatchObject({ toStatus: "in_progress", reasonCode: "completion_run_superseded" }); + }); + + it("does not close the task when a newer run takes it after arbitration but before the status commit", async () => { + const f = await acceptedDoneResult(); + let continuationRunId: string | null = null; + const { racing, acted } = actBeforeStatusCommit(async () => { continuationRunId = await stopAndContinue(f); }); + + await finalizeNativeRun({ db: racing, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + + expect(acted()).toBe(true); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", executionRunId: continuationRunId }); + expect((await decision(f.runId)).row).toMatchObject({ reasonCode: "completion_run_superseded" }); + }); + async function approveEvidenceLater(f: Awaited>) { await db.update(issueWorkProducts).set({ reviewState: "approved", updatedAt: new Date(Date.now() + 1_000) }) .where(eq(issueWorkProducts.id, f.workProductId)); diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index 1bc389852a5..225d979dc35 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -236,7 +236,7 @@ export async function reassessNativeStatusFromSource( } // New evidence from a reassigned-away run cannot close the new owner's task. if (currentIssue && !currentIssue.conversationAgentId) { - decision = supersedeStaleOwnerCompletion(decision, { ...currentIssue, status: row.issueStatus }, row.agentId); + decision = supersedeStaleOwnerCompletion(decision, { ...currentIssue, status: row.issueStatus }, row.agentId, row.runId); } let committed: Awaited>; try { diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 97eaf9079b0..ba5bf640d92 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -103,17 +103,25 @@ export function ownsNativeTaskCompletion( return issue.assigneeAgentId === agentId && !issue.assigneeUserId; } +export function heldByAnotherRun(issue: { executionRunId?: string | null }, runId: string) { + return Boolean(issue.executionRunId && issue.executionRunId !== runId); +} + export function supersedeStaleOwnerCompletion( decision: NativeStatusDecision, - issue: { status: string; assigneeAgentId: string | null; assigneeUserId: string | null }, + issue: { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; + executionRunId?: string | null }, agentId: string, + runId?: string, ): NativeStatusDecision { - if (decision.statusAction !== "done" || ownsNativeTaskCompletion(issue, agentId)) return decision; + if (decision.statusAction !== "done") return decision; + const ownerSuperseded = !ownsNativeTaskCompletion(issue, agentId); + if (!ownerSuperseded && !(runId && heldByAnotherRun(issue, runId))) return decision; return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "preserve", toStatus: issue.status as NativeAuthoritativeIssueStatus, - reasonCode: "completion_owner_superseded", + reasonCode: ownerSuperseded ? "completion_owner_superseded" : "completion_run_superseded", unblockDescriptor: null, effects: [], }; diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index b192e3f670c..29089f8ddf4 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -46,6 +46,7 @@ import { workspaceOperations, } from "@paperclipai/db"; import { + heldByAnotherRun, NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeGovernanceGate, type NativeStatusDecision, @@ -1793,6 +1794,9 @@ export async function commitNativeStatusDecision(input: { if (input.decision.statusAction === "done" && !await onboardingFirstTaskCompletionAllowed(tx as unknown as Db, { companyId: input.companyId, issueId: input.issueId, runId: input.runId, })) throw new NativeStatusRaceError(); + if (input.decision.statusAction === "done" && !isConversation(issue) && heldByAnotherRun(issue, input.runId)) { + throw new NativeStatusRaceError(); + } if (input.decision.statusAction === "done") { try { const [saved] = await tx.select({ context: heartbeatRuns.contextSnapshot, receipts: heartbeatRuns.resultJson, diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 0bc8aecc468..c00cb3b4517 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -9,6 +9,7 @@ import { import { taskClosedByAgentSql, TEMPORARY_REFUSAL_TRIGGERS } from "../work-intent-outbox.js"; import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; import { logger } from "../../middleware/logger.js"; +import { getServerProcessStartedAt } from "../../server-info.js"; import { persistActivity } from "../activity-log.js"; import { currentProcessProbe, proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js"; import { LEGACY_OWNER_REASONS, recordVersionAttestations } from "./owner-gone-attestation.js"; @@ -466,7 +467,7 @@ async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Prom * 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 }> { +async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): 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 @@ -498,8 +499,10 @@ async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number): P 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; + // when the grace ends (a saved deadline), and report only older ones. This + // process writes that wake after the comment commits, so a comment from + // before it started has no writer left and is owed now. + const graceEnd = Math.max(now.getTime() - COMMENT_WAKE_GRACE_MS, processStartedAt.getTime()); 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) @@ -594,6 +597,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: { limitPerKind?: number; probe?: ProcessProbe; now?: Date; + processStartedAt?: Date; /** Write new violations to the activity log. Defaults to true. */ record?: boolean; /** Settle proven-gone records of kinds that register a `settle`, and surface @@ -686,7 +690,8 @@ export async function runOwnershipReconciliationPass(db: Db, options: { () => checkExactlyOnceIntents(db, state, now, limit), () => checkStaleDroppedComments(db, now, limit), async () => { - const result = await checkUserCommentsWithoutWake(db, now, limit); + const result = await checkUserCommentsWithoutWake(db, now, limit, + options.processStartedAt ?? new Date(getServerProcessStartedAt())); if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt; return result.violations; },