diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 98a2007fd3e..42b90da768f 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -6355,10 +6355,30 @@ describe("ACPX engine run lifecycle corrections (F3: one teardown error policy)" }; } + it("collects instructions after confirmed close on a thrown provider turn, before workspace restore", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, anyStopped } = stubBridges(); + const order: string[] = []; + const execute = createAcpxEngineExecutor({ + stagingLocks: new Map(), warmHandles: new Map(), stagedRuntimes: new Map(), + createRuntime: () => ({ ensureSession: async () => okHandle, startTurn: () => throwingTurn(), + close: async () => { order.push("close"); } }) as never, + }); + const result = await execute({ runId: "instruction-stop-failure", + ...remoteArgs(stateDir, localCwd, executionTarget, { onProviderStopped: async () => { + expect(anyStopped(paperclipStops)).toBe(true); + order.push("collect"); + } }), + } as never); + expect(result.exitCode).toBe(1); + expect(order).toEqual(["close", "collect"]); + }); + it("test_teardown_continues_after_one_teardown_step_fails", async () => { const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); const { paperclipStops, processStops, anyStopped } = stubBridges(); const stagingLocks = new Map>(); + const collectInstructions = vi.fn(async () => {}); const logs: Array<{ stream: string; text: string }> = []; const execute = createAcpxEngineExecutor({ stagingLocks, @@ -6378,6 +6398,7 @@ describe("ACPX engine run lifecycle corrections (F3: one teardown error policy)" const result = await execute({ runId: "td-continue", ...remoteArgs(stateDir, localCwd, executionTarget, { + onProviderStopped: collectInstructions, onLog: async (stream: "stdout" | "stderr", text: string) => { logs.push({ stream, text }); }, @@ -6385,6 +6406,7 @@ describe("ACPX engine run lifecycle corrections (F3: one teardown error policy)" } as never); expect(result.exitCode).toBe(1); + expect(collectInstructions).not.toHaveBeenCalled(); // The close failure did not stop the bridge stops or the lease release. expect(anyStopped(paperclipStops)).toBe(true); expect(anyStopped(processStops)).toBe(true); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index cacae0af7e4..ccc30259c5a 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -3600,12 +3600,13 @@ async function closeWarmHandle(input: { input.handles.delete(input.key); } clearWarmHandleTimer(input.entry); - await input.entry.runtime.close({ + const stopped = await input.entry.runtime.close({ handle: input.entry.handle, reason: input.reason, discardPersistentState: input.discardPersistentState ?? false, - }).catch(() => {}); + }).then(() => true, () => false); flushChildStderr(input.entry.childStderrState); + return stopped; } function warmHandleMatches( @@ -5289,7 +5290,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { ) { // A matching warm entry closes through the warm store, which also // clears its idle timer and flushes its child stderr. - await closeWarmHandle({ + runtimeStopConfirmed = await closeWarmHandle({ handles: warmHandles, key: prepared.sessionKey, entry: existing, @@ -5335,6 +5336,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { // runs, so the host mints a `traceparent` for the provider spans, and the // per-task restore spans parent to `sandbox.syncBack`. syncBack: () => timedPhase("sync_back", async () => { + // A transport error or terminal narrative cannot authorize collecting + // mutable provider files. Require the owned close or local OS handle. + const stopped = runtimeStopConfirmed || (!prepared.processSessionBridge && capturedProcessExited(processIdentitySink?.localProcess)); + if (stopped) await ctx.onProviderStopped?.(); await runRuntimeSpan("sandbox.syncBack", async () => { const restoreOutcome = await syncBackManagedHome(prepared); if (!restoreOutcome.ok) { diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 0735cb51056..54b055bac6d 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -280,6 +280,9 @@ export interface AdapterExecutionTargetProcessOptions { onLog: (stream: "stdout" | "stderr", chunk: string) => Promise; onRuntimeProgress?: RuntimeStatusSink; onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise; + /** Trusted invocation observation, not a turn completion callback. Called only + * after local child close or a remotely observed exit, including nonzero exits. */ + onProcessStopped?: () => void; terminalResultCleanup?: TerminalResultCleanupOptions; /** * Sandbox-only: factory from the Paperclip bridge handle that streams the @@ -894,6 +897,7 @@ export async function runAdapterExecutionTargetProcess( // after the clean process completion cannot latch a false mid-run loss. A // control channel that died before this clean completion still fails the // run closed. + if (!result.timedOut && typeof result.exitCode === "number" && Number.isInteger(result.exitCode) && result.exitCode >= 0) options.onProcessStopped?.(); const settled = applyRunDispositionSeam(result, options.settleRunDisposition); if (runLogTail) { await runLogTail.finish({ stdout: result.stdout, stderr: result.stderr }); @@ -912,7 +916,7 @@ export async function runAdapterExecutionTargetProcess( ? sanitizeRemoteExecutionEnv(options.env) : options.env; - return await runChildProcess(runId, command, args, { + const result = await runChildProcess(runId, command, args, { cwd: options.cwd, env, stdin: options.stdin, @@ -924,6 +928,13 @@ export async function runAdapterExecutionTargetProcess( localProcessSandbox: target?.kind === "local" || !target ? options.localProcessSandbox : null, remoteExecution: adapterExecutionTargetToRemoteSpec(target), }); + // Closing an SSH client on timeout/disconnect does not prove the remote + // provider exited. SSH status 255 is transport failure, never a stop receipt. + if (!target || target.kind === "local" || + (!result.timedOut && !result.signal && result.exitCode !== null && result.exitCode >= 0 && result.exitCode < 255)) { + options.onProcessStopped?.(); + } + return result; } export async function runAdapterExecutionTargetShellCommand( diff --git a/packages/adapter-utils/src/provider-stopped-boundary.test.ts b/packages/adapter-utils/src/provider-stopped-boundary.test.ts new file mode 100644 index 00000000000..6fabb28aa8d --- /dev/null +++ b/packages/adapter-utils/src/provider-stopped-boundary.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it, vi, afterEach } from "vitest"; +import * as serverUtils from "./server-utils.js"; +import { runAdapterExecutionTargetProcess } from "./execution-target.js"; +import { createProviderStoppedBoundary } from "./provider-stopped-boundary.js"; + +const result = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: 12, startedAt: "now" }; +const options = (onProcessStopped: () => void) => ({ cwd: "/tmp", env: {}, timeoutSec: 1, graceSec: 1, onLog: async () => {}, onProcessStopped }); +const target = (execute: () => Promise) => ({ kind: "remote", transport: "sandbox", remoteCwd: "/workspace", runner: { execute } }) as any; +afterEach(() => vi.restoreAllMocks()); +describe("owned provider process stop boundary", () => { + it.each([0, 1])("notifies after a remote invocation returns an observed exit code %s", async exitCode => { + const stopped = vi.fn(); + await runAdapterExecutionTargetProcess("run", target(async () => ({ ...result, exitCode })), "provider", [], options(stopped)); + expect(stopped).toHaveBeenCalledOnce(); + }); + it("notifies after the local child closed on timeout", async () => { + vi.spyOn(serverUtils, "runChildProcess").mockResolvedValue({ ...result, exitCode: null, signal: "SIGTERM", timedOut: true }); + const stopped = vi.fn(); + await runAdapterExecutionTargetProcess("run", null, "provider", [], options(stopped)); + expect(stopped).toHaveBeenCalledOnce(); + }); + it.each([ + { ...result, exitCode: null, timedOut: true }, + { ...result, exitCode: null, timedOut: false }, + ])("does not treat an incomplete remote execution as a stop receipt", async incomplete => { + const stopped = vi.fn(); + await runAdapterExecutionTargetProcess("run", target(async () => incomplete), "provider", [], options(stopped)); + expect(stopped).not.toHaveBeenCalled(); + }); + it("does not infer remote termination from a transport rejection", async () => { + const stopped = vi.fn(); + await expect(runAdapterExecutionTargetProcess("run", target(async () => { throw new Error("transport lost"); }), "provider", [], options(stopped))).rejects.toThrow("transport lost"); + expect(stopped).not.toHaveBeenCalled(); + }); +}); + +describe("final invocation instruction collection fence", () => { + it("collects once, after the last completed attempt, before restore", async () => { + const order: string[] = []; + const fence = createProviderStoppedBoundary(async () => { order.push("collect"); }); + fence.beginInvocation()(); + expect(order).toEqual([]); + fence.beginInvocation()(); + await fence.collectBeforeRestore(); + order.push("restore"); + await fence.collectBeforeRestore(); + expect(order).toEqual(["collect", "restore"]); + }); + it("an uncertain later attempt invalidates an earlier stop, including late callbacks", async () => { + const collect = vi.fn(async () => {}); + const fence = createProviderStoppedBoundary(collect); + const old = fence.beginInvocation(); old(); + fence.beginInvocation(); old(); + await fence.collectBeforeRestore(); + expect(collect).not.toHaveBeenCalled(); + }); + it("does not collect before any provider invocation", async () => { + const collect = vi.fn(async () => {}); + await createProviderStoppedBoundary(collect).collectBeforeRestore(); + expect(collect).not.toHaveBeenCalled(); + }); + it("does not infer SSH provider termination from transport exit 255", async () => { + vi.spyOn(serverUtils, "runChildProcess").mockResolvedValue({ ...result, exitCode: 255 }); + const stopped = vi.fn(); + await runAdapterExecutionTargetProcess("run", { kind: "remote", transport: "ssh", spec: { host: "fixture" }, remoteCwd: "/workspace" } as any, "provider", [], options(stopped)); + expect(stopped).not.toHaveBeenCalled(); + }); + it("retains the stop observation when a subsequent log-tail flush throws", async () => { + const stopped = vi.fn(); + await expect(runAdapterExecutionTargetProcess("run", target(async () => result), "provider", [], { + ...options(stopped), runLogTail: { create: () => ({ wrapCommand: (command: string, args: string[]) => ({ command, args }), start: () => {}, finish: async () => { throw new Error("log flush failed"); }, abort: async () => {} }) } as any, + })).rejects.toThrow("log flush failed"); + expect(stopped).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/adapter-utils/src/provider-stopped-boundary.ts b/packages/adapter-utils/src/provider-stopped-boundary.ts new file mode 100644 index 00000000000..be2e9101214 --- /dev/null +++ b/packages/adapter-utils/src/provider-stopped-boundary.ts @@ -0,0 +1,19 @@ +/** Per-adapter invocation fence. A failed later attempt invalidates an earlier + * stop receipt; a late callback from that earlier attempt cannot restore it. */ +export function createProviderStoppedBoundary(onProviderStopped?: () => Promise) { + let invocation = 0; + let stopped = false; + let collected = false; + return { + beginInvocation(): () => void { + const current = ++invocation; + stopped = false; + return () => { if (invocation === current) stopped = true; }; + }, + async collectBeforeRestore() { + if (!stopped || collected || !onProviderStopped) return; + collected = true; + await onProviderStopped(); + }, + }; +} diff --git a/packages/adapter-utils/src/types.ts b/packages/adapter-utils/src/types.ts index 541f7a6b683..50b3653c24b 100644 --- a/packages/adapter-utils/src/types.ts +++ b/packages/adapter-utils/src/types.ts @@ -202,6 +202,9 @@ export interface AdapterExecutionContext { /** Host-owned stop of this run's sandbox during setup or direct CLI execution. Resolves only after * provider termination is verified; never accepts an agent-selected lease. */ stopRemoteStartup?: () => Promise; + /** Host-owned collection after the final provider invocation is confirmed stopped, + * before remote workspace restore or disposal. Never call on an unverified timeout. */ + onProviderStopped?: () => Promise; /** Server-owned, actor-attributed snapshot also rendered by legacy wake prompts. */ executionContinuation?: ExecutionContinuationEnvelope | null; runId: string; diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 8d8291fad6d..3a51c65fb68 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -1,3 +1,4 @@ +import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; import fs from "node:fs/promises"; import path from "node:path"; import { fileURLToPath } from "node:url"; @@ -397,6 +398,7 @@ export async function runClaudeLogin(input: { } export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const engineSelection = await resolveClaudeExecutionEngineForRun(ctx); if (engineSelection.unavailableReason) { return { @@ -961,6 +963,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise ); } - async function runTeardown(input: { sandboxAuth: string; hostAuth: string }) { + async function runTeardown(input: { sandboxAuth: string; hostAuth: string; onProviderStopped?: () => Promise }) { const rootDir = await mkdtemp( path.join(os.tmpdir(), "paperclip-codex-copyback-e2e-"), ); @@ -138,6 +138,7 @@ describe("codex execute — outbound auth copy-back restore contribution", () => const executionResult = await execute({ runId: "run-copyback-e2e", + onProviderStopped: input.onProviderStopped, agent: { id: "agent-1", companyId: "company-1", @@ -181,6 +182,24 @@ describe("codex execute — outbound auth copy-back restore contribution", () => }; } + it("collects stopped-provider instruction edits before a throwing remote restore", async () => { + const order: string[] = []; + prepareAdapterExecutionTargetRuntime.mockImplementationOnce(async () => ({ + target: { kind: "remote", transport: "ssh" }, workspaceRemoteDir: "/remote/workspace", + runtimeRootDir: REMOTE_RUNTIME_ROOT, assetDirs: { home: `${REMOTE_RUNTIME_ROOT}/home` }, + restoreWorkspace: async () => { order.push("restore"); throw new Error("restore failed"); }, + })); + await expect(runTeardown({ sandboxAuth: "{}", hostAuth: "{}", onProviderStopped: async () => { order.push("collect"); } })).rejects.toThrow("restore failed"); + expect(order).toEqual(["collect", "restore"]); + }); + + it("collects after a failed provider exit before restoring its workspace", async () => { + runChildProcess.mockResolvedValueOnce({ exitCode: 1, signal: null, timedOut: false, stdout: "", stderr: "provider failed", pid: 321, startedAt: new Date().toISOString() }); + const collected = vi.fn(async () => {}); + await runTeardown({ sandboxAuth: "{}", hostAuth: "{}", onProviderStopped: collected }); + expect(collected).toHaveBeenCalledOnce(); + }); + it("declares a Codex `home` asset carrying both inbound provision and outbound restore contributions", async () => { await runTeardown({ sandboxAuth: subscriptionAuth({ accountId: "acct", lastRefresh: "2026-07-09T01:00:00Z", marker: "s" }), diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index 8e8e4954e82..32b09e048c4 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1,3 +1,4 @@ +import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; import fs from "node:fs/promises"; import path from "node:path"; import { fileURLToPath } from "node:url"; @@ -567,6 +568,7 @@ export async function ensureCodexSkillsInjected( } export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const engineSelection = await resolveCodexExecutionEngineForRun(ctx); if (engineSelection.unavailableReason) { return { @@ -1321,6 +1323,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise }) => { runnerState.commands.push(input.command); - if (input.command === "sh") { - return { - exitCode: 0, - signal: null, - timedOut: false, - stdout: "", - stderr: "", - pid: 555, - startedAt: new Date().toISOString(), - }; - } - + // Exercise actual bounded file reads during managed-home restoration; + // reporting empty success for every shell command hides missing bytes. return runChildProcess(`cursor-fresh-lease-${runnerState.commands.length}`, input.command, input.args ?? [], { cwd: remoteWorkspace, - env: input.env ?? {}, + env: { ...input.env, PATH: `${input.env?.PATH ?? ""}:/usr/bin:/bin` }, timeoutSec: 30, graceSec: 5, onLog: async () => {}, diff --git a/packages/adapters/cursor-local/src/server/execute.ts b/packages/adapters/cursor-local/src/server/execute.ts index 872d3779c75..083e5ceef2d 100644 --- a/packages/adapters/cursor-local/src/server/execute.ts +++ b/packages/adapters/cursor-local/src/server/execute.ts @@ -1,3 +1,4 @@ +import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -199,6 +200,7 @@ export async function ensureCursorSkillsInjected( } export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const { runId, agent, runtime, config, context, onLog, onMeta, onSpawn, authToken } = ctx; const executionTarget = readAdapterExecutionTarget({ executionTarget: ctx.executionTarget, @@ -655,6 +657,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const engineSelection = await resolveGeminiExecutionEngineForRun(ctx); if (engineSelection.unavailableReason) { return { @@ -638,6 +640,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const { runId, agent, runtime, config, context, onLog, onMeta, onSpawn, authToken } = ctx; const executionTarget = readAdapterExecutionTarget({ executionTarget: ctx.executionTarget, @@ -616,6 +618,7 @@ async function executeTurn(ctx: AdapterExecutionContext): Promise { await restoreRemoteWorkspace?.(); }); + return await withWorkspaceRestore(executeTurn, async () => { await providerStop.collectBeforeRestore(); await restoreRemoteWorkspace?.(); }); } finally { // Cleanup runs after settlement on both success and failure. if (stagedGrokHomeDir) { diff --git a/packages/adapters/kimi-local/src/server/execute.ts b/packages/adapters/kimi-local/src/server/execute.ts index 732f4ca0a2c..b54fa55bc58 100644 --- a/packages/adapters/kimi-local/src/server/execute.ts +++ b/packages/adapters/kimi-local/src/server/execute.ts @@ -1,3 +1,4 @@ +import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -186,6 +187,7 @@ async function buildKimiSkillsDir( } export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const engineSelection = await resolveKimiExecutionEngineForRun(ctx); if (engineSelection.unavailableReason) { return { @@ -605,6 +607,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise): Promise< } export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const { runId, agent, runtime, config, context, onLog, onMeta, onSpawn, authToken } = ctx; const executionTarget = readAdapterExecutionTarget({ executionTarget: ctx.executionTarget, @@ -641,6 +643,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const providerStop = createProviderStoppedBoundary(ctx.onProviderStopped); const { runId, agent, runtime, config, context, onLog, onMeta, onSpawn, authToken } = ctx; const executionTarget = readAdapterExecutionTarget({ executionTarget: ctx.executionTarget, @@ -733,6 +735,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise