From 56c2b28c04bdfc729c4577a0a39e9bc89c0fbb1d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 18:08:17 -0600 Subject: [PATCH] Settle saved monitor results after a newer user request (#19644) 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. > - Native results must finish their durable finalization before later work can reuse a saved workspace. > - An accepted result can wait on a monitor while its workspace files are recovered. > - A newer user request can replace that old wait before finalization resumes. > - The old finalizer then reports lost monitor authority and blocks the saved-workspace lineage. > - This pull request proves the newer user request and settles only the old result bookkeeping. > - The benefit is released recovery work without replaying an old turn or changing the current request. ## Linked Issues or Issue Description Refs #17641 and #16975. **What happened?** An older successful native result has saved files and a server-issued monitor receipt. Its task is now blocked after a newer user-comment continuation failed during preparation. The old coordinator stays in retryable_failure with monitor_wait_authority_lost. The retained allocation ownership check correctly refuses the unresolved predecessor, so later file recovery also stays held. **Expected behavior** The normal finalizer can settle the old accepted result when exact newer user-request evidence proves the old wait is superseded. It must preserve the current task, user input, ownership, schedule, accepted result, and attempt history. It must not start provider work or publish the old wait summary. **Steps to reproduce** Create an accepted successful monitor result and mark workspace finalization successful. Leave the task blocked. Record a newer canonical user comment and its ordinary user wake, either queued or consumed by a same-task continuation that retains the exact comment revision and source-run lineage. Call the real native finalizer. Before the repair, both cases remain retryable_failure instead of settling. **Paperclip version or commit** e482f0f632, before this repair. ## What Changed - Add a company-scoped passive supersession proof in the native monitor owner. It applies only when the old wait is not currently eligible. The proof binds the old accepted result and monitor receipt to a newer canonical unedited user comment and matching user wake. Admitted successors must retain the exact task, comment revision, and source lineage. - Recheck the proof under the existing task transaction, with nonblocking shared locks on the comment and wake; contention uses the existing status-race retry path. Commit a preserve decision with no effects. Keep normal monitor authority, Stop, workspace finalization, and retained-allocation guards unchanged. - Add real PostgreSQL coverage for queued and consumed input, current successor ownership, newer schedules, immutable accepted results, attempt history, foreign/missing/cancelled evidence, and changes before commit. Real scheduler ticks create no old monitor wake. - Include the present superseding proof in the existing assessment identity. The old failed assessment and decision stay immutable, while the new decision gets its own assessment. Absent proof keeps the existing fingerprint unchanged. - Document passive finalization after a newer user request. ## Verification - Behavioral RED before production edits: two real PostgreSQL finalizer cases returned monitor_wait_authority_lost/retryable_failure instead of settling after a newer queued or consumed user request. Initial interrupted imports and rejected database setup were not counted as behavioral failures. - Additional real-boundary RED: the normal finalization reconciler reconstructed both persisted errors but failed on status_decisions_company_assessment_uq because the changed proof reused an older assessment identity. The owned PostgreSQL log confirmed the constraint. Including only the present proof in the existing fingerprint makes both reconciler cases pass while old assessments and decisions remain immutable. - Review RED: real comment-row and wake-row contention produced side_effect_planning_failed. A test-local 100ms lock timeout bounds this diagnostic. Nonblocking proof locks and `55P03`→NativeStatusRaceError safely defer both cases, preserve current state, and commit after the original lock owner releases. Both regressions pass without weakening the production lock or authority checks. - Final local GREEN at `0c40331d8d7f587c9a532fbc51d358bfacd1bd46`: five files, 111 passing tests (126.17s), including all 54 saved-monitor cases, actual persisted-error reconciliation with an absent retry deadline and older boot identity, the status arbiter/task-monitor/terminal-status suites, and six model fixtures. Current requests, schedules, holds, accepted results and prior assessments/decisions remain unchanged; normal attempt increments and idempotent repeat are asserted. Real scheduler ticks create no old monitor wake. - All seven server typecheck layers, module-boundary and whitespace checks pass at the final inputs. Required preflight passes in 15.348s. No new code comments were added. Local environment: macOS arm64, Node 26.4.0, pinned pnpm 9.15.4, Vitest 5.0.3, owned disposable PostgreSQL 18 with bounded 1GiB tmpfs; no shared Docker data was pruned. - [Final-head CI](https://github.com/paperclipai/paperclip/actions/runs/42311889918): 59 successful checks,three intentional skips,one neutral optional review,zero failed or pending checks. Required E2E and verify aggregates pass. Greptile reviewed this exact head at 5/5 with no unresolved findings. - The [original support job](https://github.com/paperclipai/paperclip/actions/runs/42311889918/job/128217026059) passed 6849 assertions and hit one 30000ms timeout in an unchanged private-evidence fixture. The exact case passed locally; all 41 cases passed together (42.79s). The [identical-input sibling CI control](https://github.com/paperclipai/paperclip/actions/runs/42311878548/job/128216598362) passed all 6850 assertions and the 41-case file in 5.417s. Exactly one targeted replay was authorized and dispatched after workflow completion and ordinary quota reset, preserving the other 57 passing checks. [Replay job128227417618](https://github.com/paperclipai/paperclip/actions/runs/42311889918/job/128227417618) passed 323 files/6850 assertions with four existing skips; the 41-case file passed in 5.665s. No fixture, source, assertion, timeout, credentials or full workflow was changed/replayed. The original timeout remains recorded; its precise host-level cause was not determined. - Inherited baseline failures were handled through normal integration of existing #19610 (model expectations), #19627 (deterministic Slack deadline fixture), and #19646 (current trusted Runner lane flags). All four exact lane-command dry runs and the lane-inventory regression pass. No force push or new production patch was used for those failures. - Hosted recovery remains pending deployment and acceptance through the normal pipeline; these checks do not claim the live task has already recovered. ## Risks - The new proof must remain passive. It cannot grant execution or restore an obsolete task state. Missing or changing proof leaves the old result held. - The query reads only the same company's task and agent, with requests newer than the old completion time and one selected result. It runs during active finalization and adds no timer or idle polling. - Edited comments, cancelled wakes, foreign lineage, source Stop, and workspace failure are refused. Existing workspace snapshot, stop, and handoff checks still own physical file recovery. - Hosted acceptance remains pending deployment through the normal pipeline. ## Model Used - OpenAI Codex (GPT-6 family), with code execution and repository tools. The runtime does not expose the exact serving model identifier or context window size. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 13 ++ .../native-runtime/native-monitor-delivery.ts | 62 +++++- .../native-preserved-monitor-wait.test.ts | 208 +++++++++++++++++- .../native-runtime/native-run-finalizer.ts | 12 +- .../services/native-runtime/status-arbiter.ts | 6 + .../status-decision-committer.ts | 19 +- .../native-runtime/work-assessments.ts | 4 +- 7 files changed, 316 insertions(+), 8 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6c8d070b092..f8f657777a8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -774,6 +774,19 @@ durable wake and an admitted successor run. The old run must no longer own the task's checkout or execution. This settlement preserves the task and successor ownership, and neither re-arms the timer nor starts another provider turn. Workspace persistence must still succeed first. + +When the old monitor wait is no longer eligible, a newer ordinary user-comment +wake can supersede that old saved wait. +The finalizer requires the canonical unedited user comment, the matching user wake, +and the accepted old result with its server-issued monitor receipt. An admitted +successor must retain the exact task, comment revision, and source-run lineage. +The finalizer rechecks this proof under the task lock. Assessment fingerprints +include the exact superseding request proof, so a prior failed assessment remains +immutable when the finalization decision changes. It records a passive +settlement and preserves the current task status, assignment, checkout, schedule, +and newer request. It does not re-arm the old monitor, publish its old waiting +summary, or start another provider turn. Missing, cancelled, foreign, or conflicting +proof keeps the old result held. Source Stop and workspace persistence gates remain. If a saved monitor expires during workspace recovery, settlement releases the old run into the still-persisted schedule. The scheduler then applies its timeout or attempt-limit recovery policy. Expiration still rejects new monitor-wait diff --git a/server/src/services/native-runtime/native-monitor-delivery.ts b/server/src/services/native-runtime/native-monitor-delivery.ts index b2af0e80605..2fe57b7edc7 100644 --- a/server/src/services/native-runtime/native-monitor-delivery.ts +++ b/server/src/services/native-runtime/native-monitor-delivery.ts @@ -1,5 +1,5 @@ -import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db, type issues } from "@paperclipai/db"; -import { and, desc, eq, inArray, isNotNull, ne, sql } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRuns, issueComments, nativeRunFinalizations, nativeRunResults, type Db, type issues } from "@paperclipai/db"; +import { and, desc, eq, gt, inArray, isNotNull, isNull, ne, notInArray, or, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import { boundRunIssueIdSql } from "../run-task-binding.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; @@ -21,7 +21,14 @@ export async function readNativePreservedMonitorWait( issue.executionRunId !== null || issue.checkoutRunId !== null) return null; const nextCheckAt = scheduledIssueMonitorWait(issue, binding.agentId); if (!nextCheckAt) return null; - const [source] = await db.select({ resultId: nativeRunResults.id, hasReceipt: sql`exists ( + const source = await readCompletedNativeMonitorResult(db, binding); + return source?.hasReceipt ? { agentId: binding.agentId, nextCheckAt, resultId: source.resultId } : null; +} + +type MonitorBinding = { companyId: string; issueId: string; runId: string; agentId: string }; + +async function readCompletedNativeMonitorResult(db: Db, binding: MonitorBinding) { + const [source] = await db.select({ resultId: nativeRunResults.id, finishedAt: heartbeatRuns.finishedAt, hasReceipt: sql`exists ( select 1 from jsonb_each(case when jsonb_typeof(${heartbeatRuns.resultJson}->'semanticToolReceipts') = 'object' then ${heartbeatRuns.resultJson}->'semanticToolReceipts' else '{}'::jsonb end) receipt where receipt.value->>'operationId' = 'set_task_monitor' @@ -41,7 +48,54 @@ export async function readNativePreservedMonitorWait( ...["cancellation", "nativeCancellation", "executionCancellation", "startupCancellation"].map(key => sql`coalesce(${heartbeatRuns.resultJson}->${key}, 'null'::jsonb) = 'null'::jsonb`), )).limit(1); - return source?.hasReceipt ? { agentId: binding.agentId, nextCheckAt, resultId: source.resultId } : null; + return source?.hasReceipt ? source : null; +} + +export type NativeSupersededMonitorWait = { + agentId: string; resultId: string; wakeId: string; commentId: string; successorRunId: string | null; +}; + +export async function readNativeSupersededMonitorWait(db: Db, binding: MonitorBinding, + issue: typeof issues.$inferSelect, lockProof = false): Promise { + if (issue.companyId !== binding.companyId || issue.id !== binding.issueId || + issue.executionRunId === binding.runId || issue.checkoutRunId === binding.runId || + scheduledIssueMonitorWait(issue, binding.agentId) !== null) return null; + const source = await readCompletedNativeMonitorResult(db, binding); + if (!source?.finishedAt) return null; + const successor = alias(heartbeatRuns, "monitor_superseding_run"); + const query = db.select({ wakeId: agentWakeupRequests.id, commentId: issueComments.id, + successorRunId: successor.id }).from(agentWakeupRequests).innerJoin(issueComments, and( + eq(issueComments.companyId, binding.companyId), eq(issueComments.issueId, binding.issueId), + eq(issueComments.authorType, "user"), eq(issueComments.authorUserId, agentWakeupRequests.requestedByActorId), + isNull(issueComments.authorAgentId), isNull(issueComments.createdByRunId), + isNull(issueComments.deletedAt), isNull(issueComments.sourceTrust), + gt(issueComments.createdAt, source.finishedAt), + sql`${issueComments.updatedAt} = ${issueComments.createdAt}`, + sql`length(trim(${issueComments.body})) > 0`, + sql`${agentWakeupRequests.payload}->>'commentId' = ${issueComments.id}::text`, + )).leftJoin(successor, and(eq(successor.id, agentWakeupRequests.runId), + eq(successor.companyId, binding.companyId), eq(successor.agentId, binding.agentId), + eq(successor.nativeIssueId, binding.issueId), ne(successor.id, binding.runId), + eq(successor.wakeupRequestId, agentWakeupRequests.id), + sql`${successor.contextSnapshot}#>>'{executionContinuation,companyId}' = ${binding.companyId}`, + sql`${successor.contextSnapshot}#>>'{executionContinuation,issueId}' = ${binding.issueId}`, + sql`${successor.contextSnapshot}#>>'{executionContinuation,resumeDelta,baseRunId}' = ${binding.runId}`, + sql`${successor.contextSnapshot}#>>'{executionContinuation,objectiveSource,kind}' = 'comment'`, + sql`${successor.contextSnapshot}#>>'{executionContinuation,objectiveSource,id}' = ${issueComments.id}::text`, + sql`${successor.contextSnapshot}#>>'{executionContinuation,objectiveSource,revision}' = + to_char(${issueComments.updatedAt} at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')`, + )).where(and(eq(agentWakeupRequests.companyId, binding.companyId), + eq(agentWakeupRequests.agentId, binding.agentId), gt(agentWakeupRequests.requestedAt, source.finishedAt), + inArray(agentWakeupRequests.source, ["automation", "on_demand"]), + inArray(agentWakeupRequests.reason, ["issue_commented", "issue_reopened_via_comment"]), + eq(agentWakeupRequests.requestedByActorType, "user"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${binding.issueId}`, + notInArray(agentWakeupRequests.status, ["skipped", "cancelled"]), + or(and(isNull(agentWakeupRequests.runId), inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"])), + isNotNull(successor.id)), + )).orderBy(desc(issueComments.createdAt), desc(issueComments.id)).limit(1); + const [request] = await (lockProof ? query.for("share", { of: [agentWakeupRequests, issueComments], noWait: true }) : query); + return request ? { ...request, agentId: binding.agentId, resultId: source.resultId } : null; } export async function readNativeMonitorSource( diff --git a/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts b/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts index ebd1eee0a14..d5ca40230a7 100644 --- a/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts +++ b/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { agents, agentWakeupRequests, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, - issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork } from "@paperclipai/db"; + issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork, workspaceOperations, workAssessments } from "@paperclipai/db"; import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; @@ -13,6 +13,8 @@ import * as monitorDelivery from "./native-monitor-delivery.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import { heartbeatService } from "../heartbeat.js"; import { hasPendingRunCleanup } from "../execution-blocker.js"; +import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js"; +import { issueRecoveryActionService } from "../issue-recovery-actions.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PRESERVED_MONITOR_TEST_DATABASE_URL"); const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); @@ -221,6 +223,210 @@ describe.skipIf(!support.supported)("released native monitor results", () => { expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]); }); + async function newerUserRequest(state: "queued" | "consumed", existing?: Awaited>) { + const f = existing ?? await fixture(); + if (!existing) await db.update(issues).set({ status: "blocked", statusVersion: 5 }).where(eq(issues.id, f.issueId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "operator", body: "Continue with the new objective." }).returning(); + const successorId = state === "consumed" ? randomUUID() : null; + if (successorId) await db.insert(heartbeatRuns).values({ id: successorId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "failed", errorCode: "native_continuation_requires_reconciliation", + contextSnapshot: { issueId: f.issueId, commentId: comment!.id, wakeCommentId: comment!.id, + executionContinuation: { companyId: f.companyId, issueId: f.issueId, + objectiveSource: { kind: "comment", id: comment!.id, revision: comment!.updatedAt.toISOString() }, + resumeDelta: { baseRunId: f.runId } } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_reopened_via_comment", requestedByActorType: "user", requestedByActorId: "operator", + status: successorId ? "failed" : "queued", runId: successorId, + payload: { issueId: f.issueId, commentId: comment!.id, + _paperclipWakeContext: { issueId: f.issueId, wakeCommentId: comment!.id, wakeCommentIds: [comment!.id] } }, + }).returning(); + if (successorId) await db.update(heartbeatRuns).set({ wakeupRequestId: wake!.id }).where(eq(heartbeatRuns.id, successorId)); + if (!existing) await db.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, f.runId)); + return { f, comment: comment!, wake: wake!, successorId }; + } + + it.each(["queued", "consumed", "owned_successor", "new_monitor"] as const)("settles the old saved monitor result without changing a newer user request (%s)", async state => { + const { f, comment, wake, successorId } = await newerUserRequest(state === "queued" ? "queued" : "consumed"); + if (state === "owned_successor") { + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, successorId!)); + await db.update(issues).set({ executionRunId: successorId, checkoutRunId: successorId }).where(eq(issues.id, f.issueId)); + } + if (state === "new_monitor") { + const task = await f.readIssue(), nextCheckAt = new Date(Date.now() + 7_200_000); + await db.update(issues).set({ monitorNextCheckAt: nextCheckAt, executionPolicy: { ...task.executionPolicy, + monitor: { ...(task.executionPolicy?.monitor as Record), nextCheckAt: nextCheckAt.toISOString() } } }) + .where(eq(issues.id, f.issueId)); + } + expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(true); + const before = await f.readIssue(); + await f.finalize(); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator).toMatchObject({ phase: "committed", nextAttemptAt: null, failureCode: null, attempt: 2 }); + expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false); + expect(await heartbeatService(db, { runtimeEnv: {} }).tickIssueMonitors(new Date(Date.parse(f.nextCheckAt) + 1))) + .toMatchObject({ triggered: 0 }); + expect(await f.readIssue()).toMatchObject({ status: before.status, statusVersion: before.statusVersion, + assigneeAgentId: before.assigneeAgentId, assigneeUserId: before.assigneeUserId, + executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId, + executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt, + monitorAttemptCount: before.monitorAttemptCount, workMode: before.workMode }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([wake]); + expect(await db.select().from(issueComments).where(eq(issueComments.id, comment!.id))).toEqual([comment]); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.runId))).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "succeeded", finishedAt: f.finishedAt }); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + expect(decisions.at(-1)?.decisionJson).toMatchObject({ effects: [] }); + await f.finalize(); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId))).toEqual(decisions); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].attempt).toBe(2); + }); + + it.each(["queued", "consumed"] as const)("reconstructs and settles a historical monitor error through the real finalization reconciler (%s)", async state => { + const f = await fixture(); + await db.update(issues).set({ status: "blocked", statusVersion: 5 }).where(eq(issues.id, f.issueId)); + await f.finalize(); + const [prior] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(prior).toMatchObject({ phase: "retryable_failure", failureCode: "monitor_wait_authority_lost" }); + const oldAssessment = await db.select().from(workAssessments).where(eq(workAssessments.id, prior.assessmentId!)); + const oldDecision = await db.select().from(statusDecisions).where(eq(statusDecisions.id, prior.decisionId!)); + expect(oldAssessment).toHaveLength(1); + expect(oldDecision).toHaveLength(1); + const { comment, wake, successorId } = await newerUserRequest(state, f); + let hold: typeof issueRecoveryActions.$inferSelect | undefined; + if (successorId) { + const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), + ownerType: "board", nextAction: "Recover the current user request.", evidence: { runId: successorId }, supersedeOnIdentityChange: true }); + [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)); + } + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, + phase: "workspace_finalize", status: "succeeded", finishedAt: new Date() }); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null, + controllerBootId: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId)); + const before = await f.readIssue(); + const results = await reconcileNativeFinalizations(db, [f.runId]); + expect(results).toContainEqual(expect.objectContaining({ runId: f.runId, phase: "committed" })); + const [settled] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(settled).toMatchObject({ phase: "committed", failureCode: null, nextAttemptAt: null }); + expect(settled.assessmentId).not.toBe(prior.assessmentId); + const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, settled.assessmentId!)); + expect(assessment).toMatchObject({ supersedesAssessmentId: prior.assessmentId, resultId: f.resultId, + assessmentJson: { supersededMonitorWait: { agentId: f.agentId, resultId: f.resultId, + commentId: comment.id, wakeId: wake.id, successorRunId: successorId } } }); + expect(await db.select().from(workAssessments).where(eq(workAssessments.id, prior.assessmentId!))).toEqual(oldAssessment); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.id, prior.decisionId!))).toEqual(oldDecision); + expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false); + expect(await f.readIssue()).toMatchObject({ status: before.status, statusVersion: before.statusVersion, + executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt, + executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([wake]); + expect(await db.select().from(issueComments).where(eq(issueComments.id, comment.id))).toEqual([comment]); + if (hold) expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id))).toEqual([hold]); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + expect(decisions.at(-1)?.toStatus).toBe(before.status); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result); + }); + + it.each(["missing_wake", "actor", "foreign_task", "foreign_company", "untrusted_comment", "deleted_comment", "edited_comment", + "prior_comment", "foreign_run", "foreign_base", "wrong_objective", "cancelled_request", "source_stop", "missing_monitor_receipt", + "workspace_failed"])("keeps the old monitor held without exact newer user-request proof (%s)", async state => { + const consumed = ["foreign_run", "foreign_base", "wrong_objective"].includes(state); + const { f, comment, wake, successorId } = await newerUserRequest(consumed ? "consumed" : "queued"); + if (state === "missing_wake") await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + if (state === "actor") await db.update(agentWakeupRequests).set({ requestedByActorType: "agent" }).where(eq(agentWakeupRequests.id, wake.id)); + if (state === "foreign_task") await db.update(agentWakeupRequests).set({ payload: { ...wake.payload, issueId: randomUUID() } }) + .where(eq(agentWakeupRequests.id, wake.id)); + if (state === "foreign_company") { + const other = await fixture(); + await db.update(agentWakeupRequests).set({ companyId: other.companyId, agentId: other.agentId }) + .where(eq(agentWakeupRequests.id, wake.id)); + } + if (state === "untrusted_comment") await db.update(issueComments).set({ sourceTrust: { preset: "low_trust_review", disposition: "quarantined" } }).where(eq(issueComments.id, comment.id)); + if (state === "deleted_comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id)); + if (state === "edited_comment") await db.update(issueComments).set({ updatedAt: new Date(comment.createdAt.getTime() + 1) }).where(eq(issueComments.id, comment.id)); + if (state === "prior_comment") await db.update(issueComments).set({ createdAt: new Date(f.finishedAt.getTime() - 1), + updatedAt: new Date(f.finishedAt.getTime() - 1) }).where(eq(issueComments.id, comment.id)); + if (state === "foreign_run") { + const [otherIssue] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task", status: "blocked" }).returning(); + const [otherRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + nativeIssueId: otherIssue!.id, runtimeMode: "native", status: "failed", wakeupRequestId: wake.id, + contextSnapshot: { issueId: otherIssue!.id } }).returning(); + await db.update(agentWakeupRequests).set({ runId: otherRun!.id }).where(eq(agentWakeupRequests.id, wake.id)); + } + if (state === "foreign_base" || state === "wrong_objective") await db.update(heartbeatRuns).set({ contextSnapshot: + sql`jsonb_set(${heartbeatRuns.contextSnapshot}, ${state === "foreign_base" ? "{executionContinuation,resumeDelta,baseRunId}" : "{executionContinuation,objectiveSource,id}"}::text[], ${JSON.stringify(randomUUID())}::jsonb)` }) + .where(eq(heartbeatRuns.id, successorId!)); + if (state === "cancelled_request") await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, wake.id)); + if (state === "source_stop") await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeCancellation}', '{"state":"pending"}'::jsonb)` }) + .where(eq(heartbeatRuns.id, f.runId)); + if (state === "missing_monitor_receipt") await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + const before = await f.readIssue(); + await f.finalize(state === "workspace_failed" ? "failed" : "succeeded"); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator.phase).toBe("retryable_failure"); + expect(await f.readIssue()).toMatchObject({ status: before.status, executionPolicy: before.executionPolicy, + monitorNextCheckAt: before.monitorNextCheckAt, executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId }); + expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId), + eq(statusDecisions.reasonCode, "scheduled_monitor_superseded")))).toEqual([]); + }); + + it.each(["comment", "wake"] as const)("defers superseded monitor settlement during child-row contention (%s)", async state => { + const { f, comment, wake } = await newerUserRequest("queued"); + const before = await f.readIssue(); + let release!: () => void, locked!: () => void; + const released = new Promise(resolve => { release = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const owner = db.transaction(async tx => { + if (state === "comment") await tx.select().from(issueComments).where(eq(issueComments.id, comment.id)).for("update"); + else await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)).for("update"); + locked(); + await released; + }); + await acquired; + const original = monitorDelivery.readNativeSupersededMonitorWait; + const bounded = vi.spyOn(monitorDelivery, "readNativeSupersededMonitorWait").mockImplementation(async (...args) => { + if (args[3]) await args[0].execute(sql`set local lock_timeout = '100ms'`); + return original(...args); + }); + try { + await f.finalize(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "retryable_failure", failureCode: "status_cas_exhausted" }); + expect(await f.readIssue()).toMatchObject({ status: before.status, statusVersion: before.statusVersion, + executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt }); + expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId), + eq(statusDecisions.reasonCode, "scheduled_monitor_superseded")))).toEqual([]); + } finally { bounded.mockRestore(); release(); await owner; } + await f.finalize(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("committed"); + expect(await db.select().from(issueComments).where(eq(issueComments.id, comment.id))).toEqual([comment]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id))).toEqual([wake]); + }); + + it.each(["comment", "wake", "lineage", "source_stop"])("rechecks newer user-request proof under the task lock (%s)", async state => { + const { f, comment, wake, successorId } = await newerUserRequest("consumed"); + const original = monitorDelivery.readNativeSupersededMonitorWait; + const race = vi.spyOn(monitorDelivery, "readNativeSupersededMonitorWait").mockImplementationOnce(async (...args) => { + const proof = await original(...args); + expect(proof).not.toBeNull(); + if (state === "comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id)); + if (state === "wake") await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, wake.id)); + if (state === "lineage") await db.update(heartbeatRuns).set({ contextSnapshot: {} }).where(eq(heartbeatRuns.id, successorId!)); + if (state === "source_stop") await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeCancellation}', '{"state":"pending"}'::jsonb)` }) + .where(eq(heartbeatRuns.id, f.runId)); + return proof; + }); + try { await f.finalize(); } finally { race.mockRestore(); } + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("retryable_failure"); + expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId), + eq(statusDecisions.reasonCode, "scheduled_monitor_superseded")))).toEqual([]); + }); + it.each(["cleared", "owner", "receipt", "rescheduled"])("rechecks monitor proof under the task lock (%s)", async state => { const f = await fixture(), scheduled = await f.readIssue(); const [owner] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }).returning(); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index f71a8fdd1fb..d61039733f6 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -22,7 +22,7 @@ import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.j import { assertNativeCaseHandoffResult } from "./native-case-handoff.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js"; -import { readNativeMonitorDelivery, readNativePreservedMonitorWait } from "./native-monitor-delivery.js"; +import { readNativeMonitorDelivery, readNativePreservedMonitorWait, readNativeSupersededMonitorWait } from "./native-monitor-delivery.js"; import { isNativePlanWaitResult, readNativePlanWait } from "./native-plan-wait.js"; import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js"; import { settleExternalConversation } from "../slack-conversation-lifecycle.js"; @@ -1506,12 +1506,19 @@ export async function finalizeNativeRun(input: { authoritativeIssue.executionRunId !== run.id && authoritativeIssue.checkoutRunId !== run.id ? await readNativeMonitorDelivery(input.db, { companyId: run.companyId, issueId: coordinator.issueId, runId: run.id }) : null; + const supersededMonitorWait = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" && + assessment.reportedDisposition === "yielded" && assessment.continuation?.kind === "monitor" + ? await readNativeSupersededMonitorWait(input.db, { + companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId, + }, authoritativeIssue) : null; + if (supersededMonitorWait && supersededMonitorWait.resultId !== resultRow.id) throw new NativeStatusRaceError(); const qualityReview = await readRequiredQualityReview(input.db, authoritativeIssue); const statusInput: Parameters[0] = { requiredQualityReviewSatisfied: qualityReview.satisfied, caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)), monitorWaitDispatched: monitorDelivery !== null, monitorWaitPreserved: preservedMonitorWait !== null, + monitorWaitSuperseded: supersededMonitorWait !== null, monitorWaitAuthorized: authoritativeIssue.workMode === "standard" && authoritativeIssue.executionRunId === run.id && monitorWaitAt !== null, planWaitAuthorized: planWait !== null, hasPendingChildCompletion, @@ -1607,6 +1614,7 @@ export async function finalizeNativeRun(input: { monitorDelivery, monitorWaitAt: assessment.continuation?.kind === "monitor" ? monitorWaitAt : null, preservedMonitorWait, + supersededMonitorWait, }); await input.db .update(nativeRunFinalizations) @@ -1649,6 +1657,8 @@ export async function finalizeNativeRun(input: { ? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined, requireMonitorDelivery: decision.reasonCode === "scheduled_monitor_dispatched" ? monitorDelivery ?? undefined : undefined, + requireSupersededMonitorWait: decision.reasonCode === "scheduled_monitor_superseded" + ? supersededMonitorWait ?? undefined : undefined, requirePreservedMonitorWait: decision.reasonCode === "scheduled_monitor_preserved" ? preservedMonitorWait ?? undefined : undefined, requirePlanWaitSource: diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 9d3f34ff6b7..97eaf9079b0 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -150,6 +150,7 @@ export function arbitrateNativeStatus(input: { /** The saved timer already admitted another run and this run no longer owns checkout. */ monitorWaitDispatched?: boolean; monitorWaitPreserved?: boolean; + monitorWaitSuperseded?: boolean; boardResponseWaitAuthorized?: boolean; boardResponseWaitOrigin?: boolean; isConversation?: boolean; @@ -536,6 +537,11 @@ export function arbitrateNativeStatus(input: { effects: [], }; } + if (input.monitorWaitSuperseded) return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: input.priorIssueStatus, + reasonCode: "scheduled_monitor_superseded", unblockDescriptor: null, effects: [], + }; if (input.monitorWaitPreserved) { return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 3cdb31eea0b..b192e3f670c 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -20,7 +20,7 @@ import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat import { lockNativeGovernanceSources, pendingNativeGovernance } from "./native-governance.js"; import { findSupersededExternalReviewRequests, supersedeExternalReviewRequestsInTransaction } from "./superseded-external-review-requests.js"; import { normalizePrpResultSignals, type PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js"; -import { readNativeMonitorDelivery, readNativePreservedMonitorWait, type NativeMonitorDelivery, type NativePreservedMonitorWait } from "./native-monitor-delivery.js"; +import { readNativeMonitorDelivery, readNativePreservedMonitorWait, readNativeSupersededMonitorWait, type NativeSupersededMonitorWait, type NativeMonitorDelivery, type NativePreservedMonitorWait } from "./native-monitor-delivery.js"; import { readNativePlanWait, type NativePlanWaitSource } from "./native-plan-wait.js"; import { activeIssueInteractionCondition } from "../issue-question-context.js"; import { isConversation } from "../agent-conversations.js"; @@ -1627,6 +1627,7 @@ export async function commitNativeStatusDecision(input: { requireMonitorWait?: { agentId: string; nextCheckAt: string }; requireMonitorDelivery?: NativeMonitorDelivery; requirePreservedMonitorWait?: NativePreservedMonitorWait; + requireSupersededMonitorWait?: NativeSupersededMonitorWait; requireNoPendingChildCompletion?: NativeChildCompletionRecipient; requireReassessmentSource?: { phase: string; resultId: string; assessmentId: string; decisionId: string | null; agentId: string; @@ -1850,6 +1851,22 @@ export async function commitNativeStatusDecision(input: { input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status || input.decision.effects.length !== 0) throw new NativeStatusRaceError(); } + if (reasonCode === "scheduled_monitor_superseded") { + const expected = input.requireSupersededMonitorWait; + let actual: Awaited> = null; + try { + actual = expected ? await readNativeSupersededMonitorWait(tx as unknown as Db, { + companyId: input.companyId, issueId: input.issueId, runId: input.runId, agentId: expected.agentId, + }, issue, true) : null; + } catch (error) { + if (record(error).code === "55P03" || record(record(error).cause).code === "55P03") throw new NativeStatusRaceError(); + throw error; + } + if (!expected || !actual || actual.resultId !== expected.resultId || actual.wakeId !== expected.wakeId || + actual.commentId !== expected.commentId || actual.successorRunId !== expected.successorRunId || + input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status || + input.decision.effects.length !== 0) throw new NativeStatusRaceError(); + } if (reasonCode === "scheduled_monitor_preserved") { const expected = input.requirePreservedMonitorWait; const actual = expected ? await readNativePreservedMonitorWait(tx as unknown as Db, { diff --git a/server/src/services/native-runtime/work-assessments.ts b/server/src/services/native-runtime/work-assessments.ts index fe554d33edd..efcdd3ae9de 100644 --- a/server/src/services/native-runtime/work-assessments.ts +++ b/server/src/services/native-runtime/work-assessments.ts @@ -3,7 +3,7 @@ import type { Db } from "@paperclipai/db"; import { completionContracts, nativeRunResults, workAssessments } from "@paperclipai/db"; import { classifyNativeEvidence, type NativeEvidenceAssessment } from "./evidence-classifier.js"; import { nativeSha256 } from "./canonical.js"; -import type { NativeMonitorDelivery, NativePreservedMonitorWait } from "./native-monitor-delivery.js"; +import type { NativeMonitorDelivery, NativePreservedMonitorWait, NativeSupersededMonitorWait } from "./native-monitor-delivery.js"; export async function recordNativeWorkAssessment(input: { db: Db; @@ -27,6 +27,7 @@ export async function recordNativeWorkAssessment(input: { /** A persisted schedule can regain settlement authority after a policy repair. */ monitorWaitAt?: string | null; preservedMonitorWait?: NativePreservedMonitorWait | null; + supersededMonitorWait?: NativeSupersededMonitorWait | null; supersedesAssessmentId?: string | null; }) { const assessmentJson = { @@ -34,6 +35,7 @@ export async function recordNativeWorkAssessment(input: { ...(input.monitorDelivery ? { monitorDelivery: input.monitorDelivery } : {}), ...(input.monitorWaitAt ? { monitorWaitAt: input.monitorWaitAt } : {}), ...(input.preservedMonitorWait ? { preservedMonitorWait: input.preservedMonitorWait } : {}), + ...(input.supersededMonitorWait ? { supersededMonitorWait: input.supersededMonitorWait } : {}), } as unknown as Record; const inputDigest = nativeSha256({ contractId: input.contractId,