fix(adapter-utils): fail fast when the sandbox control channel is lost mid-turn (#13158)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Adapter utilities run agent turns and report their results to the control plane > - A lost sandbox control channel can leave an agent turn without a result > - The host then waits for the full adapter timeout instead of reporting the loss > - This pull request adds a push loss signal and a bounded host wait > - The benefit is a prompt failure terminal when the agent stops answering ## Linked Issues or Issue Description **What happened?** A sandbox control channel loss during an Agent Client Protocol turn left the host waiting for the four-hour adapter execution timeout. **Expected behavior** The host should detect the terminal channel loss, stop the turn, and report a safe failure without waiting for the agent. **Steps to reproduce** 1. Start an Agent Client Protocol turn through a sandbox adapter. 2. Close the duplex control channel while the turn remains active. 3. Observe the host response before the adapter timeout expires. **Paperclip version or commit** Test the pull request commit set at `10b6bbc5525a79fd575298607dd5a25ae448fc8a`. **Deployment mode** The change applies to sandbox-backed adapter execution. ## What Changed - Add `onLoss(listener)` to the duplex bridge handle. - Register the loss listener at turn start and read losses latched before turn start. - Cancel the turn on loss and arm a 30-second host deadline. - Close the stream locally when the deadline wins and create a host terminal. - Derive the public error from the closed `DuplexLossReason` enum. - Add tests for loss order, cancellation, timeout, and safe error output. ## Verification - Run `pnpm --filter @paperclipai/adapter-utils exec tsc --noEmit`. - Run `pnpm --filter @paperclipai/adapter-utils exec vitest run src/acpx-engine/execute.test.ts -t "run-disposition seam"`. - Confirm that the full pull request workflow passes. ## Risks The new deadline changes a lost-channel path from a long wait to a host-built failure after 30 seconds. Orderly completion keeps its existing behavior. The deadline race against a pending `turn.result` has no direct test. ## Model Used OpenAI Codex, GPT-5, with tool use and code execution. The runtime does not expose a more specific deployment version or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs: #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
60ee13a0f7
commit
d1ba17eeca
|
|
@ -18,6 +18,14 @@ export const ACPX_HANDSHAKE_TIMEOUT_MS = 60_000;
|
|||
// of a channel loss.
|
||||
export const ACPX_HANDSHAKE_TRANSPORT_POLL_MS = 250;
|
||||
|
||||
// The bound on how long the host waits, after a latched terminal sandbox
|
||||
// duplex-channel loss, for the agent to answer the `turn.cancel()` request.
|
||||
// `cancel()` only asks the agent to end the turn; it does not end the turn by
|
||||
// itself. An agent that stopped answering never honors it, so this deadline
|
||||
// is the host-side bound that ends the run without the agent's help. It is
|
||||
// much smaller than the whole-adapter execution timeout.
|
||||
export const ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS = 30_000;
|
||||
|
||||
export const ACPX_ADAPTER_AGENT_IDS = {
|
||||
claude_local: "claude",
|
||||
codex_local: "codex",
|
||||
|
|
|
|||
|
|
@ -6129,6 +6129,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
let lossOrdered = false;
|
||||
let lossReason: string | null = null;
|
||||
let completionOrdered = false;
|
||||
let lossListener: ((reason: string) => void) | null = null;
|
||||
const readDisposition = () => ({ failed: lossOrdered, lossReason });
|
||||
const markOrderlyCompletion = vi.fn(() => {
|
||||
if (completionOrdered || lossOrdered) return;
|
||||
|
|
@ -6138,6 +6139,12 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
markOrderlyCompletion();
|
||||
return readDisposition();
|
||||
});
|
||||
const onLoss = vi.fn((listener: (reason: string) => void) => {
|
||||
lossListener = listener;
|
||||
return () => {
|
||||
if (lossListener === listener) lossListener = null;
|
||||
};
|
||||
});
|
||||
const stop = vi.fn(async () => {});
|
||||
const handle = {
|
||||
env: {
|
||||
|
|
@ -6148,19 +6155,25 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
readRunDisposition: () => readDisposition(),
|
||||
settleRunDisposition,
|
||||
markOrderlyCompletion,
|
||||
onLoss,
|
||||
stop,
|
||||
};
|
||||
return {
|
||||
handle,
|
||||
markOrderlyCompletion,
|
||||
settleRunDisposition,
|
||||
onLoss,
|
||||
readDisposition,
|
||||
// Record the first ordered loss. A loss ordered after a completion, or a
|
||||
// second loss, is a no-op — the same rule the real transport applies.
|
||||
// A loss that latches here (the first ordered call) also pushes the
|
||||
// reason to the one registered listener, the same way the real HTTP/2
|
||||
// transport's disposition latch does.
|
||||
emitLoss: (reason: string) => {
|
||||
if (lossOrdered || completionOrdered) return;
|
||||
lossOrdered = true;
|
||||
lossReason = reason;
|
||||
lossListener?.(reason);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -6215,6 +6228,81 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
};
|
||||
}
|
||||
|
||||
// A runtime whose one turn never resolves on its own — it hangs exactly
|
||||
// like a turn whose sandbox duplex channel died mid-turn produces no
|
||||
// terminal result. The turn only ends when something calls `cancel()`, the
|
||||
// same mechanism the push seam calls. `onCancel` observes each call.
|
||||
function hangingTurnRuntime(onCancel: (reason: string | undefined) => void) {
|
||||
let release: (() => void) | null = null;
|
||||
const released = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
return {
|
||||
ensureSession: async () => ({
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
}),
|
||||
startTurn: () => ({
|
||||
events: (async function* () {
|
||||
await released;
|
||||
})(),
|
||||
result: (async () => {
|
||||
await released;
|
||||
return { status: "cancelled" as const, stopReason: "cancelled" };
|
||||
})(),
|
||||
cancel: async (input?: { reason?: string }) => {
|
||||
onCancel(input?.reason);
|
||||
release?.();
|
||||
},
|
||||
}),
|
||||
setConfigOption: async () => {},
|
||||
close: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
// A runtime whose one turn hangs exactly like the real `acpx` shape: its
|
||||
// `cancel()` only sends the cancel request and returns. It does NOT settle
|
||||
// the turn — neither `events` nor `result` ever resolves on its own.
|
||||
// `closeStream()` ends the event drain locally, with no agent cooperation,
|
||||
// the same way the real runtime's does; it still leaves `result` pending.
|
||||
// This is the sensitivity control for the fail-fast deadline: only the
|
||||
// deadline, not the cancel request, can end this turn.
|
||||
function unresponsiveCancelTurnRuntime(input: {
|
||||
onCancel: (reason: string | undefined) => void;
|
||||
onCloseStream: (reason: string | undefined) => void;
|
||||
}) {
|
||||
let endEvents: (() => void) | null = null;
|
||||
const eventsEnded = new Promise<void>((resolve) => {
|
||||
endEvents = resolve;
|
||||
});
|
||||
return {
|
||||
ensureSession: async () => ({
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
}),
|
||||
startTurn: () => ({
|
||||
events: (async function* () {
|
||||
await eventsEnded;
|
||||
})(),
|
||||
// Never settles on its own. The real acpx result settles only when
|
||||
// the provider process returns or rejects, bounded by the adapter
|
||||
// execution timeout — not by a `session/cancel` request.
|
||||
result: new Promise<never>(() => {}),
|
||||
cancel: async (reasonInput?: { reason?: string }) => {
|
||||
input.onCancel(reasonInput?.reason);
|
||||
},
|
||||
closeStream: async (reasonInput?: { reason?: string }) => {
|
||||
input.onCloseStream(reasonInput?.reason);
|
||||
endEvents?.();
|
||||
},
|
||||
}),
|
||||
setConfigOption: async () => {},
|
||||
close: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
async function setupRemoteSandbox() {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
|
|
@ -6238,6 +6326,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
handle: unknown,
|
||||
runtime: unknown,
|
||||
sandbox: Awaited<ReturnType<typeof setupRemoteSandbox>>,
|
||||
deps: Partial<AcpxEngineExecutorOptions> = {},
|
||||
) {
|
||||
vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(
|
||||
async () => handle as never,
|
||||
|
|
@ -6247,6 +6336,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
);
|
||||
const execute = createAcpxEngineExecutor({
|
||||
createRuntime: () => runtime as never,
|
||||
...deps,
|
||||
});
|
||||
return await execute({
|
||||
runId: "run-duplex-seam",
|
||||
|
|
@ -6342,6 +6432,25 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
expect(fake.readDisposition().failed).toBe(false);
|
||||
});
|
||||
|
||||
it("keeps duplex_channel_lost precedence when the loss latches before a failed terminal", async () => {
|
||||
const sandbox = await setupRemoteSandbox();
|
||||
const fake = createFakeBridgeHandle();
|
||||
// Latch the loss before the ACP terminal resolves, and the terminal
|
||||
// itself also reports a provider failure.
|
||||
const runtime = runtimeWithFailedResult(() => fake.emitLoss("provider_exit"));
|
||||
|
||||
const result = await runRemote(fake.handle, runtime, sandbox);
|
||||
|
||||
expect(result.exitCode).not.toBe(0);
|
||||
// The duplex loss reason wins over the provider's own failed terminal.
|
||||
expect(result.errorCode).toBe("duplex_channel_lost");
|
||||
// The message carries only the typed loss reason, not the raw provider
|
||||
// failure text.
|
||||
expect(result.errorMessage).toContain("provider_exit");
|
||||
expect(result.errorMessage).not.toContain("agent failed");
|
||||
expect(result.resultJson).toMatchObject({ status: "failed" });
|
||||
});
|
||||
|
||||
it("releases the runtime locally and places no remote close call once the duplex channel is lost", async () => {
|
||||
const sandbox = await setupRemoteSandbox();
|
||||
const fake = createFakeBridgeHandle();
|
||||
|
|
@ -6435,6 +6544,170 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () =>
|
|||
expect(result.errorCode).not.toBe("acpx_session_init_failed");
|
||||
expect(result.errorCode).not.toBe("acpx_handshake_timeout");
|
||||
}, 10000);
|
||||
|
||||
it("aborts an in-flight turn and fails the run when the duplex channel latches a loss mid-turn", async () => {
|
||||
const sandbox = await setupRemoteSandbox();
|
||||
const fake = createFakeBridgeHandle();
|
||||
const cancelReasons: (string | undefined)[] = [];
|
||||
// This turn never returns a terminal result on its own: without a push
|
||||
// seam it would wait for the wall-clock adapter execution timeout. It
|
||||
// ends only once something calls `cancel()`.
|
||||
const runtime = hangingTurnRuntime((reason) => cancelReasons.push(reason));
|
||||
|
||||
const resultPromise = runRemote(fake.handle, runtime, sandbox);
|
||||
// Wait until the turn registers its loss listener, then latch the loss —
|
||||
// the same order a real mid-turn channel death follows: the turn starts,
|
||||
// then later the channel is lost.
|
||||
await vi.waitFor(() => expect(fake.onLoss).toHaveBeenCalled());
|
||||
fake.emitLoss("provider_exit");
|
||||
|
||||
const result = await resultPromise;
|
||||
|
||||
// The push seam cancelled the hanging turn instead of waiting for the
|
||||
// turn to return a terminal result on its own, so the run ends promptly
|
||||
// instead of waiting for the adapter execution timeout.
|
||||
expect(cancelReasons).toHaveLength(1);
|
||||
expect(result.exitCode).not.toBe(0);
|
||||
expect(result.errorCode).toBe("duplex_channel_lost");
|
||||
// The failure message carries only the closed loss-reason enum, never
|
||||
// raw provider text.
|
||||
expect(result.errorMessage).toContain("provider_exit");
|
||||
expect(result.resultJson).toMatchObject({ status: "failed" });
|
||||
}, 5000);
|
||||
|
||||
it("bounds the wait with a deadline when a latched loss cancel gets no agent cooperation", async () => {
|
||||
const sandbox = await setupRemoteSandbox();
|
||||
const fake = createFakeBridgeHandle();
|
||||
const cancelReasons: (string | undefined)[] = [];
|
||||
const closeStreamReasons: (string | undefined)[] = [];
|
||||
// The real acpx shape: `cancel()` only requests cancellation and returns.
|
||||
// It settles neither `events` nor `result`. Only the fail-fast deadline
|
||||
// can end this turn.
|
||||
const runtime = unresponsiveCancelTurnRuntime({
|
||||
onCancel: (reason) => cancelReasons.push(reason),
|
||||
onCloseStream: (reason) => closeStreamReasons.push(reason),
|
||||
});
|
||||
|
||||
const resultPromise = runRemote(fake.handle, runtime, sandbox, {
|
||||
// Small and fake-time-free: real-timer test, so the deadline must stay
|
||||
// short enough to run fast without waiting 60 real seconds.
|
||||
duplexLossCancelDeadlineMs: 25,
|
||||
});
|
||||
await vi.waitFor(() => expect(fake.onLoss).toHaveBeenCalled());
|
||||
fake.emitLoss("provider_exit");
|
||||
|
||||
const result = await resultPromise;
|
||||
|
||||
// The seam still tried the cooperative cancel first.
|
||||
expect(cancelReasons).toHaveLength(1);
|
||||
// The agent never answered the cancel, so the deadline ended the event
|
||||
// drain locally instead of waiting for it.
|
||||
expect(closeStreamReasons).toHaveLength(1);
|
||||
// The run reached a failure terminal within the deadline, even though
|
||||
// neither `events` nor `result` ever settled on their own.
|
||||
expect(result.exitCode).not.toBe(0);
|
||||
expect(result.errorCode).toBe("duplex_channel_lost");
|
||||
// The failure message carries only the closed loss-reason enum, never
|
||||
// raw provider text.
|
||||
expect(result.errorMessage).toContain("provider_exit");
|
||||
expect(result.resultJson).toMatchObject({ status: "failed" });
|
||||
}, 5000);
|
||||
|
||||
it("awaits stream closure and the event drain before finalizing a duplex loss deadline", async () => {
|
||||
const sandbox = await setupRemoteSandbox();
|
||||
const fake = createFakeBridgeHandle();
|
||||
let endEvents: (() => void) | null = null;
|
||||
const eventsEnded = new Promise<void>((resolve) => {
|
||||
endEvents = resolve;
|
||||
});
|
||||
let releaseCloseStream!: () => void;
|
||||
const closeStreamGate = new Promise<void>((resolve) => {
|
||||
releaseCloseStream = resolve;
|
||||
});
|
||||
let closeStreamCalls = 0;
|
||||
// `closeStream()` stays pending on a gate the test controls, and only
|
||||
// ends the event drain once the test releases that gate. If the run
|
||||
// finalizes before the gate opens, the seam did not wait for the close
|
||||
// call, so a late event on this drain could still land after the result.
|
||||
const runtime = {
|
||||
ensureSession: async () => ({
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
}),
|
||||
startTurn: () => ({
|
||||
events: (async function* () {
|
||||
await eventsEnded;
|
||||
})(),
|
||||
result: new Promise<never>(() => {}),
|
||||
cancel: async () => {},
|
||||
closeStream: async () => {
|
||||
closeStreamCalls += 1;
|
||||
await closeStreamGate;
|
||||
endEvents?.();
|
||||
},
|
||||
}),
|
||||
setConfigOption: async () => {},
|
||||
close: async () => {},
|
||||
};
|
||||
|
||||
const resultPromise = runRemote(fake.handle, runtime, sandbox, {
|
||||
duplexLossCancelDeadlineMs: 25,
|
||||
});
|
||||
await vi.waitFor(() => expect(fake.onLoss).toHaveBeenCalled());
|
||||
fake.emitLoss("provider_exit");
|
||||
|
||||
await vi.waitFor(() => expect(closeStreamCalls).toBe(1));
|
||||
// The close call has not resolved yet, so the run must still be pending.
|
||||
let settled = false;
|
||||
void resultPromise.then(() => {
|
||||
settled = true;
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
expect(settled).toBe(false);
|
||||
|
||||
releaseCloseStream();
|
||||
const result = await resultPromise;
|
||||
|
||||
expect(result.errorCode).toBe("duplex_channel_lost");
|
||||
}, 5000);
|
||||
|
||||
it("does not abort or fail an already-completed run when the duplex channel loses after an orderly completion", async () => {
|
||||
const sandbox = await setupRemoteSandbox();
|
||||
const fake = createFakeBridgeHandle();
|
||||
const cancelReasons: (string | undefined)[] = [];
|
||||
const runtime = {
|
||||
ensureSession: async () => ({
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
}),
|
||||
startTurn: () => ({
|
||||
events: (async function* () {
|
||||
yield { type: "done", stopReason: "end_turn" };
|
||||
})(),
|
||||
result: Promise.resolve({ status: "completed" as const, stopReason: "end_turn" }),
|
||||
cancel: async (input?: { reason?: string }) => {
|
||||
cancelReasons.push(input?.reason);
|
||||
},
|
||||
}),
|
||||
setConfigOption: async () => {},
|
||||
close: async () => {},
|
||||
};
|
||||
|
||||
const result = await runRemote(fake.handle, runtime, sandbox);
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(result.errorCode ?? null).toBeNull();
|
||||
|
||||
// The channel dies only after the turn already completed cleanly. The
|
||||
// loss listener the turn registered is still live at this point, but the
|
||||
// latch already marked the orderly completion, so the loss cannot relatch
|
||||
// and must never reach a cancel call on the (already-finished) turn.
|
||||
fake.emitLoss("provider_exit");
|
||||
|
||||
expect(cancelReasons).toHaveLength(0);
|
||||
expect(fake.readDisposition().failed).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("ACPX startup handshake guard and late-completion fence", () => {
|
||||
|
|
|
|||
|
|
@ -92,6 +92,7 @@ import {
|
|||
type AcpSessionStore,
|
||||
} from "acpx/runtime";
|
||||
import {
|
||||
ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS,
|
||||
ACPX_HANDSHAKE_TIMEOUT_MS,
|
||||
ACPX_HANDSHAKE_TRANSPORT_POLL_MS,
|
||||
DEFAULT_ACP_ENGINE_AGENT,
|
||||
|
|
@ -342,6 +343,14 @@ export interface AcpxRemoteManagedHomeResult {
|
|||
export interface AcpxEngineExecutorOptions {
|
||||
createRuntime?: AcpxRuntimeFactory;
|
||||
now?: () => number;
|
||||
/**
|
||||
* The bound on how long the fail-fast seam waits for a cooperative
|
||||
* `turn.cancel()` after a latched terminal sandbox duplex-channel loss,
|
||||
* before it ends the turn without the agent's help. Defaults to
|
||||
* {@link ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS}. Tests inject a small value
|
||||
* to drive the deadline without real time.
|
||||
*/
|
||||
duplexLossCancelDeadlineMs?: number;
|
||||
warmHandles?: Map<string, RuntimeCacheEntry>;
|
||||
/**
|
||||
* Per-session staged-runtime cache for the remote runner-backed lane (PR 3).
|
||||
|
|
@ -3704,6 +3713,7 @@ function openTurnSpan(
|
|||
export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
||||
const createRuntime = deps.createRuntime ?? createAcpRuntime;
|
||||
const now = deps.now ?? (() => Date.now());
|
||||
const duplexLossCancelDeadlineMs = deps.duplexLossCancelDeadlineMs ?? ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS;
|
||||
const warmHandles = deps.warmHandles ?? defaultWarmHandles;
|
||||
const stagedRuntimes = deps.stagedRuntimes ?? defaultStagedRuntimes;
|
||||
const stagingLocks = deps.stagingLocks ?? defaultStagingLocks;
|
||||
|
|
@ -3800,6 +3810,24 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
let releaseStagingLease: (() => void) | null = null;
|
||||
let stopTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let removeStopListener: (() => void) | undefined;
|
||||
// Unregisters the sandbox duplex bridge's loss listener (below, in
|
||||
// `stepTurnStart`). Set only on a sandbox target whose bridge exposes
|
||||
// `onLoss`; stays undefined everywhere else, so the cleanup call is a
|
||||
// no-op there.
|
||||
let removeLossListener: (() => void) | undefined;
|
||||
// Bounds the wait after a latched terminal duplex loss so a silent agent
|
||||
// cannot hold the run open on the cooperative `turn.cancel()` request
|
||||
// alone. `stepTurnStart` arms `lossDeadlineTimer` the moment a loss
|
||||
// latches; it stays undefined everywhere else, so the cleanup call below
|
||||
// is a no-op there. `stepEventRelay` races the turn against
|
||||
// `lossDeadline` and, once it fires, ends the event drain and hands
|
||||
// `turnFinalize` a host-built terminal instead of the agent's.
|
||||
let lossDeadlineTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let lossDeadlineTripped = false;
|
||||
let resolveLossDeadline: (() => void) | undefined;
|
||||
const lossDeadline = new Promise<void>((resolve) => {
|
||||
resolveLossDeadline = resolve;
|
||||
});
|
||||
let forcedStop = false;
|
||||
let runtimeStopConfirmed = false;
|
||||
let safeInterruptedSession = false;
|
||||
|
|
@ -4606,6 +4634,36 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
signal,
|
||||
});
|
||||
activeTurn = turn;
|
||||
// A latched sandbox duplex-channel loss otherwise has no way to reach
|
||||
// this turn: the bridge only exposes a pull read, and the engine
|
||||
// pulls it at the terminal-finalization boundary, which runs only
|
||||
// after the turn already returned a terminal result. A channel that
|
||||
// dies mid-turn then leaves the turn with no terminal result to
|
||||
// return, so it waits for the wall-clock adapter execution timeout
|
||||
// instead of failing fast. Cancel the turn the moment a terminal loss
|
||||
// latches — whether it latches from here on, or already latched
|
||||
// before this turn started — so the turn returns a terminal result
|
||||
// right away. `turnFinalize` reads the same latch and builds the
|
||||
// failure from the typed loss reason alone.
|
||||
const bridge = prepared.paperclipBridge;
|
||||
if (bridge?.onLoss) {
|
||||
const cancelForLoss = (reason: DuplexLossReason) => {
|
||||
void turn.cancel({ reason: `paperclip sandbox duplex channel lost (${reason})` }).catch(() => {});
|
||||
// `cancel()` only asks the agent to end the turn; it does not end
|
||||
// the turn by itself. Start the fail-fast deadline the moment the
|
||||
// loss latches, so the run does not wait past this bound for an
|
||||
// agent that stopped answering.
|
||||
if (!lossDeadlineTimer && !lossDeadlineTripped) {
|
||||
lossDeadlineTimer = setTimeout(() => {
|
||||
lossDeadlineTripped = true;
|
||||
resolveLossDeadline?.();
|
||||
}, duplexLossCancelDeadlineMs);
|
||||
}
|
||||
};
|
||||
removeLossListener = bridge.onLoss(cancelForLoss);
|
||||
const alreadyLatched = bridge.readRunDisposition?.();
|
||||
if (alreadyLatched?.failed) cancelForLoss(alreadyLatched.lossReason ?? "other");
|
||||
}
|
||||
// ACP can resolve the turn before its provider exits. Keep the Stop
|
||||
// deadline armed through settlement, including provider cleanup.
|
||||
const armStopDeadline = () => {
|
||||
|
|
@ -4631,40 +4689,74 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
},
|
||||
};
|
||||
};
|
||||
// The host-built terminal `stepEventRelay` hands to `turnFinalize` once
|
||||
// the fail-fast deadline fires with no agent-supplied terminal. Its
|
||||
// `status` mirrors the shape a real cooperative cancel already
|
||||
// produces, so `turnFinalize` needs no change: it reads the latched
|
||||
// loss disposition, not this `stopReason`, to build the reported
|
||||
// failure and its message.
|
||||
const LOSS_DEADLINE_TERMINAL: AcpRuntimeTurnResult = {
|
||||
status: "cancelled",
|
||||
stopReason: "paperclip_duplex_loss_deadline",
|
||||
};
|
||||
const stepEventRelay = async (): Promise<AcpRuntimeTurnResult> => {
|
||||
const turn = activeTurn as AcpRuntimeTurn;
|
||||
const toolTitles = new Map<string, string>();
|
||||
for await (const event of turn.events) {
|
||||
// ACPX currently flattens client-side filesystem/terminal receipts
|
||||
// into status text. They cannot establish complete action outcomes.
|
||||
if (event.type === "status" && /^(fs|terminal)\//.test(event.text)) incompleteToolInventory = true;
|
||||
if (event.type === "tool_call") {
|
||||
if (!event.toolCallId) incompleteToolInventory = true;
|
||||
else {
|
||||
const previous = interruptionTools.get(event.toolCallId);
|
||||
interruptionTools.set(event.toolCallId, {
|
||||
kind: event.kind ?? previous?.kind,
|
||||
status: event.status ?? previous?.status,
|
||||
});
|
||||
const drainEvents = (async (): Promise<void> => {
|
||||
for await (const event of turn.events) {
|
||||
// ACPX currently flattens client-side filesystem/terminal receipts
|
||||
// into status text. They cannot establish complete action outcomes.
|
||||
if (event.type === "status" && /^(fs|terminal)\//.test(event.text)) incompleteToolInventory = true;
|
||||
if (event.type === "tool_call") {
|
||||
if (!event.toolCallId) incompleteToolInventory = true;
|
||||
else {
|
||||
const previous = interruptionTools.get(event.toolCallId);
|
||||
interruptionTools.set(event.toolCallId, {
|
||||
kind: event.kind ?? previous?.kind,
|
||||
status: event.status ?? previous?.status,
|
||||
});
|
||||
}
|
||||
}
|
||||
if (event.type === "text_delta" && event.stream !== "thought") {
|
||||
currentOutputChunk.push(event.text);
|
||||
} else if (event.type === "tool_call" && event.tag !== "tool_call_update") {
|
||||
// ACP makes tool-call status optional. The normalized event tag is
|
||||
// the reliable boundary between an initial call and its updates,
|
||||
// so a statusless initial call must still end the preceding output
|
||||
// segment while updates must not create extra boundaries.
|
||||
flushOutputSegment();
|
||||
}
|
||||
if (event.type === "status" && event.tag === "usage_update") {
|
||||
eventBreakdown = event.breakdown ?? eventBreakdown;
|
||||
eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd;
|
||||
}
|
||||
await emitRuntimeEvent(ctx, event, toolTitles, prepared.coalescePlaceholderToolUpdates);
|
||||
}
|
||||
if (event.type === "text_delta" && event.stream !== "thought") {
|
||||
currentOutputChunk.push(event.text);
|
||||
} else if (event.type === "tool_call" && event.tag !== "tool_call_update") {
|
||||
// ACP makes tool-call status optional. The normalized event tag is
|
||||
// the reliable boundary between an initial call and its updates,
|
||||
// so a statusless initial call must still end the preceding output
|
||||
// segment while updates must not create extra boundaries.
|
||||
flushOutputSegment();
|
||||
}
|
||||
if (event.type === "status" && event.tag === "usage_update") {
|
||||
eventBreakdown = event.breakdown ?? eventBreakdown;
|
||||
eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd;
|
||||
}
|
||||
await emitRuntimeEvent(ctx, event, toolTitles, prepared.coalescePlaceholderToolUpdates);
|
||||
})();
|
||||
// A latched loss already asked the agent to cancel (above, in
|
||||
// `cancelForLoss`); that request settles neither `turn.events` nor
|
||||
// `turn.result` by itself. Race the event drain against the fail-fast
|
||||
// deadline so a silent agent cannot hold this wait open.
|
||||
const eventsEnded = await Promise.race([
|
||||
drainEvents.then(() => true as const),
|
||||
lossDeadline.then(() => false as const),
|
||||
]);
|
||||
if (!eventsEnded) {
|
||||
// The deadline won: stop waiting on the agent. `closeStream` ends
|
||||
// the event drain locally, with no agent cooperation required. Await
|
||||
// both the close call and the drain it unblocks before this step
|
||||
// returns, so no late runtime event can still mutate shared state
|
||||
// (output segments, tool inventory) after finalization reads it.
|
||||
await turn.closeStream({ reason: "paperclip duplex loss cancel deadline" }).catch(() => {});
|
||||
await drainEvents.catch(() => {});
|
||||
flushOutputSegment();
|
||||
return LOSS_DEADLINE_TERMINAL;
|
||||
}
|
||||
flushOutputSegment();
|
||||
return await turn.result;
|
||||
// `turn.result` settles only when the agent's provider process
|
||||
// returns or rejects; a latched loss that armed the deadline after
|
||||
// the event drain already ended must still bound this wait.
|
||||
return await Promise.race([turn.result, lossDeadline.then(() => LOSS_DEADLINE_TERMINAL)]);
|
||||
};
|
||||
const stepTurnFinalize = async (
|
||||
input: TurnFinalizeInput<AcpRuntimeTurnResult>,
|
||||
|
|
@ -4673,33 +4765,23 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
const terminal = input.terminal;
|
||||
const timedOut = input.timedOut;
|
||||
// Read the sandbox duplex control-channel disposition at the ACP
|
||||
// terminal-finalization boundary, before the bridge teardown. A control
|
||||
// channel that died mid-turn latches a failure with a typed loss reason;
|
||||
// a healthy channel or a normal-teardown loss reports a success. Only a
|
||||
// nominally completed, non-timed-out terminal is success-eligible, so the
|
||||
// seam reads the disposition only there. For that success-eligible
|
||||
// terminal the seam marks the host-observed orderly completion, so a later
|
||||
// teardown loss cannot flip the run to a failure. The file bridge path
|
||||
// never sets these methods, so the optional calls no-op there.
|
||||
// terminal-finalization boundary, before the bridge teardown, on every
|
||||
// terminal outcome. A control channel that died before this point
|
||||
// latches a failure with a typed loss reason; a healthy channel or a
|
||||
// normal-teardown loss reports a success. The read and the mark of the
|
||||
// host-observed orderly completion happen atomically in one broker
|
||||
// step, with no `await` between them, so a teardown loss cannot slip
|
||||
// in between. This stops a later teardown `channel_exit` from latching
|
||||
// a false loss. The mark no-ops once a loss already latched, so a real
|
||||
// mid-turn loss still fails the run — including a loss that arrived
|
||||
// through the in-flight-turn cancel this seam issues, which surfaces
|
||||
// here as a `cancelled` (not `completed`) terminal, not just through a
|
||||
// nominally completed terminal. The file bridge path never sets this
|
||||
// method, so the optional call no-ops there.
|
||||
let duplexLossReason: DuplexLossReason | null = null;
|
||||
if (terminal.status === "completed" && !timedOut) {
|
||||
// Success-eligible terminal. Atomically read the disposition and mark
|
||||
// the orderly completion in one broker step. No `await` separates the
|
||||
// read from the mark, so a teardown loss cannot slip in between them. A
|
||||
// latched loss fails the run closed; a healthy channel marks its
|
||||
// orderly completion, so a later teardown loss stays a normal teardown.
|
||||
const disposition = prepared.paperclipBridge?.settleRunDisposition?.() ?? null;
|
||||
if (disposition?.failed) {
|
||||
duplexLossReason = disposition.lossReason ?? "other";
|
||||
}
|
||||
} else {
|
||||
// Non-success-eligible terminal (failed, cancelled, or timed out). A
|
||||
// deliberate host teardown follows, so mark the orderly completion now.
|
||||
// This stops the teardown `channel_exit` from latching `lossSeq`, from
|
||||
// emitting a false loss event, and from incrementing the loss counters.
|
||||
// The mark no-ops once a loss latched, so a real mid-run loss still
|
||||
// fails the run.
|
||||
prepared.paperclipBridge?.markOrderlyCompletion?.();
|
||||
const disposition = prepared.paperclipBridge?.settleRunDisposition?.() ?? null;
|
||||
if (disposition?.failed) {
|
||||
duplexLossReason = disposition.lossReason ?? "other";
|
||||
}
|
||||
// A terminal that reports "completed" but whose duplex control channel
|
||||
// died before the completion is not a success. The seam fails it closed.
|
||||
|
|
@ -4778,12 +4860,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
signal: timedOut ? "SIGTERM" : null,
|
||||
timedOut,
|
||||
errorMessage,
|
||||
errorCode: terminal.status === "failed"
|
||||
? "acpx_turn_failed"
|
||||
: timedOut
|
||||
? "acpx_timeout"
|
||||
: channelLost
|
||||
? DUPLEX_CHANNEL_LOST_ERROR_CODE
|
||||
errorCode: timedOut
|
||||
? "acpx_timeout"
|
||||
: channelLost
|
||||
? DUPLEX_CHANNEL_LOST_ERROR_CODE
|
||||
: terminal.status === "failed"
|
||||
? "acpx_turn_failed"
|
||||
: null,
|
||||
sessionId: sessionHandle.backendSessionId ?? sessionHandle.runtimeSessionName,
|
||||
sessionParams: buildSessionParams({ prepared, handle: sessionHandle }),
|
||||
|
|
@ -4831,16 +4913,6 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
resources: emptyConsumed,
|
||||
};
|
||||
}
|
||||
if (terminal.status === "failed") {
|
||||
return {
|
||||
kind: "failed",
|
||||
cause: {
|
||||
kind: "turn_failed",
|
||||
error: terminal.error instanceof Error ? terminal.error : new Error(String(terminal.error)),
|
||||
},
|
||||
resources: emptyConsumed,
|
||||
};
|
||||
}
|
||||
if (terminal.status === "cancelled") {
|
||||
return {
|
||||
kind: "cancelled",
|
||||
|
|
@ -4848,10 +4920,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
resources: emptyConsumed,
|
||||
};
|
||||
}
|
||||
// A completed terminal whose duplex control channel died mid-turn returns
|
||||
// a failed completion, so the coordinator settles for a failure and the
|
||||
// reuse decision forbids a save. The message carries only the typed loss
|
||||
// reason, so no raw provider text rides the cause.
|
||||
// A duplex control-channel loss outranks a provider-reported failure or
|
||||
// completion: the loss reason explains why the provider terminal reads
|
||||
// the way it does, not the other way round. This also covers a
|
||||
// "completed" terminal whose channel died mid-turn. The message carries
|
||||
// only the typed loss reason, so no raw provider text rides the cause,
|
||||
// even when the provider terminal itself reports `failed`.
|
||||
if (channelLost) {
|
||||
return {
|
||||
kind: "failed",
|
||||
|
|
@ -4862,6 +4936,16 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
resources: emptyConsumed,
|
||||
};
|
||||
}
|
||||
if (terminal.status === "failed") {
|
||||
return {
|
||||
kind: "failed",
|
||||
cause: {
|
||||
kind: "turn_failed",
|
||||
error: terminal.error instanceof Error ? terminal.error : new Error(String(terminal.error)),
|
||||
},
|
||||
resources: emptyConsumed,
|
||||
};
|
||||
}
|
||||
return { kind: "finalized" };
|
||||
}
|
||||
const err = input.error;
|
||||
|
|
@ -5195,6 +5279,8 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
} finally {
|
||||
clearTimeout(stopTimer);
|
||||
removeStopListener?.();
|
||||
removeLossListener?.();
|
||||
clearTimeout(lossDeadlineTimer);
|
||||
// End the run root span exactly once, on every return and on a throw.
|
||||
runRootSpan.end(runFailed);
|
||||
// Release the per-session staging lease as the run's final act, AFTER the
|
||||
|
|
|
|||
|
|
@ -344,6 +344,20 @@ export interface AdapterExecutionTargetPaperclipBridgeHandle {
|
|||
* bridge path never sets it, so the method is absent there.
|
||||
*/
|
||||
markOrderlyCompletion?(): void;
|
||||
/**
|
||||
* Register a listener for a newly latched terminal loss. The listener
|
||||
* fires at most once, and only for a loss that flips the disposition to
|
||||
* failed — never for a clean channel end that orders after a
|
||||
* host-observed orderly completion. Returns a function that unregisters
|
||||
* the listener.
|
||||
*
|
||||
* The caller uses this to abort an in-flight Agent Client Protocol turn
|
||||
* the moment the channel dies, instead of waiting for the turn to return
|
||||
* a terminal result on its own (a dead channel can leave a turn with
|
||||
* nothing to return). The file bridge path never sets it, so the method
|
||||
* is absent there.
|
||||
*/
|
||||
onLoss?(listener: (reason: DuplexLossReason) => void): () => void;
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
|
|
@ -3661,12 +3675,19 @@ interface Http2RunDispositionLatch {
|
|||
markOrderlyCompletion(): void;
|
||||
/** Atomically mark the orderly completion and read the disposition. */
|
||||
settleRunDisposition(): DuplexBrokerRunDisposition;
|
||||
/**
|
||||
* Register a listener that fires once, only on the call to `recordLoss`
|
||||
* that actually latches a new terminal loss. Returns a function that
|
||||
* unregisters the listener.
|
||||
*/
|
||||
onLoss(listener: (reason: DuplexLossReason) => void): () => void;
|
||||
}
|
||||
|
||||
function createHttp2RunDispositionLatch(): Http2RunDispositionLatch {
|
||||
let lossOrdered = false;
|
||||
let lossReason: DuplexLossReason | null = null;
|
||||
let completionOrdered = false;
|
||||
let lossListener: ((reason: DuplexLossReason) => void) | null = null;
|
||||
const markOrderlyCompletion = (): void => {
|
||||
if (completionOrdered || lossOrdered) return;
|
||||
completionOrdered = true;
|
||||
|
|
@ -3679,6 +3700,7 @@ function createHttp2RunDispositionLatch(): Http2RunDispositionLatch {
|
|||
if (lossOrdered || completionOrdered) return false;
|
||||
lossOrdered = true;
|
||||
lossReason = reason;
|
||||
lossListener?.(reason);
|
||||
return true;
|
||||
},
|
||||
markOrderlyCompletion,
|
||||
|
|
@ -3686,6 +3708,12 @@ function createHttp2RunDispositionLatch(): Http2RunDispositionLatch {
|
|||
markOrderlyCompletion();
|
||||
return { failed: lossOrdered, lossReason };
|
||||
},
|
||||
onLoss(listener: (reason: DuplexLossReason) => void): () => void {
|
||||
lossListener = listener;
|
||||
return () => {
|
||||
if (lossListener === listener) lossListener = null;
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -4672,6 +4700,8 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
// the mark and a teardown loss cannot slip in between.
|
||||
settleRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.settleRunDisposition(),
|
||||
markOrderlyCompletion: (): void => dispositionLatch.markOrderlyCompletion(),
|
||||
onLoss: (listener: (reason: DuplexLossReason) => void): (() => void) =>
|
||||
dispositionLatch.onLoss(listener),
|
||||
stop: async () => {
|
||||
// Close the HTTP/2 server's sessions, then the channel, before
|
||||
// lease release, so no live provider session remains when the
|
||||
|
|
|
|||
Loading…
Reference in New Issue