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(