diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 48e7704aaac..441b13685a6 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -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. diff --git a/server/src/__tests__/heartbeat-instruction-working-copy.test.ts b/server/src/__tests__/heartbeat-instruction-working-copy.test.ts new file mode 100644 index 00000000000..8af787a5484 --- /dev/null +++ b/server/src/__tests__/heartbeat-instruction-working-copy.test.ts @@ -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(), revocations: new Set(), attempts: new Map() })); +vi.mock("../services/agent-instruction-working-copies.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, agentInstructionWorkingCopyService: (...args: Parameters) => { + const copies = actual.agentInstructionWorkingCopyService(...args); + return { ...copies, prepare: async (input: Parameters[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>; + let db: ReturnType; + let heartbeat: ReturnType; + const captured = new Map }>(); + const failures = new Set(); + + 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 => { + 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>) { + 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" }); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1f62017a68b..db99e407a38 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -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[0] | null = null; let nativeSessionResumeScheduled = false; let nativeOwnershipHeld = false; - let releaseWarmInstructionPreparation: (() => Promise) | null = null; - let nativeInstructionReservation: Awaited> = null; + let instructionWorkingCopy: ReturnType | null = null; let nativeDispatchStarted = false; let nativeWorkspaceFinalizeScheduled = false; let nativeWorkspaceSync: Awaited< @@ -2740,49 +2710,12 @@ export function heartbeatService( appendRunEvent, }, }); - let instructionCopy: Awaited> = null; - let instructionSave: Record | 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>>) => { - 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", diff --git a/server/src/services/heartbeat/instruction-working-copy.test.ts b/server/src/services/heartbeat/instruction-working-copy.test.ts new file mode 100644 index 00000000000..3ed57790ea7 --- /dev/null +++ b/server/src/services/heartbeat/instruction-working-copy.test.ts @@ -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(), + reserveWarmNativeInstructionDirectory: native.reserve, + claimWarmNativeInstructionCopy: native.claim, +})); + +type Input = Parameters[0]; +type Prepare = Parameters["prepare"]>[0]; +type Copy = NonNullable>>; + +function fixture() { + let run = { id: "current-run", companyId: "company", agentId: "agent", runnerProfileJson: null, status: "queued" } as ReturnType; + 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().mockResolvedValue(copy), + collectStopped: vi.fn().mockResolvedValue({ ...copy, state: "saved" }), + checkpointWarm: vi.fn().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; + 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" }); + }); +}); diff --git a/server/src/services/heartbeat/instruction-working-copy.ts b/server/src/services/heartbeat/instruction-working-copy.ts new file mode 100644 index 00000000000..87b443b92a7 --- /dev/null +++ b/server/src/services/heartbeat/instruction-working-copy.ts @@ -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; +type Selection = HeartbeatRuntimeSelectionInput; +type AppendRunEvent = ReturnType["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, + 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; + 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) | null = null; + let nativeInstructionReservation: Awaited> = null; + let instructionCopy: Awaited> = null; + let instructionSave: Record | 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>>) => { + 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; + 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; }, + }; +} diff --git a/tests/e2e/instruction-working-copy/README.md b/tests/e2e/instruction-working-copy/README.md new file mode 100644 index 00000000000..027f57ddd22 --- /dev/null +++ b/tests/e2e/instruction-working-copy/README.md @@ -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. diff --git a/tests/e2e/instruction-working-copy/launch.ts b/tests/e2e/instruction-working-copy/launch.ts new file mode 100644 index 00000000000..557799edce5 --- /dev/null +++ b/tests/e2e/instruction-working-copy/launch.ts @@ -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), INSTRUCTION_FIXTURE_PATH: input.config.instructionsFilePath }, + } }); + // execute returns only after the real child process exits. + await input.onProviderStopped?.(); + return result; + }, +}); +await testDriveCommand(options); diff --git a/tests/e2e/instruction-working-copy/lifecycle.spec.ts b/tests/e2e/instruction-working-copy/lifecycle.spec.ts new file mode 100644 index 00000000000..94b6d18856d --- /dev/null +++ b/tests/e2e/instruction-working-copy/lifecycle.spec.ts @@ -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 }); } +}); diff --git a/tests/e2e/instruction-working-copy/playwright.config.ts b/tests/e2e/instruction-working-copy/playwright.config.ts new file mode 100644 index 00000000000..d9a3dcc12b2 --- /dev/null +++ b/tests/e2e/instruction-working-copy/playwright.config.ts @@ -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"]], +}); diff --git a/tests/e2e/instruction-working-copy/worker.mjs b/tests/e2e/instruction-working-copy/worker.mjs new file mode 100644 index 00000000000..9bc3ed0e86e --- /dev/null +++ b/tests/e2e/instruction-working-copy/worker.mjs @@ -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());