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,