diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 687bbc5aaba..b96d485b48e 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -766,6 +766,17 @@ Rollback by setting the marker to `0` and restarting. No schema changes are required. This mechanism does not sleep claimed workspaces or replace a durable scheduler for their background work. +## Native finalization history and idle sleep + +Idle sleep retains committed native finalization records as audit history. +A record permits sleep only after its owned run, accepted result, assessment, +applied status decision, and effect deliveries agree on completion. Leases, +retries, recovery requests, unfinished accounting, and incomplete run projections +keep the instance awake. Historical controller identity alone does not block sleep. +The exact run also needs its successful workspace-finalization receipt. Pending +final-reply recovery shares its discovery predicate with the recovery worker, and +live warm sessions remain local work through close/checkpoint completion. + ## Feedback history and idle sleep Automatic idle checks treat private, unshared feedback as saved data. A sent diff --git a/server/src/__tests__/idle-native-finalization.test.ts b/server/src/__tests__/idle-native-finalization.test.ts new file mode 100644 index 00000000000..c6f62565a9f --- /dev/null +++ b/server/src/__tests__/idle-native-finalization.test.ts @@ -0,0 +1,214 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, companies, completionContracts, createDb, environments, environmentLeases, heartbeatRuns, issues, issueComments, + nativeRunFinalizations, nativeRunResults, statusDecisionEffects, statusDecisions, + workAssessments, workspaceOperations, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; +import { beginIdleTrackedWork } from "../services/task-admission.js"; +import { nativeResponsePresentationWork } from "../services/native-runtime/native-response-presentation-work.js"; + +const now = Date.UTC(2020, 0, 1); +const ownerId = "0e421722-e69c-45ca-ac42-80fbd0d66739"; +const future = new Date("2099-01-01T00:00:00Z"); +const none = { version: 1, backgroundWork: "none" }; +const present = { version: 1, backgroundWork: "present" }; +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping finalization idle-sleep Postgres tests: ${support.reason}`); + +(support.supported ? describe : describe.skip)("committed native finalization idle safety", () => { + let db: ReturnType; + let database: Awaited>; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("idle-native-finalization-"); + db = createDb(database.connectionString); + }, 60_000); + afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); }); + afterAll(async () => { await database?.cleanup(); }); + + const read = () => readIdleSleepSafety(db, () => ({ + ownerId, draining: true, startedAt: new Date(now - 1000), expiresAt: new Date(now + 60_000), + activeRuns: 0, pendingWakes: 0, + }), () => now, ownerId, async () => "none"); + + async function seed(status = "succeeded") { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + const runId = randomUUID(), contractId = randomUUID(), resultId = randomUUID(); + const assessmentId = randomUUID(), decisionId = randomUUID(); + const operationId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Finalization fixture", issuePrefix: "FIN" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Idle worker", adapterType: "process", status: "idle" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Completed task", status: "done" }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "fixture", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "fixture" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, + completionContractId: contractId, runtimeMode: "native", nativePhase: "committed", + status, finishedAt: new Date(now), resultJson: { finalizationPhase: "committed" }, + processPid: 123, processStartedAt: new Date(now - 1000) }); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: resultId, schemaStatus: "accepted", resultJson: { terminal: { runTerminalState: status } }, + canonicalSha256: resultId }); + await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId, runId, contractId, resultId, + triggerKind: "native_result", triggerActorCompanyId: companyId, priorIssueStatus: "in_progress", + priorStatusVersion: 0, policyVersion: "fixture", assessmentJson: {}, inputDigest: assessmentId }); + await db.insert(statusDecisions).values({ id: decisionId, companyId, issueId, runId, assessmentId, + decisionVersion: 1, policyVersion: "fixture", fromStatus: "in_progress", toStatus: "done", + reasonCode: "fixture", decisionJson: {}, decisionDigest: decisionId, applicationState: "applied", appliedAt: new Date(now) }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, resultId, assessmentId, decisionId, + phase: "committed", controllerBootId: "historical-controller", controllerPid: 123 }); + await db.insert(workspaceOperations).values({ id: operationId, companyId, issueId, heartbeatRunId: runId, + phase: "workspace_finalize", status: "succeeded", startedAt: new Date(now), finishedAt: new Date(now) }); + return { companyId, agentId, issueId, runId, resultId, assessmentId, decisionId, operationId }; + } + + it.each(["succeeded", "failed", "cancelled"])("permits a fully settled %s run and preserves its audit history", async status => { + const f = await seed(status); + const before = await db.select().from(nativeRunFinalizations); + expect(await read()).toEqual(none); + expect(await db.select().from(nativeRunFinalizations)).toEqual(before); + expect(await db.select({ id: nativeRunResults.id }).from(nativeRunResults)).toEqual([{ id: f.resultId }]); + }); + + it.each(["observed", "result_accepted", "workspace_finalizing", "arbitrating", "retryable_failure", "terminal_failure", "unknown"])( + "retains finalization phase %s", async phase => { + const f = await seed(); + await db.update(nativeRunFinalizations).set({ phase }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await read()).toEqual(present); + }, + ); + + it.each([ + { leaseOwner: "still-owned" }, { leaseExpiresAt: future }, { nextAttemptAt: future }, + { controlDeadlineAt: future }, { recoveryState: "resuming_session" }, { recoveryRequestId: "pending-recovery" }, + { failureCode: "pending_cleanup" }, { failureDetail: { cleanupPending: true } }, + { decisionId: null }, { assessmentId: null, decisionId: null }, { resultId: null, assessmentId: null, decisionId: null }, + ])("retains unresolved finalization intent %j", async change => { + const f = await seed(); + await db.update(nativeRunFinalizations).set(change).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await read()).toEqual(present); + }); + + it.each([ + { status: "running" }, { finishedAt: null }, { costAccountingPending: true }, { scheduledRetryAt: future }, + { nativePhase: "arbitrating" }, { runtimeMode: "legacy" }, { resultJson: {} }, + { resultJson: { finalizationPhase: "committed", nextAttemptAt: future.toISOString() } }, + { resultJson: { finalizationPhase: "committed", failureCode: "projection_pending" } }, + { resultJson: { finalizationPhase: "committed", originalFailureCode: "projection_pending" } }, + ])("retains an unfinished run projection %j", async change => { + const f = await seed(); + await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, f.runId)); + expect(await read()).toEqual(present); + }); + + it.each([ + { schemaStatus: "rejected" }, { resultJson: {} }, + { resultJson: { terminal: { runTerminalState: "active" } } }, + ])("retains a missing or inconsistent accepted result %j", async change => { + const f = await seed(); + await db.update(nativeRunResults).set(change).where(eq(nativeRunResults.id, f.resultId)); + expect(await read()).toEqual(present); + }); + + it.each([{ applicationState: "proposed" }, { appliedAt: null }])("retains an unapplied decision %j", async change => { + const f = await seed(); + await db.update(statusDecisions).set(change).where(eq(statusDecisions.id, f.decisionId)); + expect(await read()).toEqual(present); + }); + + it.each([ + { deliveryState: "pending", deliveredAt: null }, { deliveryState: "failed", deliveredAt: null }, + { deliveryState: "unknown", deliveredAt: new Date(now) }, { deliveryState: "delivered", deliveredAt: null }, + { deliveryState: "delivered", deliveredAt: new Date(now), nextAttemptAt: future }, + ])("retains incomplete status effects %j", async change => { + const f = await seed(); + await db.insert(statusDecisionEffects).values({ companyId: f.companyId, issueId: f.issueId, + decisionId: f.decisionId, ordinal: 0, effectKind: "release_checkout", targetType: "issue", + idempotencyKey: randomUUID(), payload: {}, ...change }); + expect(await read()).toEqual(present); + }); + + it("permits acknowledged effects without deleting their receipts", async () => { + const f = await seed(); + await db.insert(statusDecisionEffects).values({ companyId: f.companyId, issueId: f.issueId, + decisionId: f.decisionId, ordinal: 0, effectKind: "release_checkout", targetType: "issue", + idempotencyKey: randomUUID(), payload: {}, deliveryState: "delivered", deliveredAt: new Date(now) }); + expect(await read()).toEqual(none); + expect(await db.select({ state: statusDecisionEffects.deliveryState }).from(statusDecisionEffects)) + .toEqual([{ state: "delivered" }]); + }); + + it("retains an active provider allocation after the finalization commits", async () => { + const f = await seed(); + const [environment] = await db.insert(environments).values({ + name: "Sandbox fixture", driver: "sandbox", config: {}, + }).returning(); + await db.insert(environmentLeases).values({ companyId: f.companyId, + heartbeatRunId: f.runId, environmentId: environment.id, provider: "sandbox", + providerLeaseId: "live-fixture", status: "active" }); + expect(await read()).toEqual(present); + }); + + it("retains live local work after the finalization commits", async () => { + await seed(); + const finish = beginIdleTrackedWork(); + try { + expect(await read()).toEqual({ version: 1, backgroundWork: "unknown" }); + } finally { finish(); } + expect(await read()).toEqual(none); + }); + + it.each([ + { status: "failed" }, { status: "cancelled" }, { phase: "workspace_prepare" }, + { issueId: null }, { heartbeatRunId: null }, + ])("retains workspace repair without its owned successful receipt %j", async change => { + const f = await seed(); + await db.update(workspaceOperations).set(change).where(eq(workspaceOperations.id, f.operationId)); + expect(await read()).toEqual(present); + }); + + it("retains a committed run with no workspace-finalization receipt", async () => { + const f = await seed(); + await db.delete(workspaceOperations).where(eq(workspaceOperations.id, f.operationId)); + expect(await read()).toEqual(present); + }); + + async function seedPendingReply() { + const f = await seed(); + await db.update(nativeRunResults).set({ resultJson: { + result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Completed task" }, + terminal: { runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, + } }).where(eq(nativeRunResults.id, f.resultId)); + return f; + } + + it("retains final-reply recovery between result commit and comment creation", async () => { + const f = await seedPendingReply(); + expect(await nativeResponsePresentationWork(db).load()).toMatchObject([{ id: f.runId }]); + expect(await read()).toEqual(present); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorAgentId: f.agentId, createdByRunId: f.runId, body: "Final reply fixture" }).returning(); + await db.update(heartbeatRuns).set({ resultJson: { finalizationPhase: "committed", + presentationDecision: { commentAction: "create", commentId: comment.id } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await nativeResponsePresentationWork(db).load()).toEqual([]); + expect(await read()).toEqual(none); + expect(await db.select({ id: issueComments.id }).from(issueComments)).toEqual([{ id: comment.id }]); + }); + + it.each([true, "true", false])("respects the exact skipIssueComment contract %j", async skipIssueComment => { + const f = await seedPendingReply(); + await db.update(heartbeatRuns).set({ contextSnapshot: { skipIssueComment } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await read()).toEqual(skipIssueComment === true ? none : present); + }); + + it.each(["reuse", "none"])("preserves an explicit final-reply presentation choice %s", async commentAction => { + const f = await seedPendingReply(); + await db.update(heartbeatRuns).set({ resultJson: { finalizationPhase: "committed", + presentationDecision: { commentAction } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await read()).toEqual(none); + }); +}); diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index b9ee6483e52..a43a3b2f089 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -8,6 +8,7 @@ import { sql } from "drizzle-orm"; import { verifiedResourceCleanup } from "./cleanup-verification-proof.js"; import { companyHoldMismatch } from "../modules/agent-lifecycle/index.js"; import { pendingBudgetRunCancellationPredicate } from "./heartbeat/budget-cancellation.js"; +import { nativeResponsePresentationPendingQuery } from "./native-runtime/native-response-presentation-query.js"; export interface IdleSleepDrainStatus { ownerId?: string; @@ -222,6 +223,41 @@ const WORK_CHECKS = [ WHERE c.status = 'active' AND s.enabled AND s.automatic_updates AND (s.access_scope = 'company' OR EXISTS ( SELECT 1 FROM project_skill_sources b WHERE b.source_id = s.id AND b.company_id = s.company_id AND b.active))`, + `SELECT 1 FROM native_run_finalizations f WHERE ( + f.phase = 'committed' AND f.lease_owner IS NULL AND f.lease_expires_at IS NULL + AND f.next_attempt_at IS NULL AND f.control_deadline_at IS NULL + AND f.recovery_state IS NULL AND f.recovery_request_id IS NULL + AND f.failure_code IS NULL AND f.failure_detail IS NULL + AND EXISTS ( + SELECT 1 FROM heartbeat_runs r + JOIN native_run_results result_row ON result_row.id = f.result_id + AND result_row.company_id = r.company_id AND result_row.issue_id = r.native_issue_id + AND result_row.run_id = r.id AND result_row.completion_contract_id = r.completion_contract_id + JOIN work_assessments assessment ON assessment.id = f.assessment_id + AND assessment.company_id = r.company_id AND assessment.issue_id = r.native_issue_id + AND assessment.run_id = r.id AND assessment.result_id = result_row.id + JOIN status_decisions decision ON decision.id = f.decision_id + AND decision.company_id = r.company_id AND decision.issue_id = r.native_issue_id + AND decision.run_id = r.id AND decision.assessment_id = assessment.id + WHERE r.id = f.run_id AND r.company_id = f.company_id AND r.native_issue_id = f.issue_id + AND r.runtime_mode = 'native' AND r.native_phase = 'committed' + AND ${finishedRun("r")} + AND r.result_json->>'finalizationPhase' = 'committed' + AND r.result_json->>'nextAttemptAt' IS NULL AND r.result_json->>'failureCode' IS NULL + AND r.result_json->>'originalFailureCode' IS NULL + AND result_row.schema_status = 'accepted' + AND result_row.result_json->'terminal'->>'runTerminalState' = r.status + AND decision.application_state = 'applied' AND decision.applied_at IS NOT NULL) + AND EXISTS (SELECT 1 FROM workspace_operations operation + WHERE operation.company_id = f.company_id AND operation.issue_id = f.issue_id + AND operation.heartbeat_run_id = f.run_id + AND operation.phase = 'workspace_finalize' AND operation.status = 'succeeded') + AND NOT EXISTS ( + SELECT 1 FROM status_decision_effects effect WHERE effect.decision_id = f.decision_id + AND (effect.company_id <> f.company_id OR effect.issue_id <> f.issue_id + OR effect.delivery_state <> 'delivered' OR effect.delivered_at IS NULL + OR effect.next_attempt_at IS NOT NULL)) + ) IS NOT TRUE`, // These less common work sources fail closed on any retained state. Their // terminal-state exceptions can be added with tests for the owning service. ...[ @@ -229,7 +265,7 @@ const WORK_CHECKS = [ "chat_completion_deliveries", "tool_action_requests", "tool_action_deliveries", "workspace_runtime_services", "execution_workspace_runtime_leases", - "native_run_finalizations", "company_transfer_runs", "decisions", "decision_effect_executions", + "company_transfer_runs", "decisions", "decision_effect_executions", "decision_archive_notification_outbox", "browser_use_sessions", "browser_use_runs", "browser_use_browsers", "company_secret_proposals", @@ -288,6 +324,7 @@ export async function readIdleSleepSafety( checks.push(sql`EXISTS (SELECT 1 FROM agents JOIN companies ON companies.id = agents.company_id WHERE ${companyHoldMismatch()})`); checks.push(sql`EXISTS (SELECT 1 FROM heartbeat_runs WHERE ${pendingBudgetRunCancellationPredicate()})`); + checks.push(sql`EXISTS (${nativeResponsePresentationPendingQuery()})`); // Only the current worker's owned runtime drain can exempt a plugin. // Missing workers, old SDKs, and newly installed plugins still block. checks.push(quietPlugins.length === 0 diff --git a/server/src/services/native-runtime/native-response-presentation-query.ts b/server/src/services/native-runtime/native-response-presentation-query.ts new file mode 100644 index 00000000000..d4daaba61bc --- /dev/null +++ b/server/src/services/native-runtime/native-response-presentation-query.ts @@ -0,0 +1,70 @@ +import { QueryBuilder } from "drizzle-orm/pg-core"; +import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db"; +import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js"; + +export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) { + const db = new QueryBuilder(); + return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), + eq(nativeRunResults.runId, heartbeatRuns.id), eq(nativeRunResults.completionContractId, heartbeatRuns.completionContractId))) + .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId), + eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId), + eq(statusDecisions.runId, heartbeatRuns.id), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId))) + .where(and( + eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out"]), + isNotNull(heartbeatRuns.finishedAt), eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner), + eq(nativeRunResults.schemaStatus, "accepted"), + sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`, + sql`coalesce(${heartbeatRuns.resultJson} #> '{presentationDecision,commentAction}', '"create"'::jsonb) = '"create"'::jsonb`, + sql`${heartbeatRuns.contextSnapshot}->'skipIssueComment' is distinct from 'true'::jsonb`, + sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`, + sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,runTerminalState}' = 'succeeded'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`, + or( + and(eq(heartbeatRuns.status, "succeeded"), + sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, + sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`), + and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'yielded'`, + sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,summary}') = 'string' + and ${nativeRunResults.resultJson} #>> '{result,summary}' ~ '[^[:space:]]'`, + sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,continuation,idempotencyKey}') = 'string' + and ${nativeRunResults.resultJson} #>> '{result,continuation,idempotencyKey}' ~ '[^[:space:]]'`, + or( + and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'response_wake'`, + sql`${nativeRunResults.resultJson} #> '{result,attentionRequests}' = '[]'::jsonb`, + sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,continuation,summary}') = 'string' + and ${nativeRunResults.resultJson} #>> '{result,continuation,summary}' ~ '[^[:space:]]'`), + and(eq(heartbeatRuns.status, "succeeded"), or( + and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'same_agent'`, + eq(statusDecisions.reasonCode, "live_continuation_registered")), + and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'monitor'`, + inArray(statusDecisions.reasonCode, ["scheduled_monitor_waiting", "scheduled_monitor_dispatched", "scheduled_monitor_preserved"])), + )), + )), + ), + notExists(db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, nativeRunFinalizations.companyId), + eq(statusDecisionEffects.issueId, nativeRunFinalizations.issueId), + eq(statusDecisionEffects.decisionId, nativeRunFinalizations.decisionId), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + ))), + scope ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined, + after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined, + )).orderBy(asc(heartbeatRuns.id)).limit(100); +} + +export function nativeResponsePresentationPendingQuery() { + return nativeResponsePresentationCandidates(); +} diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index 6f545e2c923..aac11b54e3f 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -1,5 +1,6 @@ -import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; -import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, type Db } from "@paperclipai/db"; +import { nativeResponsePresentationCandidates } from "./native-response-presentation-query.js"; +import { asc } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js"; @@ -15,66 +16,10 @@ export function registerNativeResponsePresentationWork(db: Db, type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; export function nativeResponsePresentationWork(db: Db) { - function candidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) { - return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, - issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, - }).from(nativeRunFinalizations) - .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), - eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) - .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) - .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), - eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), - eq(nativeRunResults.runId, heartbeatRuns.id), eq(nativeRunResults.completionContractId, heartbeatRuns.completionContractId))) - .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId), - eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId), - eq(statusDecisions.runId, heartbeatRuns.id), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId))) - .where(and( - eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out"]), - isNotNull(heartbeatRuns.finishedAt), eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner), - eq(nativeRunResults.schemaStatus, "accepted"), - sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`, - sql`coalesce(${heartbeatRuns.resultJson} #> '{presentationDecision,commentAction}', '"create"'::jsonb) = '"create"'::jsonb`, - sql`${heartbeatRuns.contextSnapshot}->'skipIssueComment' is distinct from 'true'::jsonb`, - sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`, - sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`, - sql`${nativeRunResults.resultJson} #>> '{terminal,runTerminalState}' = 'succeeded'`, - sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`, - or( - and(eq(heartbeatRuns.status, "succeeded"), - sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, - sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`, - sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`), - and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, - sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'yielded'`, - sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,summary}') = 'string' - and ${nativeRunResults.resultJson} #>> '{result,summary}' ~ '[^[:space:]]'`, - sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,continuation,idempotencyKey}') = 'string' - and ${nativeRunResults.resultJson} #>> '{result,continuation,idempotencyKey}' ~ '[^[:space:]]'`, - or( - and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'response_wake'`, - sql`${nativeRunResults.resultJson} #> '{result,attentionRequests}' = '[]'::jsonb`, - sql`jsonb_typeof(${nativeRunResults.resultJson} #> '{result,continuation,summary}') = 'string' - and ${nativeRunResults.resultJson} #>> '{result,continuation,summary}' ~ '[^[:space:]]'`), - and(eq(heartbeatRuns.status, "succeeded"), or( - and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'same_agent'`, - eq(statusDecisions.reasonCode, "live_continuation_registered")), - and(sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' = 'monitor'`, - inArray(statusDecisions.reasonCode, ["scheduled_monitor_waiting", "scheduled_monitor_dispatched", "scheduled_monitor_preserved"])), - )), - )), - ), - notExists(db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( - eq(statusDecisionEffects.companyId, nativeRunFinalizations.companyId), - eq(statusDecisionEffects.issueId, nativeRunFinalizations.issueId), - eq(statusDecisionEffects.decisionId, nativeRunFinalizations.decisionId), - inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), - ))), - scope ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined, - scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, - scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined, - after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined, - )).orderBy(asc(heartbeatRuns.id)).limit(100); - } + const candidates = (scope?: NativeResponsePresentationScope, after?: string, runId?: string) => { + const query = nativeResponsePresentationCandidates(scope, after, runId).as("pending_native_response"); + return db.select().from(query).orderBy(asc(query.id)); + }; async function load(scope?: NativeResponsePresentationScope, signal?: AbortSignal, canRun = () => true): Promise { const entries: Entry[] = []; let after: string | undefined; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index db399718a65..51a4b76ecb8 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1,4 +1,5 @@ import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js"; +import { readIdleLocalWork, markIdleStartupComplete, markIdleIngressTracked } from "../idle-local-work.js"; import * as runnerCheckpoint from "./native-runner-checkpoint.js"; import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; import * as environmentTaskRuntime from "../environment-task-runtime.js"; @@ -349,6 +350,7 @@ import { nativeSessionWorkspaceScope, reserveWarmNativeInstructionDirectory, closeIdleWarmNativeSessionsForRestart, + readNativeSessionIdleWork, createGovernedWaitEventObservation, createRemoteRunnerProcessLauncher, createRunnerdBackend, @@ -9240,6 +9242,38 @@ describe("native warm session supervision", () => { expect(onGoalCheckpoint).toHaveBeenCalledOnce(); }); + it("blocks idle sleep for a released local warm session until its close checkpoint finishes", async () => { + await closeIdleWarmNativeSessionsForRestart(); + markIdleIngressTracked(); markIdleStartupComplete({ scheduledBackups: false }); + let finishClose!: () => void; + const checkpoint = new Promise(resolve => { finishClose = resolve; }); + const close = vi.fn(() => checkpoint); + const warmExecution = { ...execution, binding: { ...execution.binding, + runId: "run-local-idle-sleep", executionWorkspaceId: "workspace-local-idle-sleep" }, + session: { ...execution.session, normalizedSessionId: "session-local-idle-sleep", + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1; + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession?.({ close }); + return { result: { summary: "Completed" }, terminal: { runTerminalState: "succeeded" }, + turnId: "turn-local-idle-sleep", normalizedSessionId: warmExecution.session.normalizedSessionId, + providerSessionId: "provider-local-idle-sleep", driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + try { + expect(readNativeSessionIdleWork()).toBe("none"); + await executePaperclipNativeSession({ db: leaseDb(warmExecution), execution: warmExecution, runnerInstanceId: "runner" }); + expect(close).not.toHaveBeenCalled(); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(await readIdleLocalWork()).toBe("present"); + const closing = closeIdleWarmNativeSessionsForRestart(); + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(await readIdleLocalWork()).toBe("present"); + finishClose(); await closing; + expect(readNativeSessionIdleWork()).toBe("none"); + } finally { finishClose(); await closeIdleWarmNativeSessionsForRestart(); } + }); + it.each(["environment deletion", "controller restart"])("closes an idle warm session before %s", async (shutdownKind) => { const close = vi.fn(async () => undefined); const warmExecution = { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 924d0cf0375..6bb96323a1b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,6 +1,7 @@ import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { releaseRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; +import { registerIdleLocalWorkProbe, type IdleLocalWork } from "../idle-local-work.js"; import type { NativeCleanupOutcome } from "./retained-native-session-cleanup.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; @@ -610,6 +611,11 @@ const pluginTaskSessionOwners = new WeakMap>(); +export function readNativeSessionIdleWork(): IdleLocalWork { + return warmNativeSessions.size !== 0 || closingWarmNativeSessions.size !== 0 ? "present" : "none"; +} +registerIdleLocalWorkProbe(readNativeSessionIdleWork); + /** Reserve the existing physical copy before heartbeat composes AGENT_HOME and * task guidance. The final execution/configuration fence still owns admission. */ export async function claimWarmNativeInstructionCopy(input: {