mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
fix(server): let settled native finalizations permit idle sleep (#18184)
## Thinking Path > - Paperclip manages AI agents and their work. > - Native runs retain finalization records for recovery and audit history. > - Idle sleep checks durable work before it stops an instance. > - The current check treats every retained finalization as pending work. > - A committed record can remain after its run and status effects have finished. > - This pull request recognizes complete records without deleting them. > - Instances can sleep while unfinished execution and recovery work remain protected. ## Linked Issues or Issue Description **What happened?** A retained `native_run_finalizations` row prevents idle sleep even after the run has finished and its status decision has been applied. The current gate checks whether any row exists. **Expected behavior** Complete finalization history should remain available without counting as pending work. Active leases, retries, accounting, recovery, and incomplete status effects must still prevent sleep. **Steps to reproduce** 1. Finish a native run and retain its committed finalization. 2. Confirm that its accepted result, assessment, applied decision, and delivered effects agree. 3. Request an idle report under a valid quiet admission hold. 4. Observe that the retained row reports background work before this change. **Paperclip version or commit** Reproduced on master at `cd0c86e5e1d427c8fa8aa8c3a9499391ee9b425b` with a local PostgreSQL fixture. **Deployment mode** Server with PostgreSQL. No provider calls are needed to reproduce the gate failure. Related PR and issue searches found no duplicate for this check. ## What Changed - Replace the blanket finalization-row check with a completion proof bound to the same company, issue, run, accepted result, assessment, and applied decision. - Require the owning run's successful workspace-finalization receipt; retain unresolved leases, retries, accounting, recovery, and status effects. - Share final-reply candidate discovery with the presentation recovery worker, so a completed run cannot hide a required reply. The pure query stays with native presentation; idle inspection uses it in the existing read-only transaction and five-second statement limit. - Register open warm sessions and in-flight close/checkpoint promises with local idle inspection. Historical controller/provider process metadata remains preserved. - Add PostgreSQL and local-warm regressions and document the retained-history rule. ## Verification - PostgreSQL sleep/finalization and chat-monitor continuation suites: 300 tests passed, including 56 finalization cases. - Local-warm regression: passed; an open session and its pending close/checkpoint both block sleep until retirement completes. - Server typecheck and server build: passed after review fixes. - Full workspace typecheck and build: passed before the review additions; hosted checks cover the updated revision. - An earlier local full invocation encountered two chat recovery failures that passed in isolation, and two existing Git workspace fixture failures caused by Xcode system Git initializing `main`. Those two pass with `init.defaultBranch=master` scoped to the test subprocess. That superseded full invocation was stopped for review revisions; no full local pass is claimed. No machine configuration or unrelated source was changed. - Fresh Apex review: 5/5 on `6846203fe2383bae1ede888fbaaedf83e9946e73`, with all findings addressed. - Hosted CI: fully green on `6846203fe2383bae1ede888fbaaedf83e9946e73`, including the final verify gate. Twelve jobs were interrupted by a simultaneous runner shutdown; rerunning the failed jobs completed successfully. - No hosted deployment of this change is claimed. ## Risks This changes an automatic-stop gate. An incomplete proof could stop an instance with pending work. The SQL fails closed and retains the existing admission, local work, plugin, schedule, backup, and provider-allocation gates. PostgreSQL cases cover unresolved leases, retry/recovery intent, accounting, projections, rejected results, unapplied decisions, and pending effects. No migration or history deletion is required. Reverting the predicate restores the conservative gate. ## Model Used OpenAI GPT-6 through Codex, with repository tools and local PostgreSQL verification. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
e2cfdc7e0c
commit
5d79d88d68
7 files changed
+380
-63
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
@@ -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<Entry[]> {
|
||||
const entries: Entry[] = [];
|
||||
let after: string | undefined;
|
||||
|
||||
@@ -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<void>(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 = {
|
||||
|
||||
@@ -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<NativeSession, ReturnType<typeof cre
|
||||
// not inspect or quarantine that owner's state until the save has finished.
|
||||
const closingWarmNativeSessions = new Map<string, Promise<void>>();
|
||||
|
||||
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: {
|
||||
|
||||
Reference in new issue
Block a user