mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 16:02:36 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
96da284ed8
commit
0e111c5075
5 files changed
+340
-2
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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<typeof heartbeatRuns.$inferSelect,
|
||||
@@ -59,6 +60,9 @@ async function failedNativePreparation(db: Db, run: Run, coordinator: Coordinato
|
||||
|
||||
async function hasSettledFailedPreparationEvidence(db: Db, run: Run, historicalRetryAncestor = false): Promise<boolean> {
|
||||
const [source] = await db.select({ agentId: heartbeatRuns.agentId,
|
||||
contextSnapshot: sql<Record<string, unknown>>`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",
|
||||
|
||||
@@ -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<string, string>());
|
||||
vi.mock("./run-log-store.js", async importOriginal => {
|
||||
const original = await importOriginal<typeof import("./run-log-store.js")>();
|
||||
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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<typeof hasVerifiedPreparationLog>[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);
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> | 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<boolean> {
|
||||
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<typeof setTimeout> | 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<null>(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<string, unknown> | 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<string, unknown>)[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(); }
|
||||
}
|
||||
Reference in new issue
Block a user