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));