mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 01:50:13 +01:00
fix(runner): retry a rejected runner suspension and log every cleanup quarantine (#19653)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - A native run's close must prove that its runner durably suspended
before the session's next run may start. If it cannot, the runtime
quarantines the session (`NativeSessionCloseUnrecoverableError`)
> - In a restart chaos storm, an ACPX Claude run succeeded and its
normal close quarantined its own session. The follow-up run on that
session failed 2 s later with `native_session_cleanup_quarantined`, and
no control-plane restart came between the two runs
> - The close sent `runner.suspend` once, the ACPX sidecar rejected it
right after the turn completed, and the barrier gave up at the first
rejected command. No log line showed the quarantine or its cause
> - This pull request retries a rejected suspension within the close
deadline, and logs every cleanup quarantine with its cause when it is
created
> - The benefit is that a successful run no longer blocks its own
follow-up because of a one-time suspension rejection, and any remaining
quarantine can be traced
## Linked Issues or Issue Description
**What happened?**
Run A (ACPX Claude) succeeded. The run log of its close shows:
- `runner did not prove durable suspension before checkpoint`
- `native_session_settlement_incomplete
{"runnerSuspended":false,"providerDrained":true,"suspensionState":{"commandStatus":"failed","runnerLifecycle":"suspended",...},"semanticTools":{"pending":[]},"finalProviderState":{"activeProviderTurnId":null,"providerSettled":true}}`
The runner journal shows `runner.drain` completed, then `runner.suspend`
failed with `failed to suspend ACPX provider: ACPX sidecar command
session.suspend was rejected (retryable=false,
classification=unclassified)`. The follow-up run B on the same session
was created 200 ms after A finished and failed 2 s later at admission
(`retryQuarantinedSessionCleanups`, operator recovery branch). The
server log had no line for the quarantine. One Retry completed the task.
**Steps to reproduce**
1. Run an ACPX task whose sidecar rejects the first `session.suspend`
after the turn completes.
2. Start a follow-up run on the same session.
**Expected behavior**
The close asks again, proves suspension, and the follow-up runs. If a
quarantine is still needed, it is logged with its cause.
**Actual behavior**
The close fails on the first rejection, quarantines the session, and the
follow-up fails with no logged cause.
## What Changed
- `runner/.../durable/runner.rs` (`process_command`): a failed
`runner.suspend`, on first execution or replay, now keeps the runner in
`Continue` instead of `Suspend`. Before, the runner recorded `suspended`
over a provider whose session was still open, then exited. The RC-34
journal shows exactly that: runner lifecycle `suspended`, ACPX provider
`session_open`. A failed `runner.shutdown` still stops the runner.
- `durable-prp-control-plane.ts`: a failed or rejected `runner.suspend`
now lifts the connection's terminal-lifecycle fence, so the next pending
command (the fresh suspend) is delivered on the same connection.
Completed suspensions and every shutdown keep the fence. New test:
"delivers a fresh suspension after a rejected one on the same
connection". Before the fix, the retry was never sent and the test timed
out waiting for the frame.
- `runnerd-codex-transport.ts` (`awaitRunnerSuspensionBarrier`): after a
rejected `runner.suspend`, the barrier sends a fresh suspend command, up
to 3 attempts, 250 ms apart, inside the same close deadline. An exited
runner or a spent budget still fails closed. A rejection changes no
state: the runner returns before it records suspension.
- `native-session-runtime.ts`: new
`onNativeSessionCleanupQuarantined(listener)`. It is called when a
cleanup quarantine is created or escalates to operator recovery, with
company, run, issue, session, `operatorRecoveryRequired`, a bounded
cause (error name, code, message) and bounded settlement facts.
- `server`: `logNativeCleanupQuarantines` logs each quarantine as a
warning. The subscription is process-wide and installed once, so closes
that fail during shutdown, after HTTP closes, are still logged.
## Verification
- Red before the change: `pnpm --filter @paperclipai/paperclip-runner
exec vitest run src/live/runnerd-codex-transport.test.ts -t "rejected
suspension"`.
- "retries a rejected suspension…" got `false`.
- "stops retrying…" saw 1 attempt instead of 3.
- Green after the change: both pass.
- Rust: new test
`rejected_suspend_keeps_the_runner_serving_its_controller` covers a
failed suspend, its replay, and an accepted retry with a fresh command.
`failed_lifecycle_commands_replay_their_terminal_transition` now expects
`Continue` for suspend and `Shutdown` for shutdown. The `pnpm run
check:runner` unit suites pass (559). `acpx_park_readiness` has two
timing tests (`< 500ms`) that fail at this host load on unchanged master
as well.
- New runtime test: "announces every cleanup quarantine with its cause
when it is created" (operator-required quarantine, cause and settlement
reported once). `src/native-session-runtime.test.ts` passes (212).
- The suspension cases in `runnerd-codex-transport.test.ts` pass, except
"proves local suspension after an event backlog… (1024 suffix deltas)".
That case also times out on unchanged master at this host load (27–35);
I confirmed this by stashing the change and rerunning.
- Server `tsc --noEmit` passes after the runner package build.
## Risks
- A provider that keeps rejecting suspension now gets three tries
instead of one, within the existing close deadline. The final outcome
and quarantine are unchanged.
- The cause of the ACPX sidecar's first rejection is not proven. The
logged evidence fits the sidecar's "not at a safe suspension point"
guard (`turnId`, tools or inputs still set) racing the just-completed
turn. This change makes close tolerate it. It does not change the
sidecar.
- Log volume: one warning per quarantine.
- The Rust change ships in the runner binary. Hosted images need a
rebuild to pick it up.
- After a failed suspend, the runner stays alive with its provider. If
every retry fails, close still treats suspension as unproven and kills
the owned runner process, exactly as before.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code
execution in Claude Code.
## 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
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
Refs #19532, #19609.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
30857db55e
commit
a894b11a2b
10 files changed
+400
-11
No files matched your search
@@ -99,6 +99,17 @@ impl CommandLifecycle {
|
||||
}
|
||||
}
|
||||
|
||||
/// A rejected suspension changed no provider state: the provider still
|
||||
/// owns its session. Keep the runner serving its controller so it can
|
||||
/// retry, instead of recording `suspended` over a live provider and
|
||||
/// exiting. A failed shutdown still stops the runner.
|
||||
fn after_failure(command: &Command) -> Self {
|
||||
match command.command_type.as_str() {
|
||||
"runner.suspend" => Self::Continue,
|
||||
_ => Self::for_terminal(command),
|
||||
}
|
||||
}
|
||||
|
||||
fn merge(self, next: Self) -> Self {
|
||||
match (self, next) {
|
||||
(Self::Shutdown, _) | (_, Self::Shutdown) => Self::Shutdown,
|
||||
@@ -1974,6 +1985,23 @@ fn deliver_warm_attachment(
|
||||
}
|
||||
}
|
||||
|
||||
/// A replayed terminal command must reproduce the lifecycle its fence
|
||||
/// recorded. One legacy exception: runners built before a failed
|
||||
/// `runner.suspend` stopped being terminal persisted a `suspended` fence for
|
||||
/// it. Its replay now continues; honour that old fence so an upgraded runner
|
||||
/// can still deliver the journaled result and finish the old transition.
|
||||
fn replay_satisfies_terminal_fence(
|
||||
pending: &PendingTerminalDelivery,
|
||||
result_status: &str,
|
||||
lifecycle: CommandLifecycle,
|
||||
) -> bool {
|
||||
lifecycle.durable_state() == Some(pending.lifecycle.as_str())
|
||||
|| (pending.command_type == "runner.suspend"
|
||||
&& pending.lifecycle == "suspended"
|
||||
&& matches!(result_status, "failed" | "rejected")
|
||||
&& lifecycle == CommandLifecycle::Continue)
|
||||
}
|
||||
|
||||
fn reconcile_pending_terminal_delivery<E: CommandExecutor>(
|
||||
state: &mut DurableState,
|
||||
store: &DurableStateStore,
|
||||
@@ -1998,7 +2026,7 @@ fn reconcile_pending_terminal_delivery<E: CommandExecutor>(
|
||||
));
|
||||
}
|
||||
let (result, lifecycle) = process_command(state, store, config, executor, command)?;
|
||||
if lifecycle.durable_state() != Some(pending.lifecycle.as_str()) {
|
||||
if !replay_satisfies_terminal_fence(&pending, &result.status, lifecycle) {
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"pending terminal command did not replay its durable lifecycle",
|
||||
));
|
||||
@@ -2293,6 +2321,8 @@ fn process_command<E: CommandExecutor>(
|
||||
// the old receipt must not replace the newer command cursor
|
||||
// with another terminal-delivery fence at an older sequence.
|
||||
CommandLifecycle::Continue
|
||||
} else if result.status == "failed" {
|
||||
CommandLifecycle::after_failure(command)
|
||||
} else {
|
||||
CommandLifecycle::for_terminal(command)
|
||||
};
|
||||
@@ -2340,7 +2370,7 @@ fn process_command<E: CommandExecutor>(
|
||||
}
|
||||
let result = state.fail_command(command, failure)?;
|
||||
store.save(state)?;
|
||||
return Ok((result, CommandLifecycle::for_terminal(command)));
|
||||
return Ok((result, CommandLifecycle::after_failure(command)));
|
||||
}
|
||||
};
|
||||
if matches!(
|
||||
@@ -5685,7 +5715,10 @@ mod tests {
|
||||
#[test]
|
||||
fn failed_lifecycle_commands_replay_their_terminal_transition() {
|
||||
for (command_type, expected_lifecycle) in [
|
||||
("runner.suspend", CommandLifecycle::Suspend),
|
||||
// A rejected suspension leaves the provider's session live, so the
|
||||
// runner keeps serving its controller (which may retry) rather
|
||||
// than recording `suspended` over it and exiting.
|
||||
("runner.suspend", CommandLifecycle::Continue),
|
||||
("runner.shutdown", CommandLifecycle::Shutdown),
|
||||
] {
|
||||
let directory = std::env::temp_dir().join(format!(
|
||||
@@ -5943,6 +5976,87 @@ mod tests {
|
||||
fs::remove_dir_all(directory).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_failed_suspend_fence_still_reconciles_after_upgrade() {
|
||||
let fence = |command_type: &str, lifecycle: &str| PendingTerminalDelivery {
|
||||
command_id: "command_1".to_owned(),
|
||||
controller_seq: 1,
|
||||
command_type: command_type.to_owned(),
|
||||
lifecycle: lifecycle.to_owned(),
|
||||
};
|
||||
// An old runner persisted `suspended` for a failed suspend; the
|
||||
// upgraded replay continues, and must still satisfy that fence.
|
||||
assert!(replay_satisfies_terminal_fence(
|
||||
&fence("runner.suspend", "suspended"),
|
||||
"failed",
|
||||
CommandLifecycle::Continue
|
||||
));
|
||||
assert!(replay_satisfies_terminal_fence(
|
||||
&fence("runner.suspend", "suspended"),
|
||||
"completed",
|
||||
CommandLifecycle::Suspend
|
||||
));
|
||||
assert!(replay_satisfies_terminal_fence(
|
||||
&fence("runner.shutdown", "stopped"),
|
||||
"failed",
|
||||
CommandLifecycle::Shutdown
|
||||
));
|
||||
// A completed suspend or a shutdown must still replay its own lifecycle.
|
||||
assert!(!replay_satisfies_terminal_fence(
|
||||
&fence("runner.suspend", "suspended"),
|
||||
"completed",
|
||||
CommandLifecycle::Continue
|
||||
));
|
||||
assert!(!replay_satisfies_terminal_fence(
|
||||
&fence("runner.shutdown", "stopped"),
|
||||
"failed",
|
||||
CommandLifecycle::Continue
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejected_suspend_keeps_the_runner_serving_its_controller() {
|
||||
let directory = std::env::temp_dir().join(format!(
|
||||
"paperclip-runner-rejected-suspend-{}",
|
||||
std::process::id()
|
||||
));
|
||||
let _ = fs::remove_dir_all(&directory);
|
||||
let config = config(directory.clone());
|
||||
let store = DurableStateStore::new(&directory).unwrap();
|
||||
let (mut state, _) = store.load_or_create(&config).unwrap();
|
||||
let mut executor = FailingExecutor { calls: 0 };
|
||||
let command = command("runner.suspend");
|
||||
|
||||
let (result, lifecycle) =
|
||||
process_command(&mut state, &store, &config, &mut executor, &command).unwrap();
|
||||
let (replayed, replay_lifecycle) =
|
||||
process_command(&mut state, &store, &config, &mut executor, &command).unwrap();
|
||||
|
||||
assert_eq!(result.status, "failed");
|
||||
assert_eq!(replayed.status, "failed");
|
||||
// The provider still owns its session: no `suspended` lifecycle and no
|
||||
// exit, so a fresh runner.suspend from the controller can be served.
|
||||
assert_eq!(lifecycle, CommandLifecycle::Continue);
|
||||
assert_eq!(replay_lifecycle, CommandLifecycle::Continue);
|
||||
assert_eq!(lifecycle.durable_state(), None);
|
||||
assert_eq!(executor.calls, 1);
|
||||
let mut retry = CountingExecutor { calls: 0 };
|
||||
let (_, retry_lifecycle) = process_command(
|
||||
&mut state,
|
||||
&store,
|
||||
&config,
|
||||
&mut retry,
|
||||
&Command {
|
||||
command_id: "command-suspend-retry".to_owned(),
|
||||
controller_seq: 2,
|
||||
..command.clone()
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(retry_lifecycle, CommandLifecycle::Suspend);
|
||||
fs::remove_dir_all(directory).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_suspend_replay_remains_restartable() {
|
||||
let directory = std::env::temp_dir().join(format!(
|
||||
|
||||
@@ -5132,6 +5132,44 @@ describe("DurablePrpControlPlane", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it("delivers a fresh suspension after a rejected one on the same connection", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-rejected-suspend-"));
|
||||
const controlPlane = new DurablePrpControlPlane({
|
||||
stateDirectory: root,
|
||||
identity,
|
||||
expectedRunnerVersion,
|
||||
expectedRunnerDigest,
|
||||
});
|
||||
try {
|
||||
await controlPlane.start();
|
||||
const rejected = controlPlane.queueCommand("runner.suspend", {}, "command-suspend-rejected");
|
||||
const client = await authenticate(controlPlane, controlPlane.issueBootstrapTicket());
|
||||
const retry = controlPlane.queueCommand("runner.suspend", {}, "command-suspend-retry", true);
|
||||
expect(controlPlane.store.state.commandDeliveryCounts[retry.commandId]).toBeUndefined();
|
||||
sendSecure(client!, {
|
||||
protocol: "paperclip.runner", version: 1, kind: "command_result",
|
||||
payload: {
|
||||
commandId: rejected.commandId, commandType: rejected.type, controllerSeq: rejected.controllerSeq,
|
||||
status: "failed",
|
||||
result: { code: "command_execution_failed", message: "ACPX session is not at a safe suspension point" },
|
||||
},
|
||||
});
|
||||
await expect(receiveSecure(client!)).resolves.toMatchObject({
|
||||
kind: "command_result_ack", payload: { commandId: "command-suspend-rejected", status: "failed" },
|
||||
});
|
||||
// The runner keeps serving after a rejected suspension, so the retry is
|
||||
// delivered on the same connection instead of waiting behind the fence.
|
||||
await expect(receiveSecure(client!)).resolves.toMatchObject({
|
||||
kind: "command", payload: { commandId: "command-suspend-retry" },
|
||||
});
|
||||
expect(controlPlane.store.state.commandDeliveryCounts[retry.commandId]).toBe(1);
|
||||
client?.socket.destroy();
|
||||
} finally {
|
||||
await controlPlane.stop();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("acknowledges terminal command results after persisting them", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-terminal-ack-"));
|
||||
const controlPlane = new DurablePrpControlPlane({
|
||||
|
||||
@@ -3983,7 +3983,8 @@ export class DurablePrpControlPlane {
|
||||
this.#store.state.duplicateCommandResults += 1;
|
||||
this.#store.save();
|
||||
this.#ackTerminalCommandResult(connection, command);
|
||||
if (!this.#isTerminalLifecycleCommand(command)) {
|
||||
this.#releaseServingLifecycle(connection, command);
|
||||
if (!this.#isTerminalLifecycleCommand(command) || connection.terminalLifecycleCommandId === null) {
|
||||
this.#sendNextCommand(connection);
|
||||
}
|
||||
return;
|
||||
@@ -4028,11 +4029,31 @@ export class DurablePrpControlPlane {
|
||||
command.result = structuredClone(result);
|
||||
this.#store.save();
|
||||
this.#ackTerminalCommandResult(connection, command);
|
||||
if (!this.#isTerminalLifecycleCommand(command)) {
|
||||
this.#releaseServingLifecycle(connection, command);
|
||||
if (!this.#isTerminalLifecycleCommand(command) || connection.terminalLifecycleCommandId === null) {
|
||||
this.#sendNextCommand(connection);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A rejected runner.suspend changes no provider state, and the runner keeps
|
||||
* serving this connection. Lift the terminal-lifecycle fence it installed
|
||||
* so the controller's next command (a fresh suspend) can be delivered.
|
||||
* Completed suspensions and every shutdown keep the fence.
|
||||
*/
|
||||
#releaseServingLifecycle(
|
||||
connection: AuthorityConnection,
|
||||
command: DurableRecoveryCoreCommand,
|
||||
): void {
|
||||
if (
|
||||
command.type === "runner.suspend" &&
|
||||
(command.status === "failed" || command.status === "rejected") &&
|
||||
connection.terminalLifecycleCommandId === command.commandId
|
||||
) {
|
||||
connection.terminalLifecycleCommandId = null;
|
||||
}
|
||||
}
|
||||
|
||||
#ackWarmTransition(
|
||||
connection: AuthorityConnection,
|
||||
receipt: DurableWarmRunTransition,
|
||||
|
||||
@@ -2013,6 +2013,56 @@ it("waits for the exact durable suspension command behind prior close work", asy
|
||||
expect(pumpCount).toBeGreaterThanOrEqual(2);
|
||||
});
|
||||
|
||||
it("retries a rejected suspension within the close deadline instead of failing a settled session", async () => {
|
||||
// An ACPX sidecar can reject session.suspend while the just-completed turn
|
||||
// is still settling. One rejected command must not leave a successful run's
|
||||
// own session closed without a proven suspension.
|
||||
const commands: Array<{ commandId: string; type: string; status: string }> = [];
|
||||
let lifecycle = "ready";
|
||||
const queueSuspend = vi.fn((commandId: string) => {
|
||||
commands.push({ commandId, type: "runner.suspend", status: "pending" });
|
||||
});
|
||||
await expect(
|
||||
runnerdRecoveryInternals.awaitRunnerSuspensionBarrier({
|
||||
commands: () => commands,
|
||||
queueSuspend,
|
||||
readRunnerState: async () => ({ lifecycle }),
|
||||
runnerHasExited: async () => false,
|
||||
pump: () => {
|
||||
const pending = commands.find(command => command.status === "pending");
|
||||
if (!pending) return;
|
||||
if (commands.length === 1) pending.status = "failed";
|
||||
else { pending.status = "completed"; lifecycle = "suspended"; }
|
||||
},
|
||||
deadline: Date.now() + 2_000,
|
||||
pollIntervalMs: 0,
|
||||
}),
|
||||
).resolves.toBe(true);
|
||||
expect(queueSuspend).toHaveBeenCalledTimes(2);
|
||||
expect(new Set(commands.map(command => command.commandId)).size).toBe(2);
|
||||
});
|
||||
|
||||
it("stops retrying a rejected suspension at its attempt bound or when the runner exits", async () => {
|
||||
for (const exited of [false, true]) {
|
||||
const commands: Array<{ commandId: string; type: string; status: string }> = [];
|
||||
const queueSuspend = vi.fn((commandId: string) => {
|
||||
commands.push({ commandId, type: "runner.suspend", status: "failed" });
|
||||
});
|
||||
await expect(
|
||||
runnerdRecoveryInternals.awaitRunnerSuspensionBarrier({
|
||||
commands: () => commands,
|
||||
queueSuspend,
|
||||
readRunnerState: async () => ({ lifecycle: "ready" }),
|
||||
runnerHasExited: async () => exited,
|
||||
pump: () => undefined,
|
||||
deadline: Date.now() + 5_000,
|
||||
pollIntervalMs: 0,
|
||||
}),
|
||||
).resolves.toBe(false);
|
||||
expect(queueSuspend).toHaveBeenCalledTimes(exited ? 1 : 3);
|
||||
}
|
||||
});
|
||||
|
||||
it("does not accept process exit without durable suspension", async () => {
|
||||
const commands: Array<{
|
||||
commandId: string;
|
||||
|
||||
@@ -929,6 +929,9 @@ async function releaseRunnerProcessOwnership(input: {
|
||||
if (releaseFailure !== undefined) throw releaseFailure;
|
||||
}
|
||||
|
||||
const MAX_RUNNER_SUSPEND_ATTEMPTS = 3;
|
||||
const RUNNER_SUSPEND_RETRY_DELAY_MS = 250;
|
||||
|
||||
async function awaitRunnerSuspensionBarrier(input: {
|
||||
commands: () => readonly {
|
||||
commandId: string;
|
||||
@@ -980,10 +983,11 @@ async function awaitRunnerSuspensionBarrier(input: {
|
||||
existing = undefined;
|
||||
}
|
||||
if (Date.now() >= input.deadline) return false;
|
||||
const commandId =
|
||||
let commandId =
|
||||
existing?.commandId ??
|
||||
`command_close_suspend_${randomUUID().replaceAll("-", "")}`;
|
||||
if (!existing) input.queueSuspend(commandId);
|
||||
let suspendAttempts = 1;
|
||||
|
||||
while (Date.now() < input.deadline) {
|
||||
input.pump();
|
||||
@@ -995,7 +999,25 @@ async function awaitRunnerSuspensionBarrier(input: {
|
||||
command.status !== "pending" &&
|
||||
command.status !== "completed"
|
||||
) {
|
||||
return false;
|
||||
// A provider can reject suspension while its just-finished turn is
|
||||
// still settling (an ACPX sidecar reports "not at a safe suspension
|
||||
// point"). The rejection changed nothing, so ask again with a fresh
|
||||
// command, a bounded number of times, within the same close deadline.
|
||||
// An exited runner cannot answer, and an exhausted budget fails closed.
|
||||
if (
|
||||
suspendAttempts >= MAX_RUNNER_SUSPEND_ATTEMPTS ||
|
||||
(await input.runnerHasExited())
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
await new Promise<void>((resolveWait) =>
|
||||
setTimeout(resolveWait, input.pollIntervalMs ?? RUNNER_SUSPEND_RETRY_DELAY_MS),
|
||||
);
|
||||
if (Date.now() >= input.deadline) return false;
|
||||
suspendAttempts += 1;
|
||||
commandId = `command_close_suspend_${randomUUID().replaceAll("-", "")}`;
|
||||
input.queueSuspend(commandId);
|
||||
continue;
|
||||
}
|
||||
let lifecycle: unknown;
|
||||
try {
|
||||
|
||||
@@ -42,6 +42,7 @@ import {
|
||||
completeTerminatedRemoteNativeSessionCleanup,
|
||||
completeTerminatedLocalNativeSessionCleanup,
|
||||
nativeSessionCleanupBlock,
|
||||
onNativeSessionCleanupQuarantined,
|
||||
type ExecuteNativeSessionOptions,
|
||||
} from "./native-session-runtime.js";
|
||||
|
||||
@@ -5006,6 +5007,45 @@ describe("executeNativeSession recovery", () => {
|
||||
completeTerminatedLocalNativeSessionCleanup(binding);
|
||||
});
|
||||
|
||||
it("announces every cleanup quarantine with its cause when it is created", async () => {
|
||||
const quarantined = vi.fn();
|
||||
const unsubscribe = onNativeSessionCleanupQuarantined(quarantined);
|
||||
const scopedIdentity = { ...identity, companyId: "announce-company", runId: "announce-run", issueId: "announce-issue", sessionId: "announce-session" };
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true };
|
||||
const settlement = { runnerSuspended: false, suspensionState: { commandStatus: "failed", runnerLifecycle: "suspended" } };
|
||||
const session: NativeSession = {
|
||||
identity: () => scopedIdentity, capabilities: async () => capabilities,
|
||||
async *events() { throw new Error("announce transport stopped"); },
|
||||
startTurn: async () => ({ turnId: "announce-turn" }), result: async () => null,
|
||||
async snapshot() { return { backendKind: "mock", sessionId: "announce-driver", identity: scopedIdentity,
|
||||
providerSessionId: "announce-provider", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; },
|
||||
close: vi.fn(async () => { throw new NativeSessionCloseUnrecoverableError(settlement); }),
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
descriptor: async () => ({ kind: "runner", name: "announce-backend", version: "1", capabilities }),
|
||||
openSession: vi.fn(async () => session),
|
||||
};
|
||||
try {
|
||||
await expect(executeNativeSession({
|
||||
input: { ...input, binding: { ...input.binding, companyId: scopedIdentity.companyId, runId: scopedIdentity.runId, issueId: scopedIdentity.issueId },
|
||||
session: { ...input.session, normalizedSessionId: scopedIdentity.sessionId } },
|
||||
backend, runnerInstanceId: "announce-runner", controlPlaneInstanceId: "control", requireSessionCloseBeforeReturn: true,
|
||||
controlPlane: { openRun: async () => {}, checkpointSession: async () => {},
|
||||
appendEvent: async () => ({ cursor: 0, highestContiguousSourceSeq: 0, disposition: "committed" }),
|
||||
replayEvents: async () => ({ events: [], highestContiguousSourceSeq: 0 }), completeRun: async () => {} },
|
||||
})).rejects.toThrow("announce transport stopped");
|
||||
expect(quarantined).toHaveBeenCalledOnce();
|
||||
expect(quarantined).toHaveBeenCalledWith({
|
||||
companyId: scopedIdentity.companyId, runId: scopedIdentity.runId, issueId: scopedIdentity.issueId,
|
||||
sessionId: scopedIdentity.sessionId, operatorRecoveryRequired: true,
|
||||
cause: expect.stringContaining("native_session_close_unrecoverable"), settlement,
|
||||
});
|
||||
} finally {
|
||||
unsubscribe();
|
||||
completeTerminatedLocalNativeSessionCleanup({ companyId: scopedIdentity.companyId, runId: scopedIdentity.runId, runnerInstanceId: "announce-runner" });
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps an unrecoverable close from quarantining unrelated tasks of the same backend", async () => {
|
||||
const companyId = "unrecoverable-scope-company";
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true };
|
||||
|
||||
@@ -601,11 +601,11 @@ function retainUnadmittedSessionCleanup(
|
||||
return;
|
||||
} catch (error) {
|
||||
if (error instanceof NativeSessionCloseUnrecoverableError) {
|
||||
quarantineSessionCleanup(session, cleanupDomain, true);
|
||||
quarantineSessionCleanup(session, cleanupDomain, true, error);
|
||||
throw error;
|
||||
}
|
||||
if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) {
|
||||
quarantineSessionCleanup(session, cleanupDomain);
|
||||
quarantineSessionCleanup(session, cleanupDomain, false, error);
|
||||
throw error;
|
||||
}
|
||||
retryCount += 1;
|
||||
@@ -763,15 +763,86 @@ function retainFailedSessionCleanupOwner(
|
||||
.catch(() => undefined);
|
||||
}
|
||||
|
||||
export interface NativeSessionCleanupQuarantine {
|
||||
companyId: string;
|
||||
runId: string;
|
||||
issueId: string;
|
||||
sessionId: string;
|
||||
operatorRecoveryRequired: boolean;
|
||||
/** Bounded error name/code/message of the close failure that caused it. */
|
||||
cause: string;
|
||||
/** Bounded settlement facts of an unrecoverable close, when it reported them. */
|
||||
settlement: Record<string, unknown> | null;
|
||||
}
|
||||
type NativeSessionCleanupQuarantineListener = (
|
||||
quarantine: NativeSessionCleanupQuarantine,
|
||||
) => void;
|
||||
const cleanupQuarantineListeners = new Set<NativeSessionCleanupQuarantineListener>();
|
||||
|
||||
/** Observe every cleanup quarantine when it is created or escalates to
|
||||
* operator recovery, with its cause. Observation grants no authority. */
|
||||
export function onNativeSessionCleanupQuarantined(
|
||||
listener: NativeSessionCleanupQuarantineListener,
|
||||
): () => void {
|
||||
cleanupQuarantineListeners.add(listener);
|
||||
return () => cleanupQuarantineListeners.delete(listener);
|
||||
}
|
||||
|
||||
function boundedSettlement(cause: unknown): Record<string, unknown> | null {
|
||||
if (!(cause instanceof NativeSessionCloseUnrecoverableError) || !cause.settlement) return null;
|
||||
try {
|
||||
const text = JSON.stringify(cause.settlement);
|
||||
return text.length <= 4_000 ? (JSON.parse(text) as Record<string, unknown>) : { truncated: true };
|
||||
} catch {
|
||||
return { unreadable: true };
|
||||
}
|
||||
}
|
||||
|
||||
function announceCleanupQuarantine(
|
||||
session: NativeSession,
|
||||
operatorRecoveryRequired: boolean,
|
||||
cause: unknown,
|
||||
): void {
|
||||
if (cleanupQuarantineListeners.size === 0) return;
|
||||
let identity: ReturnType<NativeSession["identity"]> | null = null;
|
||||
try {
|
||||
identity = session.identity();
|
||||
} catch {
|
||||
// Observation must not fail quarantine.
|
||||
}
|
||||
const error = cause instanceof Error ? cause : null;
|
||||
const code = error && "code" in error && typeof error.code === "string" ? error.code : null;
|
||||
const quarantine: NativeSessionCleanupQuarantine = {
|
||||
companyId: identity?.companyId ?? "",
|
||||
runId: identity?.runId ?? "",
|
||||
issueId: identity?.issueId ?? "",
|
||||
sessionId: identity?.sessionId ?? "",
|
||||
operatorRecoveryRequired,
|
||||
cause: [error?.name, code, error?.message ?? (cause === undefined ? null : String(cause))]
|
||||
.filter((part): part is string => Boolean(part)).join(": ").slice(0, 1_000) || "unknown",
|
||||
settlement: boundedSettlement(cause),
|
||||
};
|
||||
for (const listener of cleanupQuarantineListeners) {
|
||||
try {
|
||||
listener(structuredClone(quarantine));
|
||||
} catch {
|
||||
// Observers cannot change cleanup ownership.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function quarantineSessionCleanup(
|
||||
session: NativeSession,
|
||||
cleanupDomain: NativeSessionCleanupDomain,
|
||||
operatorRecoveryRequired = false,
|
||||
cause?: unknown,
|
||||
): void {
|
||||
const existing = [...quarantinedSessionCleanups].find(
|
||||
(entry) => entry.session === session,
|
||||
);
|
||||
if (existing) {
|
||||
if (operatorRecoveryRequired && !existing.operatorRecoveryRequired)
|
||||
announceCleanupQuarantine(session, true, cause);
|
||||
if (operatorRecoveryRequired) {
|
||||
existing.operatorRecoveryRequired = true;
|
||||
if (existing.timer) clearTimeout(existing.timer);
|
||||
@@ -790,6 +861,7 @@ function quarantineSessionCleanup(
|
||||
operatorRecoveryRequired,
|
||||
};
|
||||
quarantinedSessionCleanups.add(cleanup);
|
||||
announceCleanupQuarantine(session, operatorRecoveryRequired, cause);
|
||||
if (operatorRecoveryRequired) return;
|
||||
startQuarantinedSessionCleanupRecovery(
|
||||
cleanup,
|
||||
@@ -835,6 +907,8 @@ function startQuarantinedSessionCleanupRecovery(
|
||||
announceCleanupSettled(cleanup.domain);
|
||||
} catch (error) {
|
||||
if (error instanceof NativeSessionCloseUnrecoverableError) {
|
||||
if (!cleanup.operatorRecoveryRequired)
|
||||
announceCleanupQuarantine(cleanup.session, true, error);
|
||||
cleanup.operatorRecoveryRequired = true;
|
||||
}
|
||||
// Retain the quarantine after this finite, sequential retry batch.
|
||||
@@ -2484,11 +2558,11 @@ export async function executeNativeSession(
|
||||
return;
|
||||
} catch (error) {
|
||||
if (error instanceof NativeSessionCloseUnrecoverableError) {
|
||||
quarantineSessionCleanup(session, cleanupDomain, true);
|
||||
quarantineSessionCleanup(session, cleanupDomain, true, error);
|
||||
throw error;
|
||||
}
|
||||
if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) {
|
||||
quarantineSessionCleanup(session, cleanupDomain);
|
||||
quarantineSessionCleanup(session, cleanupDomain, false, error);
|
||||
throw error;
|
||||
}
|
||||
retryCount += 1;
|
||||
|
||||
@@ -83,6 +83,7 @@ import { createSentryRunLinkResolver } from "./services/sentry-run-links.js";
|
||||
import { waitForPendingRunFailureReports } from "./services/run-failure-report.js";
|
||||
import { verifyStoppedNativeSessionForReplacement } from "./services/native-runtime/native-session-executor.js";
|
||||
import { registerNativeCleanupReleaseWake } from "./services/native-runtime/native-cleanup-release-wake.js";
|
||||
import { logNativeCleanupQuarantines } from "./services/native-runtime/native-cleanup-quarantine-log.js";
|
||||
import { embeddedPostgresOwnerPort } from "./embedded-postgres-owner.js";
|
||||
import { deliverExecutionStatuses } from "./services/execution-status-delivery.js";
|
||||
import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js";
|
||||
@@ -1352,6 +1353,10 @@ async function startServerWithDatabaseTeardown(
|
||||
if (skillVersionRetentionWorker) trackHeartbeatSchedulerWork(skillVersionRetentionWorker.ready);
|
||||
const workspaceRecoveryWork = registerWorkspaceRecoveryWork(db, deliveryWork, workspaceRecoveryRuntime,
|
||||
accountingHeartbeat.resumeRemoteStopComments, () => !heartbeatSchedulerStopped);
|
||||
// Each native cleanup quarantine is logged with its cause when created. The
|
||||
// subscription lives with the process: closes that fail during shutdown,
|
||||
// after HTTP closes, still need their cause logged.
|
||||
logNativeCleanupQuarantines(logger);
|
||||
const localLoginCleanup = localAiLoginService(db);
|
||||
const localLoginCleanupWork = registerExecutionControlWork(deliveryWork, [
|
||||
{ queue: DELIVERY_QUEUES.localLoginCleanup, run: localLoginCleanup.reapExpired, next: localLoginCleanup.nextCleanupAt },
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import { onNativeSessionCleanupQuarantined } from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
type QuarantineLogger = { warn: (details: Record<string, unknown>, message: string) => void };
|
||||
|
||||
let installed: { logger: QuarantineLogger; unsubscribe: () => void } | null = null;
|
||||
|
||||
/**
|
||||
* Log every native session cleanup quarantine when the runner creates it or
|
||||
* escalates it to operator recovery, so a later admission refusal can be
|
||||
* traced to its owning session and close failure. The subscription is
|
||||
* process-wide and installed once: native closes can still fail during
|
||||
* shutdown after the HTTP server has closed.
|
||||
*/
|
||||
export function logNativeCleanupQuarantines(logger: QuarantineLogger): () => void {
|
||||
if (installed?.logger === logger) return installed.unsubscribe;
|
||||
installed?.unsubscribe();
|
||||
const unsubscribe = onNativeSessionCleanupQuarantined((quarantine) => {
|
||||
logger.warn({ ...quarantine }, quarantine.operatorRecoveryRequired
|
||||
? "native session cleanup quarantined; operator recovery required"
|
||||
: "native session cleanup quarantined; automatic close recovery scheduled");
|
||||
});
|
||||
installed = { logger, unsubscribe: () => { unsubscribe(); installed = null; } };
|
||||
return installed.unsubscribe;
|
||||
}
|
||||
+1
@@ -193,6 +193,7 @@ export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTermi
|
||||
export const completeTerminatedLocalNativeSessionCleanup = runner.completeTerminatedLocalNativeSessionCleanup;
|
||||
export const nativeSessionCleanupBlock = runner.nativeSessionCleanupBlock;
|
||||
export const onNativeSessionCleanupSettled = runner.onNativeSessionCleanupSettled;
|
||||
export const onNativeSessionCleanupQuarantined = runner.onNativeSessionCleanupQuarantined;
|
||||
|
||||
export const probeAcpxClaudeInstallation = runner.probeAcpxClaudeInstallation;
|
||||
export const probeAcpxGrokInstallation = runner.probeAcpxGrokInstallation;
|
||||
|
||||
Reference in new issue
Block a user