From 0e111c50752cd4829a8edf171cda715df50e62d3 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:19:31 -0600 Subject: [PATCH] fix(recovery): verify host preparation logs before continuation (#19580) ## Thinking Path > - Paperclip manages AI agents and the work assigned to them. > - A task can retain saved input after a failure during execution preparation. > - Retry must prove that the old attempt stopped and its files are saved before it admits a new turn. > - The host can write a skill announcement before it dispatches a provider. > - The historical preparation verifier requires an empty log and rejects this safe host output. > - This pull request verifies that exact small log against the saved skill selection and immutable log facts. > - Eligible saved input can continue once. Uncertain execution and current policy gates stay held. ## Linked Issues or Issue Description Refs: #19004, #19464, #19552. These repairs establish the existing preparation proof. They do not accept the host skill announcement described here. A search for the exact announcement found no duplicate repair. This is a bug fix within the existing recovery contract and does not duplicate a roadmap feature. **What happened?** A legacy preparation failed before provider dispatch with a host bootstrap certificate. Its remote lease stopped and its files were recovered. A newer preparation failed while that save was in progress. An accepted Retry then remained in execution recovery because its ancestor had one host skill log record. **Expected behavior** Admit one fresh turn for the existing saved user request after all execution, workspace and policy proofs pass. Preserve old outcomes and source records. **Steps to reproduce** 1. Create the real PostgreSQL fixture in `saved-bootstrap-continuation.test.ts`. It includes a recovered remote workspace, exact stop receipt, bootstrap ancestor, later failed preparation and saved user status queue. 2. Call `requestTaskRetry`, then call the real resume path concurrently. 3. On the baseline guard, the empty-log control dispatches. Adding only the host skill announcement leaves zero queued successors and the same execution-recovery wait. **Paperclip version or commit** Reproduced on master `65aee998a3`. The final branch rebases cleanly onto `c650f36007`. **Deployment mode** Source checkout with real embedded PostgreSQL and controlled log storage. The reported runtime is a server with remote sandbox workspaces. ## What Changed - Add a narrow verifier for one host skill-announcement record. It binds the complete bytes to the run log reference, digest, size, sequence, timestamp, output excerpt and recorded skill selection. Each announced skill must have a saved version; unrelated configured skills may also be selected. - Read at most 2,049 bytes with a two-second cancellation deadline. Missing, slow, changed, additional or provider output fails closed. - Use this proof only for historical bootstrap ancestors that already pass the no-provider, settlement, stop and saved-file checks. - Add real admission regressions for one successor, concurrent resume, duplicate Retry and delivery of the original saved comment. Add current-gate and storage-boundary negative cases. - Document the continuation contract. ## Verification - **RED:** On the unchanged baseline guard, the exact final fixture passes the empty-log control and fails the host-announcement case: zero successors instead of one. The saved comment and Retry remain deferred. - **GREEN:** 486/486 tests pass across `task-retry.test.ts`, `explicit-native-continuation.test.ts`, `saved-bootstrap-continuation.test.ts` and `verified-preparation-log.test.ts`. - The new cases cover provider-started evidence, provider events, a native coordinator, missing stop proof, missing skill selection, changed hash, unknown output, pending approval and operator Stop. - Read-boundary cases cover foreign references, changed bytes, truncation, extra records, malformed selection and cancellation. An unrelated recovery completes while another read is stalled. - After the rebase, the original 27 new cases passed again at head `be640292fe136232e3db0dc6938233f53b6ccb43`. The review fix adds three cases, and the full 486-test pass includes all 30 new cases. - The exact previously observed historical input also passes the production log verifier: one read, 240 bytes matching the stored size and SHA, the same timestamp and output sequence, and a recorded version for the announced skill. This uses captured bytes for transport and does not claim live hosted admission. - Full local server layered typecheck passes after building this checkout's Runner declarations and normal excluded-plugin SDK links. The typing correction also passes all 30 new cases and the actual captured historical log proof. The prior CI build/typecheck failure was TS7006 in the JSON parser; it is retained as failed evidence and corrected in this revision. - Module boundary and whitespace checks pass. Local verification selection requires broad CI because the execution contract changed; focused local suites are recorded above. [Required final-head CI](https://github.com/paperclipai/paperclip/actions/runs/42281590449) typecheck, build and tests pass at `c557cdc4dc57000cb478f9be0717675dcac7364e` (56 successful checks, no required check pending or failed). Greptile rates this final revision 5/5; the addressed thread is resolved. The advisory hosted Paperclip Review is paused with its request saved and has not supplied a final-head assessment. Its pending status is disclosed separately from required CI and Greptile. - Controlled fixtures exercise the admission and dispatch receipts. Hosted continuation still requires rollout and reassessment of the existing request. ## Risks This adds one bounded storage read to an active historical admission check. Failed or uncertain reads keep the existing hold. The host bootstrap certificate remains the primary no-provider proof. Log text alone cannot authorize continuation. No schema, retry budget, provider setting, account selection or scheduler changes. ## Model Used OpenAI GPT-6 family via Codex, with coding and tool use. The exact model ID and context window are unavailable in this session's metadata. ## 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 --- doc/execution-semantics.md | 8 + .../src/services/cancelled-native-startup.ts | 12 +- .../saved-bootstrap-continuation.test.ts | 186 ++++++++++++++++++ .../services/verified-preparation-log.test.ts | 84 ++++++++ .../src/services/verified-preparation-log.ts | 52 +++++ 5 files changed, 340 insertions(+), 2 deletions(-) create mode 100644 server/src/services/saved-bootstrap-continuation.test.ts create mode 100644 server/src/services/verified-preparation-log.test.ts create mode 100644 server/src/services/verified-preparation-log.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 0fad16bcc3e..c321e01b5e2 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3237,6 +3237,14 @@ must agree with the absence of a native result, coordinator, environment lease, controller, process or provider output. A previous session display hint does not establish that the failed attempt started a provider. +A recovered historical bootstrap ancestor may contain the host's run-scoped +skill announcement before dispatch. Admission verifies the complete small log +against its stored digest, byte count, output watermark and recorded skill +selection. Missing, slow, changed or additional output remains a hold. This +exception still requires the host's no-provider bootstrap certificate, exact +settlement events, stopped leases and verified saved files; it cannot certify +provider output or authorize replay of an earlier session. + That intent pins the saved comment and accepted identity. Queue admission rechecks the source; dispatch rechecks the coalesced intent and its exact claimed successor receipt before inheriting the accepted identity. Current permissions, diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index bd6eaac678f..c8102eeb6eb 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -11,6 +11,7 @@ import { legacyControllerBootId } from "./legacy-controller-lease.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; import { readFailedPreparationSettlement } from "./failed-preparation-settlement.js"; +import { hasVerifiedPreparationLog } from "./verified-preparation-log.js"; import { executionStartedEventPredicate, noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js"; type Run = Pick { const [source] = await db.select({ agentId: heartbeatRuns.agentId, + contextSnapshot: sql>`jsonb_build_object( + 'skillSelectionSnapshot', ${heartbeatRuns.contextSnapshot}->'skillSelectionSnapshot', + 'skillVersions', ${heartbeatRuns.contextSnapshot}->'skillVersions')`, runtimeModeResolverVersion: heartbeatRuns.runtimeModeResolverVersion, runtimeModeReason: heartbeatRuns.runtimeModeReason, runnerInstanceId: heartbeatRuns.runnerInstanceId, nativePhase: heartbeatRuns.nativePhase, driverKind: heartbeatRuns.driverKind, driverVersion: heartbeatRuns.driverVersion, @@ -75,9 +79,13 @@ async function hasSettledFailedPreparationEvidence(db: Db, run: Run, historicalR && source.logRef === `${run.companyId}/${source.agentId}/${run.id}` && source.logBytes === 0 && source.logSha256 === EMPTY_LOG_SHA256 && [source.stdoutExcerpt, source.stderrExcerpt].every(value => value === null || value === ""); - if (!source || source.logCompressed || source.lastOutputSeq !== 0 || Object.entries(source).some(([key, value]) => - !["agentId", "logCompressed", "lastOutputSeq", "usageJson"].includes(key) && + const preparationLog = historicalRetryAncestor && source && + await hasVerifiedPreparationLog({ ...source, id: run.id, companyId: run.companyId }); + if (!source || source.logCompressed || !preparationLog && source.lastOutputSeq !== 0 || Object.entries(source).some(([key, value]) => + !["agentId", "contextSnapshot", "logCompressed", "lastOutputSeq", "usageJson"].includes(key) && !(historicalRetryAncestor && key === "sessionIdBefore") && + !(preparationLog && ["logStore", "logRef", "logBytes", "logSha256", "stdoutExcerpt", "stderrExcerpt", + "lastOutputAt", "lastOutputStream", "lastOutputBytes"].includes(key)) && !(emptyLog && ["logStore", "logRef", "logBytes", "logSha256", "stdoutExcerpt", "stderrExcerpt"].includes(key)) && value !== null) || !hasOnlyPreparationAccountingContext(source.usageJson, run.issueId ?? run.contextSnapshot?.issueId) || Object.keys(run.runnerProfileJson ?? {}).some(key => !["adapterDispatch", "remoteWorkspace", diff --git a/server/src/services/saved-bootstrap-continuation.test.ts b/server/src/services/saved-bootstrap-continuation.test.ts new file mode 100644 index 00000000000..c62a6200c38 --- /dev/null +++ b/server/src/services/saved-bootstrap-continuation.test.ts @@ -0,0 +1,186 @@ +import { createHash, randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, createDb, environmentLeases, executionWorkspaces, + executionWorkspaceSnapshots, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, + workspaceOperations, nativeRunFinalizations, approvals, issueApprovals, issueComments } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { requestTaskRetry } from "./task-retry.js"; +import { heartbeatService } from "./heartbeat.js"; +import { buildExecutionContinuation } from "./execution-continuation.js"; +import { createSystemRecovery } from "./system-recovery-state.js"; +import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { isFailedNativePreparationRetryAncestor } from "./cancelled-native-startup.js"; +import { withQueuedCommentIdsInWakePayload } from "./issue-queued-comment-queue.js"; + +const logs = vi.hoisted(() => new Map()); +vi.mock("./run-log-store.js", async importOriginal => { + const original = await importOriginal(); + return { ...original, getRunLogStore: () => ({ ...original.getRunLogStore(), + read: async (handle: { logRef: string }) => { + const content = logs.get(handle.logRef); + if (content === undefined) throw new Error("Fixture log missing"); + return { content }; + } }) }; +}); + +describe("saved bootstrap continuation admission", () => { + let postgres: Awaited>; + let db: ReturnType; + beforeAll(async () => { + postgres = await startEmbeddedPostgresTestDatabase("saved-bootstrap-"); + db = createDb(postgres.connectionString); + }, 30_000); + afterAll(async () => { await db?.$client.end(); await postgres?.cleanup(); }); + + async function seed(hostLog: boolean, configured = false) { + const companyId = randomUUID(), issueId = randomUUID(), agentId = randomUUID(); + const sourceId = randomUUID(), childId = randomUUID(), actionId = randomUUID(); + const workspaceId = randomUUID(), leaseId = randomUUID(), snapshotId = randomUUID(); + const snapshotOperationId = randomUUID(), recoveryOperationId = randomUUID(); + const providerLeaseId = randomUUID(), actorId = "board", now = Date.now(); + const sha256 = "a".repeat(64), error = "native_harness_storage_owner_missing"; + const startedAt = new Date(now - 60_000), finishedAt = new Date(now - 40_000); + await db.insert(companies).values({ id: companyId, name: "Saved bootstrap", issuePrefix: `S${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", + status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Report the existing repair status", + status: "todo", assigneeAgentId: agentId }); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, + mode: "isolated_workspace", strategyType: "isolated_workspace", name: "Recovered files", + metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId, + source: { kind: "empty" }, location: null, writerRunId: null, dirty: false } } }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issueId)); + const restore = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }; + const chunk = "[paperclip] Enabled run-scoped skills from issue mentions: fixture/status-report\n"; + const log = JSON.stringify({ ts: startedAt.toISOString(), attemptId: randomUUID(), stream: "stdout", chunk, seq: 1 }) + "\n"; + logs.set(`${companyId}/${agentId}/${sourceId}`, log); + const [source] = await db.insert(heartbeatRuns).values({ id: sourceId, companyId, issueId, agentId, + status: "failed", runtimeMode: "legacy", errorCode: "adapter_failed", error, executionStage: "preparing", + startedAt, finishedAt, createdAt: startedAt, controllerBootId: randomUUID(), + sessionIdBefore: randomUUID(), + contextSnapshot: { issueId, skillSelectionSnapshot: true, skillVersions: { "fixture/status-report": randomUUID(), + ...(configured ? { "fixture/configured-skill": randomUUID() } : {}) } }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" }, + remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId, + remoteCwd: "/fixture/saved-work", epoch: 1, state: "saved", snapshotId } }, + resultJson: { stopReason: "adapter_failed", executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: restore }, + ...(hostLog ? { logStore: "s3_segments", logRef: `${companyId}/${agentId}/${sourceId}`, + logBytes: Buffer.byteLength(log), logSha256: createHash("sha256").update(log).digest("hex"), + stdoutExcerpt: chunk, stderrExcerpt: "", lastOutputAt: startedAt, lastOutputSeq: 1, + lastOutputStream: "stdout", lastOutputBytes: Buffer.byteLength(log) } : {}), + }).returning(); + await db.insert(heartbeatRunEvents).values([ + { companyId, agentId, runId: sourceId, seq: 1, eventType: "lifecycle", level: "info", stream: "system", message: "run started" }, + { companyId, agentId, runId: sourceId, seq: 2, eventType: "error", level: "error", stream: "system", message: error }, + { companyId, agentId, runId: sourceId, seq: 3, eventType: "lifecycle", level: "info", stream: "system", message: "task.run.measured", + payload: { phase: "task.run.measured", outcome: "failed", durationMs: 20_000, durableSettlement: true } }, + ]); + await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId, heartbeatRunId: sourceId, + executionWorkspaceId: workspaceId, provider: "daytona", providerLeaseId, status: "released", + cleanupStatus: "success", leasePolicy: "retain_on_failure", releasedAt: finishedAt, + metadata: { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1", companyId, + runId: sourceId, leaseId, provider: "daytona", providerLeaseId, state: "stopped", confirmedAt: finishedAt.toISOString() } } }); + await db.insert(executionWorkspaceSnapshots).values({ id: snapshotId, companyId, executionWorkspaceId: workspaceId, + sourceOperationId: snapshotOperationId, sourceRunId: sourceId, generation: 1, epoch: 1, status: "committed", sha256 }); + const scope = { companyId, issueId, heartbeatRunId: sourceId, executionWorkspaceId: workspaceId, + status: "succeeded", finishedAt }; + await db.insert(workspaceOperations).values([ + { ...scope, id: snapshotOperationId, phase: "remote_workspace_snapshot", metadata: { snapshotId, sha256, epoch: 1, generation: 1 } }, + { ...scope, id: recoveryOperationId, phase: "workspace_recovery", metadata: { recoveryActionId: actionId, + storage: "remote_workspace", leaseId, providerLeaseId, snapshotId, sha256, epoch: 1, generation: 1 } }, + { ...scope, phase: "workspace_finalize", metadata: { owningService: "workspace_recovery", preservationOperationIds: [recoveryOperationId] } }, + ]); + await db.insert(issueRecoveryActions).values({ id: actionId, companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: sourceId, + status: "resolved", outcome: "blocked", nextAction: "Continue the saved request.", + evidence: { runId: sourceId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" }, + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: restore, + systemRecovery: { ...createSystemRecovery(sourceId), state: "complete", phase: "reconcile_effects", attempts: 1, + preservationOperationIds: [recoveryOperationId] }, continuationDelivery: "invalidated", + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: sourceId, status: "recovered", + stage: "saving_receipts", attempt: 1, attemptLimit: 6, requestIds: [] } } }); + const [child] = await db.insert(heartbeatRuns).values({ id: childId, companyId, issueId, agentId, + status: "failed", runtimeMode: "legacy", errorCode: "setup_failed", executionStage: "preparing", + error: "The workspace still has an unfinished writer or save operation.", retryOfRunId: sourceId, + startedAt: new Date(now - 30_000), finishedAt: new Date(now - 20_000), createdAt: new Date(now - 30_000), + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { stopReason: "adapter_failed", executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + contextSnapshot: { issueId, retryOfRunId: sourceId, retryReason: "native_session_interrupted" }, + }).returning(); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }); + const [comment] = await db.insert(issueComments).values({ companyId, issueId, + authorType: "user", authorUserId: actorId, body: "What is the latest status?", createdAt: new Date(now - 10_000) }).returning(); + const [savedQueue] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "on_demand", + reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: actorId, + payload: withQueuedCommentIdsInWakePayload({ issueId, executionWait: { reason: "execution_recovery" } }, [comment.id]) }).returning(); + expect(await hasVerifiedWorkspaceRepair(db, source)).toBe(true); + return { source, child, companyId, issueId, agentId, actorId, leaseId, comment, savedQueue }; + } + + it.each([{ hostLog: false, configured: false }, { hostLog: true, configured: false }, { hostLog: true, configured: true }])( + "dispatches one saved-status turn after a certified bootstrap with host log=$hostLog, configured=$configured", async ({ hostLog, configured }) => { + const f = await seed(hostLog, configured); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, + actorId: f.actorId, observedRunId: f.child.id, idempotencyKey: randomUUID(), actorSource: "local_implicit" }); + expect((await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: f.actorId, + observedRunId: f.child.id, idempotencyKey: randomUUID(), actorSource: "local_implicit" })).requestId).toBe(receipt.requestId); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!)]); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(queued, JSON.stringify(wake.payload)).toHaveLength(1); + expect(wake).toMatchObject({ status: "coalesced", runId: queued[0].id }); + expect(queued[0].contextSnapshot?.wakeCommentIds).toContain(f.comment.id); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.savedQueue.id)))[0]) + .toMatchObject({ status: "coalesced", runId: queued[0].id }); + expect(await isFailedNativePreparationRetryAncestor(db, f.source, undefined)).toBe(true); + await expect(buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: queued[0].id, context: queued[0].contextSnapshot!, summary: null, exposeLowTrustRaw: false })) + .resolves.toMatchObject({ interruptedRunId: f.child.id }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.source.id)))[0]) + .toMatchObject({ status: "failed", resultJson: f.source.resultJson }); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)))[0].evidence.automaticRecovery) + .toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" }); + }); + + it.each(["provider_started", "provider_event", "native_checkpoint", "missing_stop", "missing_selection", + "changed_hash", "unknown_output", "pending_approval", "operator_stop"] as const)( + "does not admit the host-log continuation after a late gate changes: %s", async gate => { + const f = await seed(true); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, + actorId: f.actorId, observedRunId: f.child.id, idempotencyKey: randomUUID(), actorSource: "local_implicit" }); + if (gate === "provider_started") await db.update(heartbeatRuns).set({ resultJson: { ...f.source.resultJson, + executionRecovery: { kind: "bootstrap", providerWorkStarted: true } } }).where(eq(heartbeatRuns.id, f.source.id)); + if (gate === "provider_event") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, + agentId: f.agentId, runId: f.source.id, seq: 4, eventType: "tool.execution.started" }); + if (gate === "native_checkpoint") { + await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId }).where(eq(heartbeatRuns.id, f.source.id)); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, + runId: f.source.id, phase: "running", attempt: 1, controllerGeneration: 1 }); + } + if (gate === "missing_stop") await db.update(environmentLeases).set({ metadata: {} }).where(eq(environmentLeases.id, f.leaseId)); + if (gate === "missing_selection") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId } }) + .where(eq(heartbeatRuns.id, f.source.id)); + if (gate === "changed_hash") await db.update(heartbeatRuns).set({ logSha256: "b".repeat(64) }).where(eq(heartbeatRuns.id, f.source.id)); + if (gate === "unknown_output") { + const record = JSON.parse(logs.get(f.source.logRef!)!); + record.chunk = "Provider says it completed an external action\n"; + const content = JSON.stringify(record) + "\n"; + logs.set(f.source.logRef!, content); + await db.update(heartbeatRuns).set({ stdoutExcerpt: record.chunk, logBytes: Buffer.byteLength(content), + lastOutputBytes: Buffer.byteLength(content), logSha256: createHash("sha256").update(content).digest("hex") }) + .where(eq(heartbeatRuns.id, f.source.id)); + } + if (gate === "pending_approval") { + const [approval] = await db.insert(approvals).values({ companyId: f.companyId, type: "hire_agent", status: "pending", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId: approval.id }); + } + if (gate === "operator_stop") await db.update(issues).set({ status: "cancelled", cancelledAt: new Date() }).where(eq(issues.id, f.issueId)); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!)]); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))) + .toHaveLength(0); + }); +}); diff --git a/server/src/services/verified-preparation-log.test.ts b/server/src/services/verified-preparation-log.test.ts new file mode 100644 index 00000000000..0ab4e7096e9 --- /dev/null +++ b/server/src/services/verified-preparation-log.test.ts @@ -0,0 +1,84 @@ +import { createHash, randomUUID } from "node:crypto"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { hasVerifiedPreparationLog } from "./verified-preparation-log.js"; + +const fixture = vi.hoisted(() => ({ read: vi.fn() })); +vi.mock("./run-log-store.js", () => ({ getRunLogStore: () => ({ read: fixture.read }) })); + +function source() { + const companyId = randomUUID(), agentId = randomUUID(), id = randomUUID(), ts = new Date(); + const chunk = "[paperclip] Enabled run-scoped skills from issue mentions: fixture/status-report\n"; + const record = { ts: ts.toISOString(), attemptId: randomUUID(), stream: "stdout", chunk, seq: 1 }; + const content = JSON.stringify(record) + "\n"; + const run: Parameters[0] = { companyId, agentId, id, + contextSnapshot: { skillSelectionSnapshot: true, skillVersions: { "fixture/status-report": randomUUID() } }, + logStore: "s3_segments", logRef: `${companyId}/${agentId}/${id}`, logBytes: Buffer.byteLength(content), + logSha256: createHash("sha256").update(content).digest("hex"), logCompressed: false, + stdoutExcerpt: chunk, stderrExcerpt: "", lastOutputAt: ts, lastOutputSeq: 1, + lastOutputStream: "stdout", lastOutputBytes: Buffer.byteLength(content) }; + return { run, content, record }; +} + +describe("verified host preparation output", () => { + beforeEach(() => { fixture.read.mockReset(); }); + it("verifies the complete canonical bytes against the persisted selection and output watermark", async () => { + const f = source(); + fixture.read.mockResolvedValue({ content: f.content }); + expect(await hasVerifiedPreparationLog(f.run)).toBe(true); + expect(fixture.read).toHaveBeenCalledTimes(1); + expect(fixture.read.mock.calls[0][1]).toMatchObject({ offset: 0 }); + expect(fixture.read.mock.calls[0][1].limitBytes).toBeLessThan(4096); + expect(fixture.read.mock.calls[0][1].signal.aborted).toBe(true); + }); + + it.each(["versioned", "unversioned"])("verifies mentioned skills when the saved selection also has %s configured skills", async kind => { + const f = source(); + f.run.contextSnapshot!.skillVersions = { "fixture/configured-skill": kind === "versioned" ? randomUUID() : null, + "fixture/status-report": randomUUID() }; + fixture.read.mockResolvedValue({ content: f.content }); + expect(await hasVerifiedPreparationLog(f.run)).toBe(true); + }); + + it.each(["missing_selection", "different_selection", "malformed_version", "foreign_log", "changed_hash", + "truncated", "extra_record", "provider_output", "changed_timestamp", "extra_provenance", "larger_log", + "other_stream", "changed_watermark", "missing_log"] as const)("keeps uncertain output held: %s", async state => { + const f = source(); + let content = f.content; + if (state === "missing_selection") f.run.contextSnapshot = null; + if (state === "different_selection") f.run.contextSnapshot!.skillVersions = { "other/skill": randomUUID() }; + if (state === "malformed_version") f.run.contextSnapshot!.skillVersions = { "fixture/status-report": "unbound" }; + if (state === "foreign_log") f.run.logRef = `${randomUUID()}/${f.run.agentId}/${f.run.id}`; + if (state === "changed_hash") f.run.logSha256 = "b".repeat(64); + if (state === "larger_log") f.run.logBytes = 4096; + if (state === "other_stream") f.run.lastOutputStream = "stderr"; + if (state === "changed_watermark") f.run.lastOutputSeq = 2; + if (state === "extra_record") content += JSON.stringify({ ...f.record, chunk: "Provider performed a tool action", seq: 2 }) + "\n"; + if (state === "provider_output") { f.record.chunk = "Provider completed an action\n"; content = JSON.stringify(f.record) + "\n"; f.run.stdoutExcerpt = f.record.chunk; } + if (state === "changed_timestamp") content = JSON.stringify({ ...f.record, ts: new Date(0).toISOString() }) + "\n"; + if (state === "extra_provenance") content = JSON.stringify({ ...f.record, sourceCursor: 1 }) + "\n"; + if (["extra_record", "provider_output", "changed_timestamp", "extra_provenance"].includes(state)) { + f.run.logSha256 = createHash("sha256").update(content).digest("hex"); + f.run.logBytes = f.run.lastOutputBytes = Buffer.byteLength(content); + } + if (state === "missing_log") fixture.read.mockRejectedValue(new Error("Object log not found")); + else fixture.read.mockResolvedValue({ content, ...(state === "truncated" ? { nextOffset: f.run.logBytes } : {}) }); + expect(await hasVerifiedPreparationLog(f.run)).toBe(false); + }); + + it("cancels a slow read, fails closed, and leaves an unrelated recovery responsive", async () => { + const slow = source(), fast = source(); + let signal: AbortSignal | undefined; + fixture.read.mockImplementation((handle, options) => { + if (handle.logRef === fast.run.logRef) return Promise.resolve({ content: fast.content }); + signal = options.signal; + return new Promise((_, reject) => options.signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true })); + }); + const pending = hasVerifiedPreparationLog(slow.run); + const ready = hasVerifiedPreparationLog(fast.run); + expect(await Promise.race([pending.then(() => "slow"), ready.then(() => "fast")])).toBe("fast"); + expect(await ready).toBe(true); + expect(await pending).toBe(false); + expect(signal?.aborted).toBe(true); + expect(fixture.read).toHaveBeenCalledTimes(2); + }); +}); diff --git a/server/src/services/verified-preparation-log.ts b/server/src/services/verified-preparation-log.ts new file mode 100644 index 00000000000..161d18441ea --- /dev/null +++ b/server/src/services/verified-preparation-log.ts @@ -0,0 +1,52 @@ +import { createHash } from "node:crypto"; +import { isUuidLike } from "@paperclipai/shared"; +import { getRunLogStore } from "./run-log-store.js"; + +const MAX_BYTES = 2048; +const READ_TIMEOUT_MS = 2000; +type PreparationLog = { + companyId: string; agentId: string; id: string; + contextSnapshot: Record | null; + logStore: string | null; logRef: string | null; logBytes: number | null; logSha256: string | null; + logCompressed: boolean; stdoutExcerpt: string | null; stderrExcerpt: string | null; + lastOutputAt: Date | null; lastOutputSeq: number; lastOutputStream: string | null; lastOutputBytes: number | null; +}; + +export async function hasVerifiedPreparationLog(run: PreparationLog): Promise { + const versions = run.contextSnapshot?.skillVersions; + if (run.contextSnapshot?.skillSelectionSnapshot !== true || !versions || typeof versions !== "object" + || Array.isArray(versions) || Object.keys(versions).length === 0 + || run.logStore !== "s3_segments" || run.logRef !== `${run.companyId}/${run.agentId}/${run.id}` + || !Number.isSafeInteger(run.logBytes) || run.logBytes! <= 0 || run.logBytes! > MAX_BYTES + || !run.logSha256 || !/^[a-f0-9]{64}$/.test(run.logSha256) || run.logCompressed + || run.lastOutputSeq !== 1 || run.lastOutputStream !== "stdout" || !run.lastOutputAt + || run.lastOutputBytes !== run.logBytes || run.stderrExcerpt !== "") return false; + const controller = new AbortController(); + let timer: ReturnType | undefined; + try { + const read = getRunLogStore().read({ store: "s3_segments", logRef: run.logRef }, + { offset: 0, limitBytes: MAX_BYTES + 1, signal: controller.signal }); + const result = await Promise.race([read, new Promise(resolve => { + timer = setTimeout(() => { controller.abort(); resolve(null); }, READ_TIMEOUT_MS); + timer.unref(); + })]); + if (!result || result.nextOffset !== undefined || Buffer.byteLength(result.content) !== run.logBytes + || createHash("sha256").update(result.content).digest("hex") !== run.logSha256) return false; + const record = JSON.parse(result.content) as Record | null; + if (!record || Array.isArray(record) || typeof record !== "object" + || Object.keys(record).some(key => !["ts", "attemptId", "stream", "chunk", "seq"].includes(key)) + || record.ts !== run.lastOutputAt.toISOString() || typeof record.attemptId !== "string" || !isUuidLike(record.attemptId) + || record.seq !== 1 || record.stream !== "stdout" || typeof record.chunk !== "string" + || record.chunk !== run.stdoutExcerpt) return false; + const prefix = "[paperclip] Enabled run-scoped skills from issue mentions: "; + if (!record.chunk.startsWith(prefix) || !record.chunk.endsWith("\n")) return false; + const keys = record.chunk.slice(prefix.length, -1).split(", "); + return keys.length > 0 && new Set(keys).size === keys.length + && keys.every(key => { + const version = (versions as Record)[key]; + return key.length > 0 && Object.hasOwn(versions, key) && !/[\r\n]/.test(key) + && typeof version === "string" && isUuidLike(version); + }); + } catch { return false; } + finally { if (timer) clearTimeout(timer); controller.abort(); } +}