From 90b8190c33e8f8eae44cae1f57721251fe1e1343 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 00:52:01 -0600 Subject: [PATCH] fix(native-runtime): detect half-open runner ingress and project Stop for live runs (#19407) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs in a managed sandbox talk to the server through a provider-ingress WebSocket. The server opens this socket to the runner (`paperclip-runnerd`) through the sandbox preview proxy. > - A staging run ran a 20-minute foreground shell command that printed nothing. After the first second, no run events arrived. The run stayed "Working" for over 35 minutes. No watchdog acted, and the run projection said `canStop: false`. > - The server only detected a lost runner connection from a `close` or `error` event. It never sent pings. A half-open route, for example a proxy that drops the connection without a FIN, never emits `close`. The run then waits forever and the 7-minute reconnect grace never starts. > - A long silent tool call is legitimate. The fix must tell "quiet but alive" apart from "nothing alive". A protocol ping does that: runnerd answers every ping with a pong while its provider waits on the tool. > - This pull request pings authenticated runner connections and terminates the socket after 90 seconds of total silence. The existing reconnect grace and transport-failure path then recover the run. It also makes the execution projection report `canStop: true` for every queued or running run. > - The benefit is that a run with a dead runner connection now ends in a bounded time instead of hanging. API readers also see the Stop that the board already offers. ## Linked Issues or Issue Description **What happened?** A native-runner run (managed sandbox, OpenCode with an OpenRouter model) ran a foreground shell command for about 20 minutes with no output. The last run event came at about T+1s. The command finished on disk, but the run never got another event and stayed `running` / "Working" for over 35 minutes. Its projection had `canStop: false` and `permittedActions: ["inspect_run"]`. A board cancel through the API worked. **Expected behavior** The server detects a dead runner connection and recovers or fails the run in a bounded time. A runner that is alive but quiet stays attached. The projection reports that a live run can be stopped. **Steps to reproduce** 1. Start a native run in a managed sandbox. 2. Make the agent run a long foreground command that prints nothing. 3. Let the preview route drop the WebSocket without a close frame. 4. Before this fix, the run stays "Working" forever. ## What Changed - `server/src/realtime/runner-prp-outbound.ts`: after PRP authentication, `WsJsonWireConnection.monitorLiveness` sends a WebSocket ping every 20 s. Any inbound frame or pong counts as liveness. After 90 s with nothing inbound, it terminates the socket. That emits an ordinary `close`, so the PRP authority and the reconnect loop use the existing 7-minute grace and then fail the transport. If the server's own event loop stalls, the window restarts instead of blaming the runner. The 90 s bound is longer than runnerd's own 60 s connection lease, which already requires runnerd to read the socket. - `server/src/services/execution-projection.ts`: `canStop` is now `true` for every `queued` or `running` run. The old failed-native-retry rule is unchanged. - `server/src/services/execution-projection.ts`: a queued run now shows "Waiting for previous run to stop" (cause `previous_run_settling`) when a predecessor that admission checks (the task execution owner, the retried run, or the agent's latest settled run on the task) still has pending cleanup by `hasPendingRunCleanup`. Before, a turn queued after a Stop showed only "Queued" with no reason while the stopped run's controller still waited on the dead runner. That in-process wait (`claimQueuedRun` settling-owner gate) also made Retry a no-op. The liveness bound above makes that controller settle. Its existing cleanup then calls `startNextQueuedRunForAgent`, which starts the queued turn. - `server/src/services/run-cancellation.ts` and `server/src/modules/run-dispatch/adapters/postgres.ts`: a user-stopped native run whose runner was unreachable can later settle as `failed` (`native_provider_terminal_failed`) when its transport fails closed. The dispatch gate's message deferral required a `cancelled` source and compared the message with `finishedAt`. The user's follow-up message was therefore marked `skipped`, and the task asked for a second message. The new `recordedUserStop` helper honors the recorded user Stop on such a failed source, so the gate saves the message for the deferred-receipt worker. - `packages/shared/src/types/execution-projection.ts`: documents the `canStop` meaning. - `doc/execution-semantics.md`: documents the ingress liveness bound and the new `canStop` meaning. ## Verification - Red then green: `runner-prp-outbound.test.ts` "detects a half-open authenticated ingress and fails the run within a bounded window". On the old code it failed because no ping was sent and the socket was never terminated. It now passes and the active-run `failure` rejects with `runner_ingress_unavailable`. - Safety: "keeps a quiet but live runner attached through a long silent tool call" checks that a runner answering pings stays attached for 20 simulated minutes with no termination and no failure. - Red then green: `execution-projection.test.ts` "offers Stop for every live run, including a long-silent native turn". On the old code it failed with `canStop: false`. - Red then green: `execution-projection-recovery.test.ts` "names the wait when a stopped predecessor still holds the task's runner" (real Postgres). On the old code the successor projected a plain "Queued". It also checks that the wait clears on lease release and ignores another agent's run. - Red then green: `workspace-recovery.test.ts` "continues a message sent after Stop when the stopped run later failed closed". On the old code it failed 3 of 3 runs with the receipt `skipped`. It now passes, and the full file passes twice (392 tests). The test now pins the message after the recorded Stop. Before, the fixture could write both in the same millisecond, which made the sibling variants flaky. - Passed: run-cancellation, explicit-native-continuation, run-dispatch and run-control suites (622 tests). - Passed: `npx vitest run src/realtime/ src/services/execution-projection*.test.ts src/__tests__/heartbeat-run-privacy-routes.test.ts`, the native-runner-execution-driver tests, and the plugin-runner-task tests. - Passed: `pnpm --filter @paperclipai/server typecheck`. - I read runnerd (`runner-core/src/durable/transport.rs`) and confirmed it answers `Ping` with `Pong` in its non-blocking read loop. - Not run: a live managed-sandbox reproduction. Staging and Daytona were not reachable from this environment. `plugin-runner-probe.test.ts` needs a local cargo build of runnerd, and it did not start in this worktree. ## Risks - A false termination is possible if a runner fails to answer pings for 90 s while it is still alive. runnerd already loses its own connection lease after 60 s without reading, so this bound does not add a stricter requirement. A terminated live runner reconnects within the existing grace. - If the preview proxy drops WebSocket control frames, ping alone does not prove liveness. Inbound PRP frames (runnerd's lease renewals, about every 30 s) still count, so a live runner stays attached. - `canStop` now means "the board can Stop now" for live runs. The UI already shows Stop for running and queued runs without reading this hint, so the UI does not change. - Out of scope: the inbound runner WebSocket server (`runner-prp-ws.ts`), and a lost OpenCode SSE event while the runner is alive. A stopped run whose controller hangs for a reason other than a dead transport still holds the queued successor. The projection now names that wait, but this PR does not force-release it, because the controller has no termination evidence. A queued reason for the agent concurrency cap is also not covered. The wait label calls the admission predicate (`hasPendingRunCleanup`) for up to three predecessor runs per queued run. This adds a few reads to projections of queued runs only. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell, file edits) 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 24 ++- .../shared/src/types/execution-projection.ts | 1 + .../modules/run-dispatch/adapters/postgres.ts | 8 +- .../src/realtime/runner-prp-outbound.test.ts | 143 ++++++++++++++++++ server/src/realtime/runner-prp-outbound.ts | 67 +++++++- .../execution-projection-recovery.test.ts | 43 ++++++ .../src/services/execution-projection.test.ts | 10 ++ server/src/services/execution-projection.ts | 68 ++++++++- server/src/services/run-cancellation.ts | 15 +- .../src/services/workspace-recovery.test.ts | 20 ++- 10 files changed, 384 insertions(+), 15 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f7b7c588a76..06def5c40dc 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2477,6 +2477,26 @@ A runner receipt that its reconnect grace expired ends that incident. The transport recognizes both the coded error and the runner's plain-text message. It must not open another recovery window after that receipt. +An authenticated provider-ingress WebSocket is pinged every 20 seconds. Any +frame or pong counts as liveness, so a runner that stays quiet through a long +foreground tool call remains attached. When nothing arrives for 90 seconds the +server terminates the socket. A half-open route, such as a preview proxy that +drops the connection without a close, then follows the ordinary reconnect grace +and fails the transport if the runner cannot be reached. Without this, such a +run stayed "Working" indefinitely. The 90-second bound exceeds the runner's own +60-second connection lease, which already requires it to read and renew. + +A stopped native run can keep its execution lease while its controller still +waits on an unreachable runner. A queued turn of the same agent and task cannot +start until that controller settles. The queued turn's projection names this +wait ("Waiting for previous run to stop", cause `previous_run_settling`) while +admission's pending-cleanup check holds a predecessor it checks. The ingress liveness +bound above makes that controller settle, and its cleanup starts the queued turn. +Such a stopped run can then settle as `failed` when its transport fails closed. +Its recorded user Stop still marks later messages as new direction. When the +dispatch gate withholds the queued turn behind saved-work recovery, it saves +that message for the deferred-receipt worker, as it does for a cancelled source. + A bounded transport close can return before artifact staging settles. That return does not prove cleanup completed or release the retained-state obligation. Launch attempts, authenticated runner connections, provider process entries, @@ -4124,8 +4144,8 @@ and no result has been accepted. Unknown, legacy and accepted-result states remain ineligible. Grok failed retries keep their existing external Stop contract and do not gain this action; a pre-assignment failed retry is not supported by that endpoint. Existing Grok running/queued Stop and assignment fences stay in -place. A false or absent hint does not remove the existing queued -or running Cancel action. This presentation hint grants no permission; the existing +place. Every queued or running run also projects `canStop: true`, so API +readers see the same Stop that the run detail and task composer offer. This presentation hint grants no permission; the existing company, actor and locked cancellation checks remain authoritative. A changed retry can reject Stop, which the run detail displays without claiming cleanup. Failed-native Stop keeps its request identity by company, run and current board diff --git a/packages/shared/src/types/execution-projection.ts b/packages/shared/src/types/execution-projection.ts index ce2a1e3a67a..7818cc27f97 100644 --- a/packages/shared/src/types/execution-projection.ts +++ b/packages/shared/src/types/execution-projection.ts @@ -28,6 +28,7 @@ export interface ExecutionBlocker { /** Presentation of existing execution records, not a second task status machine. */ export interface ExecutionProjection { + /** The board can Stop this run now: any queued or running run, or a failed native run whose retry is still pending. */ canStop?: boolean; phase: | "working" diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 9733401c99a..ca98461535b 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -12,7 +12,7 @@ import { notifyInstructionCollection } from "../../../services/instruction-colle import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; import { readClaimedNativeStopRecovery } from "../../../services/native-runtime/native-cancellation-request.js"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; -import { userStopBoundaryAt } from "../../../services/run-cancellation.js"; +import { recordedUserStop, userStopBoundaryAt } from "../../../services/run-cancellation.js"; import { getExecutionBlocker } from "../../../services/execution-blocker.js"; import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js"; import { getNativeReviewAssignment } from "../../../services/native-runtime/native-review-participant.js"; @@ -996,10 +996,12 @@ export function createPostgresRunDispatchAdapter( )).for("update"); const [source] = preservation ? await tx.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), - eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "cancelled"), + eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["cancelled", "failed"]), sql`${heartbeatRuns.id}::text = ${preservation.evidence.runId}`, )) : []; - if (!preservation || !source) return null; + // A user-stopped run can settle as failed when its unreachable runner's + // transport fails closed. Only that recorded Stop makes a failed source eligible. + if (!preservation || !source || (source.status === "failed" && !recordedUserStop(source))) return null; const baseline = readNativeWorkspaceSyncReference(source.runnerProfileJson?.nativeWorkspaceSync); // Copyback from a native baseline replays the saved turn. Without one, only // a message sent after the Stop is saved; admission still applies all gates. diff --git a/server/src/realtime/runner-prp-outbound.test.ts b/server/src/realtime/runner-prp-outbound.test.ts index e0bf46c6395..06ab2f12a0a 100644 --- a/server/src/realtime/runner-prp-outbound.test.ts +++ b/server/src/realtime/runner-prp-outbound.test.ts @@ -26,6 +26,7 @@ vi.mock("ws", async (importOriginal) => { class FakeSocket extends EventEmitter { readyState: number = WebSocket.OPEN; send = vi.fn(); + ping = vi.fn(); terminate = vi.fn(() => { this.readyState = WebSocket.CLOSED; this.emit("close", 1006, Buffer.alloc(0)); @@ -393,4 +394,146 @@ describe("runner provider-ingress WebSocket wire", () => { expect(failureObserver).not.toHaveBeenCalled(); expect(endpoint.close).toHaveBeenCalledOnce(); }); + it("detects a half-open authenticated ingress and fails the run within a bounded window", async () => { + vi.useFakeTimers(); + const halfOpen = new FakeSocket(); + const create = sockets.create; + let opens = 0; + // The first socket authenticates and then goes silent without a FIN, as a + // preview proxy that drops the connection does. Reconnects cannot reach a + // runner whose process already exited. + sockets.create = () => { + opens += 1; + if (opens === 1) return openingSocket(halfOpen); + const refused = new FakeSocket(); + refused.readyState = WebSocket.CONNECTING; + queueMicrotask(() => { + refused.readyState = WebSocket.CLOSED; + refused.emit("error", new Error("fixture runner gone")); + }); + return refused; + }; + const endpoint = { + kind: "authenticated_websocket" as const, + websocketUrl: "wss://preview.invalid/api/runner/v1/connect/run-half-open", + secretHeaders: [], + generation: "generation-half-open", + refresh: vi.fn(async () => endpoint), + close: vi.fn(async () => undefined), + }; + const states: string[] = []; + const handle = connectRunnerPrpIngress({ + authority: { + attachWireConnection: vi.fn(() => ({ isAuthenticated: () => true })), + activeRunnerConnectionCount: () => 0, + } as unknown as DurablePrpControlPlane, + endpoint, + recoveryGraceMs: 60_000, + livenessIntervalMs: 10_000, + livenessTimeoutMs: 30_000, + onStateChange: (state) => states.push(state), + }); + const failure = vi.fn(); + void handle.failure.catch(failure); + try { + await vi.advanceTimersByTimeAsync(0); + await handle.ready; + await vi.advanceTimersByTimeAsync(29_000); + expect(halfOpen.terminate).not.toHaveBeenCalled(); + expect(halfOpen.ping).toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(11_000); + expect(halfOpen.terminate).toHaveBeenCalledOnce(); + expect(states).toContain("reconnecting"); + await vi.advanceTimersByTimeAsync(61_000); + expect(failure).toHaveBeenCalledWith( + expect.objectContaining({ code: "runner_ingress_unavailable" }), + ); + } finally { + await handle.close(); + sockets.create = create; + vi.useRealTimers(); + } + }); + + it("keeps a quiet but live runner attached through a long silent tool call", async () => { + vi.useFakeTimers(); + const live = new FakeSocket(); + // A live runnerd answers protocol pings even while its provider is + // blocked on a minutes-long foreground command that prints nothing. + live.ping = vi.fn(() => { + queueMicrotask(() => live.emit("pong", Buffer.alloc(0))); + }); + const create = sockets.create; + sockets.create = () => openingSocket(live); + const endpoint = { + kind: "authenticated_websocket" as const, + websocketUrl: "wss://preview.invalid/api/runner/v1/connect/run-quiet", + secretHeaders: [], + generation: "generation-quiet", + refresh: vi.fn(async () => endpoint), + close: vi.fn(async () => undefined), + }; + const attachWireConnection = vi.fn(() => ({ isAuthenticated: () => true })); + const handle = connectRunnerPrpIngress({ + authority: { attachWireConnection, activeRunnerConnectionCount: () => 1 } as unknown as DurablePrpControlPlane, + endpoint, + livenessIntervalMs: 10_000, + livenessTimeoutMs: 30_000, + }); + const failure = vi.fn(); + void handle.failure.catch(failure); + try { + await vi.advanceTimersByTimeAsync(0); + await handle.ready; + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(live.terminate).not.toHaveBeenCalled(); + expect(attachWireConnection).toHaveBeenCalledOnce(); + expect(failure).not.toHaveBeenCalled(); + } finally { + await handle.close(); + sockets.create = create; + vi.useRealTimers(); + } + }); + it("does not blame the runner for a server event-loop stall", () => { + vi.useFakeTimers(); + try { + const socket = new FakeSocket(); + const wire = new WsJsonWireConnection(socket as unknown as WebSocket); + const stop = wire.monitorLiveness({ intervalMs: 10_000, timeoutMs: 30_000 }); + expect(socket.ping).toHaveBeenCalledOnce(); + // The server's own loop is blocked for 5 minutes: no timers fire and + // the runner's pongs sit unread. The next tick must restart the window. + vi.setSystemTime(Date.now() + 5 * 60_000); + vi.advanceTimersByTime(10_000); + expect(socket.terminate).not.toHaveBeenCalled(); + expect(socket.ping).toHaveBeenCalledTimes(2); + // Genuine silence after the stall is still detected. + vi.advanceTimersByTime(30_000); + expect(socket.terminate).toHaveBeenCalledOnce(); + stop(); + } finally { + vi.useRealTimers(); + } + }); + + it("pings a healthy peer before any deadline, even with an interval longer than the timeout", () => { + vi.useFakeTimers(); + try { + const socket = new FakeSocket(); + socket.ping = vi.fn(() => { queueMicrotask(() => socket.emit("pong", Buffer.alloc(0))); }); + const wire = new WsJsonWireConnection(socket as unknown as WebSocket); + const stop = wire.monitorLiveness({ intervalMs: 60_000, timeoutMs: 30_000 }); + expect(socket.ping).toHaveBeenCalledOnce(); + for (let i = 0; i < 20; i++) { + vi.advanceTimersByTime(15_000); + // Deliver the queued pong before the next tick, as a live peer would. + socket.emit("pong", Buffer.alloc(0)); + } + expect(socket.terminate).not.toHaveBeenCalled(); + stop(); + } finally { + vi.useRealTimers(); + } + }); }); diff --git a/server/src/realtime/runner-prp-outbound.ts b/server/src/realtime/runner-prp-outbound.ts index 3b40546b35d..08db6258698 100644 --- a/server/src/realtime/runner-prp-outbound.ts +++ b/server/src/realtime/runner-prp-outbound.ts @@ -11,6 +11,13 @@ const DEFAULT_STARTUP_DEADLINE_MS = 60_000; const DEFAULT_RECOVERY_GRACE_MS = REMOTE_RUNNER_RECONNECT_GRACE_MS; const MAX_FRAME_BYTES = 4 * 1024 * 1024; const RECONNECT_DELAYS_MS = [250, 500, 1_000, 2_000, 5_000] as const; +// A runner can legitimately stay silent for many minutes while its provider +// waits on a long foreground tool call. Protocol pings distinguish that quiet +// but live runner from a half-open connection (for example a preview proxy that +// drops the route without a FIN), which otherwise never emits `close` and leaves +// the run "running" forever. runnerd answers every ping with a pong. +const DEFAULT_LIVENESS_INTERVAL_MS = 20_000; +const DEFAULT_LIVENESS_TIMEOUT_MS = 90_000; export type RunnerIngressFailureCode = | "runner_ingress_unavailable" @@ -32,9 +39,14 @@ export class WsJsonWireConnection implements PrpWireConnection { #onClose = new Set<(reason: TransportCloseReason) => void>(); #pendingJson: unknown[] = []; #closeReason: TransportCloseReason | null = null; + #lastInboundAt = Date.now(); constructor(readonly socket: WebSocket) { + socket.on("pong", () => { + this.#lastInboundAt = Date.now(); + }); socket.on("message", (data, isBinary) => { + this.#lastInboundAt = Date.now(); if (isBinary) { this.close(1003); return; @@ -73,6 +85,47 @@ export class WsJsonWireConnection implements PrpWireConnection { } } + /** + * Ping the runner and terminate the socket when nothing at all (frame or + * pong) arrives within the timeout. Termination emits `close`, so the PRP + * authority and the reconnect loop observe an ordinary transport loss and + * the fixed recovery grace applies. Returns a stop function. + */ + monitorLiveness(input: { intervalMs: number; timeoutMs: number }): () => void { + this.#lastInboundAt = Date.now(); + // At least two pings fit inside the timeout, so a healthy peer always has + // a chance to answer before the deadline, whatever the caller passes. + const intervalMs = Math.max(1, Math.min(input.intervalMs, Math.floor(input.timeoutMs / 2))); + let lastTickAt = Date.now(); + const ping = () => { + try { + this.socket.ping(); + } catch { + // A failing ping is itself a transport loss; ws reports it via close. + } + }; + if (this.socket.readyState === WebSocket.OPEN) ping(); + const timer = setInterval(() => { + if (this.#closeReason || this.socket.readyState !== WebSocket.OPEN) return; + const now = Date.now(); + // A blocked server event loop delays our own timers and the runner's + // queued pongs alike. That silence is not runner evidence; restart the + // window instead of terminating a live runner. + if (now - lastTickAt > intervalMs * 2) this.#lastInboundAt = now; + lastTickAt = now; + if (now - this.#lastInboundAt >= input.timeoutMs) { + clearInterval(timer); + this.socket.terminate(); + return; + } + ping(); + }, intervalMs); + timer.unref?.(); + const stop = () => clearInterval(timer); + this.onClose(stop); + return stop; + } + onJson(listener: (value: unknown) => void): void { this.#onJson = listener; for (const value of this.#pendingJson.splice(0)) listener(value); @@ -263,6 +316,10 @@ export function connectRunnerPrpIngress(input: { endpoint: RunnerIngressEndpoint; startupDeadlineMs?: number; recoveryGraceMs?: number; + /** How often to ping an authenticated runner connection. */ + livenessIntervalMs?: number; + /** Silence after which an authenticated connection is treated as lost. */ + livenessTimeoutMs?: number; random?: () => number; onStateChange?: ( state: "connecting" | "authenticated" | "reconnecting" | "failed" | "closed", @@ -330,7 +387,15 @@ export function connectRunnerPrpIngress(input: { } input.onStateChange?.("authenticated"); attempt = 0; - await socketClosed(opened.wire); + const stopLiveness = opened.wire.monitorLiveness({ + intervalMs: input.livenessIntervalMs ?? DEFAULT_LIVENESS_INTERVAL_MS, + timeoutMs: input.livenessTimeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS, + }); + try { + await socketClosed(opened.wire); + } finally { + stopLiveness(); + } recoveryDeadline = Date.now() + recoveryGraceMs; } catch (error) { // close() owns intentional shutdown. openEndpoint rejects its pending diff --git a/server/src/services/execution-projection-recovery.test.ts b/server/src/services/execution-projection-recovery.test.ts index 5f1628d6a96..603d59bdc0c 100644 --- a/server/src/services/execution-projection-recovery.test.ts +++ b/server/src/services/execution-projection-recovery.test.ts @@ -75,4 +75,47 @@ describe.skipIf(!support.supported)("persisted stopped-workspace projection", () expect((await executionProjectionsForRuns(db, companyId, [runId])).get(runId)?.canStop).toBe(false); }); + it("names the wait when a stopped predecessor still holds the task's runner", async () => { + // Staging: a hung native run was stopped by the board, then a resume + // comment queued a successor. The stopped run's controller still held its + // execution lease, so the successor sat "Queued" with no reason. + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + const stoppedRunId = randomUUID(), successorRunId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Stopped predecessor", issuePrefix: "SPR" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Long foreground tool", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: stoppedRunId, companyId, agentId, nativeIssueId: issueId, + runtimeMode: "native", status: "cancelled", finishedAt: new Date(), contextSnapshot: { issueId } }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: stoppedRunId, phase: "observed", + leaseOwner: "controller-a", leaseExpiresAt: new Date(Date.now() + 60_000) }); + await db.insert(heartbeatRuns).values({ id: successorRunId, companyId, agentId, + runtimeMode: "native", status: "queued", contextSnapshot: { issueId } }); + expect((await executionProjectionsForRuns(db, companyId, [successorRunId])).get(successorRunId)).toMatchObject({ + phase: "queued", label: "Waiting for previous run to stop", cause: "previous_run_settling", canStop: true, + }); + // An expired lease still owned by a controller still holds admission. + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 60_000) }) + .where(eq(nativeRunFinalizations.runId, stoppedRunId)); + expect((await executionProjectionsForRuns(db, companyId, [successorRunId])).get(successorRunId)?.cause) + .toBe("previous_run_settling"); + // A released claim whose finalization is still unsettled keeps holding admission. + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, stoppedRunId)); + expect((await executionProjectionsForRuns(db, companyId, [successorRunId])).get(successorRunId)?.cause) + .toBe("previous_run_settling"); + // Once finalization settles, the wait is ordinary. + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }) + .where(eq(nativeRunFinalizations.runId, stoppedRunId)); + expect((await executionProjectionsForRuns(db, companyId, [successorRunId])).get(successorRunId)).toMatchObject({ + phase: "queued", label: "Queued", cause: null, + }); + // Another agent's held run on the same task is not this turn's predecessor. + const otherAgentId = randomUUID(), otherRunId = randomUUID(); + await db.insert(agents).values({ id: otherAgentId, companyId, name: "Other", adapterType: "paperclip_runner" }); + await db.insert(heartbeatRuns).values({ id: otherRunId, companyId, agentId: otherAgentId, nativeIssueId: issueId, + runtimeMode: "native", status: "cancelled", finishedAt: new Date(), contextSnapshot: { issueId } }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: otherRunId, phase: "observed", + leaseOwner: "controller-b", leaseExpiresAt: new Date(Date.now() + 60_000) }); + expect((await executionProjectionsForRuns(db, companyId, [successorRunId])).get(successorRunId)?.cause).toBeNull(); + }); }); diff --git a/server/src/services/execution-projection.test.ts b/server/src/services/execution-projection.test.ts index ca5995027c1..7c98716f02f 100644 --- a/server/src/services/execution-projection.test.ts +++ b/server/src/services/execution-projection.test.ts @@ -38,6 +38,16 @@ const project = ( ) => projectExecution(r, c, pending, undefined, now); describe("execution truth projection", () => { + it("offers Stop for every live run, including a long-silent native turn", () => { + const silentSince = new Date(now.getTime() - 35 * 60_000); + const silent = run({ lastUsefulActionAt: silentSince, startedAt: silentSince }); + expect(project(silent)).toMatchObject({ phase: "working", canStop: true }); + expect(project(run({ runtimeMode: "legacy" }), undefined).canStop).toBe(true); + expect(projectExecution(run({ status: "queued", startedAt: null }), undefined, [], undefined, now).canStop).toBe(true); + for (const status of ["succeeded", "cancelled", "timed_out", "failed"] as const) { + expect(project(run({ status }), undefined).canStop).toBe(false); + } + }); it("shows maintenance only for admission that has not started", () => { startTaskDrain(); try { diff --git a/server/src/services/execution-projection.ts b/server/src/services/execution-projection.ts index 12a4520a578..857bef5dc09 100644 --- a/server/src/services/execution-projection.ts +++ b/server/src/services/execution-projection.ts @@ -1,6 +1,7 @@ -import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull, ne, sql } from "drizzle-orm"; import { heartbeatRuns, + issues, dotAgentBindings, dotRunnerAssignments, museAgentBindings, @@ -16,9 +17,12 @@ import { executionFailureRetryCount } from "./execution-recovery-attempt.js"; import { retryIdempotentDatabaseOperation } from "../database-retry.js"; import { workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; import { readTaskDrain } from "./task-admission.js"; +import { hasPendingRunCleanup } from "./execution-blocker.js"; +const MAX_PREDECESSOR_CLEANUP_CHECKS = 16; const text = (v: unknown) => (typeof v === "string" ? v : null); const executionRunColumns = { id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, driverKind: heartbeatRuns.driverKind, errorCode: heartbeatRuns.errorCode, executionControlDeadlineAt: heartbeatRuns.executionControlDeadlineAt, @@ -176,6 +180,54 @@ export async function executionProjectionsForRuns( ))); for (const row of offered) waitingByRun.set(row.runId, "openai_dot"); } + // A queued turn cannot start while its predecessor's cleanup is pending + // (for example a stopped run whose runner has not released yet). Name that + // wait instead of an unexplained "Queued". The predecessors are the ones + // admission checks: the task's execution owner, the run this one retries, + // and the agent's latest settled run on the same task, whose in-process + // executor the claim also waits for. Pending uses admission's own predicate. + const settlingPredecessorKeys = new Set(); + const queuedRuns = runs.filter(run => run.status === "queued"); + const issueIdOf = (run: Run) => run.nativeIssueId ?? text(run.contextSnapshot?.issueId); + const queuedIssueIds = [...new Set(queuedRuns.map(issueIdOf).filter((id): id is string => !!id))]; + if (queuedIssueIds.length) { + const issueKey = sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`; + // Two set-based reads for the whole batch, then one bounded, parallel + // admission check per distinct predecessor. + const [owners, latestSettled] = await Promise.all([ + read(async () => db + .select({ id: issues.id, executionRunId: issues.executionRunId }) + .from(issues) + .where(and(eq(issues.companyId, companyId), inArray(issues.id, queuedIssueIds)))), + read(async () => db + .selectDistinctOn([heartbeatRuns.agentId, issueKey], { id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, issueId: issueKey }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.agentId, [...new Set(queuedRuns.map(run => run.agentId))]), + inArray(issueKey, queuedIssueIds), + inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out"]), + )) + .orderBy(heartbeatRuns.agentId, issueKey, desc(heartbeatRuns.createdAt))), + ]); + const ownerByIssue = new Map(owners.map(row => [row.id, row.executionRunId])); + const latestByAgentIssue = new Map(latestSettled.map(row => [`${row.agentId}:${row.issueId}`, row.id])); + const candidatesByRun = new Map(); + for (const run of queuedRuns) { + const issueId = issueIdOf(run); + if (!issueId) continue; + candidatesByRun.set(run.id, [...new Set([ownerByIssue.get(issueId), run.retryOfRunId, + latestByAgentIssue.get(`${run.agentId}:${issueId}`)].filter((id): id is string => !!id && id !== run.id))]); + } + const distinct = [...new Set([...candidatesByRun.values()].flat())]; + // A label is not worth an unbounded fan-out on large list reads. + if (distinct.length <= MAX_PREDECESSOR_CLEANUP_CHECKS) { + const pendingIds = new Set((await Promise.all(distinct.map(async id => + (await read(() => hasPendingRunCleanup(db, companyId, id))) ? id : null))).filter((id): id is string => !!id)); + for (const [runId, candidates] of candidatesByRun) + if (candidates.some(id => pendingIds.has(id))) settlingPredecessorKeys.add(runId); + } + } const coordinatorByRun = new Map(coordinators.map((row) => [row.runId, row])); for (const run of runs) { const issueId = run.nativeIssueId ?? text(run.contextSnapshot?.issueId); @@ -196,6 +248,7 @@ export async function executionProjectionsForRuns( action, now, waitingByRun.get(run.id), + settlingPredecessorKeys.has(run.id), ), ); } @@ -220,13 +273,17 @@ export function projectExecution( recoveryAction: Recovery | undefined, now = new Date(), waitingForProvider?: WaitingExternalProvider, + waitingForPredecessor = false, ): ExecutionProjection { const detail = coordinator?.failureDetail ?? {}; const successorRunId = text(detail.successorRunId); const projection: ExecutionProjection = { - canStop: run.runtimeMode === "native" && run.status === "failed" && run.driverKind !== "grok_bot_mcp" && !!run.nativeIssueId + // A live run is always stoppable by the board (POST /heartbeat-runs/:id/cancel); + // a failed native run is stoppable only while its retry is still pending. + canStop: run.status === "running" || run.status === "queued" || ( + run.runtimeMode === "native" && run.status === "failed" && run.driverKind !== "grok_bot_mcp" && !!run.nativeIssueId && coordinator?.runId === run.id && coordinator.issueId === run.nativeIssueId && coordinator.resultId === null - && coordinator.phase === "retryable_failure", + && coordinator.phase === "retryable_failure"), phase: "queued", label: "Queued", cause: run.errorCode ?? coordinator?.failureCode ?? null, @@ -398,5 +455,10 @@ export function projectExecution( projection.nextAction = "This turn will start automatically when maintenance finishes."; return set("queued", "Waiting for maintenance"); } + if (run.status === "queued" && waitingForPredecessor) { + projection.cause = "previous_run_settling"; + projection.nextAction = "The previous run on this task is still releasing its runner. This turn starts automatically when it settles."; + return set("queued", "Waiting for previous run to stop"); + } return projection; } diff --git a/server/src/services/run-cancellation.ts b/server/src/services/run-cancellation.ts index 009735273e3..9ebe8e4032b 100644 --- a/server/src/services/run-cancellation.ts +++ b/server/src/services/run-cancellation.ts @@ -135,6 +135,16 @@ export function canContinueCancelledRun(run: Run): boolean { run.resultJson?.acpToolInventoryComplete === true && run.resultJson?.acpPendingToolCount === 0; } +/** The user's recorded Stop or interrupt on a settled run. A stopped native + * run whose runner was unreachable can settle as `failed` once its transport + * fails closed; the user's Stop request still bounds which messages are new. */ +export function recordedUserStop(run: Pick) { + if (run.status !== "cancelled" && run.status !== "failed") return null; + const cancellation = readRunCancellation(run.resultJson); + return cancellation && cancellation.initiator.type === "user" && + ["operator", "queued_message"].includes(cancellation.source) ? cancellation : null; +} + /** When a user's Stop or interrupt took effect for message ordering. A native * run finishes only after its provider stop and workspace save settle, which * can take many seconds. A message written after the Stop request is a new @@ -143,9 +153,8 @@ export function canContinueCancelledRun(run: Run): boolean { * `finishedAt`. */ export function userStopBoundaryAt(run: Pick): Date | null { if (!run.finishedAt) return null; - const cancellation = run.status === "cancelled" ? readRunCancellation(run.resultJson) : null; - if (!cancellation || cancellation.initiator.type !== "user" || - !["operator", "queued_message"].includes(cancellation.source)) return run.finishedAt; + const cancellation = recordedUserStop(run); + if (!cancellation) return run.finishedAt; const requestedAt = new Date(cancellation.recordedAt); return Number.isFinite(requestedAt.getTime()) && requestedAt < run.finishedAt ? requestedAt : run.finishedAt; } diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 0722a8ae155..f6e82ce4404 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -4935,7 +4935,11 @@ describe("automatic retained workspace recovery", () => { it.each([ { retried: false, title: "continues a message written during Stop settlement once, without a second message, after recovery already resolved" }, { retried: true, title: "continues a message carried by a workspace-busy retry once, without a second message, after recovery already resolved" }, - ])("$title", async ({ retried }) => { + // S52: the user stopped a hung native run whose runner was unreachable. + // Its transport later failed closed, so the stopped run settled as failed + // (native_provider_terminal_failed) instead of cancelled. + { retried: false, failedClosed: true, title: "continues a message sent after Stop when the stopped run later failed closed" }, + ])("$title", async ({ retried, failedClosed }) => { // S47: the user stops (or interrupts) a native run and sends a message. // The native source finishes only after its stop and workspace save // settle, so its finishedAt is later than the message. Recovery resolves @@ -4947,8 +4951,13 @@ describe("automatic retained workspace recovery", () => { const { f, original, reset: message, wake, queued, deps, control, execution, queueDeps, queue } = t; queueDeps.resumeExecutionWaitComments.mockImplementation(control.resumeExecutionWaitComments); queueDeps.releaseIssueExecutionAndPromote.mockImplementation(control.releaseIssueExecutionAndPromote); - await db.update(issueComments).set({ body: "Actually, give me a 5-bullet summary instead." }).where(eq(issueComments.id, message.id)); - // The Stop was requested before the message (fixture); settlement finished after it. + // The Stop was requested before the message; settlement finished after it. + // Pin the message after the recorded Stop: the fixture writes both within + // the same millisecond on a fast machine, which made this ordering flaky. + const stopRecordedAt = new Date(String(((await run(f.runId)).resultJson?.cancellation as { recordedAt?: string })?.recordedAt)); + message.createdAt = new Date(stopRecordedAt.getTime() + 100); + await db.update(issueComments).set({ body: "Actually, give me a 5-bullet summary instead.", createdAt: message.createdAt }) + .where(eq(issueComments.id, message.id)); await db.update(heartbeatRuns).set({ finishedAt: new Date(message.createdAt.getTime() + 1_000) }) .where(eq(heartbeatRuns.id, f.runId)); let pending = queued; @@ -5009,6 +5018,11 @@ describe("automatic retained workspace recovery", () => { }) ?? []); expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); }, { timeout: 10_000, interval: 10 }); + if (failedClosed) { + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "native_provider_terminal_failed", + error: "Notification transport failed closed: runner_process_termination_unconfirmed" }) + .where(eq(heartbeatRuns.id, f.runId)); + } expect(await queue.claimQueuedRun(await run(pending.id))).toBeNull(); expect(await run(pending.id)).toMatchObject({ status: "cancelled", errorCode: "execution_reconciliation_required", startedAt: null }); const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));