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