mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
Merge origin/master instruction working-copy extraction
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
52ea8711ef
10 files changed
+633
-179
No files matched your search
@@ -1284,6 +1284,14 @@ services and primary URLs, and posts workspace-ready notices. Process registries
|
||||
and leases stay in `workspace-runtime.ts`; the executor keeps release ordering in
|
||||
its existing cleanup path. Public-path tests use real local service processes.
|
||||
|
||||
Instruction working-copy coordination is in
|
||||
`server/src/services/heartbeat/instruction-working-copy.ts`. Its per-run controller
|
||||
prepares copies, tracks save receipts, and coordinates warm-directory handoffs.
|
||||
Live getters preserve updated copies and expose acquired reservations even if
|
||||
preparation fails. The executor keeps release and environment-cleanup ordering;
|
||||
a separate release closure defers cleanup failures without replacing run results.
|
||||
See `tests/e2e/instruction-working-copy/README.md` for isolated browser acceptance.
|
||||
|
||||
Task assignment Markdown is rendered by `server/src/services/heartbeat/task-markdown.ts`.
|
||||
`heartbeat.ts` re-exports it for existing callers. Keep prompt
|
||||
formatting changes in the renderer and its tests, separate from run orchestration.
|
||||
|
||||
@@ -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";
|
||||
@@ -217,7 +218,6 @@ import {
|
||||
import {
|
||||
prepareChatCompletionTurn,
|
||||
} from "./chat-completion-delivery.js";
|
||||
import { AgentDirectoryReuseInvalidatedError, isAgentDirectoryCopy } from "./agent-directory-working-copies.js";
|
||||
|
||||
import {
|
||||
NativeWorkspaceFinalizationBusyError,
|
||||
@@ -284,15 +284,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,
|
||||
@@ -337,9 +334,6 @@ import {
|
||||
} from "./issue-queued-comment-queue.js";
|
||||
|
||||
import {
|
||||
claimWarmNativeInstructionCopy,
|
||||
nativeSessionWorkspaceScope,
|
||||
reserveWarmNativeInstructionDirectory,
|
||||
dispatchNativeSessionResumptions,
|
||||
NativeCancellationPendingRecoveryError,
|
||||
NativeControllerDetachedForRestartError,
|
||||
@@ -350,7 +344,6 @@ import {
|
||||
|
||||
import { nativeChatWorkspaceCwd } from "./native-runtime/native-chat-workspace.js";
|
||||
import {
|
||||
parseNativeExecutionInput,
|
||||
type NativeExecutionInput,
|
||||
type NativeSessionBackend,
|
||||
} from "../vendor/paperclip-runner/index.js";
|
||||
@@ -1836,29 +1829,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 &&
|
||||
@@ -1959,8 +1930,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<
|
||||
@@ -2775,49 +2745,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` } }); }
|
||||
@@ -3327,100 +3260,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,
|
||||
@@ -3466,7 +3306,7 @@ export function heartbeatService(
|
||||
runtimeConfig,
|
||||
runtimeSkillEntries,
|
||||
runScopedMentionedSkillKeys,
|
||||
instructionCopy,
|
||||
instructionCopy: preparedInstructions.instructionCopy,
|
||||
agentIdentity,
|
||||
issueAssigneeOverrides,
|
||||
resolvedConfig,
|
||||
@@ -3540,12 +3380,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,
|
||||
@@ -3814,7 +3654,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 {
|
||||
@@ -3902,7 +3742,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; },
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
# Instruction working-copy acceptance
|
||||
|
||||
Run a disposable server with an instruction-capable fixture around the real
|
||||
process adapter. The capability override exists only in this launcher. The child
|
||||
reads and edits its registered private copy and completes a task through the
|
||||
run-scoped API. No provider calls are made.
|
||||
|
||||
```sh
|
||||
node cli/node_modules/tsx/dist/cli.mjs tests/e2e/instruction-working-copy/launch.ts
|
||||
```
|
||||
|
||||
Use its printed ready URL:
|
||||
|
||||
```sh
|
||||
INSTRUCTION_COPY_TEST_URL=http://127.0.0.1:3100 \
|
||||
pnpm exec playwright test --config tests/e2e/instruction-working-copy/playwright.config.ts
|
||||
```
|
||||
|
||||
The browser checks completed tasks across two turns, exact saved bytes through
|
||||
the agent-file API, the next process's restored contents, each run's save receipt,
|
||||
and private-copy cleanup. The fixture refuses non-loopback servers and requires
|
||||
its dedicated company. Screenshots and failure traces are in
|
||||
`test-results/instruction-working-copy`. Stop the launcher with Ctrl-C; its
|
||||
isolated data remains available for inspection.
|
||||
@@ -0,0 +1,23 @@
|
||||
import { prepareTestDriveEnvironment, testDriveCommand } from "../../../cli/src/commands/test-drive.js";
|
||||
|
||||
const options = { harness: "codex", apiKeyEnv: "INSTRUCTION_FIXTURE_KEY", companyName: "Instruction Lifecycle QA", browser: false } as const;
|
||||
process.env.INSTRUCTION_FIXTURE_KEY = "not-a-provider-key";
|
||||
await prepareTestDriveEnvironment(options);
|
||||
process.env.NODE_ENV = "test";
|
||||
// Only this disposable server exposes the process adapter as an instruction-
|
||||
// capable fixture. Production process-adapter capabilities are unchanged.
|
||||
const { registerServerAdapter } = await import("../../../server/src/adapters/index.js");
|
||||
const { processAdapter } = await import("../../../server/src/adapters/process/index.js");
|
||||
registerServerAdapter({
|
||||
...processAdapter,
|
||||
supportsInstructionsBundle: true,
|
||||
execute: async input => {
|
||||
const result = await processAdapter.execute({ ...input, config: { ...input.config,
|
||||
env: { ...(input.config.env as Record<string, unknown>), INSTRUCTION_FIXTURE_PATH: input.config.instructionsFilePath },
|
||||
} });
|
||||
// execute returns only after the real child process exits.
|
||||
await input.onProviderStopped?.();
|
||||
return result;
|
||||
},
|
||||
});
|
||||
await testDriveCommand(options);
|
||||
@@ -0,0 +1,47 @@
|
||||
import { test, expect } from "@playwright/test";
|
||||
import { mkdtemp, rm, stat } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
const base = process.env.INSTRUCTION_COPY_TEST_URL;
|
||||
test.skip(!base, "Launch this suite's disposable instruction-capable test-drive; see README.md");
|
||||
|
||||
test("saves stopped-process edits, shows completed tasks, and restores saved bytes on the next task", async ({ page }, info) => {
|
||||
expect(new URL(base!).hostname).toBe("127.0.0.1");
|
||||
const api = async (route: string, method = "GET", body?: unknown) => {
|
||||
const response = await page.request.fetch(`${base}/api${route}`, { method, data: body });
|
||||
expect(response.ok(), await response.text()).toBe(true);
|
||||
return response.json();
|
||||
};
|
||||
expect((await api("/health")).deploymentMode).toBe("local_trusted");
|
||||
const [company] = await api("/companies");
|
||||
expect(company.name).toBe("Instruction Lifecycle QA");
|
||||
const cwd = await mkdtemp(path.join(os.tmpdir(), "instruction-acceptance-"));
|
||||
try {
|
||||
const agent = await api(`/companies/${company.id}/agents`, "POST", {
|
||||
name: `Instruction worker ${path.basename(cwd)}`, role: "engineer", adapterType: "process",
|
||||
adapterConfig: { command: process.execPath, args: [path.resolve(import.meta.dirname, "worker.mjs")], cwd },
|
||||
instructionsBundle: { entryFile: "AGENTS.md", files: { "AGENTS.md": "# Original\n" } },
|
||||
runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true } },
|
||||
});
|
||||
for (const turn of [0, 1]) {
|
||||
const issue = await api(`/companies/${company.id}/issues`, "POST", { title: `Persist instructions turn ${turn}`, status: "todo", assigneeAgentId: agent.id });
|
||||
await page.goto(`${base}/${company.issuePrefix}/issues/${issue.identifier}`);
|
||||
const runs = async () => (await api(`/companies/${company.id}/heartbeat-runs`)).filter((run: { contextSnapshot: { issueId?: string } }) => run.contextSnapshot?.issueId === issue.id);
|
||||
await expect.poll(async () => (await runs()).some((run: { status: string }) => run.status === "succeeded"), { timeout: 45_000 }).toBe(true);
|
||||
const history = await runs();
|
||||
expect(history).toHaveLength(1);
|
||||
const run = await api(`/heartbeat-runs/${history[0].id}`);
|
||||
expect(run.resultJson.instructionSave).toMatchObject({ state: turn === 0 ? "saved" : "unchanged", contract: "agent_files", entryFile: "AGENTS.md" });
|
||||
const outputLine = run.stdoutExcerpt.split("\n").find((line: string) => line.startsWith('{"before":'));
|
||||
expect(outputLine).toBeTruthy();
|
||||
const output = JSON.parse(outputLine!);
|
||||
expect(output.before).toBe(turn === 0 ? "# Original\n" : "# Saved by the real worker\n☃\n");
|
||||
expect((await api(`/agents/${agent.id}/instructions-bundle/file?path=AGENTS.md`)).content).toBe("# Saved by the real worker\n☃\n");
|
||||
await expect.poll(async () => stat(output.instructionPath).then(() => true, () => false)).toBe(false);
|
||||
await page.reload();
|
||||
await expect(page.getByRole("button", { name: /^Change status \(current: Done/ }).first()).toBeVisible();
|
||||
await page.screenshot({ path: info.outputPath(`completed-turn-${turn}.png`), fullPage: true });
|
||||
}
|
||||
} finally { await rm(cwd, { recursive: true, force: true }); }
|
||||
});
|
||||
@@ -0,0 +1,6 @@
|
||||
import { defineConfig } from "@playwright/test";
|
||||
export default defineConfig({
|
||||
testDir: ".", testMatch: "*.spec.ts", workers: 1, timeout: 90_000,
|
||||
use: { viewport: { width: 1440, height: 1080 }, screenshot: "only-on-failure", trace: "retain-on-failure" },
|
||||
outputDir: "../../../test-results/instruction-working-copy", reporter: [["list"]],
|
||||
});
|
||||
@@ -0,0 +1,14 @@
|
||||
import { readFile, writeFile } from "node:fs/promises";
|
||||
import assert from "node:assert/strict";
|
||||
const env = process.env;
|
||||
assert(env.INSTRUCTION_FIXTURE_PATH, "The adapter must receive the registered copy");
|
||||
const before = await readFile(env.INSTRUCTION_FIXTURE_PATH, "utf8");
|
||||
assert(["# Original\n", "# Saved by the real worker\n☃\n"].includes(before));
|
||||
await writeFile(env.INSTRUCTION_FIXTURE_PATH, "# Saved by the real worker\n☃\n");
|
||||
console.log(JSON.stringify({ before, instructionPath: env.INSTRUCTION_FIXTURE_PATH }));
|
||||
const headers = { Authorization: `Bearer ${env.PAPERCLIP_API_KEY}`, "X-Paperclip-Run-Id": env.PAPERCLIP_RUN_ID, "Content-Type": "application/json" };
|
||||
const runResponse = await fetch(`${env.PAPERCLIP_API_URL}/api/heartbeat-runs/${env.PAPERCLIP_RUN_ID}`, { headers });
|
||||
assert(runResponse.ok);
|
||||
const run = await runResponse.json();
|
||||
const response = await fetch(`${env.PAPERCLIP_API_URL}/api/issues/${run.contextSnapshot.issueId}`, { method: "PATCH", headers, body: JSON.stringify({ status: "done", comment: "Read and updated the registered instruction working copy." }) });
|
||||
assert(response.ok, await response.text());
|
||||
Reference in new issue
Block a user