From 332e4c5df9199b41e5d4b3ff757cdc4644179f8c Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 25 Oct 2026 10:47:02 -0700 Subject: [PATCH] fix(native): settle completed results that retain a valid monitor (#17641) Preserve a valid saved monitor after its successful source releases execution and checkout. Recheck accepted result identity, receipts, ownership and governance under the task lock. Notify the existing scheduler once and preserve accepted live and recovered replies. Apex 5/5 after fixing its reply-presentation finding. All CI checks, 191 focused tests, 293 broader native tests, workspace typecheck and server build pass. Hosted effects require staging verification. Co-Authored-By: Paperclip --- doc/architecture/native-status-arbitration.md | 24 ++ .../__tests__/heartbeat-run-summary.test.ts | 4 +- server/src/services/heartbeat-run-summary.ts | 2 +- .../native-runtime/native-monitor-delivery.ts | 38 ++- .../native-preserved-monitor-wait.test.ts | 217 ++++++++++++++++++ .../native-runtime/native-run-finalizer.ts | 13 +- .../native-runtime/status-arbiter.test.ts | 2 +- .../services/native-runtime/status-arbiter.ts | 10 +- .../status-decision-committer.ts | 15 +- 9 files changed, 315 insertions(+), 10 deletions(-) create mode 100644 server/src/services/native-runtime/native-preserved-monitor-wait.test.ts diff --git a/doc/architecture/native-status-arbitration.md b/doc/architecture/native-status-arbitration.md index f56014fa5b0..7fa6d5782ed 100644 --- a/doc/architecture/native-status-arbitration.md +++ b/doc/architecture/native-status-arbitration.md @@ -374,3 +374,27 @@ and records a recovery action for the reviewer without an automatic retry loop. Both Codex and ACPX providers wait for server completion feedback before their terminal tool call succeeds. A rejected completion report stays correctable in the same provider turn. The runner does not latch it as an accepted result. + +## Completed runs with saved monitors + +A successful accepted result can report a monitor wait after the run has +already released execution and checkout. Finalization can preserve that wait +when the same agent still owns a valid ordinary task monitor and the completed +run has a server-recorded `set_task_monitor` receipt for the task. The current +monitor can have a later deadline; finalization does not restore an older timer. + +The server rechecks the accepted source, current monitor, empty execution and +checkout slots, and governance under the coordinator and task locks. The +`scheduled_monitor_preserved` decision keeps task status and has no task effects. +It records the completed result without dispatching another provider turn. +The existing monitor scheduler receives a transaction-aware notification when +this finalization blocker clears. Replaying the committed decision sends no +second notification. +Live response selection and committed response repair retain the accepted +monitor explanation under the existing reply and publication authorization rules. + +Missing receipts, changed ownership, cancellation intents, cleared or quota +monitors, and unfinished workspace finalization retain their recovery paths. +Due and exhausted monitors still belong to the scheduler. This does not change +their deadlines or attempts, clear unrelated recovery holds, or relax the +Cloud sleep and wake contract. diff --git a/server/src/__tests__/heartbeat-run-summary.test.ts b/server/src/__tests__/heartbeat-run-summary.test.ts index 33b03986b83..c0ed3d6b113 100644 --- a/server/src/__tests__/heartbeat-run-summary.test.ts +++ b/server/src/__tests__/heartbeat-run-summary.test.ts @@ -25,10 +25,10 @@ describe("model-facing run errors", () => { }); }); -describe("scheduled continuation presentation", () => { +describe.each(["scheduled_monitor_waiting", "scheduled_monitor_dispatched", "scheduled_monitor_preserved"])("scheduled continuation presentation (%s)", finalizationReasonCode => { const resultJson = { finalizationPhase: "committed", - finalizationReasonCode: "scheduled_monitor_waiting", + finalizationReasonCode, nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", diff --git a/server/src/services/heartbeat-run-summary.ts b/server/src/services/heartbeat-run-summary.ts index 25e0de11276..97368cb1da6 100644 --- a/server/src/services/heartbeat-run-summary.ts +++ b/server/src/services/heartbeat-run-summary.ts @@ -298,7 +298,7 @@ export function readAcceptedTaskContinuationSummary( const result = record(resultJson); if (result.finalizationPhase !== "committed") return null; const kind = result.finalizationReasonCode === "live_continuation_registered" ? "same_agent" - : ["scheduled_monitor_waiting", "scheduled_monitor_dispatched"].includes(String(result.finalizationReasonCode)) ? "monitor" : null; + : ["scheduled_monitor_waiting", "scheduled_monitor_dispatched", "scheduled_monitor_preserved"].includes(String(result.finalizationReasonCode)) ? "monitor" : null; if (!kind) return null; for (const candidate of semanticResultCandidates(result)) { const continuation = record(candidate.continuation); diff --git a/server/src/services/native-runtime/native-monitor-delivery.ts b/server/src/services/native-runtime/native-monitor-delivery.ts index d38e41d8bc1..b2af0e80605 100644 --- a/server/src/services/native-runtime/native-monitor-delivery.ts +++ b/server/src/services/native-runtime/native-monitor-delivery.ts @@ -1,7 +1,8 @@ -import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db"; -import { and, desc, eq, inArray, ne, sql } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db, type issues } from "@paperclipai/db"; +import { and, desc, eq, inArray, isNotNull, ne, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import { boundRunIssueIdSql } from "../run-task-binding.js"; +import { scheduledIssueMonitorWait } from "../issue-monitors.js"; function record(value: unknown): Record { return value !== null && typeof value === "object" && !Array.isArray(value) @@ -9,6 +10,39 @@ function record(value: unknown): Record { } export type NativeMonitorDelivery = { wakeId: string; successorRunId: string }; +export type NativePreservedMonitorWait = { agentId: string; nextCheckAt: string; resultId: string }; + +export async function readNativePreservedMonitorWait( + db: Db, + binding: { companyId: string; issueId: string; runId: string; agentId: string }, + issue: typeof issues.$inferSelect, +): Promise { + if (issue.companyId !== binding.companyId || issue.id !== binding.issueId || issue.workMode !== "standard" || + 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 ( + 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' + and receipt.value->'result'->>'taskId' = ${binding.issueId} + and receipt.value->'result'->'monitor'->>'status' = 'scheduled' + and jsonb_typeof(receipt.value->'result'->'monitor'->'nextCheckAt') = 'string' + )` }).from(heartbeatRuns).innerJoin(nativeRunFinalizations, and( + eq(nativeRunFinalizations.companyId, binding.companyId), eq(nativeRunFinalizations.issueId, binding.issueId), + eq(nativeRunFinalizations.runId, heartbeatRuns.id), + )).innerJoin(nativeRunResults, and( + eq(nativeRunResults.id, nativeRunFinalizations.resultId), eq(nativeRunResults.companyId, binding.companyId), + eq(nativeRunResults.issueId, binding.issueId), eq(nativeRunResults.runId, heartbeatRuns.id), + eq(nativeRunResults.schemaStatus, "accepted"), + )).where(and(eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId), + eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded"), isNotNull(heartbeatRuns.finishedAt), + ...["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; +} export async function readNativeMonitorSource( db: Pick, 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 new file mode 100644 index 00000000000..ad269c7467c --- /dev/null +++ b/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts @@ -0,0 +1,217 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, completionContracts, createDb, heartbeatRuns, + issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork } from "@paperclipai/db"; +import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; +import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-run-finalizer.js"; +import { nativeSha256 } from "./canonical.js"; +import * as monitorDelivery from "./native-monitor-delivery.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { hasPendingRunCleanup } from "../execution-blocker.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("released native monitor results", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-preserved-monitor-"); + db = createDb(temporary.connectionString); + }); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const contractId = randomUUID(), resultId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Saved monitor", issuePrefix: `P${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Saved follow-up", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, + runtimeMode: "native", status: "running", contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const nextCheckAt = new Date(Date.now() + 3_600_000).toISOString(); + const authority = new PaperclipRunnerToolAuthority(db, { companyId, agentId, issueId, runId }); + await authority.execute({ tool: "set_task_monitor", callId: randomUUID(), arguments: { + idempotencyKey: "saved-monitor", monitor: { nextCheckAt, notes: "Check the saved result." }, + } }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "test", criteria: [] }, canonicalSha256: randomUUID(), + createdByActorType: "system", createdByActorId: "test" }); + const result = { result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", + summary: "Wait for the saved check.", completionClaim: { contractRevision: "test", objectiveSatisfied: false, + criteria: [], remainingWork: [{ description: "Check again", blocksCompletion: true }] }, + continuation: { kind: "monitor", summary: "Wait for the check", idempotencyKey: "provider-selected" }, + evidence: [], verification: [], attentionRequests: [], artifacts: [] }, + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "yielded" } }; + const finishedAt = new Date("2020-01-01T00:00:00Z"); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt, completionContractId: contractId }).where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: randomUUID(), canonicalSha256: randomUUID(), schemaStatus: "accepted", resultJson: result }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "retryable_failure", resultId }); + await db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId)); + const readIssue = async () => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]!; + const finalize = (workspaceFinalizeStatus: "succeeded" | "failed" = "succeeded") => finalizeNativeRun({ + db, runId, workspaceFinalizeStatus, projectRunStatus: true, + }); + return { companyId, agentId, issueId, runId, resultId, result, finishedAt, nextCheckAt, readIssue, finalize }; + } + + it.each(["original", "rescheduled", "past_due"])("keeps the saved monitor and settles a released successful result (%s)", async state => { + const f = await fixture(); + if (state !== "original") { + const issue = await f.readIssue(); + const nextCheckAt = state === "past_due" ? "2020-01-01T01:00:00Z" : new Date(Date.now() + 7_200_000).toISOString(); + await db.update(issues).set({ monitorNextCheckAt: new Date(nextCheckAt), executionPolicy: { + ...issue.executionPolicy, monitor: { ...(issue.executionPolicy?.monitor as Record), nextCheckAt }, + } }).where(eq(issues.id, f.issueId)); + } + const issueBefore = await f.readIssue(); + const [hold] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: randomUUID(), + status: "active", ownerType: "board", nextAction: "Preserve independent recovery.", evidence: { runId: randomUUID() }, + }).returning(); + await f.finalize(); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator.phase).toBe("committed"); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + expect(decisions.at(-1)).toMatchObject({ reasonCode: "scheduled_monitor_preserved", toStatus: issueBefore.status }); + expect(decisions.at(-1)?.decisionJson).toMatchObject({ effects: [] }); + expect(await f.readIssue()).toMatchObject({ status: issueBefore.status, executionPolicy: issueBefore.executionPolicy, + monitorNextCheckAt: issueBefore.monitorNextCheckAt, monitorAttemptCount: issueBefore.monitorAttemptCount, + executionRunId: null, checkoutRunId: null }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].finishedAt).toEqual(f.finishedAt); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold!.id))).toEqual([hold]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toEqual([{ id: f.runId }]); + await f.finalize(); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId))).toEqual(decisions); + }); + + it.each(["cleared", "owner", "checkout", "other_agent", "no_receipt", "foreign_receipt", "unfinished", "failed_source", "quota", "workspace_failed", + "cancellation", "nativeCancellation", "executionCancellation", "startupCancellation"])( + "does not settle a released monitor without current proof (%s)", async state => { + const f = await fixture(); + const issue = await f.readIssue(); + if (state === "cleared") await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: {} }).where(eq(issues.id, f.issueId)); + if (state === "quota") await db.update(issues).set({ executionPolicy: { ...issue.executionPolicy, + monitor: { ...(issue.executionPolicy?.monitor as Record), serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME }, + } }).where(eq(issues.id, f.issueId)); + if (state === "other_agent") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + if (state === "owner" || state === "checkout") { + const [owner] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }).returning(); + await db.update(issues).set(state === "owner" ? { executionRunId: owner!.id } : { checkoutRunId: owner!.id }).where(eq(issues.id, f.issueId)); + } + if (state === "unfinished") await db.update(heartbeatRuns).set({ finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); + if (state === "failed_source") await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId)); + if (state === "no_receipt") await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + if (["cancellation", "nativeCancellation", "executionCancellation", "startupCancellation"].includes(state)) { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, [state]: { state: "pending" } } }).where(eq(heartbeatRuns.id, f.runId)); + } + if (state === "foreign_receipt") await db.update(heartbeatRuns).set({ resultJson: { semanticToolReceipts: { + invalid: { operationId: "set_task_monitor", result: { taskId: randomUUID(), monitor: { status: "scheduled", nextCheckAt: f.nextCheckAt } } }, + } } }).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({ executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt, + executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId, assigneeAgentId: before.assigneeAgentId }); + expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId), eq(statusDecisions.reasonCode, "scheduled_monitor_preserved")))).toEqual([]); + }, + ); + + it("supersedes an earlier authority error without replaying the accepted result", async () => { + const f = await fixture(), scheduled = await f.readIssue(); + await db.update(issues).set({ executionPolicy: {}, monitorNextCheckAt: null }).where(eq(issues.id, f.issueId)); + await f.finalize(); + const old = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + expect(old.some(decision => decision.reasonCode === "monitor_wait_authority_lost")).toBe(true); + await db.update(issues).set({ executionPolicy: scheduled.executionPolicy, monitorNextCheckAt: scheduled.monitorNextCheckAt }) + .where(eq(issues.id, f.issueId)); + await f.finalize(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("committed"); + const after = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + for (const before of old) expect(after.find(decision => decision.id === before.id)).toEqual(before); + expect(after.some(decision => decision.reasonCode === "scheduled_monitor_preserved")).toBe(true); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]); + }); + + it("notifies the existing monitor scheduler once when finalization stops blocking it", async () => { + const f = await fixture(), events: string[] = []; + expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(true); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); }); + try { + await f.finalize(); + expect(events).toEqual(["intent", "settled"]); + expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false); + expect((await f.readIssue()).monitorNextCheckAt?.toISOString()).toBe(f.nextCheckAt); + await f.finalize(); + expect(events).toEqual(["intent", "settled"]); + } finally { unsubscribe(); } + }); + + it.each(["live", "recovered"])("publishes the accepted monitor explanation once (%s)", async mode => { + const f = await fixture(), before = await f.readIssue(); + const completionContractSha256 = `sha256:${nativeSha256({ revision: "test", criteria: [] })}`; + const canonicalSha256 = `sha256:${nativeSha256({ ...f.result, turnId: null })}`; + const serverFingerprint = `sha256:${nativeSha256({ runId: f.runId, completionContractSha256, canonicalSha256 })}`; + await db.update(nativeRunResults).set({ canonicalSha256, serverFingerprint }).where(eq(nativeRunResults.id, f.resultId)); + await db.update(heartbeatRuns).set({ completionContractSha256, + contextSnapshot: { issueId: f.issueId, ...(mode === "recovered" ? { skipIssueComment: true } : {}) }, + }).where(eq(heartbeatRuns.id, f.runId)); + const comments = () => db.select({ id: issueComments.id, body: issueComments.body }).from(issueComments) + .where(and(eq(issueComments.companyId, f.companyId), eq(issueComments.createdByRunId, f.runId))); + const repair = () => repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId }); + await f.finalize(); + if (mode === "recovered") { + expect(await comments()).toEqual([]); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await repair()).toBe(true); + } + const published = await comments(); + expect(published).toEqual([{ id: expect.any(String), body: f.result.result.summary }]); + expect(await repair()).toBe(false); + await f.finalize(); + expect(await comments()).toEqual(published); + expect((await f.readIssue()).monitorNextCheckAt).toEqual(before.monitorNextCheckAt); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).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(); + const nextCheckAt = new Date(Date.now() + 7_200_000).toISOString(); + const original = monitorDelivery.readNativePreservedMonitorWait; + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); }); + const race = vi.spyOn(monitorDelivery, "readNativePreservedMonitorWait").mockImplementationOnce(async (...args) => { + const proof = await original(...args); + expect(proof).not.toBeNull(); + if (state === "cleared") await db.update(issues).set({ executionPolicy: {}, monitorNextCheckAt: null }).where(eq(issues.id, f.issueId)); + if (state === "owner") await db.update(issues).set({ executionRunId: owner!.id }).where(eq(issues.id, f.issueId)); + if (state === "receipt") await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + if (state === "rescheduled") await db.update(issues).set({ monitorNextCheckAt: new Date(nextCheckAt), executionPolicy: { + ...scheduled.executionPolicy, monitor: { ...(scheduled.executionPolicy?.monitor as Record), nextCheckAt }, + } }).where(eq(issues.id, f.issueId)); + return proof; + }); + try { await f.finalize(); } finally { race.mockRestore(); unsubscribe(); } + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator.phase).toBe(state === "rescheduled" ? "committed" : "retryable_failure"); + expect(events).toEqual(state === "rescheduled" ? ["intent", "settled"] : []); + const task = await f.readIssue(); + expect(task.executionRunId).toBe(state === "owner" ? owner!.id : null); + expect(task.monitorNextCheckAt?.toISOString() ?? null).toBe(state === "cleared" ? null : state === "rescheduled" ? nextCheckAt : f.nextCheckAt); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner!.id)))[0].status).toBe("running"); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]); + }); +}); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 805fd623d18..1f8b830ff03 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -4,7 +4,7 @@ import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.j import { assertNativeCaseHandoffResult } from "./native-case-handoff.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js"; -import { readNativeMonitorDelivery } from "./native-monitor-delivery.js"; +import { readNativeMonitorDelivery, readNativePreservedMonitorWait } 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"; @@ -1331,10 +1331,16 @@ export async function finalizeNativeRun(input: { }) : "not_applicable"; const monitorWaitAt = scheduledIssueMonitorWait(authoritativeIssue, run.agentId); + const preservedMonitorWait = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" && + assessment.reportedDisposition === "yielded" && assessment.continuation?.kind === "monitor" + ? await readNativePreservedMonitorWait(input.db, { + companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId, + }, authoritativeIssue) : null; + if (preservedMonitorWait && preservedMonitorWait.resultId !== resultRow.id) throw new NativeStatusRaceError(); const supersededReviews = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" && assessment.attentionRequests.length === 0 && (assessment.reportedDisposition === "done" || (assessment.reportedDisposition === "yielded" && (externalChatResponseWaitAuthorization === "authorized" || - (assessment.continuation?.kind === "monitor" && monitorWaitAt !== null)))) + (assessment.continuation?.kind === "monitor" && monitorWaitAt !== null && !preservedMonitorWait)))) ? await findSupersededExternalReviewRequests(input.db, { companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId, }) : []; @@ -1419,6 +1425,7 @@ export async function finalizeNativeRun(input: { const proposedDecision = resolveNativeFinalizerStatus({ caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)), monitorWaitDispatched: monitorDelivery !== null, + monitorWaitPreserved: preservedMonitorWait !== null, monitorWaitAuthorized: authoritativeIssue.workMode === "standard" && authoritativeIssue.executionRunId === run.id && monitorWaitAt !== null, planWaitAuthorized: planWait !== null, hasPendingChildCompletion, @@ -1522,6 +1529,8 @@ export async function finalizeNativeRun(input: { ? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined, requireMonitorDelivery: decision.reasonCode === "scheduled_monitor_dispatched" ? monitorDelivery ?? undefined : undefined, + requirePreservedMonitorWait: decision.reasonCode === "scheduled_monitor_preserved" + ? preservedMonitorWait ?? undefined : undefined, requirePlanWaitSource: decision.reasonCode === "native_plan_accepted_waiting_for_continuation" ? planWait?.source : undefined, diff --git a/server/src/services/native-runtime/status-arbiter.test.ts b/server/src/services/native-runtime/status-arbiter.test.ts index 6262e01adaf..9a3609f9a2c 100644 --- a/server/src/services/native-runtime/status-arbiter.test.ts +++ b/server/src/services/native-runtime/status-arbiter.test.ts @@ -560,7 +560,7 @@ describe("native status authority", () => { expect.objectContaining({ statusAction: "blocked", toStatus: "blocked", - policyVersion: "phase6-v12", + policyVersion: "phase6-v13", reasonCode: "current_track_blocker_waiting", unblockDescriptor: { owner: "board", diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 5c6e4679e12..4fdc5496cc1 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -1,7 +1,7 @@ import type { NativeEvidenceAssessment } from "./evidence-classifier.js"; import { NATIVE_MODEL_REJECTION_MESSAGE, NATIVE_PROVIDER_CAPACITY_MAX_RETRIES, NATIVE_PROVIDER_OVERLOADED_CODE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js"; -export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v12"; +export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v13"; export type NativeAuthoritativeIssueStatus = | "backlog" @@ -118,6 +118,7 @@ export function arbitrateNativeStatus(input: { caseHandoffAuthorized?: boolean; /** The saved timer already admitted another run and this run no longer owns checkout. */ monitorWaitDispatched?: boolean; + monitorWaitPreserved?: boolean; boardResponseWaitAuthorized?: boolean; boardResponseWaitOrigin?: boolean; isConversation?: boolean; @@ -483,6 +484,13 @@ export function arbitrateNativeStatus(input: { effects: [], }; } + if (input.monitorWaitPreserved) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: input.priorIssueStatus, + reasonCode: "scheduled_monitor_preserved", unblockDescriptor: null, effects: [], + }; + } if (input.monitorWaitAuthorized) { 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 eafd64245c9..c7684448cbf 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -3,7 +3,7 @@ import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js import { 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, type NativeMonitorDelivery } from "./native-monitor-delivery.js"; +import { readNativeMonitorDelivery, readNativePreservedMonitorWait, 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"; @@ -1565,6 +1565,7 @@ export async function commitNativeStatusDecision(input: { requirePlanWaitSource?: NativePlanWaitSource; requireMonitorWait?: { agentId: string; nextCheckAt: string }; requireMonitorDelivery?: NativeMonitorDelivery; + requirePreservedMonitorWait?: NativePreservedMonitorWait; requireNoPendingChildCompletion?: NativeChildCompletionRecipient; requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; @@ -1675,6 +1676,18 @@ 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_preserved") { + const expected = input.requirePreservedMonitorWait; + const actual = expected ? await readNativePreservedMonitorWait(tx as unknown as Db, { + companyId: input.companyId, issueId: input.issueId, runId: input.runId, agentId: expected.agentId, + }, issue) : null; + if (!expected || !actual || actual.nextCheckAt !== expected.nextCheckAt || actual.resultId !== expected.resultId || + input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status || + input.decision.effects.length !== 0 || await pendingNativeGovernance({ db: tx as unknown as Db, + companyId: input.companyId, issueId: input.issueId, runId: input.runId, executionState: record(issue.executionState), + })) throw new NativeStatusRaceError(); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + } // Recheck under the parent status lock: a child can finish between the // finalizer's read and this commit. Re-arbitrate instead of discarding its wake. if (input.requireNoPendingChildCompletion && await hasPendingNativeChildCompletion(