mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 00:05:49 +02:00
refactor(heartbeat): extract instruction working-copy lifecycle (#15984)
## Thinking Path > - Paperclip runs agent tasks and preserves their managed agent files. > - Heartbeat must prepare a private copy, collect stopped-provider edits, and retain ownership during warm-session handoff. > - That lifecycle is spread across preparation, adapter callbacks, and cleanup. > - A per-run controller can own the copy and receipt while the executor retains teardown ordering. > - This PR moves that coordination into a dedicated module. > - Live getters preserve updated state and partially acquired resources. ## Linked Issues or Issue Description **What existing behavior does this improve?** Maintenance of heartbeat instruction preparation. This is the second and final PR in this batch, originally stacked on #15980. That dependency is now merged; this PR is rebased onto its squash merge with an identical patch and identical touched-file contents. Related-PR search found no duplicate extraction. **Current behavior** After #15980, `heartbeat.ts` has 4,215 lines and 173,657 bytes. It coordinates instruction-copy preparation, save receipts, warm handoffs, and deferred cleanup inline. **Proposed behavior** `heartbeat/instruction-working-copy.ts` owns the per-run lifecycle in 233 lines. `heartbeat.ts` becomes 4,055 lines and 162,520 bytes. The executor retains the outer release order, unavailable-copy reporting, and environment/service cleanup policy. **Reason and benefit** The copy, save receipt, and warm ownership are readable together. The remaining executor names preparation and dispatch phases without obscuring teardown decisions. **Breaking changes** None intended. All 140 named public heartbeat exports remain unchanged. ## What Changed - Extract copy preparation, save/checkpoint callbacks, warm reservation/adoption, prompt guidance, and the deferred-release closure. - Read the executor's current run row through a callback; expose current copies, receipts, and acquired cleanup handles through live getters. - Keep cleanup handles available after preparation failure, and preserve release-before-retry on invalidated warm reuse. - Add four public-path characterization tests with real disposable PostgreSQL and files, plus four direct boundary tests. - Add isolated browser acceptance with a real process that edits instructions, completes its task, and reads the saved bytes on its next task. - Document the boundary and opt-in acceptance command. ## Verification - **Before extraction:** four new public-path tests pass on the original implementation, as do 108 existing instruction-copy/directory/native-cleanup tests. The two-task browser journey also passes on the baseline server. - **After extraction:** all 112 baseline tests pass; all four new boundary tests pass; 12 selected native-session ownership/retirement cases pass (the other 670 cases were outside this focused selection). - **Browser acceptance:** the fresh extracted-code server passes the full two-task journey (25.9 seconds): real child-process edits, saved/unchanged receipts, canonical bytes through the agent-file API, next-task restore, completed task UI, and copy cleanup. - `pnpm -r typecheck`, final server typecheck, and `pnpm build` pass. Module-boundary and public-export checks pass. - Normalized syntax comparison confirms all moved bodies and the reconstructed executor match the base, accounting explicitly for live run getters and delegation. - Final rebased head: `a08872c97e587a21ffc0fc6b78f34b81e484e7b0`. Combined entry-point and boundary regressions pass again (12/12); fresh Greptile review is 5/5 with no findings. All 55 current-head checks pass (two configured skips), including the hosted full-suite tests, build, typecheck, and all eight browser shards. There are no open review threads. ## Risks - Acquisition can succeed before preparation throws. Boundary tests verify that both the reservation and a transferred copy remain visible to the cleanup owner. - Callbacks can replace copy/save state during execution. Public tests verify saved receipts; direct tests verify the current run row, warm checkpoint receipt, and permission-root preservation. - Missing edit authorization and cleanup failure must remain nonfatal. Characterization tests revoke actual membership after dispatch admission and inject a cleanup error while preserving the saved result. - The browser uses a test-launcher-only instruction-capable wrapper around the real process adapter. Production adapter capabilities are unchanged. No live provider calls are made; native warm ownership is covered by focused module and native-session tests rather than a live provider browser run. ## Model Used OpenAI Codex, based on GPT-6, assisted with reasoning, code edits, tool execution, and verification. The exact deployment model ID and context-window size are not exposed in this session. ## 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:
DottaandPaperclip authored and GitHub committed
1 parent
1e0cfcb4fc
commit
ce08872dfb
10 files changed
+633
-179
No files matched your search
@@ -0,0 +1,159 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, heartbeatRunEvents, issues, principalPermissionGrants } from "@paperclipai/db";
|
||||
import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const cleanup = vi.hoisted(() => ({ failures: new Set<string>(), revocations: new Set<string>(), attempts: new Map<string, number>() }));
|
||||
vi.mock("../services/agent-instruction-working-copies.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("../services/agent-instruction-working-copies.js")>();
|
||||
return { ...actual, agentInstructionWorkingCopyService: (...args: Parameters<typeof actual.agentInstructionWorkingCopyService>) => {
|
||||
const copies = actual.agentInstructionWorkingCopyService(...args);
|
||||
return { ...copies, prepare: async (input: Parameters<typeof copies.prepare>[0]) => {
|
||||
if (cleanup.revocations.has(input.companyId)) {
|
||||
await args[0].update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.companyId, input.companyId));
|
||||
}
|
||||
return copies.prepare(input);
|
||||
}, release: async (companyId: string, runId: string) => {
|
||||
cleanup.attempts.set(runId, (cleanup.attempts.get(runId) ?? 0) + 1);
|
||||
if (cleanup.failures.has(companyId)) throw new Error("Fixture cleanup unavailable");
|
||||
return copies.release(companyId, runId);
|
||||
} };
|
||||
} };
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describePostgres = support.supported ? describe : describe.skip;
|
||||
if (!support.supported) console.warn(`Skipping heartbeat instruction copies tests: ${support.reason}`);
|
||||
|
||||
describePostgres("heartbeat instruction working-copy lifecycle", () => {
|
||||
const adapterType = "instruction_working_copy_test";
|
||||
const entryFile = "policy/INSTRUCTIONS.txt";
|
||||
const initial = "\uFEFF# Original\r\n☃\n", edited = "\uFEFF# Saved\r\nKeep across runs. ☃\n";
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
let home: string;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let heartbeat: ReturnType<typeof heartbeatService>;
|
||||
const captured = new Map<string, { instructionPath: string | null; before: string | null; context: Record<string, unknown> }>();
|
||||
const failures = new Set<string>();
|
||||
|
||||
beforeAll(async () => {
|
||||
home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "heartbeat-instruction-copies-")));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
temporary = await startEmbeddedPostgresTestDatabase("heartbeat-instruction-copies-db-");
|
||||
db = createDb(temporary.connectionString);
|
||||
heartbeat = heartbeatService(db);
|
||||
registerServerAdapter({
|
||||
type: adapterType, supportsLocalAgentJwt: false, supportsInstructionsBundle: true,
|
||||
execute: async input => {
|
||||
const instructionPath = typeof input.config.instructionsFilePath === "string" ? input.config.instructionsFilePath : null;
|
||||
captured.set(input.runId, { instructionPath, before: instructionPath ? await fs.readFile(instructionPath, "utf8") : null, context: structuredClone(input.context) });
|
||||
if (instructionPath) await fs.writeFile(instructionPath, edited);
|
||||
await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, String(input.context.issueId)));
|
||||
await input.onProviderStopped?.();
|
||||
if (failures.has(input.agent.id)) throw new Error("Invalid fixture adapter configuration");
|
||||
return { exitCode: 0, signal: null, timedOut: false, resultJson: {} };
|
||||
},
|
||||
testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }),
|
||||
});
|
||||
}, 30_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await heartbeat?.drainActiveRunExecutions();
|
||||
unregisterServerAdapter(adapterType);
|
||||
await db?.$client.end({ timeout: 0 });
|
||||
await temporary?.cleanup();
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
|
||||
if (home) {
|
||||
const makeRemovable = async (directory: string): Promise<void> => {
|
||||
await fs.chmod(directory, 0o700);
|
||||
for (const child of await fs.readdir(directory, { withFileTypes: true })) if (child.isDirectory()) await makeRemovable(path.join(directory, child.name));
|
||||
};
|
||||
await makeRemovable(home);
|
||||
await fs.rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
async function seed() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), userId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Instruction lifecycle fixture", issuePrefix: `I${companyId.slice(0, 6)}` });
|
||||
await db.insert(authUsers).values({ id: userId, name: "Editor", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "operator" });
|
||||
await db.insert(principalPermissionGrants).values({ companyId, principalType: "user", principalId: userId, permissionKey: "agents:configure", scope: { agentIds: [agentId] } });
|
||||
const root = resolveManagedInstructionsRoot({ companyId, id: agentId, name: "Editor", adapterConfig: {} });
|
||||
await fs.mkdir(path.dirname(path.join(root, entryFile)), { recursive: true });
|
||||
await fs.writeFile(path.join(root, entryFile), initial);
|
||||
const cwd = await fs.mkdtemp(path.join(home, "workspace-"));
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Editor", adapterType, status: "idle", adapterConfig: { cwd, instructionsBundleMode: "managed", instructionsRootPath: root, instructionsEntryFile: entryFile } });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "agent", principalId: agentId, membershipRole: "member" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Save agent instructions", status: "todo", assigneeAgentId: agentId, responsibleUserId: userId, issueNumber: 1 });
|
||||
return { companyId, agentId, issueId, root };
|
||||
}
|
||||
|
||||
async function execute(f: Awaited<ReturnType<typeof seed>>) {
|
||||
const run = await heartbeat.wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "manual", payload: { issueId: f.issueId }, contextSnapshot: { issueId: f.issueId } });
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [finished] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, run!.id));
|
||||
return { finished: finished!, events, input: captured.get(run!.id)! };
|
||||
}
|
||||
|
||||
it("dispatches a private copy, saves exact bytes on provider stop, and restores them on the next run", async () => {
|
||||
const f = await seed();
|
||||
const first = await execute(f);
|
||||
expect(first.finished.status, first.finished.error ?? undefined).toBe("succeeded");
|
||||
expect(first.input.before).toBe(initial);
|
||||
expect(first.input.instructionPath).not.toBe(path.join(f.root, entryFile));
|
||||
expect(first.input.context.paperclipTaskMarkdown).toContain("Your persistent agent directory is");
|
||||
expect(first.finished.resultJson?.instructionSave).toMatchObject({ state: "saved", contract: "agent_files", entryFile });
|
||||
expect(first.events.some(event => event.eventType === "instruction_save" && event.payload?.state === "saved")).toBe(true);
|
||||
expect(await fs.readFile(path.join(f.root, entryFile), "utf8")).toBe(edited);
|
||||
await expect(fs.stat(first.input.instructionPath!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
const second = await execute(f);
|
||||
expect(second.finished.status, second.finished.error ?? undefined).toBe("succeeded");
|
||||
expect(second.input.before).toBe(edited);
|
||||
expect(second.finished.resultJson?.instructionSave).toMatchObject({ state: "unchanged" });
|
||||
});
|
||||
|
||||
it("continues without a copy or save claim when access is revoked after dispatch admission", async () => {
|
||||
const f = await seed();
|
||||
cleanup.revocations.add(f.companyId);
|
||||
const { finished, input, events } = await execute(f);
|
||||
expect(finished.status, finished.error ?? undefined).toBe("succeeded");
|
||||
expect(input.instructionPath).toBeNull();
|
||||
expect(input.context.paperclipTaskMarkdown).toContain("No editable agent instruction working copy is registered");
|
||||
expect(finished.resultJson?.instructionSave).toBeUndefined();
|
||||
expect(events.some(event => event.eventType === "instruction_save" && event.payload?.state === "unavailable")).toBe(true);
|
||||
expect(await fs.readFile(path.join(f.root, entryFile), "utf8")).toBe(initial);
|
||||
});
|
||||
|
||||
it("keeps the successful save and result when cleanup fails, without repeating that failure in outer teardown", async () => {
|
||||
const f = await seed();
|
||||
cleanup.failures.add(f.companyId);
|
||||
const { finished, events } = await execute(f);
|
||||
expect(finished.status, finished.error ?? undefined).toBe("succeeded");
|
||||
expect(finished.resultJson?.instructionSave).toMatchObject({ state: "saved" });
|
||||
expect(await fs.readFile(path.join(f.root, entryFile), "utf8")).toBe(edited);
|
||||
expect(cleanup.attempts.get(finished.id)).toBe(1);
|
||||
expect(events.filter(event => event.eventType === "instruction_cleanup")).toEqual([expect.objectContaining({ level: "warn", payload: { state: "deferred" } })]);
|
||||
});
|
||||
|
||||
it("preserves stopped-provider edits and releases the copy even if adapter dispatch then fails", async () => {
|
||||
const f = await seed();
|
||||
failures.add(f.agentId);
|
||||
const { finished, input, events } = await execute(f);
|
||||
expect(finished.status).toBe("failed");
|
||||
expect(finished.error).toContain("Invalid fixture adapter configuration");
|
||||
expect(events.some(event => event.eventType === "instruction_save" && event.payload?.state === "saved")).toBe(true);
|
||||
expect(await fs.readFile(path.join(f.root, entryFile), "utf8")).toBe(edited);
|
||||
await expect(fs.stat(input.instructionPath!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
});
|
||||
@@ -1,3 +1,4 @@
|
||||
import { createHeartbeatInstructionCopyRelease, createHeartbeatInstructionWorkingCopy } from "./heartbeat/instruction-working-copy.js";
|
||||
import { prepareHeartbeatRuntimeServiceIntents, startHeartbeatRuntimeServices, recordHeartbeatAdapterRuntimeServices } from "./heartbeat/runtime-services.js";
|
||||
import { prepareHeartbeatExecutionConfiguration } from "./heartbeat/execution-configuration.js";
|
||||
import { createHeartbeatRunOutput } from "./heartbeat/run-output.js";
|
||||
@@ -216,7 +217,6 @@ import {
|
||||
import {
|
||||
prepareChatCompletionTurn,
|
||||
} from "./chat-completion-delivery.js";
|
||||
import { AgentDirectoryReuseInvalidatedError, isAgentDirectoryCopy } from "./agent-directory-working-copies.js";
|
||||
|
||||
import {
|
||||
NativeWorkspaceFinalizationBusyError,
|
||||
@@ -283,15 +283,12 @@ import {
|
||||
prepareGitHubExecutionEnvironment,
|
||||
startAdapterExecutionTargetPaperclipBridge,
|
||||
} from "@paperclipai/adapter-utils/execution-target";
|
||||
import { agentInstructionWorkingCopyService, collectStoppedInstructionCopyWithRetries, instructionWorkingCopyGuidance } from "./agent-instruction-working-copies.js";
|
||||
import { agentInstructionWorkingCopyService } from "./agent-instruction-working-copies.js";
|
||||
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import {
|
||||
createHash,
|
||||
} from "node:crypto";
|
||||
import {
|
||||
and,
|
||||
eq,
|
||||
@@ -336,9 +333,6 @@ import {
|
||||
} from "./issue-queued-comment-queue.js";
|
||||
|
||||
import {
|
||||
claimWarmNativeInstructionCopy,
|
||||
nativeSessionWorkspaceScope,
|
||||
reserveWarmNativeInstructionDirectory,
|
||||
dispatchNativeSessionResumptions,
|
||||
NativeCancellationPendingRecoveryError,
|
||||
NativeControllerDetachedForRestartError,
|
||||
@@ -348,7 +342,6 @@ import {
|
||||
} from "./native-runtime/index.js";
|
||||
|
||||
import {
|
||||
parseNativeExecutionInput,
|
||||
type NativeExecutionInput,
|
||||
type NativeSessionBackend,
|
||||
} from "../vendor/paperclip-runner/index.js";
|
||||
@@ -1855,29 +1848,7 @@ export function heartbeatService(
|
||||
run = claimed;
|
||||
}
|
||||
|
||||
const instructionCleanupRun = run;
|
||||
let instructionCleanupDeferred = false;
|
||||
const releaseInstructionCopy = async () => {
|
||||
// Cleanup is retried from the durable working-copy receipt by the
|
||||
// recovery sweep. It must not replace the provider result (or prevent
|
||||
// lease release), and a timeout must not be repeated in outer teardown.
|
||||
if (instructionCleanupDeferred) return;
|
||||
try {
|
||||
await instructionCopies.release(instructionCleanupRun.companyId, instructionCleanupRun.id);
|
||||
} catch (err) {
|
||||
instructionCleanupDeferred = true;
|
||||
logger.warn({ err, runId: instructionCleanupRun.id }, "Agent file cleanup deferred; run outcome preserved");
|
||||
await appendRunEvent(instructionCleanupRun, {
|
||||
eventType: "instruction_cleanup",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "Agent file cleanup was deferred. The run outcome and file-save receipt are unchanged.",
|
||||
payload: { state: "deferred" },
|
||||
}).catch((eventError) => {
|
||||
logger.warn({ err: eventError, runId: instructionCleanupRun.id }, "Failed to record deferred agent file cleanup");
|
||||
});
|
||||
}
|
||||
};
|
||||
const releaseInstructionCopy = createHeartbeatInstructionCopyRelease(run, instructionCopies, appendRunEvent);
|
||||
|
||||
if (
|
||||
runOptions.nativeLeaseOwner &&
|
||||
@@ -1978,8 +1949,7 @@ export function heartbeatService(
|
||||
Parameters<typeof cleanupGitHubOperationLaunchers>[0] | null = null;
|
||||
let nativeSessionResumeScheduled = false;
|
||||
let nativeOwnershipHeld = false;
|
||||
let releaseWarmInstructionPreparation: (() => Promise<void>) | null = null;
|
||||
let nativeInstructionReservation: Awaited<ReturnType<typeof reserveWarmNativeInstructionDirectory>> = null;
|
||||
let instructionWorkingCopy: ReturnType<typeof createHeartbeatInstructionWorkingCopy> | null = null;
|
||||
let nativeDispatchStarted = false;
|
||||
let nativeWorkspaceFinalizeScheduled = false;
|
||||
let nativeWorkspaceSync: Awaited<
|
||||
@@ -2740,49 +2710,12 @@ export function heartbeatService(
|
||||
appendRunEvent,
|
||||
},
|
||||
});
|
||||
let instructionCopy: Awaited<ReturnType<typeof instructionCopies.prepare>> = null;
|
||||
let instructionSave: Record<string, unknown> | null = null;
|
||||
const instructionPreparationKey = createHash("sha256").update(JSON.stringify({
|
||||
adapterType: agent.adapterType, adapterConfig: agent.adapterConfig, runtimeConfig: agent.runtimeConfig, sessionConfigMetadata,
|
||||
workspace: nativeSessionWorkspaceScope({
|
||||
binding: { runId: run.id, executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id },
|
||||
workspace: executionWorkspace,
|
||||
}), cwd: executionWorkspace.cwd,
|
||||
})).digest("hex");
|
||||
const recordInstructionSave = async (saved: NonNullable<Awaited<ReturnType<typeof instructionCopies.get>>>) => {
|
||||
const receipt = parseObject(saved.receipt);
|
||||
const storageWarning = readNonEmptyString(receipt.storageWarning);
|
||||
const state = saved.errorCode === "AGENT_FILES_CHECKPOINT_UNSTABLE" ? "pending_collection"
|
||||
: saved.state === "warm_saved" ? readNonEmptyString(receipt.checkpointState) ?? "saved" : saved.state;
|
||||
instructionSave = { state, entryFile: saved.entryFile,
|
||||
...(isAgentDirectoryCopy(saved) ? { contract: "agent_files", appliedCandidateHash: saved.candidateHash, checkpointStats: receipt.checkpointStats }
|
||||
: { revisionId: parseObject(receipt.revision).id ?? null }), storageWarning, errorCode: saved.errorCode, errorMessage: saved.errorMessage };
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system",
|
||||
level: !saved.errorCode && !storageWarning && ["saved", "unchanged", "resolved"].includes(state) ? "info" : "warn",
|
||||
message: storageWarning ?? (state === "saved" ? "Agent files saved."
|
||||
: state === "unchanged" ? "Instruction working copy is unchanged."
|
||||
: saved.errorMessage ?? "Instruction edits were not saved."), payload: instructionSave });
|
||||
};
|
||||
const collectStoppedInstructions = async () => {
|
||||
if (!instructionCopy) return;
|
||||
const saved = await collectStoppedInstructionCopyWithRetries(() => instructionCopies.collectStopped({
|
||||
companyId: agent.companyId, runId: run.id, target: executionTarget,
|
||||
}));
|
||||
if (!saved) return;
|
||||
if (saved.state !== "superseded") await recordInstructionSave(saved);
|
||||
};
|
||||
const nativeInstructionWorkingCopy = () => instructionCopy ? {
|
||||
...(isAgentDirectoryCopy(instructionCopy) ? { runId: run.id, preparationKey: instructionPreparationKey } : {}),
|
||||
root: instructionCopy.executionRoot,
|
||||
...(instructionCopy.receipt?.warm === true ? { checkpointWarm: async () => {
|
||||
const saved = await instructionCopies.checkpointWarm({ companyId: agent.companyId, runId: run.id, target: executionTarget });
|
||||
if (saved) await recordInstructionSave(saved);
|
||||
return saved?.state === "warm_saved" && saved.errorCode === null;
|
||||
} } : {}),
|
||||
hasChanges: () => instructionCopies.hasChanges({ companyId: agent.companyId, runId: run.id, target: executionTarget }),
|
||||
collectStopped: collectStoppedInstructions,
|
||||
retirementFailed: async () => { await instructionCopies.reportRetirementUnconfirmed(agent.companyId, run.id); },
|
||||
} : undefined;
|
||||
const preparedInstructions = instructionWorkingCopy = createHeartbeatInstructionWorkingCopy({
|
||||
db, instructionCopies, readRun: () => run!, agent, context,
|
||||
session: { taskSession, taskSessionForRun, sessionConfigMetadata },
|
||||
workspace: { executionTarget, executionWorkspace, persistedExecutionWorkspace },
|
||||
appendRunEvent,
|
||||
});
|
||||
if (managedAiRuntime && aiBinding) {
|
||||
try { await assertManagedAiProjectAuth({ ...resolvedConfig, cwd: executionWorkspace.cwd }, aiBinding.provider, executionTarget); }
|
||||
catch { throw new ConfigurationIncompleteFailure("Project authentication conflicts with this agent’s managed AI connection", { configurationIncomplete: { reason: "ai_connection_incompatible", actionUrl: `/agents/${agent.id}/runtime` } }); }
|
||||
@@ -3287,100 +3220,7 @@ export function heartbeatService(
|
||||
const hasInstructionFilesystem = nativeRuntimeResolution.kind !== "native"
|
||||
? adapter.supportsInstructionsBundle === true
|
||||
: !["claude_managed_agents_api", "aws_agentcore_harness_api", "openai_dot_mcp"].includes(nativeRuntimeResolution.profile.backend);
|
||||
if (hasInstructionFilesystem) {
|
||||
try {
|
||||
// Missing contract fields on a restored session mean the deployed
|
||||
// legacy format. New sessions opt into whole-directory persistence.
|
||||
const priorFileRun = taskSession?.lastRunId
|
||||
? await db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, taskSession.lastRunId), eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id))).then(rows => rows[0])
|
||||
: null;
|
||||
const savedFileInput = parseObject(parseObject(run.runnerProfileJson).nativeExecutionInput);
|
||||
const priorFileInput = Object.keys(savedFileInput).length ? savedFileInput : parseObject(parseObject(priorFileRun?.profile).nativeExecutionInput);
|
||||
const priorWorkingCopy = parseObject(parseObject(parseObject(priorFileInput.runtimeContext).instructions).workingCopy);
|
||||
const warmFiles = nativeRuntimeResolution.kind === "native" && (nativeRuntimeResolution.profile.backend === "codex_app_server" ||
|
||||
(nativeRuntimeResolution.profile.backend === "acpx_runtime" && parseObject(agent.adapterConfig).acpxAgent === "cursor")) &&
|
||||
(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox"
|
||||
? executionTarget.runnerLifecyclePolicy?.mode === "warm"
|
||||
: parseObject(agent.adapterConfig).lifecycleMode === "warm");
|
||||
if (warmFiles && taskSessionForRun?.lastRunId) {
|
||||
nativeInstructionReservation = await reserveWarmNativeInstructionDirectory({ companyId: agent.companyId, agentId: agent.id,
|
||||
previousRunId: taskSessionForRun.lastRunId, runId: run.id, target: executionTarget,
|
||||
canReuse: () => instructionCopies.canReuseWarm(agent.companyId, agent.id, taskSessionForRun!.lastRunId!),
|
||||
});
|
||||
}
|
||||
if (!warmFiles && nativeRuntimeResolution.kind === "native" && priorWorkingCopy.kind === "agent_files" && taskSession) {
|
||||
releaseWarmInstructionPreparation = await claimWarmNativeInstructionCopy({
|
||||
priorExecution: parseNativeExecutionInput(priorFileInput), companyId: agent.companyId, agentId: agent.id,
|
||||
executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id, workspace: executionWorkspace,
|
||||
environmentId: executionTarget?.environmentId ?? null, runId: run.id, preparationKey: instructionPreparationKey, forceRetirement: !taskSessionForRun,
|
||||
adopt: async previousRunId => {
|
||||
instructionCopy = await instructionCopies.adopt({ companyId: agent.companyId, agentId: agent.id,
|
||||
runId: run.id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd, allowRetirementHandoff: true });
|
||||
return nativeInstructionWorkingCopy() ?? null;
|
||||
},
|
||||
});
|
||||
// A collection-only handoff was retired. Re-enter normal
|
||||
// preparation instead of composing a root already collected.
|
||||
if (!releaseWarmInstructionPreparation) instructionCopy = null;
|
||||
}
|
||||
const prepareInstructions = (reuseRunId?: string) => instructionCopies.prepare({
|
||||
companyId: agent.companyId, agentId: agent.id, runId: run.id,
|
||||
target: executionTarget, cwd: executionWorkspace.cwd,
|
||||
legacy: Object.keys(priorFileInput).length > 0 && priorWorkingCopy.kind !== "agent_files",
|
||||
warm: warmFiles, reuseRunId,
|
||||
onWarmHandoff: copy => {
|
||||
instructionCopy = copy;
|
||||
nativeInstructionReservation?.adopt(copy.executionRoot, collectStoppedInstructions);
|
||||
},
|
||||
});
|
||||
try {
|
||||
instructionCopy ??= await prepareInstructions(nativeInstructionReservation?.reuseRunId);
|
||||
} catch (error) {
|
||||
if (!(error instanceof AgentDirectoryReuseInvalidatedError)) throw error;
|
||||
// prepare has released its canonical lock. Retirement can now
|
||||
// collect under that same lock before a fresh restore starts.
|
||||
await nativeInstructionReservation?.release();
|
||||
nativeInstructionReservation = null;
|
||||
instructionCopy = await prepareInstructions();
|
||||
}
|
||||
} catch (error) {
|
||||
if ((error as { status?: number }).status !== 403) throw error;
|
||||
// Missing write identity must not break a background run's read-only
|
||||
// prompt. It must also never imply that ordinary file edits will save.
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
message: "Persistent instruction editing is unavailable. Use an authenticated user with instruction edit access and a managed instruction bundle.",
|
||||
payload: { state: "unavailable", code: "INSTRUCTION_COPY_UNAVAILABLE" } });
|
||||
const guidance = "No editable agent instruction working copy is registered for this turn. Use authenticated agent file tools for persistent edits; do not edit a private copy named in an earlier turn or claim its changes will persist.";
|
||||
for (const key of ["paperclipTaskMarkdown", "paperclipTaskMarkdownCompact"]) {
|
||||
context[key] = [readNonEmptyString(context[key]), guidance].filter(Boolean).join("\n\n");
|
||||
}
|
||||
}
|
||||
if (instructionCopy) {
|
||||
const storageWarning = readNonEmptyString(instructionCopy.receipt?.storageWarning);
|
||||
if (storageWarning) {
|
||||
instructionSave = { state: "prepared", contract: "agent_files", storageWarning };
|
||||
// This is an advisory on the run, never an agent pause, execution
|
||||
// failure, or scheduling gate. Keep it visible while work runs.
|
||||
await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ instructionSave })}::jsonb` }).where(eq(heartbeatRuns.id, run.id));
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
message: storageWarning, payload: instructionSave });
|
||||
}
|
||||
runtimeConfig = { ...runtimeConfig, instructionsFilePath: path.join(instructionCopy.localRoot, instructionCopy.entryFile) };
|
||||
if (isAgentDirectoryCopy(instructionCopy)) {
|
||||
const workspace = parseObject(context.paperclipWorkspace);
|
||||
context.paperclipWorkspace = { ...workspace, agentHome: instructionCopy.executionRoot,
|
||||
// Keep the pre-existing permission root stable for ACP session
|
||||
// identity. The per-run copy is already under the company root.
|
||||
agentHomeForPermissions: workspace.agentHome,
|
||||
};
|
||||
}
|
||||
const guidance = instructionWorkingCopyGuidance(instructionCopy);
|
||||
for (const key of ["paperclipTaskMarkdown", "paperclipTaskMarkdownCompact"]) {
|
||||
context[key] = [readNonEmptyString(context[key]), guidance].filter(Boolean).join("\n\n");
|
||||
}
|
||||
}
|
||||
}
|
||||
runtimeConfig = await preparedInstructions.prepare({ runtimeConfig, nativeRuntimeResolution, hasInstructionFilesystem });
|
||||
const runtimeSelection = await selectHeartbeatRuntime(db, {
|
||||
run,
|
||||
agent,
|
||||
@@ -3416,7 +3256,7 @@ export function heartbeatService(
|
||||
runtimeConfig,
|
||||
runtimeSkillEntries,
|
||||
runScopedMentionedSkillKeys,
|
||||
instructionCopy,
|
||||
instructionCopy: preparedInstructions.instructionCopy,
|
||||
agentIdentity,
|
||||
issueAssigneeOverrides,
|
||||
resolvedConfig,
|
||||
@@ -3490,12 +3330,12 @@ export function heartbeatService(
|
||||
goalCheckpointSession,
|
||||
},
|
||||
instructions: {
|
||||
nativeInstructionWorkingCopy,
|
||||
collectStoppedInstructions,
|
||||
nativeInstructionReservation,
|
||||
nativeInstructionWorkingCopy: preparedInstructions.nativeInstructionWorkingCopy,
|
||||
collectStoppedInstructions: preparedInstructions.collectStoppedInstructions,
|
||||
nativeInstructionReservation: preparedInstructions.nativeInstructionReservation,
|
||||
releaseInstructionCopy,
|
||||
get instructionCopy() { return instructionCopy; },
|
||||
getInstructionSave: () => instructionSave,
|
||||
get instructionCopy() { return preparedInstructions.instructionCopy; },
|
||||
getInstructionSave: preparedInstructions.getInstructionSave,
|
||||
},
|
||||
trace: {
|
||||
providerTraceCapture,
|
||||
@@ -3764,7 +3604,7 @@ export function heartbeatService(
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
await nativeInstructionReservation?.release().catch(error => logger.warn({ runId: run.id, err: error }, "Managed warm session preparation cleanup failed"));
|
||||
await instructionWorkingCopy?.nativeInstructionReservation?.release().catch(error => logger.warn({ runId: run.id, err: error }, "Managed warm session preparation cleanup failed"));
|
||||
if (managedAiRuntime) await managedAiRuntime.cleanup().catch(() => logger.warn({ runId: run.id }, "AI connection refresh or cleanup failed"));
|
||||
let latestRun = await getRun(run.id).catch(() => null);
|
||||
try {
|
||||
@@ -3852,7 +3692,7 @@ export function heartbeatService(
|
||||
}
|
||||
// A retained or unverified process stays above this release boundary.
|
||||
// If no stopped-copy capture occurred, preserve an explicit loss report.
|
||||
await releaseWarmInstructionPreparation?.();
|
||||
await instructionWorkingCopy?.releaseWarmInstructionPreparation?.();
|
||||
const uncapturedInstructions = await instructionCopies.reportUnavailable(run.companyId, run.id);
|
||||
if (uncapturedInstructions?.state === "unavailable") {
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { createHeartbeatInstructionWorkingCopy } from "./instruction-working-copy.js";
|
||||
import { AGENT_FILES_CONTRACT } from "../agent-file-store.js";
|
||||
import { AgentDirectoryReuseInvalidatedError } from "../agent-directory-working-copies.js";
|
||||
|
||||
const native = vi.hoisted(() => ({ reserve: vi.fn(), claim: vi.fn() }));
|
||||
vi.mock("../native-runtime/index.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../native-runtime/index.js")>(),
|
||||
reserveWarmNativeInstructionDirectory: native.reserve,
|
||||
claimWarmNativeInstructionCopy: native.claim,
|
||||
}));
|
||||
|
||||
type Input = Parameters<typeof createHeartbeatInstructionWorkingCopy>[0];
|
||||
type Prepare = Parameters<ReturnType<typeof createHeartbeatInstructionWorkingCopy>["prepare"]>[0];
|
||||
type Copy = NonNullable<Awaited<ReturnType<Input["instructionCopies"]["prepare"]>>>;
|
||||
|
||||
function fixture() {
|
||||
let run = { id: "current-run", companyId: "company", agentId: "agent", runnerProfileJson: null, status: "queued" } as ReturnType<Input["readRun"]>;
|
||||
const copy = { runId: run.id, companyId: run.companyId, agentId: run.agentId,
|
||||
localRoot: "/workspace/copy", executionRoot: "/workspace/copy", entryFile: "AGENTS.md",
|
||||
state: "prepared", receipt: { schema: AGENT_FILES_CONTRACT, warm: true }, errorCode: null, errorMessage: null,
|
||||
} as unknown as Copy;
|
||||
const copies = {
|
||||
prepare: vi.fn<Input["instructionCopies"]["prepare"]>().mockResolvedValue(copy),
|
||||
collectStopped: vi.fn<Input["instructionCopies"]["collectStopped"]>().mockResolvedValue({ ...copy, state: "saved" }),
|
||||
checkpointWarm: vi.fn<Input["instructionCopies"]["checkpointWarm"]>().mockResolvedValue({ ...copy, state: "warm_saved", receipt: { ...copy.receipt, checkpointState: "unchanged" } }),
|
||||
hasChanges: vi.fn().mockResolvedValue(false), canReuseWarm: vi.fn().mockResolvedValue(true),
|
||||
reportRetirementUnconfirmed: vi.fn().mockResolvedValue(null),
|
||||
};
|
||||
const appendRunEvent = vi.fn().mockResolvedValue(undefined);
|
||||
const context = { paperclipWorkspace: { agentHome: "/canonical" } } as Record<string, unknown>;
|
||||
const controller = createHeartbeatInstructionWorkingCopy({
|
||||
db: {} as Input["db"], instructionCopies: copies as unknown as Input["instructionCopies"],
|
||||
readRun: () => run,
|
||||
agent: { id: "agent", companyId: "company", adapterType: "codex_local", adapterConfig: { lifecycleMode: "warm" }, runtimeConfig: {} } as unknown as Input["agent"],
|
||||
context,
|
||||
session: { taskSession: null, taskSessionForRun: { lastRunId: "previous-run" }, sessionConfigMetadata: {} as Input["session"]["sessionConfigMetadata"] },
|
||||
workspace: { executionTarget: null, executionWorkspace: { cwd: "/workspace" } as Input["workspace"]["executionWorkspace"], persistedExecutionWorkspace: null },
|
||||
appendRunEvent,
|
||||
});
|
||||
const prepare: Prepare = { runtimeConfig: { retained: true }, hasInstructionFilesystem: true,
|
||||
nativeRuntimeResolution: { kind: "native", profile: { backend: "codex_app_server" } } as Prepare["nativeRuntimeResolution"] };
|
||||
const reservation = { reuseRunId: "previous-run", adopt: vi.fn(), release: vi.fn().mockResolvedValue(undefined) };
|
||||
native.reserve.mockResolvedValue(reservation);
|
||||
return { controller, copies, copy, context, prepare, reservation, appendRunEvent,
|
||||
replaceRun: () => { run = { ...run, status: "running" }; return run; } };
|
||||
}
|
||||
|
||||
beforeEach(() => { vi.clearAllMocks(); native.claim.mockResolvedValue(null); });
|
||||
|
||||
describe("instruction working-copy extraction boundary", () => {
|
||||
it("retains a warm reservation for outer cleanup when subsequent preparation throws", async () => {
|
||||
const f = fixture();
|
||||
f.copies.prepare.mockRejectedValue(new Error("preparation interrupted"));
|
||||
await expect(f.controller.prepare(f.prepare)).rejects.toThrow("preparation interrupted");
|
||||
expect(f.controller.nativeInstructionReservation).toBe(f.reservation);
|
||||
expect(f.controller.instructionCopy).toBeNull();
|
||||
await f.controller.nativeInstructionReservation?.release();
|
||||
expect(f.reservation.release).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("publishes transferred copy ownership before fallible preparation bookkeeping", async () => {
|
||||
const f = fixture();
|
||||
f.copies.prepare.mockImplementation(async input => {
|
||||
input.onWarmHandoff?.(f.copy);
|
||||
throw new Error("bookkeeping interrupted");
|
||||
});
|
||||
await expect(f.controller.prepare(f.prepare)).rejects.toThrow("bookkeeping interrupted");
|
||||
expect(f.controller.instructionCopy).toBe(f.copy);
|
||||
expect(f.reservation.adopt).toHaveBeenCalledWith(f.copy.executionRoot, f.controller.collectStoppedInstructions);
|
||||
await f.controller.nativeInstructionReservation?.release();
|
||||
expect(f.reservation.release).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("releases invalidated reuse before fresh preparation and exposes the replacement copy", async () => {
|
||||
const f = fixture(), order: string[] = [];
|
||||
f.reservation.release.mockImplementation(async () => { order.push("release"); });
|
||||
f.copies.prepare.mockImplementationOnce(async () => { order.push("reuse"); throw new AgentDirectoryReuseInvalidatedError("canonical changed"); })
|
||||
.mockImplementationOnce(async () => { order.push("fresh"); return f.copy; });
|
||||
const config = await f.controller.prepare(f.prepare);
|
||||
expect(order).toEqual(["reuse", "release", "fresh"]);
|
||||
expect(f.copies.prepare.mock.calls.map(([input]) => input.reuseRunId)).toEqual(["previous-run", undefined]);
|
||||
expect(f.controller.nativeInstructionReservation).toBeNull();
|
||||
expect(config).toMatchObject({ retained: true, instructionsFilePath: "/workspace/copy/AGENTS.md" });
|
||||
expect(f.controller.instructionCopy).toBe(f.copy);
|
||||
});
|
||||
|
||||
it("reads the current run row and receipt during stopped collection and warm checkpoints", async () => {
|
||||
const f = fixture();
|
||||
await f.controller.prepare(f.prepare);
|
||||
const running = f.replaceRun();
|
||||
await f.controller.collectStoppedInstructions();
|
||||
expect(f.appendRunEvent.mock.calls.at(-1)?.[0]).toBe(running);
|
||||
expect(f.controller.getInstructionSave()).toMatchObject({ state: "saved" });
|
||||
expect(await f.controller.nativeInstructionWorkingCopy()?.checkpointWarm?.()).toBe(true);
|
||||
expect(f.controller.getInstructionSave()).toMatchObject({ state: "unchanged" });
|
||||
expect(f.appendRunEvent.mock.calls.at(-1)?.[0]).toBe(running);
|
||||
expect(f.context.paperclipWorkspace).toMatchObject({ agentHome: f.copy.executionRoot, agentHomeForPermissions: "/canonical" });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,233 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import path from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { parseObject } from "../../adapters/utils.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { AgentDirectoryReuseInvalidatedError, isAgentDirectoryCopy } from "../agent-directory-working-copies.js";
|
||||
import { collectStoppedInstructionCopyWithRetries, instructionWorkingCopyGuidance, type agentInstructionWorkingCopyService } from "../agent-instruction-working-copies.js";
|
||||
import { claimWarmNativeInstructionCopy, nativeSessionWorkspaceScope, reserveWarmNativeInstructionDirectory } from "../native-runtime/index.js";
|
||||
import { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import type { HeartbeatRuntimeSelectionInput } from "./runtime-selection.js";
|
||||
import type { createHeartbeatLifecycle } from "./run-lifecycle.js";
|
||||
import type { EffectiveRunSessionConfigMetadata } from "./workspaces.js";
|
||||
|
||||
type Copies = ReturnType<typeof agentInstructionWorkingCopyService>;
|
||||
type Selection = HeartbeatRuntimeSelectionInput;
|
||||
type AppendRunEvent = ReturnType<typeof createHeartbeatLifecycle>["appendRunEvent"];
|
||||
|
||||
function readNonEmptyString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
/** A failed cleanup is deferred once per run; it cannot replace the run result. */
|
||||
export function createHeartbeatInstructionCopyRelease(
|
||||
instructionCleanupRun: Selection["run"],
|
||||
instructionCopies: Pick<Copies, "release">,
|
||||
appendRunEvent: AppendRunEvent,
|
||||
) {
|
||||
let instructionCleanupDeferred = false;
|
||||
const releaseInstructionCopy = async () => {
|
||||
// Cleanup is retried from the durable working-copy receipt by the
|
||||
// recovery sweep. It must not replace the provider result (or prevent
|
||||
// lease release), and a timeout must not be repeated in outer teardown.
|
||||
if (instructionCleanupDeferred) return;
|
||||
try {
|
||||
await instructionCopies.release(instructionCleanupRun.companyId, instructionCleanupRun.id);
|
||||
} catch (err) {
|
||||
instructionCleanupDeferred = true;
|
||||
logger.warn({ err, runId: instructionCleanupRun.id }, "Agent file cleanup deferred; run outcome preserved");
|
||||
await appendRunEvent(instructionCleanupRun, {
|
||||
eventType: "instruction_cleanup",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "Agent file cleanup was deferred. The run outcome and file-save receipt are unchanged.",
|
||||
payload: { state: "deferred" },
|
||||
}).catch((eventError) => {
|
||||
logger.warn({ err: eventError, runId: instructionCleanupRun.id }, "Failed to record deferred agent file cleanup");
|
||||
});
|
||||
}
|
||||
};
|
||||
return releaseInstructionCopy;
|
||||
}
|
||||
|
||||
interface InstructionWorkingCopyInput {
|
||||
db: Db;
|
||||
instructionCopies: Copies;
|
||||
// The executor replaces its run row before preparation. Callbacks must see it.
|
||||
readRun: () => Selection["run"];
|
||||
agent: Selection["agent"];
|
||||
context: Selection["task"]["context"];
|
||||
session: {
|
||||
taskSession: { lastRunId: string | null } | null;
|
||||
taskSessionForRun: Selection["session"]["taskSessionForRun"];
|
||||
sessionConfigMetadata: EffectiveRunSessionConfigMetadata;
|
||||
};
|
||||
workspace: Pick<Selection["workspace"], "executionTarget" | "executionWorkspace" | "persistedExecutionWorkspace">;
|
||||
appendRunEvent: AppendRunEvent;
|
||||
}
|
||||
|
||||
/** Per-run copy, receipt, and warm ownership; the executor orders final releases. */
|
||||
export function createHeartbeatInstructionWorkingCopy(input: InstructionWorkingCopyInput) {
|
||||
const { db, instructionCopies, readRun, agent, context, appendRunEvent } = input;
|
||||
const { taskSession, taskSessionForRun, sessionConfigMetadata } = input.session;
|
||||
const { executionTarget, executionWorkspace, persistedExecutionWorkspace } = input.workspace;
|
||||
let releaseWarmInstructionPreparation: (() => Promise<void>) | null = null;
|
||||
let nativeInstructionReservation: Awaited<ReturnType<typeof reserveWarmNativeInstructionDirectory>> = null;
|
||||
let instructionCopy: Awaited<ReturnType<typeof instructionCopies.prepare>> = null;
|
||||
let instructionSave: Record<string, unknown> | null = null;
|
||||
const instructionPreparationKey = createHash("sha256").update(JSON.stringify({
|
||||
adapterType: agent.adapterType, adapterConfig: agent.adapterConfig, runtimeConfig: agent.runtimeConfig, sessionConfigMetadata,
|
||||
workspace: nativeSessionWorkspaceScope({
|
||||
binding: { runId: readRun().id, executionWorkspaceId: persistedExecutionWorkspace?.id ?? readRun().id },
|
||||
workspace: executionWorkspace,
|
||||
}), cwd: executionWorkspace.cwd,
|
||||
})).digest("hex");
|
||||
const recordInstructionSave = async (saved: NonNullable<Awaited<ReturnType<typeof instructionCopies.get>>>) => {
|
||||
const receipt = parseObject(saved.receipt);
|
||||
const storageWarning = readNonEmptyString(receipt.storageWarning);
|
||||
const state = saved.errorCode === "AGENT_FILES_CHECKPOINT_UNSTABLE" ? "pending_collection"
|
||||
: saved.state === "warm_saved" ? readNonEmptyString(receipt.checkpointState) ?? "saved" : saved.state;
|
||||
instructionSave = { state, entryFile: saved.entryFile,
|
||||
...(isAgentDirectoryCopy(saved) ? { contract: "agent_files", appliedCandidateHash: saved.candidateHash, checkpointStats: receipt.checkpointStats }
|
||||
: { revisionId: parseObject(receipt.revision).id ?? null }), storageWarning, errorCode: saved.errorCode, errorMessage: saved.errorMessage };
|
||||
await appendRunEvent(readRun(), { eventType: "instruction_save", stream: "system",
|
||||
level: !saved.errorCode && !storageWarning && ["saved", "unchanged", "resolved"].includes(state) ? "info" : "warn",
|
||||
message: storageWarning ?? (state === "saved" ? "Agent files saved."
|
||||
: state === "unchanged" ? "Instruction working copy is unchanged."
|
||||
: saved.errorMessage ?? "Instruction edits were not saved."), payload: instructionSave });
|
||||
};
|
||||
const collectStoppedInstructions = async () => {
|
||||
if (!instructionCopy) return;
|
||||
const saved = await collectStoppedInstructionCopyWithRetries(() => instructionCopies.collectStopped({
|
||||
companyId: agent.companyId, runId: readRun().id, target: executionTarget,
|
||||
}));
|
||||
if (!saved) return;
|
||||
if (saved.state !== "superseded") await recordInstructionSave(saved);
|
||||
};
|
||||
const nativeInstructionWorkingCopy = () => instructionCopy ? {
|
||||
...(isAgentDirectoryCopy(instructionCopy) ? { runId: readRun().id, preparationKey: instructionPreparationKey } : {}),
|
||||
root: instructionCopy.executionRoot,
|
||||
...(instructionCopy.receipt?.warm === true ? { checkpointWarm: async () => {
|
||||
const saved = await instructionCopies.checkpointWarm({ companyId: agent.companyId, runId: readRun().id, target: executionTarget });
|
||||
if (saved) await recordInstructionSave(saved);
|
||||
return saved?.state === "warm_saved" && saved.errorCode === null;
|
||||
} } : {}),
|
||||
hasChanges: () => instructionCopies.hasChanges({ companyId: agent.companyId, runId: readRun().id, target: executionTarget }),
|
||||
collectStopped: collectStoppedInstructions,
|
||||
retirementFailed: async () => { await instructionCopies.reportRetirementUnconfirmed(agent.companyId, readRun().id); },
|
||||
} : undefined;
|
||||
|
||||
async function prepare({ runtimeConfig, nativeRuntimeResolution, hasInstructionFilesystem }: {
|
||||
runtimeConfig: Record<string, unknown>;
|
||||
nativeRuntimeResolution: Selection["nativeRuntimeResolution"];
|
||||
hasInstructionFilesystem: boolean;
|
||||
}) {
|
||||
if (hasInstructionFilesystem) {
|
||||
try {
|
||||
// Missing contract fields on a restored session mean the deployed
|
||||
// legacy format. New sessions opt into whole-directory persistence.
|
||||
const priorFileRun = taskSession?.lastRunId
|
||||
? await db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, taskSession.lastRunId), eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id))).then(rows => rows[0])
|
||||
: null;
|
||||
const savedFileInput = parseObject(parseObject(readRun().runnerProfileJson).nativeExecutionInput);
|
||||
const priorFileInput = Object.keys(savedFileInput).length ? savedFileInput : parseObject(parseObject(priorFileRun?.profile).nativeExecutionInput);
|
||||
const priorWorkingCopy = parseObject(parseObject(parseObject(priorFileInput.runtimeContext).instructions).workingCopy);
|
||||
const warmFiles = nativeRuntimeResolution.kind === "native" && (nativeRuntimeResolution.profile.backend === "codex_app_server" ||
|
||||
(nativeRuntimeResolution.profile.backend === "acpx_runtime" && parseObject(agent.adapterConfig).acpxAgent === "cursor")) &&
|
||||
(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox"
|
||||
? executionTarget.runnerLifecyclePolicy?.mode === "warm"
|
||||
: parseObject(agent.adapterConfig).lifecycleMode === "warm");
|
||||
if (warmFiles && taskSessionForRun?.lastRunId) {
|
||||
nativeInstructionReservation = await reserveWarmNativeInstructionDirectory({ companyId: agent.companyId, agentId: agent.id,
|
||||
previousRunId: taskSessionForRun.lastRunId, runId: readRun().id, target: executionTarget,
|
||||
canReuse: () => instructionCopies.canReuseWarm(agent.companyId, agent.id, taskSessionForRun!.lastRunId!),
|
||||
});
|
||||
}
|
||||
if (!warmFiles && nativeRuntimeResolution.kind === "native" && priorWorkingCopy.kind === "agent_files" && taskSession) {
|
||||
releaseWarmInstructionPreparation = await claimWarmNativeInstructionCopy({
|
||||
priorExecution: parseNativeExecutionInput(priorFileInput), companyId: agent.companyId, agentId: agent.id,
|
||||
executionWorkspaceId: persistedExecutionWorkspace?.id ?? readRun().id, workspace: executionWorkspace,
|
||||
environmentId: executionTarget?.environmentId ?? null, runId: readRun().id, preparationKey: instructionPreparationKey, forceRetirement: !taskSessionForRun,
|
||||
adopt: async previousRunId => {
|
||||
instructionCopy = await instructionCopies.adopt({ companyId: agent.companyId, agentId: agent.id,
|
||||
runId: readRun().id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd, allowRetirementHandoff: true });
|
||||
return nativeInstructionWorkingCopy() ?? null;
|
||||
},
|
||||
});
|
||||
// A collection-only handoff was retired. Re-enter normal
|
||||
// preparation instead of composing a root already collected.
|
||||
if (!releaseWarmInstructionPreparation) instructionCopy = null;
|
||||
}
|
||||
const prepareInstructions = (reuseRunId?: string) => instructionCopies.prepare({
|
||||
companyId: agent.companyId, agentId: agent.id, runId: readRun().id,
|
||||
target: executionTarget, cwd: executionWorkspace.cwd,
|
||||
legacy: Object.keys(priorFileInput).length > 0 && priorWorkingCopy.kind !== "agent_files",
|
||||
warm: warmFiles, reuseRunId,
|
||||
onWarmHandoff: copy => {
|
||||
instructionCopy = copy;
|
||||
nativeInstructionReservation?.adopt(copy.executionRoot, collectStoppedInstructions);
|
||||
},
|
||||
});
|
||||
try {
|
||||
instructionCopy ??= await prepareInstructions(nativeInstructionReservation?.reuseRunId);
|
||||
} catch (error) {
|
||||
if (!(error instanceof AgentDirectoryReuseInvalidatedError)) throw error;
|
||||
// prepare has released its canonical lock. Retirement can now
|
||||
// collect under that same lock before a fresh restore starts.
|
||||
await nativeInstructionReservation?.release();
|
||||
nativeInstructionReservation = null;
|
||||
instructionCopy = await prepareInstructions();
|
||||
}
|
||||
} catch (error) {
|
||||
if ((error as { status?: number }).status !== 403) throw error;
|
||||
// Missing write identity must not break a background run's read-only
|
||||
// prompt. It must also never imply that ordinary file edits will save.
|
||||
await appendRunEvent(readRun(), { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
message: "Persistent instruction editing is unavailable. Use an authenticated user with instruction edit access and a managed instruction bundle.",
|
||||
payload: { state: "unavailable", code: "INSTRUCTION_COPY_UNAVAILABLE" } });
|
||||
const guidance = "No editable agent instruction working copy is registered for this turn. Use authenticated agent file tools for persistent edits; do not edit a private copy named in an earlier turn or claim its changes will persist.";
|
||||
for (const key of ["paperclipTaskMarkdown", "paperclipTaskMarkdownCompact"]) {
|
||||
context[key] = [readNonEmptyString(context[key]), guidance].filter(Boolean).join("\n\n");
|
||||
}
|
||||
}
|
||||
if (instructionCopy) {
|
||||
const storageWarning = readNonEmptyString(instructionCopy.receipt?.storageWarning);
|
||||
if (storageWarning) {
|
||||
instructionSave = { state: "prepared", contract: "agent_files", storageWarning };
|
||||
// This is an advisory on the run, never an agent pause, execution
|
||||
// failure, or scheduling gate. Keep it visible while work runs.
|
||||
await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ instructionSave })}::jsonb` }).where(eq(heartbeatRuns.id, readRun().id));
|
||||
await appendRunEvent(readRun(), { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
message: storageWarning, payload: instructionSave });
|
||||
}
|
||||
runtimeConfig = { ...runtimeConfig, instructionsFilePath: path.join(instructionCopy.localRoot, instructionCopy.entryFile) };
|
||||
if (isAgentDirectoryCopy(instructionCopy)) {
|
||||
const workspace = parseObject(context.paperclipWorkspace);
|
||||
context.paperclipWorkspace = { ...workspace, agentHome: instructionCopy.executionRoot,
|
||||
// Keep the pre-existing permission root stable for ACP session
|
||||
// identity. The per-run copy is already under the company root.
|
||||
agentHomeForPermissions: workspace.agentHome,
|
||||
};
|
||||
}
|
||||
const guidance = instructionWorkingCopyGuidance(instructionCopy);
|
||||
for (const key of ["paperclipTaskMarkdown", "paperclipTaskMarkdownCompact"]) {
|
||||
context[key] = [readNonEmptyString(context[key]), guidance].filter(Boolean).join("\n\n");
|
||||
}
|
||||
}
|
||||
}
|
||||
return runtimeConfig;
|
||||
}
|
||||
|
||||
return {
|
||||
prepare,
|
||||
collectStoppedInstructions,
|
||||
nativeInstructionWorkingCopy,
|
||||
get instructionCopy() { return instructionCopy; },
|
||||
getInstructionSave: () => instructionSave,
|
||||
// These getters remain usable after partial acquisition failures.
|
||||
get nativeInstructionReservation() { return nativeInstructionReservation; },
|
||||
get releaseWarmInstructionPreparation() { return releaseWarmInstructionPreparation; },
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user