From cd501499a2fa8fd02b64efca3934f0d72a3087bb Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sat, 15 Aug 2026 21:36:59 -0700 Subject: [PATCH] test: add ACPX run lifecycle characterization baselines (#11461) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The adapter runtime starts, turns, settles, and composes ACPX runs > - Recent lifecycle corrections changed several order and cleanup rules > - Those rules need regression coverage before the planned engine refactor > - This pull request adds characterization suites for the corrected behavior > - The benefit is a clear test baseline for the next refactor ## Linked Issues or Issue Description **What existing behavior does this improve?** The ACPX adapter runtime and server heartbeat lifecycle need stable regression coverage for their current corrected behavior. **Subsystem affected** Cross-cutting (multiple of the above): `packages/adapter-utils` and `server` test suites. **Current behavior** The runtime has corrected rules for startup, turns, settlement, composed results, and heartbeat terminalization. The repository lacks a single characterization baseline for these rules. **Proposed behavior** Keep the current lifecycle rules pinned by five test suites. Let the later engine refactor change behavior only when it updates these tests with a clear reason. **Reason and benefit** The suites expose order, cleanup, transport, timeout, retry, result, and lease-release changes during the refactor. They also record one known latent defect as current behavior. **Breaking changes** None. This pull request adds tests only. ## What Changed - Add startup characterization coverage for commands, launch values, session fingerprints, sync order, bridge overlap, and cleanup paths. - Add turn characterization coverage for inputs, events, transports, timeout and cancel behavior, retry rules, errors, and usage. - Add settlement characterization coverage for teardown, adapter sync-back, workspace restore order, native sync, and error policy. - Add composed-run characterization coverage for result forms, finalization sets, and host-lane warm save and warm hit behavior. - Add server coverage that checks run terminalization before environment lease release. ## Verification - Run `npx vitest run packages/adapter-utils/src/acpx-engine/startup-characterization.test.ts packages/adapter-utils/src/acpx-engine/turn-characterization.test.ts packages/adapter-utils/src/acpx-engine/settlement-characterization.test.ts packages/adapter-utils/src/acpx-engine/composed-run-characterization.test.ts packages/adapter-utils/src/acpx-engine/execute.test.ts`. - Run `npx vitest run server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts`. - The adapter-utils run passes 178 tests, and the server run passes 4 tests. - Check `pnpm --filter @paperclipai/adapter-utils typecheck`. - Check `pnpm --filter @paperclipai/server typecheck`. ## Risks Low risk. The change adds test files and does not change production code. One known cold ensure-session cleanup defect remains pinned as current behavior. ## Model Used OpenAI Codex, GPT-5, with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../composed-run-characterization.test.ts | 877 ++++++++++++++ .../settlement-characterization.test.ts | 1061 +++++++++++++++++ .../startup-characterization.test.ts | 913 ++++++++++++++ .../acpx-engine/turn-characterization.test.ts | 685 +++++++++++ ...eat-run-terminalize-before-release.test.ts | 312 +++++ server/src/services/heartbeat.ts | 4 +- 6 files changed, 3851 insertions(+), 1 deletion(-) create mode 100644 packages/adapter-utils/src/acpx-engine/composed-run-characterization.test.ts create mode 100644 packages/adapter-utils/src/acpx-engine/settlement-characterization.test.ts create mode 100644 packages/adapter-utils/src/acpx-engine/startup-characterization.test.ts create mode 100644 packages/adapter-utils/src/acpx-engine/turn-characterization.test.ts create mode 100644 server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts diff --git a/packages/adapter-utils/src/acpx-engine/composed-run-characterization.test.ts b/packages/adapter-utils/src/acpx-engine/composed-run-characterization.test.ts new file mode 100644 index 0000000000..b9723bebc5 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/composed-run-characterization.test.ts @@ -0,0 +1,877 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { AcpRuntimeOptions } from "acpx/runtime"; +import type { AdapterExecutionContext, AdapterRuntimeMcpAccess } from "@paperclipai/adapter-utils"; +import { + prepareAdapterExecutionTargetRuntime, + startAdapterExecutionTargetPaperclipBridge, + startAdapterExecutionTargetProcessSessionBridge, +} from "@paperclipai/adapter-utils/execution-target"; + +// This file is a characterization test. It pins the engine boundary's CURRENT +// behavior; it never changes production code. Each test states the observed +// contract of `executeAcpxEngine` as data, so a later refactor that alters an +// exit path's result form, its finalization set, or its warm-lane decision +// fails here. +// +// Adapter-boundary automatic CLI fallback note (File 1b): the codex, claude, and +// gemini adapters wrap this engine and fall back to their CLI lane when an +// AUTOMATIC ACP selection fails, but RETHROW when ACP was EXPLICITLY selected. +// Those three adapters live in higher packages that depend on this one, so an +// adapter-utils test cannot import them (a reverse dependency does not exist in +// package.json). That branch already has REAL coverage in each adapter package: +// - packages/adapters/codex-local/src/server/execute.acp-fallback.test.ts +// - packages/adapters/claude-local/src/server/execute.acp-fallback.test.ts +// - packages/adapters/gemini-local/src/server/execute.acp-fallback.test.ts +// and each adapter's `normalizeEngine` / `resolveXxxExecutionEngineForRun` +// classification (explicit acp/cli vs automatic acp) is pinned in each +// package's acp.test.ts. This file pins the engine-level exit paths those +// adapters wrap. + +// Wrap the staging seam + both sandbox bridges in call-recording spies that +// still delegate to the real implementations (a runner-backed sandbox test +// exercises them end-to-end against a local runner). This lets the staging +// tests assert the exact `runtimeRootDir`/`workspaceLocalDir`/`assets` the +// engine threads without changing any real behavior for the other tests. +vi.mock("@paperclipai/adapter-utils/execution-target", async (importActual) => { + const actual = await importActual(); + return { + ...actual, + prepareAdapterExecutionTargetRuntime: vi.fn(actual.prepareAdapterExecutionTargetRuntime), + startAdapterExecutionTargetPaperclipBridge: vi.fn(actual.startAdapterExecutionTargetPaperclipBridge), + startAdapterExecutionTargetProcessSessionBridge: vi.fn(actual.startAdapterExecutionTargetProcessSessionBridge), + }; +}); +import { + createAcpxEngineExecutor, + type AcpxEngineExecutorOptions, +} from "./execute.js"; +import { runChildProcess } from "../server-utils.js"; + +const tempRoots: string[] = []; + +async function makeTempRoot() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-acpx-skills-")); + tempRoots.push(root); + return root; +} + +afterEach(async () => { + // A remote run stages a process-session bridge whose detached event writer can + // still be flushing a trailing event file into `.../process-sessions//events` + // when the run's own best-effort `client.remove(sessionDir)` (which production + // catch-wraps) has already returned. Under CI load that trailing write can land + // between this recursive delete's directory snapshot and its `rmdir`, surfacing as + // `ENOTEMPTY`. `maxRetries`/`retryDelay` make the cleanup ride out that window the + // same way production tolerates it, instead of failing the just-passed test. + await Promise.all( + tempRoots.splice(0).map((root) => + fs.rm(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }), + ), + ); +}); + +function createLocalSandboxRunner( + onExecute?: (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + }) => void, +) { + let counter = 0; + return { + execute: async (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + stdin?: string; + timeoutMs?: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }) => { + counter += 1; + onExecute?.(input); + const command = input.command === "bash" ? "/bin/bash" : input.command; + return await runChildProcess(`acpx-sandbox-run-${counter}`, command, input.args ?? [], { + cwd: input.cwd ?? process.cwd(), + env: input.env ?? {}, + stdin: input.stdin, + timeoutSec: Math.max(1, Math.ceil((input.timeoutMs ?? 30_000) / 1000)), + graceSec: 5, + onLog: input.onLog ?? (async () => {}), + onSpawn: input.onSpawn + ? async (meta) => input.onSpawn?.({ pid: meta.pid, startedAt: meta.startedAt }) + : undefined, + }); + }, + }; +} + +function buildRuntime( + onSetConfigOption?: (input: { key: string; value: string }) => void, + onEnsureSession?: (input: Record) => void, +) { + return { + ensureSession: async (input: Record) => { + onEnsureSession?.(input); + return ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }); + }, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + setConfigOption: async (input: { key: string; value: string }) => { + onSetConfigOption?.(input); + }, + close: async () => {}, + }; +} + +async function runExecutor( + config: Record, + options: { + context?: Record; + executionTransport?: Record; + authToken?: string; + executionTarget?: Record; + runtimeMcp?: AdapterRuntimeMcpAccess; + prepareRemoteManagedHome?: AcpxEngineExecutorOptions["prepareRemoteManagedHome"]; + startupTraceContext?: AdapterExecutionContext["startupTraceContext"]; + } = {}, +) { + const runtimeOptions: Record[] = []; + const configOptions: Array<{ key: string; value: string }> = []; + const sessionInputs: Record[] = []; + const meta: Record[] = []; + const logs: Array<{ stream: string; text: string }> = []; + const events: Array<{ eventType: string; payload?: Record }> = []; + const execute = createAcpxEngineExecutor({ + ...(options.prepareRemoteManagedHome + ? { prepareRemoteManagedHome: options.prepareRemoteManagedHome } + : {}), + createRuntime: (options) => { + runtimeOptions.push(options as unknown as Record); + return buildRuntime( + ({ key, value }) => configOptions.push({ key, value }), + (input) => sessionInputs.push(input), + ) as never; + }, + }); + + const result = await execute({ + runId: "run-1", + agent: { + id: "agent-1", + companyId: "company-1", + }, + runtime: {}, + config, + context: options.context ?? {}, + executionTransport: options.executionTransport, + authToken: options.authToken, + executionTarget: options.executionTarget, + runtimeMcp: options.runtimeMcp, + startupTraceContext: options.startupTraceContext, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async (payload: unknown) => { + meta.push(payload as Record); + }, + onEvent: async (event: { eventType: string; payload?: Record }) => { + events.push(event); + }, + } as never); + + expect(result.exitCode).toBe(0); + return { logs, meta, events, runtimeOptions, configOptions, sessionInputs, result }; +} + +// A remote sandbox setup that stages the host worktree through the real local +// runner, so a run reaches the post-build window with live bridges and a held +// staging lease. +async function setupRemoteSandbox() { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + }; + return { root, stateDir, localCwd, remoteCwd, executionTarget }; +} + +const okHandle = { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", +}; + +function completedTurn() { + return { + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; +} + +function throwingTurn() { + return { + events: (async function* () { + throw new Error("turn upstream boom"); + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; +} + +// A context whose only prompt-read field throws when read, so the prompt build +// fails after the session handshake succeeds and before the turn starts. +function throwingHandoffContext(): Record { + const context: Record = {}; + Object.defineProperty(context, "paperclipSessionHandoffMarkdown", { + enumerable: false, + get() { + throw new Error("prompt build boom"); + }, + }); + return context; +} + +// Stub both sandbox bridges with stop spies collected per start, so a test can +// assert the bridges stopped without running the real bridge transport. +function stubBridges() { + const paperclipStops: Array> = []; + const processStops: Array> = []; + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementation(async () => { + const stop = vi.fn(async () => {}); + paperclipStops.push(stop); + return { env: {}, stop } as never; + }); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementation(async () => { + const stop = vi.fn(async () => {}); + processStops.push(stop); + return { agentCommand: null, stop } as never; + }); + const anyStopped = (stops: Array>) => + stops.some((stop) => stop.mock.calls.length > 0); + const stoppedCount = (stops: Array>) => + stops.filter((stop) => stop.mock.calls.length > 0).length; + return { paperclipStops, processStops, anyStopped, stoppedCount }; +} + +function remoteArgs( + stateDir: string, + localCwd: string, + executionTarget: unknown, + overrides: Record = {}, +) { + return { + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + ...overrides, + }; +} + +describe("composed ACPX run: engine-boundary result form per exit path", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("returns (never throws) a create_runtime error result when the post-build runtime construction fails", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + }); + + // The engine RETURNS a settled error result on this path; it does not throw. + const result = await execute({ + runId: "boundary-create-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.signal).toBe(null); + expect(result.timedOut).toBe(false); + expect(result.resultJson?.phase).toBe("create_runtime"); + expect(result.summary).toContain("createRuntime boom"); + }); + + it("returns an ensure_session error result when the handshake fails", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => { + throw new Error("ensureSession boom"); + }, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + }); + + const result = await execute({ + runId: "boundary-ensure-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("ensure_session"); + }); + + it("returns a runtime-error result with no clearSession and no errorMeta when ensureSession yields no handle", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + // A runtime that returns no session handle drives the missing-handle path. + ensureSession: async () => undefined, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + }); + + const result = await execute({ + runId: "boundary-missing-handle", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("ensure_session"); + expect(result.errorCode).toBe("acpx_runtime_error"); + // The missing-handle result carries neither clearSession nor errorMeta, unlike + // the other pre-turn error paths. + expect(result.clearSession).toBeUndefined(); + expect(result.errorMeta).toBeUndefined(); + }); + + it("returns a configure_session error result that carries the requested session fields", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + // A custom agent with a requested model applies a session config option; + // a throwing setConfigOption drives the configure_session failure path. + setConfigOption: async () => { + throw new Error("setConfigOption boom"); + }, + close: async () => {}, + }) as never, + }); + + const result = await execute({ + runId: "boundary-configure-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { + agent: "custom", + agentCommand: "node ./fake-acp.js", + model: "custom-model-x", + stateDir: path.join(root, "state"), + }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("configure_session"); + // The configure_session result echoes the requested session identity fields. + expect(result.resultJson).toHaveProperty("agent"); + expect(result.resultJson?.requestedModel).toBe("custom-model-x"); + expect(result.resultJson).toHaveProperty("requestedThinkingEffort"); + expect(result.resultJson).toHaveProperty("fastMode"); + }); + + it("returns a prepare_turn error result when the prompt build fails before the turn starts", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + }); + + const result = await execute({ + runId: "boundary-prepare-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: throwingHandoffContext(), + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("prepare_turn"); + }); + + it("returns a turn error result with acpx_turn_failed when the running turn fails", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => {}, + }) as never, + }); + + const result = await execute({ + runId: "boundary-turn-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("turn"); + expect(result.errorCode).toBe("acpx_turn_failed"); + }); + + it("returns the terminal turn-success shape (the one non-error-shaped return)", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => buildRuntime() as never, + }); + + const result = await execute({ + runId: "boundary-success", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + // A completed, non-timed-out turn returns exitCode 0, no signal, no errorCode, + // and the session identity fields, with resultJson.status = "completed". + expect(result.exitCode).toBe(0); + expect(result.signal).toBe(null); + expect(result.timedOut).toBe(false); + expect(result.errorCode).toBe(null); + expect(result.resultJson?.status).toBe("completed"); + expect(result.sessionId).toBe("backend-session"); + expect(result.sessionDisplayId).toBe("agent-session"); + expect(result.sessionParams).toBeTruthy(); + }); + + it("THROWS (does not return) when buildRuntime fails before it settles", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + // A staging failure throws inside buildRuntime before it settles, so the engine + // rethrows through the `if (!buildRuntimeSettled) throw err` guard. This is one + // of exactly two engine-boundary throw paths. + vi.mocked(prepareAdapterExecutionTargetRuntime).mockImplementationOnce(async () => { + throw new Error("staging boom"); + }); + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => buildRuntime() as never, + }); + + await expect( + execute({ + runId: "boundary-build-throw", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never), + ).rejects.toThrow("staging boom"); + }); + + it("THROWS (does not return) on a partial bridge failure and stops the started sibling bridge once", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + // The paperclip bridge fails while the concurrently-started process-session + // bridge resolves a live handle. This is the second engine-boundary throw path; + // the abandon path must stop the started sibling so no bridge leaks. + const stop = vi.fn(async () => {}); + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(async () => { + throw new Error("paperclip bridge boom"); + }); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementationOnce( + async () => ({ agentCommand: null, stop }) as never, + ); + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => buildRuntime() as never, + }); + + await expect( + execute({ + runId: "boundary-bridge-throw", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never), + ).rejects.toThrow("paperclip bridge boom"); + expect(stop).toHaveBeenCalledTimes(1); + }); +}); + +describe("composed ACPX run: finalization set fires exactly once per exit path", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("stops both bridges once and releases the staging lease on every exit path", async () => { + const scenarios: Array<{ + name: string; + createRuntime: AcpxEngineExecutorOptions["createRuntime"]; + config?: Record; + context?: Record; + }> = [ + { + name: "create_runtime", + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + }, + { + name: "ensure_session", + createRuntime: () => + ({ + ensureSession: async () => { + throw new Error("ensure boom"); + }, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + }, + { + name: "configure_session", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + setConfigOption: async () => { + throw new Error("config boom"); + }, + close: async () => {}, + }) as never, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", model: "custom-model-x" }, + }, + { + name: "prepare_turn", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + context: throwingHandoffContext(), + }, + { + name: "turn", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => {}, + }) as never, + }, + { + name: "success", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + }, + ]; + + for (const scenario of scenarios) { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, processStops, stoppedCount } = stubBridges(); + const stagingLocks = new Map>(); + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: scenario.createRuntime, + }); + + const overrides: Record = {}; + if (scenario.config) { + overrides.config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + cwd: localCwd, + ...scenario.config, + }; + } + if (scenario.context) overrides.context = scenario.context; + + await execute({ + runId: `finalize-${scenario.name}`, + ...remoteArgs(stateDir, localCwd, executionTarget, overrides), + } as never).catch(() => {}); + + // The finalization set fires exactly once: each bridge stops once and the + // per-session staging lease releases, so the lock map never strands the next + // same-session run. + expect(stoppedCount(paperclipStops), `paperclip bridge must stop once on ${scenario.name}`).toBe(1); + expect(stoppedCount(processStops), `process-session bridge must stop once on ${scenario.name}`).toBe(1); + expect(stagingLocks.size, `lease must release on ${scenario.name}`).toBe(0); + } + }); + + it("does not re-run the turn teardown when the result mapping throws after the close", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, processStops, stoppedCount } = stubBridges(); + let closeCount = 0; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + // A completed turn whose result mapping throws after the close is read. + result: Promise.resolve({ + status: "completed", + get stopReason(): string { + throw new Error("mapping boom"); + }, + }), + cancel: async () => {}, + }), + close: async () => { + closeCount += 1; + }, + }) as never, + }); + + await execute({ + runId: "finalize-map-throw", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never).catch(() => {}); + + // The completed turn closed the runtime once; the mapping throw did not re-run + // the teardown through the turn catch. + expect(closeCount).toBe(1); + expect(stoppedCount(paperclipStops)).toBe(1); + expect(stoppedCount(processStops)).toBe(1); + }); +}); + +describe("composed ACPX run: host-lane warm handle set", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("warm-saves a persistent local runtime and reuses it for the next compatible run", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let createCount = 0; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: () => { + createCount += 1; + return buildRuntime() as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + + const first = await execute({ + runId: "warm-save-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + // A persistent local completed turn warm-saves the runtime handle. + expect(warmHandles.size).toBe(1); + expect(createCount).toBe(1); + + const second = await execute({ + runId: "warm-save-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + expect(second.exitCode).toBe(0); + // The compatible second run reuses the warm runtime, so createRuntime is not + // called again. + expect(createCount).toBe(1); + }); + + it("closes (does not warm-save) a completed non-persistent runtime, so the next run re-creates", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let createCount = 0; + let closeCount = 0; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: () => { + createCount += 1; + return { + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + setConfigOption: async () => {}, + close: async () => { + closeCount += 1; + }, + } as never; + }, + }); + // No persistent mode, so the completed turn closes the runtime instead of + // warm-saving it. + const config = { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }; + + const first = await execute({ + runId: "warm-none-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + expect(warmHandles.size).toBe(0); + expect(closeCount).toBe(1); + + const second = await execute({ + runId: "warm-none-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + expect(second.exitCode).toBe(0); + // No warm handle survived, so the second run constructs a fresh runtime. + expect(createCount).toBe(2); + }); + + it("emits a skipped acp.handshake event on a warm-handle hit and does not re-create the runtime", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let createCount = 0; + const warmHandles = new Map(); + const secondEvents: Array<{ eventType: string; payload?: Record }> = []; + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: () => { + createCount += 1; + return buildRuntime() as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + const first = await execute({ + runId: "warm-hit-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + + await execute({ + runId: "warm-hit-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: (first as { sessionParams?: unknown }).sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async (event: { eventType: string; payload?: Record }) => { + secondEvents.push(event); + }, + } as never); + + // The warm-hit skips the handshake work: it emits exactly one acp.handshake + // event with outcome = skipped and a zero wall time, and never re-creates the + // runtime. + const handshakeEvents = secondEvents.filter( + (event) => event.eventType === "run.startup.step" && event.payload?.step === "acp.handshake", + ); + expect(handshakeEvents).toHaveLength(1); + expect(handshakeEvents[0]!.payload?.outcome).toBe("skipped"); + expect(handshakeEvents[0]!.payload?.durationMs).toBe(0); + expect(createCount).toBe(1); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/settlement-characterization.test.ts b/packages/adapter-utils/src/acpx-engine/settlement-characterization.test.ts new file mode 100644 index 0000000000..d5f9754a88 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/settlement-characterization.test.ts @@ -0,0 +1,1061 @@ +// Characterization baselines for ACP run settlement/teardown. These tests PIN +// the current behavior of the engine's teardown orchestration (Layer A), the +// `restoreWorkspace` internal order plus native-sync selection (Layer B), and +// the per-adapter sync-back registration seam (Layer C). They never change +// production code; each expectation records what the code does today. +// +// The harness top (imports, the execution-target mock, `makeTempRoot`, the local +// sandbox runner, `buildRuntime`, `runExecutor`) is copied from +// `execute.test.ts` (lines 1-336) so this file drives the same engine the same +// way. The teardown helpers (`stubBridges`, `throwingHandoffContext`, +// `completedTurn`, `throwingTurn`, `setupRemoteSandbox`, `remoteArgs`) mirror the +// F2/F3 describes in `execute.test.ts` (~:2602, ~:4858). +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { execFile as execFileCallback } from "node:child_process"; +import { promisify } from "node:util"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { AdapterExecutionContext, AdapterRuntimeMcpAccess } from "@paperclipai/adapter-utils"; +import { + startAdapterExecutionTargetPaperclipBridge, + startAdapterExecutionTargetProcessSessionBridge, +} from "@paperclipai/adapter-utils/execution-target"; + +// Wrap the staging seam + both sandbox bridges in call-recording spies that +// still delegate to the real implementations. A runner-backed sandbox test +// exercises them end-to-end against a local runner, while a teardown test can +// override just the bridges with stop spies. (Copied from execute.test.ts.) +vi.mock("@paperclipai/adapter-utils/execution-target", async (importActual) => { + const actual = await importActual(); + return { + ...actual, + prepareAdapterExecutionTargetRuntime: vi.fn(actual.prepareAdapterExecutionTargetRuntime), + startAdapterExecutionTargetPaperclipBridge: vi.fn(actual.startAdapterExecutionTargetPaperclipBridge), + startAdapterExecutionTargetProcessSessionBridge: vi.fn(actual.startAdapterExecutionTargetProcessSessionBridge), + }; +}); +import { + createAcpxEngineExecutor, + type AcpxEngineExecutorOptions, +} from "./execute.js"; +import { runChildProcess } from "../server-utils.js"; +import { + mirrorDirectory, + prepareSandboxManagedRuntime, + type SandboxManagedRuntimeClient, + type SandboxSyncOperation, + type SandboxSyncResult, +} from "../sandbox-managed-runtime.js"; + +const execFile = promisify(execFileCallback); +const tempRoots: string[] = []; + +async function makeTempRoot() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-acpx-skills-")); + tempRoots.push(root); + return root; +} + +afterEach(async () => { + await Promise.all( + tempRoots.splice(0).map((root) => + fs.rm(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }), + ), + ); +}); + +function createLocalSandboxRunner( + onExecute?: (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + }) => void, +) { + let counter = 0; + return { + execute: async (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + stdin?: string; + timeoutMs?: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }) => { + counter += 1; + onExecute?.(input); + const command = input.command === "bash" ? "/bin/bash" : input.command; + return await runChildProcess(`acpx-sandbox-run-${counter}`, command, input.args ?? [], { + cwd: input.cwd ?? process.cwd(), + env: input.env ?? {}, + stdin: input.stdin, + timeoutSec: Math.max(1, Math.ceil((input.timeoutMs ?? 30_000) / 1000)), + graceSec: 5, + onLog: input.onLog ?? (async () => {}), + onSpawn: input.onSpawn + ? async (meta) => input.onSpawn?.({ pid: meta.pid, startedAt: meta.startedAt }) + : undefined, + }); + }, + }; +} + +function buildRuntime( + onSetConfigOption?: (input: { key: string; value: string }) => void, + onEnsureSession?: (input: Record) => void, +) { + return { + ensureSession: async (input: Record) => { + onEnsureSession?.(input); + return { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + }, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + setConfigOption: async (input: { key: string; value: string }) => { + onSetConfigOption?.(input); + }, + close: async () => {}, + }; +} + +async function runExecutor( + config: Record, + options: { + context?: Record; + executionTransport?: Record; + authToken?: string; + executionTarget?: Record; + runtimeMcp?: AdapterRuntimeMcpAccess; + prepareRemoteManagedHome?: AcpxEngineExecutorOptions["prepareRemoteManagedHome"]; + startupTraceContext?: AdapterExecutionContext["startupTraceContext"]; + } = {}, +) { + const runtimeOptions: Record[] = []; + const configOptions: Array<{ key: string; value: string }> = []; + const sessionInputs: Record[] = []; + const meta: Record[] = []; + const logs: Array<{ stream: string; text: string }> = []; + const events: Array<{ eventType: string; payload?: Record }> = []; + const execute = createAcpxEngineExecutor({ + ...(options.prepareRemoteManagedHome + ? { prepareRemoteManagedHome: options.prepareRemoteManagedHome } + : {}), + createRuntime: (createOptions) => { + runtimeOptions.push(createOptions as unknown as Record); + return buildRuntime( + ({ key, value }) => configOptions.push({ key, value }), + (input) => sessionInputs.push(input), + ) as never; + }, + }); + + const result = await execute({ + runId: "run-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: options.context ?? {}, + executionTransport: options.executionTransport, + authToken: options.authToken, + executionTarget: options.executionTarget, + runtimeMcp: options.runtimeMcp, + startupTraceContext: options.startupTraceContext, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async (payload: unknown) => { + meta.push(payload as Record); + }, + onEvent: async (event: { eventType: string; payload?: Record }) => { + events.push(event); + }, + } as never); + + expect(result.exitCode).toBe(0); + return { logs, meta, events, runtimeOptions, configOptions, sessionInputs, result }; +} + +// --------------------------------------------------------------------------- +// Shared teardown-test helpers (mirror execute.test.ts F3 describe, ~:4858). +// --------------------------------------------------------------------------- + +const okHandle = { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", +}; + +async function setupRemoteSandbox() { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + }; + return { root, stateDir, localCwd, remoteCwd, executionTarget }; +} + +// Stub both sandbox bridges with stop spies collected per start, so a test can +// assert the bridges stopped without running the real bridge transport. +function stubBridges() { + const paperclipStops: Array> = []; + const processStops: Array> = []; + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementation(async () => { + const stop = vi.fn(async () => {}); + paperclipStops.push(stop); + return { env: {}, stop } as never; + }); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementation(async () => { + const stop = vi.fn(async () => {}); + processStops.push(stop); + return { agentCommand: null, stop } as never; + }); + const anyStopped = (stops: Array>) => + stops.some((stop) => stop.mock.calls.length > 0); + return { paperclipStops, processStops, anyStopped }; +} + +function throwingHandoffContext(): Record { + const context: Record = {}; + Object.defineProperty(context, "paperclipSessionHandoffMarkdown", { + enumerable: false, + get() { + throw new Error("prompt build boom"); + }, + }); + return context; +} + +function completedTurn() { + return { + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; +} + +function throwingTurn() { + return { + events: (async function* () { + throw new Error("turn upstream boom"); + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; +} + +function remoteArgs( + stateDir: string, + localCwd: string, + executionTarget: unknown, + overrides: Record = {}, +) { + return { + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + ...overrides, + }; +} + +// =========================================================================== +// Layer A — engine teardown orchestration. +// =========================================================================== +describe("ACP settlement — Layer A: engine teardown orchestration", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("test_clean_completed_remote_teardown_runs_bridge_stop_then_sync_back_then_lease_release", async () => { + // cleanupRemoteBridges (execute.ts:2306-2326) fixes the sub-order for a clean + // exit: stop both bridges (allSettled) → run the managed-home sync-back + // (remoteManagedHomeTeardown) → release the per-session staging lease LAST in + // a finally. Record the order through spies threaded into the stubbed bridges + // and the seam teardown, and read the still-held lease during the sync-back. + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const order: string[] = []; + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementation(async () => ({ + env: {}, + stop: vi.fn(async () => { + order.push("bridge-stop"); + }), + }) as never); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementation(async () => ({ + agentCommand: null, + stop: vi.fn(async () => { + order.push("bridge-stop"); + }), + }) as never); + + const stagingLocks = new Map>(); + let leaseSizeDuringSyncBack = -1; + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: vi.fn(async () => {}), + }) as never, + prepareRemoteManagedHome: async (input) => { + const stagedRuntime = await input.stage([]); + return { + stagedRuntime, + teardown: async () => { + order.push("sync-back"); + // The lease is still held while the sync-back runs; it releases only + // afterward, in the cleanupRemoteBridges finally. + leaseSizeDuringSyncBack = stagingLocks.size; + }, + }; + }, + }); + + const result = await execute({ + runId: "clean-order", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never); + + expect(result.exitCode).toBe(0); + // Both bridge stops precede the sync-back; the sync-back is the last recorded + // step before the lease release. + expect(order).toEqual(["bridge-stop", "bridge-stop", "sync-back"]); + // The lease was still held during the sync-back... + expect(leaseSizeDuringSyncBack).toBe(1); + // ...and released last, so the lock map never strands the next run. + expect(stagingLocks.size).toBe(0); + }); + + it("test_clean_persistent_local_run_warm_saves_instead_of_closing", async () => { + // execute.ts:3886 keeps a clean persistent local turn warm (no runtime.close) + // when warmIdleMs>0 and there is no process-session bridge. + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + setConfigOption: async () => {}, + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: "warm-save", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + mode: "persistent", + // A long idle window so the scheduled cleanup timer cannot fire during the + // assertions and evict the warm handle. The warm-save decision only needs + // warmIdleMs>0. The test clears the timer below, so it never outlives the + // test. + warmHandleIdleMs: 60_000, + }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + } as never); + + expect(result.exitCode).toBe(0); + // The runtime stayed warm: no close, one warm entry. + expect(closeSpy).not.toHaveBeenCalled(); + expect(warmHandles.size).toBe(1); + + // Clear the scheduled idle-cleanup timer so no timer outlives the test. The + // long idle window kept the timer from racing the assertions above; clearing + // it here makes cleanup deterministic instead of a timed drain. + for (const entry of warmHandles.values()) { + if (entry.cleanupTimer) clearTimeout(entry.cleanupTimer); + } + }); + + it("test_completed_non_persistent_local_run_closes_runtime_once", async () => { + // The default (non-persistent) completed turn closes the runtime with the + // clean-completion reason (execute.ts:3922-3926). + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: "clean-close", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(0); + expect(closeSpy).toHaveBeenCalledTimes(1); + expect((closeSpy.mock.calls[0]! as unknown[])[0]).toMatchObject({ + reason: "paperclip completed turn cleanup", + discardPersistentState: false, + }); + }); + + it("test_pre_turn_failure_after_handle_closes_runtime_except_create_runtime", async () => { + // Once a runtime handle exists, every pre-turn failure closes the runtime — + // missing handle (execute.ts:3633, synthesized handle), configure_session + // (:3716), prepare_turn (:4016) — EXCEPT create_runtime, where no runtime was + // ever constructed (:3662-3697). Each path returns a settled error with its + // phase. (The handshake close path needs a handle obtained first via the warm + // hit — pinned separately below, mirroring the F2 test.) + const scenarios: Array<{ + name: string; + phase: string; + expectClose: boolean; + // Builds the engine `createRuntime` given the shared close spy. The + // create_runtime scenario ignores the spy and throws before a runtime exists. + makeCreateRuntime: (close: () => Promise) => AcpxEngineExecutorOptions["createRuntime"]; + config: Record; + context: Record; + }> = [ + { + name: "missing_handle", + phase: "ensure_session", + expectClose: true, + makeCreateRuntime: (close) => () => + ({ + ensureSession: async () => undefined, + startTurn: () => completedTurn(), + close, + }) as never, + config: {}, + context: {}, + }, + { + name: "configure_session", + phase: "configure_session", + expectClose: true, + makeCreateRuntime: (close) => () => + ({ + ensureSession: async () => okHandle, + // A configured model on a non-claude/non-codex agent yields one session + // config option; a throwing setConfigOption drives the configure failure. + setConfigOption: async () => { + throw new Error("config boom"); + }, + startTurn: () => completedTurn(), + close, + }) as never, + config: { model: "custom-model-x" }, + context: {}, + }, + { + name: "prepare_turn", + phase: "prepare_turn", + expectClose: true, + makeCreateRuntime: (close) => () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close, + }) as never, + config: {}, + context: throwingHandoffContext(), + }, + { + name: "create_runtime", + phase: "create_runtime", + expectClose: false, + makeCreateRuntime: () => () => { + throw new Error("createRuntime boom"); + }, + config: {}, + context: {}, + }, + ]; + + for (const scenario of scenarios) { + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: scenario.makeCreateRuntime(closeSpy), + }); + + const result = await execute({ + runId: `preturn-${scenario.name}`, + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + ...scenario.config, + }, + context: scenario.context, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode, `${scenario.name} exit`).toBe(1); + expect(result.resultJson?.phase, `${scenario.name} phase`).toBe(scenario.phase); + if (scenario.expectClose) { + expect(closeSpy, `${scenario.name} closes the runtime`).toHaveBeenCalledTimes(1); + } else { + // create_runtime failed before a runtime existed, so there is nothing to close. + expect(closeSpy, `${scenario.name} never constructs a runtime`).not.toHaveBeenCalled(); + } + } + }); + + it("test_handshake_failure_after_warm_hit_closes_runtime_and_removes_warm_entry", async () => { + // Mirror execute.test.ts F2 test_handshake_failure_closes_runtime_and_removes_warm_entry + // (:4666). The first run warms a handle; the second run reuses it and fails + // while persisting process identity (onSpawn throws). Because the warm hit + // already holds the handle, the handshake failure closes the reused runtime + // (execute.ts:3591) and removes the warm entry. + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const startedAt = "2026-01-01T00:00:00.000Z"; + const closeSpy = vi.fn(async () => {}); + let created = 0; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: (options) => { + created += 1; + const opts = options as { + onAgentSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }; + return { + ensureSession: async () => { + await opts.onAgentSpawn?.({ pid: 4242, startedAt }); + return okHandle; + }, + startTurn: () => completedTurn(), + close: closeSpy, + } as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + + const first = await execute({ + runId: "warm-handshake-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + expect(warmHandles.size).toBe(1); + expect(created).toBe(1); + + const second = await execute({ + runId: "warm-handshake-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => { + throw new Error("onSpawn boom"); + }, + } as never); + + // The reused runtime (no new create) is closed and the warm entry removed. + expect(created).toBe(1); + expect(second.exitCode).toBe(1); + expect(second.resultJson?.phase).toBe("ensure_session"); + expect(closeSpy).toHaveBeenCalledTimes(1); + expect(warmHandles.size).toBe(0); + }); + + it("test_terminal_close_reason_and_discard_persistent_state_per_outcome", async () => { + // The terminal block (execute.ts:3869-3928) closes with an outcome-specific + // reason and discards persistent state only on cancel/timeout. + const cases: Array<{ + status: "completed" | "failed" | "cancelled"; + reason: string; + discard: boolean; + exitCode: number; + }> = [ + { status: "completed", reason: "paperclip completed turn cleanup", discard: false, exitCode: 0 }, + { status: "failed", reason: "paperclip turn failed", discard: false, exitCode: 1 }, + { status: "cancelled", reason: "paperclip turn cancelled", discard: true, exitCode: 1 }, + ]; + + for (const testCase of cases) { + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: + testCase.status === "failed" + ? Promise.resolve({ status: "failed", error: new Error("turn boom") }) + : testCase.status === "cancelled" + ? Promise.resolve({ status: "cancelled", stopReason: "cancelled" }) + : Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: `terminal-${testCase.status}`, + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode, `${testCase.status} exit`).toBe(testCase.exitCode); + expect(closeSpy, `${testCase.status} closes once`).toHaveBeenCalledTimes(1); + expect((closeSpy.mock.calls[0]! as unknown[])[0], `${testCase.status} close args`).toMatchObject({ + reason: testCase.reason, + discardPersistentState: testCase.discard, + }); + } + }); + + it("test_teardown_continues_after_one_step_fails_and_records_it", async () => { + // Corrected F3 policy (execute.ts:3374-3393): a failing teardown step is + // recorded and swallowed; later steps still run and the lease still releases. + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, processStops, anyStopped } = stubBridges(); + const stagingLocks = new Map>(); + const logs: Array<{ stream: string; text: string }> = []; + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => { + throw new Error("close boom"); + }, + }) as never, + }); + + const result = await execute({ + runId: "td-continue", + ...remoteArgs(stateDir, localCwd, executionTarget, { + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + }), + } as never); + + expect(result.exitCode).toBe(1); + expect(anyStopped(paperclipStops)).toBe(true); + expect(anyStopped(processStops)).toBe(true); + expect(stagingLocks.size).toBe(0); + expect( + logs.some( + (entry) => + entry.stream === "stderr" && entry.text.includes('teardown step "runtime-close" failed'), + ), + ).toBe(true); + }); + + it("test_teardown_failure_does_not_change_external_result", async () => { + // A teardown fault never leaks into the external result; the exit cause (the + // turn failure) stands (execute.ts F3 policy). + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + stubBridges(); + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => { + throw new Error("close boom"); + }, + }) as never, + }); + + const result = await execute({ + runId: "td-result", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("turn"); + expect(result.errorCode).toBe("acpx_turn_failed"); + expect(result.errorMessage).toContain("turn upstream boom"); + expect(result.errorMessage).not.toContain("close boom"); + }); +}); + +// =========================================================================== +// Layer B — restoreWorkspace internal order + native-sync selection. +// =========================================================================== + +function toArrayBuffer(bytes: Buffer): ArrayBuffer { + return bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer; +} + +// A filesystem-backed managed-runtime client with the non-native base64-tar +// FALLBACK `syncIn` (mirrors sandbox-managed-runtime.test.ts `makeFilesystemClient` +// / `attachFallbackSyncIn`). `runCommands` records every `run` so a test can prove +// which restore transfer path the orchestrator took. When `withNativeSyncOut` is +// set, a native `syncOut` is attached so the client advertises BOTH directions. +function makeFsClient(options: { + runCommands?: string[]; + syncOutCalls?: { count: number }; + withNativeSyncOut?: boolean; +}): SandboxManagedRuntimeClient { + const client: SandboxManagedRuntimeClient = { + makeDir: async (remotePath) => { + await fs.mkdir(remotePath, { recursive: true }); + }, + writeFile: async (remotePath, bytes) => { + await fs.mkdir(path.dirname(remotePath), { recursive: true }); + await fs.writeFile(remotePath, Buffer.from(bytes)); + }, + readFile: async (remotePath) => await fs.readFile(remotePath), + listFiles: async (remotePath) => { + const entries = await fs.readdir(remotePath, { withFileTypes: true }).catch(() => []); + return entries + .filter((entry) => entry.isFile()) + .map((entry) => entry.name) + .sort((left, right) => left.localeCompare(right)); + }, + remove: async (remotePath) => { + await fs.rm(remotePath, { recursive: true, force: true }); + }, + run: async (command) => { + options.runCommands?.push(command); + await execFile("sh", ["-c", command], { maxBuffer: 32 * 1024 * 1024 }); + }, + }; + client.syncIn = async (operations: SandboxSyncOperation[]): Promise => { + const resultOperations: SandboxSyncResult["operations"] = []; + for (const operation of operations) { + let filesTransferred = 0; + let bytesTransferred = 0; + for (const mapping of operation.files) { + const bytes = await fs.readFile(mapping.sourcePath); + await client.makeDir(path.posix.dirname(mapping.targetPath)); + await client.writeFile(mapping.targetPath, toArrayBuffer(bytes)); + filesTransferred += 1; + bytesTransferred += bytes.byteLength; + } + for (const command of operation.postUploadCommands ?? []) { + await client.run(command.command, { timeoutMs: command.timeoutMs ?? 30_000 }); + } + resultOperations.push({ operationId: operation.operationId, filesTransferred, bytesTransferred }); + } + return { operations: resultOperations }; + }; + if (options.withNativeSyncOut) { + // Native outbound: materialize the sandbox workspace tree directly into the + // host destination (a directory file mapping), never through the tarball run. + client.syncOut = async (operations: SandboxSyncOperation[]): Promise => { + if (options.syncOutCalls) options.syncOutCalls.count += 1; + const resultOperations: SandboxSyncResult["operations"] = []; + for (const operation of operations) { + for (const mapping of operation.files) { + await fs.mkdir(mapping.targetPath, { recursive: true }); + await mirrorDirectory(mapping.sourcePath, mapping.targetPath); + } + resultOperations.push({ operationId: operation.operationId, filesTransferred: 1, bytesTransferred: 0 }); + } + return { operations: resultOperations }; + }; + } + return client; +} + +async function git(cwd: string, args: string[]): Promise { + const { stdout } = await execFile("git", ["-C", cwd, ...args], { maxBuffer: 32 * 1024 * 1024 }); + return stdout.trim(); +} + +describe("ACP settlement — Layer B: restoreWorkspace order + native-sync selection", () => { + it("test_non_git_workspace_restore_phase_order_and_preserve_absent_via_tarball_fallback", async () => { + // Model on sandbox-managed-runtime.test.ts "syncs workspace and assets through + // a provider-neutral sandbox client" (~:342). A client that exposes only the + // fallback `syncIn` (no `syncOut`) drives the remote-tarball restore transfer + // (sandbox-managed-runtime.ts:1192-1218) and the finalize ordering. + const root = await makeTempRoot(); + const localWorkspaceDir = path.join(root, "local-workspace"); + const remoteWorkspaceDir = path.join(root, "remote-workspace"); + await fs.mkdir(path.join(localWorkspaceDir, ".claude"), { recursive: true }); + await fs.writeFile(path.join(localWorkspaceDir, "README.md"), "local workspace\n", "utf8"); + await fs.writeFile(path.join(localWorkspaceDir, ".claude", "settings.json"), '{"local":true}\n', "utf8"); + + const runCommands: string[] = []; + const client = makeFsClient({ runCommands }); + // A fallback-only client never advertises native outbound. + expect(client.syncOut).toBeUndefined(); + + const runtimeStatuses: string[] = []; + const prepared = await prepareSandboxManagedRuntime({ + spec: { + transport: "sandbox", + provider: "test", + sandboxId: "sandbox-1", + remoteCwd: remoteWorkspaceDir, + timeoutMs: 30_000, + apiKey: null, + }, + adapterKey: "test-adapter", + client, + workspaceLocalDir: localWorkspaceDir, + workspaceExclude: [".claude"], + preserveAbsentOnRestore: [".claude"], + onRuntimeProgress: async (status) => { + runtimeStatuses.push(`${status.phase}:${status.message}`); + }, + }); + + // Staging excluded the host-managed `.claude` from the sandbox. + await expect( + fs.readFile(path.join(remoteWorkspaceDir, ".claude", "settings.json"), "utf8"), + ).rejects.toMatchObject({ code: "ENOENT" }); + + await fs.writeFile(path.join(remoteWorkspaceDir, "README.md"), "remote workspace\n", "utf8"); + await fs.writeFile(path.join(remoteWorkspaceDir, "remote-only.txt"), "sync back\n", "utf8"); + await prepared.restoreWorkspace(); + + // Sync-back applied the remote edits; the preserved-absent `.claude` stayed. + await expect(fs.readFile(path.join(localWorkspaceDir, "README.md"), "utf8")).resolves.toBe( + "remote workspace\n", + ); + await expect(fs.readFile(path.join(localWorkspaceDir, "remote-only.txt"), "utf8")).resolves.toBe( + "sync back\n", + ); + await expect( + fs.readFile(path.join(localWorkspaceDir, ".claude", "settings.json"), "utf8"), + ).resolves.toBe('{"local":true}\n'); + + // The fallback restore built the remote workspace-download tarball via `run`. + expect(runCommands.some((command) => command.includes("workspace-download.tar"))).toBe(true); + // Phase order: config_sync (staging) → restore → finalize (finalize last). + expect(runtimeStatuses).toEqual( + expect.arrayContaining([ + "config_sync:Syncing workspace to sandbox", + "restore:Restoring workspace from sandbox", + "finalize:Finalizing sandbox workspace", + ]), + ); + expect(runtimeStatuses.at(-1)).toBe("finalize:Finalizing sandbox workspace"); + }); + + it("test_git_backed_workspace_restore_phase_order", async () => { + // Model on sandbox-managed-runtime.test.ts "syncs git-backed workspaces through + // a shallow standalone clone…" (~:441). A git workspace adds the git_sync and + // export phases around the same config_sync/restore/finalize spine. + const root = await makeTempRoot(); + const sourceRepoDir = path.join(root, "source-repo"); + const localWorkspaceDir = path.join(root, "local-worktree"); + const remoteWorkspaceDir = path.join(root, "remote-workspace"); + + await fs.mkdir(sourceRepoDir, { recursive: true }); + await git(sourceRepoDir, ["init"]); + await git(sourceRepoDir, ["checkout", "-b", "main"]); + await git(sourceRepoDir, ["config", "user.name", "Paperclip Test"]); + await git(sourceRepoDir, ["config", "user.email", "test@paperclip.dev"]); + await fs.writeFile(path.join(sourceRepoDir, "tracked.txt"), "base\n", "utf8"); + await git(sourceRepoDir, ["add", "tracked.txt"]); + await git(sourceRepoDir, ["commit", "-m", "base"]); + await git(sourceRepoDir, ["worktree", "add", "-b", "work", localWorkspaceDir, "HEAD"]); + await fs.writeFile(path.join(localWorkspaceDir, "tracked.txt"), "dirty local\n", "utf8"); + + const client = makeFsClient({}); + const phases: string[] = []; + const prepared = await prepareSandboxManagedRuntime({ + spec: { + transport: "sandbox", + provider: "test", + sandboxId: "sandbox-1", + remoteCwd: remoteWorkspaceDir, + timeoutMs: 30_000, + apiKey: null, + }, + adapterKey: "test-adapter", + client, + workspaceLocalDir: localWorkspaceDir, + onRuntimeProgress: async (status) => { + phases.push(status.phase); + }, + }); + + // The sandbox holds a real git worktree seeded from the host history. + expect((await git(remoteWorkspaceDir, ["rev-list", "--count", "HEAD"]))).toBe("1"); + await git(remoteWorkspaceDir, ["config", "user.name", "Paperclip Sandbox"]); + await git(remoteWorkspaceDir, ["config", "user.email", "sandbox@paperclip.dev"]); + await git(remoteWorkspaceDir, ["add", "-A"]); + await git(remoteWorkspaceDir, ["commit", "-m", "sandbox update"]); + await fs.writeFile(path.join(remoteWorkspaceDir, "remote-only.txt"), "from sandbox\n", "utf8"); + + await prepared.restoreWorkspace(); + + // The sandbox commit imported back onto the host worktree. + expect(await git(localWorkspaceDir, ["log", "-1", "--pretty=%s"])).toBe("sandbox update"); + await expect(fs.readFile(path.join(localWorkspaceDir, "remote-only.txt"), "utf8")).resolves.toBe( + "from sandbox\n", + ); + expect(phases).toEqual( + expect.arrayContaining(["git_sync", "config_sync", "export", "restore", "finalize"]), + ); + expect(phases.at(-1)).toBe("finalize"); + }); + + it("test_native_syncOut_client_restores_through_native_transfer_not_tarball", async () => { + // The consumer reads `nativeSyncOut = typeof client.syncOut === "function"` + // (sandbox-managed-runtime.ts:775) and branches at :1163: with both directions + // native it uses `client.syncOut`, never the workspace-download tarball. + const root = await makeTempRoot(); + const localWorkspaceDir = path.join(root, "local-workspace"); + const remoteWorkspaceDir = path.join(root, "remote-workspace"); + await fs.mkdir(localWorkspaceDir, { recursive: true }); + await fs.writeFile(path.join(localWorkspaceDir, "README.md"), "local workspace\n", "utf8"); + + const runCommands: string[] = []; + const syncOutCalls = { count: 0 }; + const client = makeFsClient({ runCommands, syncOutCalls, withNativeSyncOut: true }); + // With both directions the client advertises native outbound. + expect(client.syncOut).toBeTypeOf("function"); + + const prepared = await prepareSandboxManagedRuntime({ + spec: { + transport: "sandbox", + provider: "test", + sandboxId: "sandbox-1", + remoteCwd: remoteWorkspaceDir, + timeoutMs: 30_000, + apiKey: null, + }, + adapterKey: "test-adapter", + client, + workspaceLocalDir: localWorkspaceDir, + }); + + await fs.writeFile(path.join(remoteWorkspaceDir, "README.md"), "remote workspace\n", "utf8"); + await fs.writeFile(path.join(remoteWorkspaceDir, "remote-only.txt"), "native sync\n", "utf8"); + await prepared.restoreWorkspace(); + + // The native outbound transfer ran and applied the remote edits... + expect(syncOutCalls.count).toBeGreaterThanOrEqual(1); + await expect(fs.readFile(path.join(localWorkspaceDir, "README.md"), "utf8")).resolves.toBe( + "remote workspace\n", + ); + await expect(fs.readFile(path.join(localWorkspaceDir, "remote-only.txt"), "utf8")).resolves.toBe( + "native sync\n", + ); + // ...and the tarball fallback was NOT used. + expect(runCommands.some((command) => command.includes("workspace-download.tar"))).toBe(false); + }); +}); + +// =========================================================================== +// Layer C — sync-back registration at the engine seam. +// +// The three real adapters each register a restoreWorkspace-backed teardown via +// their managed-home seam: codex packages/adapters/codex-local/src/server/acp.ts +// (teardown ~:239-258, wired by withCodexAcpDefaults ~:277); claude +// packages/adapters/claude-local/src/server/acp.ts (registerWorkspaceSyncBack +// ~:201-215, wired by withClaudeAcpDefaults ~:292); gemini +// packages/adapters/gemini-local/src/server/acp.ts (registerWorkspaceSyncBack +// ~:163-177, wired by withGeminiAcpDefaults ~:249). Those adapter packages are +// NOT dependencies of adapter-utils and resolve only to the canonical app +// checkout (not this worktree), and the plugin-sdk env-sync-negotiation helpers +// (definePlugin/startWorkerRpcHost) are not public exports — importing them from +// an adapter-utils test is infeasible/fragile. So this pins the adapter-agnostic +// sync-back at the engine seam: `prepareRemoteManagedHome` returns a `teardown` +// (the same shape each adapter registers), and the engine fires it exactly once +// on the exit/cleanup path (mirrors execute.test.ts +// test_remote_seam_teardown_fires_once_on_exit :2556). +// =========================================================================== +describe("ACP settlement — Layer C: per-adapter sync-back teardown fires once at the engine seam", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("test_engine_fires_each_adapter_named_sync_back_teardown_exactly_once", async () => { + for (const adapter of ["codex", "claude", "gemini"] as const) { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + let teardownCalls = 0; + await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { + authToken: "real-run-jwt", + executionTarget, + // The seam stands in for `${adapter}`'s registerWorkspaceSyncBack: it + // returns a teardown that would call stagedRuntime.restoreWorkspace(...). + prepareRemoteManagedHome: async (input) => { + const stagedRuntime = await input.stage([]); + return { + stagedRuntime, + teardown: async () => { + teardownCalls += 1; + }, + }; + }, + }, + ); + + expect(teardownCalls, `${adapter} sync-back teardown fires exactly once`).toBe(1); + } + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/startup-characterization.test.ts b/packages/adapter-utils/src/acpx-engine/startup-characterization.test.ts new file mode 100644 index 0000000000..8f55ec0ae6 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/startup-characterization.test.ts @@ -0,0 +1,913 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { AcpRuntimeOptions } from "acpx/runtime"; +import type { AdapterExecutionContext, AdapterRuntimeMcpAccess } from "@paperclipai/adapter-utils"; +import { + prepareAdapterExecutionTargetRuntime, + startAdapterExecutionTargetPaperclipBridge, + startAdapterExecutionTargetProcessSessionBridge, +} from "@paperclipai/adapter-utils/execution-target"; + +// Wrap the staging seam + both sandbox bridges in call-recording spies that +// still delegate to the real implementations. This copies the execute.test.ts +// harness verbatim so a startup test asserts the exact staging args and bridge +// hand-off the engine threads without changing any real behavior. +vi.mock("@paperclipai/adapter-utils/execution-target", async (importActual) => { + const actual = await importActual(); + return { + ...actual, + prepareAdapterExecutionTargetRuntime: vi.fn(actual.prepareAdapterExecutionTargetRuntime), + startAdapterExecutionTargetPaperclipBridge: vi.fn(actual.startAdapterExecutionTargetPaperclipBridge), + startAdapterExecutionTargetProcessSessionBridge: vi.fn(actual.startAdapterExecutionTargetProcessSessionBridge), + }; +}); +import { createAcpxEngineExecutor, type AcpxEngineExecutorOptions } from "./execute.js"; +import { runChildProcess } from "../server-utils.js"; + +const tempRoots: string[] = []; + +async function makeTempRoot() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-acpx-skills-")); + tempRoots.push(root); + return root; +} + +afterEach(async () => { + // A remote run stages a process-session bridge whose detached event writer can + // still be flushing a trailing event file into `.../process-sessions//events` + // when the run's own best-effort `client.remove(sessionDir)` (which production + // catch-wraps) has already returned. Under CI load that trailing write can land + // between this recursive delete's directory snapshot and its `rmdir`, surfacing as + // `ENOTEMPTY`. `maxRetries`/`retryDelay` make the cleanup ride out that window the + // same way production tolerates it, instead of failing the just-passed test. + await Promise.all( + tempRoots.splice(0).map((root) => + fs.rm(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }), + ), + ); +}); + +async function pathExists(candidate: string): Promise { + return fs.access(candidate).then(() => true).catch(() => false); +} + +void pathExists; + +function createLocalSandboxRunner( + onExecute?: (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + }) => void, +) { + let counter = 0; + return { + execute: async (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + stdin?: string; + timeoutMs?: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }) => { + counter += 1; + onExecute?.(input); + const command = input.command === "bash" ? "/bin/bash" : input.command; + return await runChildProcess(`acpx-sandbox-run-${counter}`, command, input.args ?? [], { + cwd: input.cwd ?? process.cwd(), + env: input.env ?? {}, + stdin: input.stdin, + timeoutSec: Math.max(1, Math.ceil((input.timeoutMs ?? 30_000) / 1000)), + graceSec: 5, + onLog: input.onLog ?? (async () => {}), + onSpawn: input.onSpawn + ? async (meta) => input.onSpawn?.({ pid: meta.pid, startedAt: meta.startedAt }) + : undefined, + }); + }, + }; +} + +function buildRuntime( + onSetConfigOption?: (input: { key: string; value: string }) => void, + onEnsureSession?: (input: Record) => void, +) { + return { + ensureSession: async (input: Record) => { + onEnsureSession?.(input); + return ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }); + }, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + setConfigOption: async (input: { key: string; value: string }) => { + onSetConfigOption?.(input); + }, + close: async () => {}, + }; +} + +async function runExecutor( + config: Record, + options: { + context?: Record; + executionTransport?: Record; + authToken?: string; + executionTarget?: Record; + runtimeMcp?: AdapterRuntimeMcpAccess; + prepareRemoteManagedHome?: AcpxEngineExecutorOptions["prepareRemoteManagedHome"]; + startupTraceContext?: AdapterExecutionContext["startupTraceContext"]; + } = {}, +) { + const runtimeOptions: Record[] = []; + const configOptions: Array<{ key: string; value: string }> = []; + const sessionInputs: Record[] = []; + const meta: Record[] = []; + const logs: Array<{ stream: string; text: string }> = []; + const events: Array<{ eventType: string; payload?: Record }> = []; + const execute = createAcpxEngineExecutor({ + ...(options.prepareRemoteManagedHome + ? { prepareRemoteManagedHome: options.prepareRemoteManagedHome } + : {}), + createRuntime: (options) => { + runtimeOptions.push(options as unknown as Record); + return buildRuntime( + ({ key, value }) => configOptions.push({ key, value }), + (input) => sessionInputs.push(input), + ) as never; + }, + }); + + const result = await execute({ + runId: "run-1", + agent: { + id: "agent-1", + companyId: "company-1", + }, + runtime: {}, + config, + context: options.context ?? {}, + executionTransport: options.executionTransport, + authToken: options.authToken, + executionTarget: options.executionTarget, + runtimeMcp: options.runtimeMcp, + startupTraceContext: options.startupTraceContext, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async (payload: unknown) => { + meta.push(payload as Record); + }, + onEvent: async (event: { eventType: string; payload?: Record }) => { + events.push(event); + }, + } as never); + + expect(result.exitCode).toBe(0); + return { logs, meta, events, runtimeOptions, configOptions, sessionInputs, result }; +} + +// The staging-seam describe helper from execute.test.ts (~:2152). It builds a +// runner-backed remote target: the local runner extracts the staged tar into +// `remoteCwd`, so the run really ships the HOST worktree into the sandbox. +async function setupRemoteSandbox() { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + // A file present only in the HOST worktree proves the workspace is shipped + // into the sandbox: the local runner extracts the staged tar into remoteCwd. + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + const runner = createLocalSandboxRunner(); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner, + }; + return { root, stateDir, localCwd, remoteCwd, executionTarget }; +} + +// Read the `configFingerprint` off a settled run result. +function fpOf(result: { sessionParams?: unknown }): string | undefined { + return (result.sessionParams as { configFingerprint?: string } | undefined)?.configFingerprint; +} + +const okHandle = { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", +}; + +function completedTurn() { + return { + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; +} + +describe("ACPX engine startup characterization", () => { + // Item 1 + 8: the remote launch env and its finalization point. + describe("remote launch environment values", () => { + beforeEach(() => vi.clearAllMocks()); + + it("mints the bridge launch env into the process-session command payload", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + // Decode the process-session LAUNCH payload (the base64 command blob). The + // in-sandbox process env rides there, not in the exec's own `env`. + let launchPayload: Record | null = null; + (executionTarget as { runner: unknown }).runner = createLocalSandboxRunner((input) => { + if (input.env?.PAPERCLIP_SANDBOX_EXEC_CHANNEL === "bridge") { + const script = input.args?.[1] ?? ""; + const match = script.match(/PAPERCLIP_PROCESS_SESSION_COMMAND_B64='([^']+)'/); + if (match) { + launchPayload = JSON.parse(Buffer.from(match[1]!, "base64").toString("utf8")) as Record< + string, + unknown + >; + } + } + }); + + await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { authToken: "real-run-jwt", executionTarget }, + ); + + // The launch payload carries the MERGED paperclip bridge env: the queue + // transport mode, a loopback bridge base URL, and a minted bridge token. + const payloadEnv = ((launchPayload as Record | null)?.env ?? {}) as Record< + string, + unknown + >; + expect(payloadEnv).toMatchObject({ PAPERCLIP_API_BRIDGE_MODE: "queue_v1" }); + expect(String(payloadEnv.PAPERCLIP_API_URL ?? "")).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/); + // The minted bridge token is present and is NOT the host run JWT. + expect(payloadEnv.PAPERCLIP_API_KEY).toBeTruthy(); + expect(payloadEnv.PAPERCLIP_API_KEY).not.toBe("real-run-jwt"); + }); + + it("finalizes the launch env at the bridge merge: the process env carries the merged bridge values", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + // Capture the launch payload AND the bridge-channel exec's own `env`. + let launchPayload: Record | null = null; + let bridgeExecEnv: Record | undefined; + (executionTarget as { runner: unknown }).runner = createLocalSandboxRunner((input) => { + if (input.env?.PAPERCLIP_SANDBOX_EXEC_CHANNEL === "bridge") { + bridgeExecEnv = input.env; + const script = input.args?.[1] ?? ""; + const match = script.match(/PAPERCLIP_PROCESS_SESSION_COMMAND_B64='([^']+)'/); + if (match) { + launchPayload = JSON.parse(Buffer.from(match[1]!, "base64").toString("utf8")) as Record< + string, + unknown + >; + } + } + }); + + await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { authToken: "real-run-jwt", executionTarget }, + ); + + // The launch payload is the finalized carrier of the process env. It already + // holds the merged bridge values at the point the run emits it, so no later + // write mutates the process env after the bridge merge. + const payloadEnv = ((launchPayload as Record | null)?.env ?? {}) as Record< + string, + unknown + >; + expect(payloadEnv.PAPERCLIP_API_BRIDGE_MODE).toBe("queue_v1"); + expect(payloadEnv.PAPERCLIP_API_KEY).toBeTruthy(); + // The bridge-channel exec's OWN env is the sandbox transport channel, not the + // agent process env: it does not carry the minted agent bridge key. This pins + // that the merged agent env lives only in the finalized launch payload. + expect(bridgeExecEnv?.PAPERCLIP_SANDBOX_EXEC_CHANNEL).toBe("bridge"); + expect(bridgeExecEnv?.PAPERCLIP_API_KEY).toBeUndefined(); + }); + }); + + // Item 2: the 17 fingerprint fields folded into `configFingerprint`, and the + // outer session key form that embeds it. + describe("session fingerprint and session key", () => { + beforeEach(() => vi.clearAllMocks()); + + it("forms the session key as paperclip:company:agent:taskKey:fingerprint and embeds the fingerprint", async () => { + const root = await makeTempRoot(); + const { result } = await runExecutor({ + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + cwd: path.join(root, "workspace"), + }); + + const fp = fpOf(result); + expect(fp).toBeTypeOf("string"); + expect(fp).toBeTruthy(); + // No taskId/issueId/workspaceId in the default context, so taskKey is "default". + const sessionKey = (result.sessionParams as { sessionKey?: string }).sessionKey; + expect(sessionKey).toBe(`paperclip:company-1:agent-1:default:${fp}`); + }); + + it("keeps the fingerprint stable across two identical runs and a same-config new wake", async () => { + const root = await makeTempRoot(); + const baseConfig = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + cwd: path.join(root, "workspace"), + }; + const first = await runExecutor(baseConfig, { + context: { taskId: "issue-1", wakeReason: "issue_assigned" }, + }); + const identical = await runExecutor(baseConfig, { + context: { taskId: "issue-1", wakeReason: "issue_assigned" }, + }); + // A fresh heartbeat with a different wake reason but the same config env. + const newWake = await runExecutor(baseConfig, { + context: { taskId: "issue-1", wakeReason: "comment", wakeCommentId: "c-9" }, + }); + + expect(fpOf(first.result)).toBeTruthy(); + expect(fpOf(identical.result)).toBe(fpOf(first.result)); + // Per-wake PAPERCLIP_* churn does not reset the session fingerprint. + expect(fpOf(newWake.result)).toBe(fpOf(first.result)); + }); + + it("busts the fingerprint when any representative folded dimension changes", async () => { + const root = await makeTempRoot(); + const cwd = path.join(root, "workspace"); + const stateDir = path.join(root, "state"); + const context = { context: { taskId: "issue-1", wakeReason: "issue_assigned" } }; + const base = { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd }; + + const baseFp = fpOf((await runExecutor(base, context)).result); + expect(baseFp).toBeTruthy(); + + // Each edit changes exactly one folded dimension and must bust the fingerprint. + // agentCommand. + expect(fpOf((await runExecutor({ ...base, agentCommand: "node ./other-acp.js" }, context)).result)).not.toBe(baseFp); + // cwd. + expect(fpOf((await runExecutor({ ...base, cwd: path.join(root, "other-cwd") }, context)).result)).not.toBe(baseFp); + // requestedModel. + expect(fpOf((await runExecutor({ ...base, model: "some-model" }, context)).result)).not.toBe(baseFp); + // requestedThinkingEffort. + expect(fpOf((await runExecutor({ ...base, thinkingEffort: "high" }, context)).result)).not.toBe(baseFp); + // mode. + expect(fpOf((await runExecutor({ ...base, mode: "oneshot" }, context)).result)).not.toBe(baseFp); + // adapterEnvHash (a resolved adapter env value). + expect(fpOf((await runExecutor({ ...base, env: { FOO: "bar" } }, context)).result)).not.toBe(baseFp); + // mcpServers identity (injected runtime MCP set). + expect( + fpOf( + ( + await runExecutor(base, { + ...context, + runtimeMcp: { + getServers: () => [ + { name: "github", url: "https://x.test/mcp", connectionId: "c-1", token: "t-1" }, + ], + }, + }) + ).result, + ), + ).not.toBe(baseFp); + // secretManifestHash. + expect( + fpOf( + ( + await runExecutor(base, { + ...context, + context: { + taskId: "issue-1", + wakeReason: "issue_assigned", + paperclipSecrets: { + manifest: [ + { + configPath: "env.API_TOKEN", + envKey: "API_TOKEN", + secretId: "secret-1", + bindingId: "binding-1", + secretKey: "api-token", + version: 1, + provider: "local_encrypted", + }, + ], + }, + }, + }) + ).result, + ), + ).not.toBe(baseFp); + // additionalSourcesIdentity (referenced-project set). + expect( + fpOf( + ( + await runExecutor(base, { + ...context, + context: { + taskId: "issue-1", + wakeReason: "issue_assigned", + paperclipWorkspace: { + cwd, + realization: { + additional: [ + { + path: "/host/project-a", + projectId: "a", + projectWorkspaceId: "ws-a", + repoUrl: "https://example.test/a.git", + repoRef: "ref-a-1", + }, + ], + }, + }, + }, + }) + ).result, + ), + ).not.toBe(baseFp); + + // fastMode folds only for codex, so pin it with a codex-vs-codex pair. + const codexBase = { agent: "codex", agentCommand: "node ./fake-acp.js", stateDir, cwd }; + const codexFp = fpOf((await runExecutor(codexBase, context)).result); + expect(fpOf((await runExecutor({ ...codexBase, fastMode: true }, context)).result)).not.toBe(codexFp); + }); + }); + + // Item 3: the staging seam call, its arguments, and its order (workspace then + // assets, serial). Modeled on the PR-1 staging-seam tests. + describe("staging seam calls, arguments, and order", () => { + beforeEach(() => vi.clearAllMocks()); + + it("stages the host workspace with no assets, exactly once, before the process launch", async () => { + const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); + const { sessionInputs, events } = await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { authToken: "real-run-jwt", executionTarget }, + ); + + // The staging seam crossed exactly once. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(1); + const stageArgs = vi.mocked(prepareAdapterExecutionTargetRuntime).mock.calls[0]![0]; + // The HOST worktree is shipped first; no per-adapter home asset in this lane. + expect(stageArgs.workspaceLocalDir).toBe(localCwd); + expect(stageArgs.assets ?? []).toEqual([]); + expect(stageArgs.installCommand ?? null).toBeNull(); + expect(stageArgs.target).toMatchObject({ kind: "remote", transport: "sandbox" }); + + // The workspace really landed in the sandbox workspace dir. + await expect(fs.readFile(path.join(remoteCwd, "hello.txt"), "utf8")).resolves.toBe("hi"); + // A per-step timing event proves the sync ran inside its timed boundary. + const stageEvent = events.find( + (event) => event.eventType === "run.startup.step" && event.payload?.step === "stage.sync", + ); + expect(stageEvent).toBeTruthy(); + // And session/new binds to the in-sandbox workspace cwd the seam returned. + expect(sessionInputs[0]?.cwd).toBe(remoteCwd); + }); + + it("threads a managed-home asset through the same seam after the workspace", async () => { + const { root, stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const managedHomeDir = path.join(root, "managed-home"); + await fs.mkdir(managedHomeDir, { recursive: true }); + await fs.writeFile(path.join(managedHomeDir, "config.json"), "{}", "utf8"); + + await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { + authToken: "real-run-jwt", + executionTarget, + prepareRemoteManagedHome: async (input) => { + const stagedRuntime = await input.stage([ + { key: "home", localDir: managedHomeDir, followSymlinks: true }, + ]); + return { stagedRuntime }; + }, + }, + ); + + // The seam's home asset is threaded through the SAME staging seam, keyed by + // the workspace-local dir. Workspace ships first; the asset rides alongside it. + const stageArgs = vi.mocked(prepareAdapterExecutionTargetRuntime).mock.calls[0]![0]; + expect(stageArgs.workspaceLocalDir).toBe(localCwd); + expect(stageArgs.assets).toEqual([ + { key: "home", localDir: managedHomeDir, followSymlinks: true }, + ]); + }); + }); + + // Item 4: the two-bridge overlap and the ACP-initialization ordering. + describe("two-bridge overlap and ACP initialization order", () => { + beforeEach(() => vi.clearAllMocks()); + + it("defers the process bridge env, shares one runtimeRootDir, and runs session/new on the sandbox cwd", async () => { + const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); + const { sessionInputs, runtimeOptions } = await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { authToken: "real-run-jwt", executionTarget }, + ); + + // The process-session bridge receives its launch env as a DEFERRED thunk, the + // seam that lets its env-independent setup overlap the paperclip bridge start. + const processArgs = vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mock.calls[0]![0]; + expect(typeof processArgs.env).toBe("function"); + + // Both bridges receive the SAME real (non-null) runtimeRootDir from staging. + const paperclipArgs = vi.mocked(startAdapterExecutionTargetPaperclipBridge).mock.calls[0]![0]; + expect(paperclipArgs.runtimeRootDir).toBeTruthy(); + expect(String(paperclipArgs.runtimeRootDir)).toContain(".paperclip-runtime"); + expect(processArgs.runtimeRootDir).toBe(paperclipArgs.runtimeRootDir); + + // The ACP runtime + session/new both bind to the in-sandbox workspace cwd, + // which the run resolves only after the bridges bring the sandbox up. + expect(runtimeOptions[0]?.cwd).toBe(remoteCwd); + expect(sessionInputs[0]?.cwd).toBe(remoteCwd); + expect(sessionInputs[0]?.cwd).not.toBe(localCwd); + }); + }); + + // Item 5 + 6: every startup exit path, its result phase, and the cleanup-call + // set (bridges stop / lease releases / runtime closes). + describe("startup exit paths and cleanup", () => { + beforeEach(() => vi.clearAllMocks()); + + it("create_runtime failure: settles an error result, stops both bridges, releases the lease", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const paperclipStop = vi.fn(async () => {}); + const processStop = vi.fn(async () => {}); + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce( + async () => ({ env: {}, stop: paperclipStop }) as never, + ); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementationOnce( + async () => ({ agentCommand: null, stop: processStop }) as never, + ); + const stagingLocks = new Map>(); + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + }); + + const result = await execute({ + runId: "run-create-fail-remote", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + + // The post-build runtime-creation failure returns a settled error result. + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("create_runtime"); + // Both live bridges stop exactly once and the per-session lease releases. + expect(paperclipStop).toHaveBeenCalledTimes(1); + expect(processStop).toHaveBeenCalledTimes(1); + expect(stagingLocks.size).toBe(0); + }); + + it("partial-bridge failure: throws and stops the concurrently-started bridge exactly once", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const stop = vi.fn(async () => {}); + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(async () => { + throw new Error("paperclip bridge boom"); + }); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementationOnce( + async () => ({ agentCommand: null, stop }) as never, + ); + + const execute = createAcpxEngineExecutor({ + createRuntime: () => buildRuntime() as never, + }); + + // A partial bridge failure inside buildRuntime is one of the only two throw + // paths, so the run rethrows instead of settling a result. + await expect( + execute({ + runId: "run-bridge-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never), + ).rejects.toThrow("paperclip bridge boom"); + + // The concurrently-started process-session bridge was stopped exactly once. + expect(stop).toHaveBeenCalledTimes(1); + }); + + it("cold ensure_session throw: ensure_session error, and the handshake-catch close does NOT fire (no handle yet)", async () => { + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => { + throw new Error("ensureSession boom"); + }, + startTurn: () => completedTurn(), + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: "handshake-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("ensure_session"); + // Current behavior (execute.ts ~:3590): the handshake catch closes the runtime + // only `if (handle)`. A cold `ensureSession` throw never assigns a handle, so + // this close never fires. The warm-hit path below has a cached handle and does + // close. This pins the real cold-path behavior, not an aspiration. + expect(closeSpy).not.toHaveBeenCalled(); + }); + + it("missing session handle: ensure_session error and the minimal runtime closes", async () => { + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + // A runtime that returns no session handle drives the missing-handle path. + ensureSession: async () => undefined, + startTurn: () => completedTurn(), + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: "missing-handle", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("ensure_session"); + expect(result.errorCode).toBe("acpx_runtime_error"); + expect(closeSpy).toHaveBeenCalledTimes(1); + }); + + it("configure_session failure: configure_session error and the runtime closes", async () => { + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => okHandle, + // Gemini model/effort drive a session config option; a throwing setter + // fails the configure_session phase after the handshake succeeds. + setConfigOption: async () => { + throw new Error("setConfigOption boom"); + }, + startTurn: () => completedTurn(), + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: "configure-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { + agent: "gemini", + model: "gemini-2.5-pro", + thinkingEffort: "high", + stateDir: path.join(root, "state"), + }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("configure_session"); + expect(closeSpy).toHaveBeenCalledTimes(1); + }); + + it("warm-hit failure: reuses the runtime, fails with ensure_session, closes it, drops the warm entry", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const startedAt = "2026-01-01T00:00:00.000Z"; + const closeSpy = vi.fn(async () => {}); + let created = 0; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: (options) => { + created += 1; + const opts = options as AcpRuntimeOptions & { + onAgentSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }; + return { + ensureSession: async () => { + await opts.onAgentSpawn?.({ pid: 4242, startedAt }); + return okHandle; + }, + startTurn: () => completedTurn(), + close: closeSpy, + } as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + + const first = await execute({ + runId: "warm-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + expect(warmHandles.size).toBe(1); + expect(created).toBe(1); + + // The warm-hit reuses runtime #1 and fails while persisting process identity. + const second = await execute({ + runId: "warm-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => { + throw new Error("onSpawn boom"); + }, + } as never); + + // No new runtime was created; the reused runtime is closed and the warm entry + // removed, with the failure reported on the ensure_session phase. + expect(created).toBe(1); + expect(second.exitCode).toBe(1); + expect(second.resultJson?.phase).toBe("ensure_session"); + expect(closeSpy).toHaveBeenCalledTimes(1); + expect(warmHandles.size).toBe(0); + }); + }); + + // Item 7: the per-lane resource set. + describe("per-lane resource set", () => { + beforeEach(() => vi.clearAllMocks()); + + it("local lane: crosses no staging seam, starts no bridge, keeps session/new on the host cwd", async () => { + const root = await makeTempRoot(); + const localCwd = path.join(root, "worktree"); + await fs.mkdir(localCwd, { recursive: true }); + const { sessionInputs, runtimeOptions } = await runExecutor({ + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + cwd: localCwd, + }); + + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).not.toHaveBeenCalled(); + expect(vi.mocked(startAdapterExecutionTargetPaperclipBridge)).not.toHaveBeenCalled(); + expect(vi.mocked(startAdapterExecutionTargetProcessSessionBridge)).not.toHaveBeenCalled(); + expect(sessionInputs[0]?.cwd).toBe(localCwd); + expect(runtimeOptions[0]?.cwd).toBe(localCwd); + }); + + it("persistent host lane: warm-saves the handle so a second run reuses the runtime", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let created = 0; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: () => { + created += 1; + return buildRuntime() as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + const base = { + agent: { id: "agent-1", companyId: "company-1" }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + }; + + const first = await execute({ runId: "host-warm-1", runtime: {}, ...base } as never); + const second = await execute({ + runId: "host-warm-2", + runtime: { sessionParams: first.sessionParams }, + ...base, + } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // The warm handle survives, so the second run reuses runtime #1. + expect(created).toBe(1); + expect(warmHandles.size).toBe(1); + }); + + it("remote process-session lane: does NOT warm-save the handle, so a second run re-creates the runtime", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + let created = 0; + const ensureInputs: Array> = []; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => { + created += 1; + return buildRuntime(undefined, (input) => ensureInputs.push(input)) as never; + }, + }); + const base = { + agent: { id: "agent-1", companyId: "company-1" }, + config: { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + cwd: localCwd, + mode: "persistent", + warmHandleIdleMs: 60_000, + }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + }; + + const first = await execute({ runId: "remote-warm-1", runtime: {}, ...base } as never); + const second = await execute({ + runId: "remote-warm-2", + runtime: { sessionParams: first.sessionParams }, + ...base, + } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // The remote process-session lane never keeps the handle warm, so the second + // run re-creates the runtime and runs a fresh handshake instead of reusing one. + expect(created).toBe(2); + expect(warmHandles.size).toBe(0); + expect(ensureInputs).toHaveLength(2); + }); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/turn-characterization.test.ts b/packages/adapter-utils/src/acpx-engine/turn-characterization.test.ts new file mode 100644 index 0000000000..01eea58322 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/turn-characterization.test.ts @@ -0,0 +1,685 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { AdapterExecutionContext, AdapterRuntimeMcpAccess } from "@paperclipai/adapter-utils"; +import { + prepareAdapterExecutionTargetRuntime, + startAdapterExecutionTargetPaperclipBridge, + startAdapterExecutionTargetProcessSessionBridge, +} from "@paperclipai/adapter-utils/execution-target"; + +// Wrap the staging seam + both sandbox bridges in call-recording spies that +// still delegate to the real implementations. This mirrors the execute.test.ts +// harness so the turn characterization tests share the same mocked module graph. +vi.mock("@paperclipai/adapter-utils/execution-target", async (importActual) => { + const actual = await importActual(); + return { + ...actual, + prepareAdapterExecutionTargetRuntime: vi.fn(actual.prepareAdapterExecutionTargetRuntime), + startAdapterExecutionTargetPaperclipBridge: vi.fn(actual.startAdapterExecutionTargetPaperclipBridge), + startAdapterExecutionTargetProcessSessionBridge: vi.fn(actual.startAdapterExecutionTargetProcessSessionBridge), + }; +}); +import { + createAcpxEngineExecutor, + summarizeAcpxTurnUsage, + type AcpxEngineExecutorOptions, +} from "./execute.js"; +import { runChildProcess } from "../server-utils.js"; + +const tempRoots: string[] = []; + +async function makeTempRoot() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-acpx-skills-")); + tempRoots.push(root); + return root; +} + +afterEach(async () => { + // A remote run stages a process-session bridge whose detached event writer can + // still be flushing a trailing event file into `.../process-sessions//events` + // when the run's own best-effort `client.remove(sessionDir)` (which production + // catch-wraps) has already returned. Under CI load that trailing write can land + // between this recursive delete's directory snapshot and its `rmdir`, surfacing as + // `ENOTEMPTY`. `maxRetries`/`retryDelay` make the cleanup ride out that window the + // same way production tolerates it, instead of failing the just-passed test. + await Promise.all( + tempRoots.splice(0).map((root) => + fs.rm(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }), + ), + ); +}); + +function createLocalSandboxRunner( + onExecute?: (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + }) => void, +) { + let counter = 0; + return { + execute: async (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + stdin?: string; + timeoutMs?: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }) => { + counter += 1; + onExecute?.(input); + const command = input.command === "bash" ? "/bin/bash" : input.command; + return await runChildProcess(`acpx-sandbox-run-${counter}`, command, input.args ?? [], { + cwd: input.cwd ?? process.cwd(), + env: input.env ?? {}, + stdin: input.stdin, + timeoutSec: Math.max(1, Math.ceil((input.timeoutMs ?? 30_000) / 1000)), + graceSec: 5, + onLog: input.onLog ?? (async () => {}), + onSpawn: input.onSpawn + ? async (meta) => input.onSpawn?.({ pid: meta.pid, startedAt: meta.startedAt }) + : undefined, + }); + }, + }; +} + +function buildRuntime( + onSetConfigOption?: (input: { key: string; value: string }) => void, + onEnsureSession?: (input: Record) => void, +) { + return { + ensureSession: async (input: Record) => { + onEnsureSession?.(input); + return ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }); + }, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + setConfigOption: async (input: { key: string; value: string }) => { + onSetConfigOption?.(input); + }, + close: async () => {}, + }; +} + +async function runExecutor( + config: Record, + options: { + context?: Record; + executionTransport?: Record; + authToken?: string; + executionTarget?: Record; + runtimeMcp?: AdapterRuntimeMcpAccess; + prepareRemoteManagedHome?: AcpxEngineExecutorOptions["prepareRemoteManagedHome"]; + startupTraceContext?: AdapterExecutionContext["startupTraceContext"]; + } = {}, +) { + const runtimeOptions: Record[] = []; + const configOptions: Array<{ key: string; value: string }> = []; + const sessionInputs: Record[] = []; + const meta: Record[] = []; + const logs: Array<{ stream: string; text: string }> = []; + const events: Array<{ eventType: string; payload?: Record }> = []; + const execute = createAcpxEngineExecutor({ + ...(options.prepareRemoteManagedHome + ? { prepareRemoteManagedHome: options.prepareRemoteManagedHome } + : {}), + createRuntime: (options) => { + runtimeOptions.push(options as unknown as Record); + return buildRuntime( + ({ key, value }) => configOptions.push({ key, value }), + (input) => sessionInputs.push(input), + ) as never; + }, + }); + + const result = await execute({ + runId: "run-1", + agent: { + id: "agent-1", + companyId: "company-1", + }, + runtime: {}, + config, + context: options.context ?? {}, + executionTransport: options.executionTransport, + authToken: options.authToken, + executionTarget: options.executionTarget, + runtimeMcp: options.runtimeMcp, + startupTraceContext: options.startupTraceContext, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async (payload: unknown) => { + meta.push(payload as Record); + }, + onEvent: async (event: { eventType: string; payload?: Record }) => { + events.push(event); + }, + } as never); + + expect(result.exitCode).toBe(0); + return { logs, meta, events, runtimeOptions, configOptions, sessionInputs, result }; +} + +// A stub ACP runtime that yields a controlled event stream and terminal result. +// The tests supply the events/result/cancel behavior; the harness above only +// drives always-happy sessions, so the turn tests build their own runtime here. +function turnRuntime(input: { + events: () => AsyncGenerator>; + result: Promise>; + onStartTurn?: (turnInput: Record) => void; + onCancel?: (reason: string) => void; + onClose?: () => void; + getStatus?: () => Promise>; + onEnsureSession?: (session: Record) => void; + ensureSession?: (session: Record) => Promise>; +}) { + const runtime: Record = { + ensureSession: + input.ensureSession ?? + (async (session: Record) => { + input.onEnsureSession?.(session); + return { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + }), + startTurn: (turnInput: Record) => { + input.onStartTurn?.(turnInput); + return { + events: input.events(), + result: input.result, + cancel: async ({ reason }: { reason: string }) => { + input.onCancel?.(reason); + }, + }; + }, + close: async () => { + input.onClose?.(); + }, + }; + if (input.getStatus) runtime.getStatus = input.getStatus; + return runtime; +} + +describe("ACPX engine turn characterization", () => { + // The stub session handle every ensureSession returns. startTurn must receive + // this exact handle object. + const SESSION_HANDLE = { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + + it("passes exactly the six turn inputs to startTurn", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let captured: Record | null = null; + let metaPrompt = ""; + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + onStartTurn: (turnInput) => { + captured = turnInput; + }, + events: async function* () { + yield { type: "done", stopReason: "end_turn" }; + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + + const result = await execute({ + runId: "run-six-inputs", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, timeoutSec: 120 }, + context: {}, + onLog: async () => {}, + onMeta: async (payload: { prompt?: string }) => { + metaPrompt = payload.prompt ?? ""; + }, + } as never); + + expect(result.exitCode).toBe(0); + const input = captured!; + // The handle is the exact session handle ensureSession returned. + expect(input.handle).toEqual(SESSION_HANDLE); + // The text is the built run prompt, the same string reported to onMeta. + expect(input.text).toBe(metaPrompt); + expect(typeof input.text).toBe("string"); + expect((input.text as string).length).toBeGreaterThan(0); + // The turn always runs in prompt mode. + expect(input.mode).toBe("prompt"); + // The request id is the run id. + expect(input.requestId).toBe("run-six-inputs"); + // A positive timeoutSec becomes timeoutMs in milliseconds. + expect(input.timeoutMs).toBe(120_000); + // The abort signal is present and not yet aborted. + const signal = input.signal as AbortSignal; + expect(signal).toBeInstanceOf(AbortSignal); + expect(signal.aborted).toBe(false); + // Exactly the six documented keys are threaded. + expect(Object.keys(input).sort()).toEqual( + ["handle", "mode", "requestId", "signal", "text", "timeoutMs"].sort(), + ); + }); + + it("joins text_delta events into the trimmed result summary", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + events: async function* () { + yield { type: "text_delta", text: " Hello, ", stream: "output", tag: "agent_message_chunk" }; + yield { type: "text_delta", text: "world ", stream: "output", tag: "agent_message_chunk" }; + yield { type: "done", stopReason: "end_turn" }; + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + + const result = await execute({ + runId: "run-summary", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(0); + // The summary is the concatenation of the deltas, trimmed. + expect(result.summary).toBe("Hello, world"); + }); + + it("falls back to the stop reason for the summary when no text streams", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + events: async function* () { + yield { type: "done", stopReason: "end_turn" }; + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + + const result = await execute({ + runId: "run-no-text", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(0); + // With no streamed text, the summary is the terminal stop reason. + expect(result.summary).toBe("end_turn"); + }); + + it("pins one event sequence across the log transcript and structured event transports", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const logs: Array<{ stream: string; text: string }> = []; + const events: Array<{ eventType: string; payload?: Record }> = []; + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + events: async function* () { + yield { type: "text_delta", text: "streamed hello", stream: "output", tag: "agent_message_chunk" }; + yield { type: "done", stopReason: "end_turn" }; + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + + const result = await execute({ + runId: "run-transports", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async () => {}, + onEvent: async (event: { eventType: string; payload?: Record }) => { + events.push(event); + }, + } as never); + + expect(result.exitCode).toBe(0); + // Transport 1 — the stdout transcript carries the text_delta as an acpx record. + expect(logs).toContainEqual({ + stream: "stdout", + text: `${JSON.stringify({ + type: "acpx.text_delta", + text: "streamed hello", + channel: "output", + tag: "agent_message_chunk", + })}\n`, + }); + // The joined text also lands in the result summary. + expect(result.summary).toBe("streamed hello"); + // Transport 2 — the structured onEvent stream reflects the run bring-up + // sequence as run.startup.step events, including the acp handshake. + const steps = events.filter((event) => event.eventType === "run.startup.step"); + const stepNames = steps.map((event) => String(event.payload?.step)); + expect(stepNames).toContain("acp.handshake"); + expect(stepNames).toContain("workspace.resolve"); + }); + + it("aborts a hung turn on the wall-clock timer and cancels with the timeout message", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const cwd = path.join(root, "worktree"); + await fs.mkdir(cwd, { recursive: true }); + + const cancelReasons: string[] = []; + let releaseTurn: (() => void) | null = null; + const turnCancelled = new Promise((resolve) => { + releaseTurn = resolve; + }); + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + // The stream never yields on its own. Only the wall-clock timer's cancel + // unblocks it, simulating a hung run. + events: async function* () { + await turnCancelled; + }, + result: turnCancelled.then(() => ({ status: "cancelled", stopReason: "cancelled" })), + onCancel: (reason) => { + cancelReasons.push(reason); + releaseTurn?.(); + }, + }) as never, + }); + + const result = await execute({ + runId: "run-timeout", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd, timeoutSec: 1 }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + const expectedMessage = + "Run exceeded the adapter execution timeout (timeoutSec=1, configured via adapterConfig.timeoutSec). " + + "Set adapterConfig.timeoutSec to raise it."; + expect(result.timedOut).toBe(true); + expect(result.signal).toBe("SIGTERM"); + expect(result.errorCode).toBe("acpx_timeout"); + expect(result.errorMessage).toBe(expectedMessage); + // The cancel ran with the formatted timeout message. + expect(cancelReasons).toContain(expectedMessage); + }, 15_000); + + it("cancels the turn before closing the runtime when the turn throws", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const order: string[] = []; + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + // The event stream throws mid-turn, so the catch path runs teardown. + events: async function* () { + throw new Error("turn boom"); + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + onCancel: () => order.push("cancel"), + onClose: () => order.push("close"), + }) as never, + }); + + const result = await execute({ + runId: "run-throw", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + // A failure after startTurn returned reports phase "turn". + expect((result.resultJson as Record)?.phase).toBe("turn"); + // Cancel runs before close, in that exact order. + expect(order).toEqual(["cancel", "close"]); + }); + + it("retries the session resume once and never retries the turn", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const cwd = path.join(root, "worktree"); + await fs.mkdir(cwd, { recursive: true }); + const config = { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd }; + + const ensureInputs: Array> = []; + let startTurnCalls = 0; + let resumeLogged = false; + + // A first clean run mints the session params the second run resumes from. + const firstExecute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + onEnsureSession: (session) => ensureInputs.push(session), + onStartTurn: () => { + startTurnCalls += 1; + }, + events: async function* () { + yield { type: "done", stopReason: "end_turn" }; + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + const first = await firstExecute({ + runId: "run-resume-a", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + + // The second run resumes. Its first ensureSession (the resume) fails with a + // resume-shaped error; the fresh retry (no resumeSessionId) succeeds. + const secondExecute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + ensureSession: async (session: Record) => { + ensureInputs.push(session); + if (session.resumeSessionId) { + throw new Error("resume session not found"); + } + return { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + }, + onStartTurn: () => { + startTurnCalls += 1; + }, + events: async function* () { + yield { type: "done", stopReason: "end_turn" }; + }, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + const second = await secondExecute({ + runId: "run-resume-b", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: (first as { sessionParams?: unknown }).sessionParams }, + config, + context: {}, + onLog: async (_stream: string, text: string) => { + if (text.includes("is unavailable; retrying with a fresh session")) resumeLogged = true; + }, + onMeta: async () => {}, + } as never); + + expect(second.exitCode).toBe(0); + // The first run's single ensureSession, plus the second run's resume and + // fresh retry, make three ensureSession calls total; two on the second run. + expect(ensureInputs).toHaveLength(3); + expect(ensureInputs[1]?.resumeSessionId).toBe(first.sessionId); + expect(ensureInputs[2]?.resumeSessionId).toBeUndefined(); + // The turn is never retried: one startTurn per run, two across both runs. + expect(startTurnCalls).toBe(2); + // The engine logged the resume fallback. + expect(resumeLogged).toBe(true); + // The fresh retry clears the stale session for the caller. + expect(second.clearSession).toBe(true); + }); + + it("maps a failed terminal to acpx_turn_failed and folds in the reported usage", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + events: async function* () { + yield { + type: "status", + text: "usage", + tag: "usage_update", + cost: { amount: 0.31, currency: "USD" }, + breakdown: { inputTokens: 40, outputTokens: 700, cachedReadTokens: 60 }, + }; + yield { type: "done", stopReason: "failed" }; + }, + result: Promise.resolve({ status: "failed", error: new Error("boom") }), + }) as never, + }); + + const result = await execute({ + runId: "run-failed", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.errorCode).toBe("acpx_turn_failed"); + // The failed error message becomes the terminal stop reason and the summary. + expect(result.summary).toBe("boom"); + expect((result.resultJson as Record)?.stopReason).toBe("boom"); + // The usage math folds the usage_update event into per-run usage and cost. + expect(result.usage).toEqual({ inputTokens: 40, outputTokens: 700, cachedInputTokens: 60 }); + expect(result.usageBasis).toBe("per_run"); + expect(result.costUsd).toBeCloseTo(0.31); + }); + + it("computes usage the same way summarizeAcpxTurnUsage does for the event fallback", () => { + // Pin the exported helper the turn path calls: with no getStatus snapshots the + // event breakdown and cost drive the per-run usage. + const summary = summarizeAcpxTurnUsage({ + preStatus: null, + postStatus: null, + eventBreakdown: { inputTokens: 40, outputTokens: 700, cachedReadTokens: 60 }, + eventCostUsd: 0.31, + }); + expect(summary.usage).toEqual({ inputTokens: 40, outputTokens: 700, cachedInputTokens: 60 }); + expect(summary.costUsd).toBeCloseTo(0.31); + }); + + it("returns a prepare_turn error result when the prompt build throws", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let startTurnCalls = 0; + let closes = 0; + + const execute = createAcpxEngineExecutor({ + createRuntime: () => + turnRuntime({ + onStartTurn: () => { + startTurnCalls += 1; + }, + onClose: () => { + closes += 1; + }, + events: async function* () {}, + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + }) as never, + }); + + // A throwing accessor on a field only buildPrompt reads makes the prompt build + // fail after the session handshake succeeds but before startTurn runs. + const context: Record = {}; + Object.defineProperty(context, "paperclipSessionHandoffMarkdown", { + enumerable: false, + get() { + throw new Error("prompt build boom"); + }, + }); + + const result = await execute({ + runId: "run-prepare-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + // The pre-turn failure reports phase "prepare_turn". + expect((result.resultJson as Record)?.phase).toBe("prepare_turn"); + // The turn never started, and the runtime closed once. + expect(startTurnCalls).toBe(0); + expect(closes).toBe(1); + }); + + it("threads the same happy-path turn through the shared runExecutor harness", async () => { + // A smoke pin on the copied harness: the always-happy buildRuntime turn exits + // clean with the done stop reason as the summary. + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const { result } = await runExecutor({ + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + }); + expect(result.exitCode).toBe(0); + expect(result.summary).toBe("end_turn"); + }); +}); diff --git a/server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts b/server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts new file mode 100644 index 0000000000..ae6af37121 --- /dev/null +++ b/server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts @@ -0,0 +1,312 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + companies, + createDb, + heartbeatRunEvents, + heartbeatRuns, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() })); +vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient })); + +import { + heartbeatService, + leaseReleaseStatusForRunStatus, + type HeartbeatEnvironmentRuntime, +} from "../services/heartbeat.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres terminalize-before-release tests on this host: ${ + embeddedPostgresSupport.reason ?? "unsupported environment" + }`, + ); +} + +// This file is a characterization test. It pins the CURRENT run-teardown order +// in server/src/services/heartbeat.ts:16586-16607: the teardown finally +// terminalizes the run FIRST, then releases the environment lease using the +// terminalized status. The production order is: +// latestRun = await terminalizeRunOnLeaseRelease(latestRun); // :16593 first +// await releaseEnvironmentLeasesForRun({ status: latestRun?.status, ... }); // :16601 second +// +// The enclosing teardown finally is not reasonably invokable in isolation: it +// lives deep in the heartbeat run body and needs a full sandbox, adapter, and +// workspace bring-up to reach. So this test drives the two real production +// functions in the same order against the embedded database: +// `terminalizeRunOnLeaseRelease` and `releaseEnvironmentLeasesForRun`. It thus +// executes the real lease-release boundary: the terminalized run status flows +// through the real run-status → lease-status mapping +// (`leaseReleaseStatusForRunStatus`) and the real environment orchestrator +// (`envOrchestrator.releaseForRun`) down to the runtime leaf. The test injects a +// fake `environmentRuntime` that records the mapped lease status at that leaf, so +// a wrong terminal state or a broken mapping fails the suite. +// +// The two additive test seams keep production behavior unchanged. The service now +// exposes `releaseEnvironmentLeasesForRun` (next to the existing +// `terminalizeRunOnLeaseRelease`), and `leaseReleaseStatusForRunStatus` is now +// exported for the direct mapping assertions below. +describeEmbeddedPostgres("heartbeat teardown terminalizes the run before releasing the lease", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-terminalize-before-release-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(heartbeatRunEvents); + await db.delete(issues); + await db.delete(heartbeatRuns); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seed(input: { issueStatus: string; runStatus: string }) { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Coder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Terminalize before release", + status: input.issueStatus, + priority: "high", + assigneeAgentId: agentId, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: input.runStatus, + invocationSource: "manual", + startedAt: new Date(), + contextSnapshot: { issueId }, + }); + + const run = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0]!); + + return { companyId, agentId, issueId, runId, run }; + } + + async function runStatus(runId: string) { + return db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0]?.status ?? null); + } + + // Drive the two real production functions in the teardown order at + // heartbeat.ts:16586-16607 and report what the real release step observes. + // Terminalize runs first. Then `releaseEnvironmentLeasesForRun` runs with the + // terminalized run status. That real call maps the run status through + // `leaseReleaseStatusForRunStatus` and passes it to the real environment + // orchestrator, which reaches the runtime leaf. The injected fake + // `environmentRuntime` records the run id and the mapped lease status at that + // leaf, so the test observes the actual boundary, not a reproduction. + async function runTeardownSequenceObservingRelease(input: { + runId: string; + companyId: string; + agentId: string; + }) { + const { runId, companyId, agentId } = input; + const releaseLeafCalls: Array<{ runId: string; status: string }> = []; + const fakeEnvironmentRuntime = { + // The orchestrator's `releaseForRun` calls this leaf with the mapped lease + // status. There are no seeded leases, so return an empty release set. + releaseRunLeases: async ( + heartbeatRunId: string, + status: "released" | "expired" | "failed", + ) => { + releaseLeafCalls.push({ runId: heartbeatRunId, status }); + return []; + }, + } as unknown as HeartbeatEnvironmentRuntime; + const heartbeat = heartbeatService(db, { environmentRuntime: fakeEnvironmentRuntime }); + + let latestRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + const statusBeforeTerminalize = latestRun?.status ?? null; + if (latestRun) latestRun = await heartbeat.terminalizeRunOnLeaseRelease(latestRun); + // The status production passes into releaseEnvironmentLeasesForRun (:16605). + const statusThreadedToRelease = latestRun?.status ?? null; + // The run row as the later release step observes it in the database. + const dbStatusAtRelease = await runStatus(runId); + + // Execute the real release step exactly as the teardown does at :16601. + await heartbeat.releaseEnvironmentLeasesForRun({ + runId, + companyId, + agentId, + status: statusThreadedToRelease, + }); + const orchestratorObservedRunId = releaseLeafCalls.at(-1)?.runId ?? null; + const orchestratorObservedLeaseStatus = releaseLeafCalls.at(-1)?.status ?? null; + + return { + statusBeforeTerminalize, + statusThreadedToRelease, + dbStatusAtRelease, + releaseCallCount: releaseLeafCalls.length, + orchestratorObservedRunId, + orchestratorObservedLeaseStatus, + terminalRun: latestRun, + }; + } + + it("terminalizes a running run to succeeded before release when the issue reached done", async () => { + const { companyId, agentId, issueId, runId } = await seed({ issueStatus: "done", runStatus: "running" }); + + const observed = await runTeardownSequenceObservingRelease({ runId, companyId, agentId }); + + // The run was still running before terminalize, but release observes the + // terminalized status, proving terminalize ran first. + expect(observed.statusBeforeTerminalize).toBe("running"); + expect(observed.statusThreadedToRelease).toBe("succeeded"); + expect(observed.dbStatusAtRelease).toBe("succeeded"); + + // The real orchestrator ran once and received the run id plus the mapped + // lease status for a succeeded run. + expect(observed.releaseCallCount).toBe(1); + expect(observed.orchestratorObservedRunId).toBe(runId); + expect(observed.orchestratorObservedLeaseStatus).toBe("released"); + + // The issue outcome is preserved and the lifecycle event records the reason. + const issueStatus = await db + .select({ status: issues.status }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]?.status); + expect(issueStatus).toBe("done"); + + const event = await db + .select({ message: heartbeatRunEvents.message, payload: heartbeatRunEvents.payload }) + .from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)) + .then((rows) => rows[0]); + expect(event?.message).toContain("lease release"); + expect((event?.payload as { terminalStatus?: string } | null)?.terminalStatus).toBe("succeeded"); + }); + + it("terminalizes a running run to interrupted before release when the issue is not terminal", async () => { + const { companyId, agentId, runId } = await seed({ issueStatus: "in_progress", runStatus: "running" }); + + const observed = await runTeardownSequenceObservingRelease({ runId, companyId, agentId }); + + expect(observed.statusBeforeTerminalize).toBe("running"); + expect(observed.statusThreadedToRelease).toBe("interrupted"); + expect(observed.dbStatusAtRelease).toBe("interrupted"); + + // An interrupted run maps to a normal lease release. + expect(observed.releaseCallCount).toBe(1); + expect(observed.orchestratorObservedLeaseStatus).toBe("released"); + + const row = await db + .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0]); + expect(row?.status).toBe("interrupted"); + expect(row?.errorCode).toBe("lease_released_before_terminal"); + }); + + it("terminalizes a still-queued run to interrupted before release", async () => { + // A queued run holds a lease but never reached running. Release must observe a + // terminal status, not the queued phantom-live status. + const { companyId, agentId, runId } = await seed({ issueStatus: "in_progress", runStatus: "queued" }); + + const observed = await runTeardownSequenceObservingRelease({ runId, companyId, agentId }); + + expect(observed.statusBeforeTerminalize).toBe("queued"); + expect(observed.statusThreadedToRelease).toBe("interrupted"); + expect(observed.dbStatusAtRelease).toBe("interrupted"); + expect(observed.orchestratorObservedLeaseStatus).toBe("released"); + }); + + it("threads an already-terminal run's status through unchanged and writes no new event", async () => { + // When another path already made the run terminal, terminalize is a no-op, so + // release still observes that authoritative terminal status. + const { companyId, agentId, runId } = await seed({ issueStatus: "done", runStatus: "failed" }); + + const observed = await runTeardownSequenceObservingRelease({ runId, companyId, agentId }); + + expect(observed.statusBeforeTerminalize).toBe("failed"); + expect(observed.statusThreadedToRelease).toBe("failed"); + expect(observed.dbStatusAtRelease).toBe("failed"); + + // A failed run maps to a failed lease release at the orchestrator. + expect(observed.orchestratorObservedLeaseStatus).toBe("failed"); + + const eventCount = await db + .select({ id: heartbeatRunEvents.id }) + .from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)) + .then((rows) => rows.length); + expect(eventCount).toBe(0); + }); +}); + +// Pin the real run-status → lease-release-status mapping the teardown threads +// into the environment orchestrator (heartbeat.ts:16601-16606). The database +// tests above reach the "released" and "failed" branches; this direct test also +// pins the "expired" and "timed_out" branches. It needs no database, so it runs +// on every host. +describe("run-status to lease-release-status mapping", () => { + it("maps each terminal run status to the lease-release status the orchestrator receives", () => { + // A normal or in-progress run releases the lease. + expect(leaseReleaseStatusForRunStatus("succeeded")).toBe("released"); + expect(leaseReleaseStatusForRunStatus("interrupted")).toBe("released"); + expect(leaseReleaseStatusForRunStatus("running")).toBe("released"); + expect(leaseReleaseStatusForRunStatus("queued")).toBe("released"); + expect(leaseReleaseStatusForRunStatus(null)).toBe("released"); + expect(leaseReleaseStatusForRunStatus(undefined)).toBe("released"); + // A failed or timed-out run marks the lease release as failed. + expect(leaseReleaseStatusForRunStatus("failed")).toBe("failed"); + expect(leaseReleaseStatusForRunStatus("timed_out")).toBe("failed"); + // A cancelled run expires the lease. + expect(leaseReleaseStatusForRunStatus("cancelled")).toBe("expired"); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index a096530402..358fe7357f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1332,7 +1332,7 @@ async function resolveRunScopedMentionedSkillKeys(input: { .filter((skillKey): skillKey is string => Boolean(skillKey)); } -function leaseReleaseStatusForRunStatus( +export function leaseReleaseStatusForRunStatus( status: string | null | undefined, ): Extract { if (status === "cancelled") return "expired"; @@ -19274,6 +19274,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) terminalizeRunOnLeaseRelease, + releaseEnvironmentLeasesForRun, + sweepStaleIssueLocks, buildIssueGraphLivenessAutoRecoveryPreview,