diff --git a/.github/workflows/devpc-managed-release.yml b/.github/workflows/devpc-managed-release.yml index 23425781f9b5..b5fcc6650ef5 100644 --- a/.github/workflows/devpc-managed-release.yml +++ b/.github/workflows/devpc-managed-release.yml @@ -128,7 +128,7 @@ jobs: activity="$(curl --connect-timeout 2 --max-time 5 --fail --silent --show-error \ -H 'x-devpc-gateway-token: release-smoke-managed-gateway-token' \ http://127.0.0.1:4311/api/_devpc/activity)" - test "$activity" = '{"active":false}' || { + test "$activity" = '{"active":false,"working":false,"pendingWork":false}' || { echo "Unexpected managed activity response: $activity" exit 1 } diff --git a/apps/server/src/managedDevPcActivity.test.ts b/apps/server/src/managedDevPcActivity.test.ts index 47e8d2630c13..ec733d354714 100644 --- a/apps/server/src/managedDevPcActivity.test.ts +++ b/apps/server/src/managedDevPcActivity.test.ts @@ -1,7 +1,14 @@ import type { OrchestrationThreadShell } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import { describe, expect, it } from "vite-plus/test"; -import { hasRunningManagedTurn } from "./managedDevPcActivity.ts"; +import { + hasPendingManagedWork, + hasQueuedManagedTurnStart, + hasRunningManagedTurn, + hasStartingManagedSession, + QUEUED_TURN_START_GRACE_MS, +} from "./managedDevPcActivity.ts"; import { clearPrimeAgentSessionActivity, hasRunningPrimeAgentSubagents, @@ -14,11 +21,34 @@ const thread = ( ) => ({ latestTurn: state === null ? null : { state }, + latestUserMessageAt: null, + session: null, hasPendingApprovals: waiting === "approval", hasPendingUserInput: waiting === "input", hasActionableProposedPlan: waiting === "plan", }) as OrchestrationThreadShell; +const NOW_MS = Date.parse("2026-08-09T12:00:00.000Z"); +const iso = (offsetMs: number) => DateTime.formatIso(DateTime.makeUnsafe(NOW_MS + offsetMs)); + +const queuedThread = ( + overrides: Partial<{ + latestUserMessageAt: string | null; + latestTurn: OrchestrationThreadShell["latestTurn"]; + sessionStatus: "idle" | "starting" | "running" | "ready" | "interrupted" | "stopped" | "error"; + }> = {}, +) => + ({ + latestTurn: overrides.latestTurn ?? null, + latestUserMessageAt: + overrides.latestUserMessageAt === undefined ? iso(-5_000) : overrides.latestUserMessageAt, + session: + overrides.sessionStatus === undefined ? null : ({ status: overrides.sessionStatus } as never), + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + }) as OrchestrationThreadShell; + describe("managed DevPC activity", () => { it("reports activity only while an AI turn is running", () => { expect(hasRunningManagedTurn([thread("completed"), thread("running")])).toBe(true); @@ -30,6 +60,68 @@ describe("managed DevPC activity", () => { expect(hasRunningManagedTurn([])).toBe(false); }); + it("counts a fresh user message no turn adopted yet as queued work", () => { + expect(hasQueuedManagedTurnStart([queuedThread()], NOW_MS)).toBe(true); + // Adopted: the turn's requestedAt is newer than the message. + expect( + hasQueuedManagedTurnStart( + [ + queuedThread({ + latestTurn: { + state: "running", + requestedAt: iso(-1_000), + startedAt: null, + completedAt: null, + } as never, + }), + ], + NOW_MS, + ), + ).toBe(false); + // Stale: outside the grace window in either direction. + expect( + hasQueuedManagedTurnStart( + [queuedThread({ latestUserMessageAt: iso(-QUEUED_TURN_START_GRACE_MS - 1_000) })], + NOW_MS, + ), + ).toBe(false); + expect( + hasQueuedManagedTurnStart( + [queuedThread({ latestUserMessageAt: iso(QUEUED_TURN_START_GRACE_MS + 1_000) })], + NOW_MS, + ), + ).toBe(false); + // An errored session cannot adopt the message. + expect(hasQueuedManagedTurnStart([queuedThread({ sessionStatus: "error" })], NOW_MS)).toBe( + false, + ); + expect(hasQueuedManagedTurnStart([queuedThread({ latestUserMessageAt: null })], NOW_MS)).toBe( + false, + ); + }); + + it("counts a booting provider session as work in progress", () => { + expect( + hasStartingManagedSession([ + queuedThread({ latestUserMessageAt: null, sessionStatus: "starting" }), + ]), + ).toBe(true); + expect( + hasStartingManagedSession([ + queuedThread({ latestUserMessageAt: null, sessionStatus: "ready" }), + ]), + ).toBe(false); + expect(hasStartingManagedSession([queuedThread({ latestUserMessageAt: null })])).toBe(false); + }); + + it("reports human-blocked work as pending, separately from running work", () => { + expect(hasPendingManagedWork([thread("running", "approval")])).toBe(true); + expect(hasPendingManagedWork([thread("completed", "input")])).toBe(true); + expect(hasPendingManagedWork([thread("interrupted", "plan")])).toBe(true); + expect(hasPendingManagedWork([thread("running")])).toBe(false); + expect(hasPendingManagedWork([])).toBe(false); + }); + it("keeps a workspace active for detached Prime subagents, not a resident process", () => { const sessionKey = "primeAgent:thread-managed"; expect(hasRunningPrimeAgentSubagents()).toBe(false); diff --git a/apps/server/src/managedDevPcActivity.ts b/apps/server/src/managedDevPcActivity.ts index 49ecd01af112..cec6d270c151 100644 --- a/apps/server/src/managedDevPcActivity.ts +++ b/apps/server/src/managedDevPcActivity.ts @@ -1,6 +1,7 @@ import * as NodeCrypto from "node:crypto"; import type { OrchestrationThreadShell } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"; @@ -31,6 +32,60 @@ export function hasRunningManagedTurn(threads: ReadonlyArray, + nowMs: number, +): boolean { + return threads.some((thread) => { + if (thread.session?.status === "error") return false; + if (thread.latestUserMessageAt === null) return false; + const messageAtMs = Date.parse(thread.latestUserMessageAt); + if (!Number.isFinite(messageAtMs)) return false; + const latestTurnAtMs = + thread.latestTurn === null + ? Number.NEGATIVE_INFINITY + : Math.max( + ...[ + thread.latestTurn.requestedAt, + thread.latestTurn.startedAt, + thread.latestTurn.completedAt, + ].map((candidate) => + candidate == null ? Number.NEGATIVE_INFINITY : Date.parse(candidate), + ), + ); + return ( + messageAtMs > latestTurnAtMs && Math.abs(nowMs - messageAtMs) <= QUEUED_TURN_START_GRACE_MS + ); + }); +} + +/** A provider session mid-boot is about to run a turn; that is agent work too. */ +export function hasStartingManagedSession( + threads: ReadonlyArray, +): boolean { + return threads.some((thread) => thread.session?.status === "starting"); +} + +/** + * Work that exists but is blocked on the human: pause-safe (the workspace may + * idle out while an approval waits), yet worth surfacing so the platform can + * distinguish "nothing to do" from "waiting on the user". + */ +export function hasPendingManagedWork(threads: ReadonlyArray): boolean { + return threads.some( + (thread) => + thread.hasPendingApprovals || thread.hasPendingUserInput || thread.hasActionableProposedPlan, + ); +} + const handleManagedDevPcActivity = Effect.gen(function* () { const request = yield* HttpServerRequest.HttpServerRequest; const config = yield* ServerConfig.ServerConfig; @@ -45,6 +100,7 @@ const handleManagedDevPcActivity = Effect.gen(function* () { } const snapshots = yield* ProjectionSnapshotQuery; + const nowMs = yield* Clock.currentTimeMillis; return yield* snapshots.getShellSnapshot().pipe( Effect.match({ onFailure: () => @@ -52,13 +108,26 @@ const handleManagedDevPcActivity = Effect.gen(function* () { { error: { code: "ACTIVITY_UNAVAILABLE", message: "Activity is unavailable." } }, { status: 503, headers: { "cache-control": "no-store" } }, ), - onSuccess: (snapshot) => - HttpServerResponse.jsonUnsafe( + onSuccess: (snapshot) => { + // `active` keeps its original meaning (a genuinely running turn) so + // control planes reading only that field see unchanged behavior. + // `working` widens it with imminent work — a queued turn start or a + // booting session — which must hold a work claim before the turn's + // running state lands. `pendingWork` is human-blocked work: pause-safe + // but not "idle". + const active = hasRunningManagedTurn(snapshot.threads) || hasRunningPrimeAgentSubagents(); + return HttpServerResponse.jsonUnsafe( { - active: hasRunningManagedTurn(snapshot.threads) || hasRunningPrimeAgentSubagents(), + active, + working: + active || + hasQueuedManagedTurnStart(snapshot.threads, nowMs) || + hasStartingManagedSession(snapshot.threads), + pendingWork: hasPendingManagedWork(snapshot.threads), }, { status: 200, headers: { "cache-control": "no-store" } }, - ), + ); + }, }), ); }); diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index 9cee195371a9..272ea66011e7 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -18,6 +18,7 @@ import { } from "../Services/TurnLivenessWatchdog.ts"; import { applyRuntimeEvent, + rebaseAfterSuspend, seededTurnLiveness, stalledTurns, type StalledTurn, @@ -31,6 +32,18 @@ const DEFAULT_MAX_AUTO_RETRIES = 2; const DEFAULT_RETRY_DELAY_MS = 30 * 1000; /** Failed attempts stop counting against the budget after this long. */ const RETRY_BUDGET_EXPIRY_MS = 24 * 60 * 60 * 1000; +/** + * A clock reading arriving this many sweep intervals past the previous one + * means the VM was suspended, not that the event loop hiccuped. Two full + * intervals of slack keeps load-induced jitter from reading as a suspend. + */ +const SUSPEND_GAP_SWEEP_INTERVALS = 2; +/** + * Window after a detected resume in which a turn failure is attributed to + * the suspend (its provider connection died with the VM's TCP state) and is + * therefore auto-retried within the normal budget. + */ +const POST_RESUME_FAILURE_WINDOW_MS = 3 * 60 * 1000; export interface TurnLivenessWatchdogLiveOptions { readonly modelSilenceMs?: number; @@ -89,6 +102,35 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const entries = new Map(); const retryStateByThread = new Map(); + let lastObservedClockMs: number | null = null; + let resumedAtMs: number | null = null; + + /** + * Advance the shared clock baseline and detect a VM suspend: a reading + * far past the last one plus a sweep interval means this machine slept + * in between. Runs in BOTH the sweep and the runtime-event path — the + * first thing delivered after a resume is often the dying provider + * stream's terminal event, up to a full sweep interval before the timer + * fires, and retry attribution must already know about the resume by + * then. + */ + const observeClock = (nowMs: number) => + Effect.gen(function* () { + const previous = lastObservedClockMs; + lastObservedClockMs = nowMs; + if ( + previous === null || + nowMs - previous - sweepIntervalMs < SUSPEND_GAP_SWEEP_INTERVALS * sweepIntervalMs + ) { + return; + } + resumedAtMs = nowMs; + rebaseAfterSuspend(entries, nowMs); + yield* Effect.logInfo("turn.watchdog.resumed-after-suspend", { + suspendedForMs: nowMs - previous - sweepIntervalMs, + rebasedTurnCount: entries.size, + }); + }); const watchdogCommandId = (tag: string) => crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`watchdog:${tag}:${uuid}`))); @@ -136,7 +178,9 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const summary = stalled.reason === "orphaned-after-restart" ? "This turn did not survive a server restart and was marked interrupted." - : `No provider activity for ${describeSilence(stalled.silentForMs)} — the turn was interrupted.`; + : stalled.reason === "suspend-silence" + ? "The workspace was paused mid-turn and the provider connection did not survive the resume — the turn was interrupted." + : `No provider activity for ${describeSilence(stalled.silentForMs)} — the turn was interrupted.`; const activityCommandId = yield* watchdogCommandId("activity"); const activityId = yield* crypto.randomUUIDv4.pipe( @@ -179,11 +223,12 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => silentForMs: stalled.silentForMs, }); - // A model-silence stall is usually a dead provider connection, which - // a fresh turn recovers from — schedule one bounded retry. Restart - // orphans are not retried: the budget bookkeeping died with the old - // process, and retrying there could loop across repeated crashes. - if (stalled.reason !== "model-silence" || maxAutoRetries < 1) return; + // A model-silence or suspend-silence stall is a dead provider + // connection, which a fresh turn recovers from — schedule one bounded + // retry. Restart orphans are not retried: the budget bookkeeping died + // with the old process, and retrying there could loop across + // repeated crashes. + if (stalled.reason === "orphaned-after-restart" || maxAutoRetries < 1) return; const nowMs = yield* Clock.currentTimeMillis; const retryState = currentRetryState(stalled.threadId, nowMs); if (retryState.attempts >= maxAutoRetries) { @@ -214,7 +259,10 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => // Only retry if the interrupt actually landed and the human has not // already moved the thread on: the failed turn must still be the - // latest, in a failed state, with nothing else running. + // latest, in a failed state, with nothing else running. A user + // message newer than the failed turn, or a session already booting + // for one, is the human moving on — re-driving the old message + // would race their newer work and repeat side effects. const thread = yield* projectionSnapshotQuery .getThreadShellById(threadId) .pipe(Effect.map(Option.getOrUndefined)); @@ -224,10 +272,26 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => latestTurn == null || latestTurn.turnId !== scheduled.turnId || (latestTurn.state !== "interrupted" && latestTurn.state !== "error") || + thread.session?.status === "starting" || (thread.session?.status === "running" && thread.session.activeTurnId !== null) ) { return; } + const latestTurnAtMs = Math.max( + Date.parse(latestTurn.requestedAt), + latestTurn.startedAt === null + ? Number.NEGATIVE_INFINITY + : Date.parse(latestTurn.startedAt), + latestTurn.completedAt === null + ? Number.NEGATIVE_INFINITY + : Date.parse(latestTurn.completedAt), + ); + if ( + thread.latestUserMessageAt !== null && + Date.parse(thread.latestUserMessageAt) > latestTurnAtMs + ) { + return; + } const nowMs = yield* Clock.currentTimeMillis; retryState.attempts += 1; @@ -278,6 +342,7 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const sweep = Effect.gen(function* () { const nowMs = yield* Clock.currentTimeMillis; + yield* observeClock(nowMs); const stalled = stalledTurns(entries, nowMs, { modelSilenceMs, recoveryGraceMs }); for (const turn of stalled) { yield* interruptStalledTurn(turn).pipe( @@ -310,12 +375,117 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const followRuntimeEvents = Stream.runForEach(providerService.streamEvents, (event) => Effect.gen(function* () { - // A completed turn is the proof of recovery that refills the - // auto-retry budget; anything less keeps counting toward the cap. - if (event.type === "turn.completed") { + // Some providers report a broken turn as a completion carrying a + // failed state (ahead of their own error / session-exit events), so + // "completed" alone is not proof of recovery. + const failedCompletion = + event.type === "turn.completed" && + "state" in event.payload && + event.payload.state === "failed"; + // A successfully completed turn is the proof of recovery that + // refills the auto-retry budget; anything less keeps counting + // toward the cap. A stale completion flushed for a superseded turn + // proves nothing and must not defeat the cap — nor cancel a retry + // already scheduled for a different failed turn. + const trackedForRefill = entries.get(event.threadId); + const scheduledRetryTurnId = + retryStateByThread.get(event.threadId)?.scheduled?.turnId ?? null; + const completionMatchesKnownTurn = + trackedForRefill !== undefined + ? event.turnId === undefined || event.turnId === trackedForRefill.turnId + : scheduledRetryTurnId !== null + ? event.turnId === undefined || event.turnId === scheduledRetryTurnId + : true; + if (event.type === "turn.completed" && !failedCompletion && completionMatchesKnownTurn) { retryStateByThread.delete(event.threadId); } const nowMs = yield* Clock.currentTimeMillis; + // The resume must be known BEFORE this event is judged: the first + // thing delivered after a suspend is often the dying stream's own + // terminal event, well ahead of the next sweep timer. + yield* observeClock(nowMs); + // A probation turn dying shortly after a resume-from-suspend is the + // suspend's doing (its provider stream went down with the VM's TCP + // state), not the model's — schedule a budgeted retry so the work + // survives the pause instead of ending in a manual-retry error. + // Turns started after the resume (or already proven live) fail for + // their own reasons and are not second-guessed here. + if ( + (event.type === "turn.aborted" || + event.type === "session.exited" || + event.type === "runtime.error" || + failedCompletion) && + resumedAtMs !== null && + nowMs - resumedAtMs <= POST_RESUME_FAILURE_WINDOW_MS && + maxAutoRetries > 0 + ) { + const tracked = entries.get(event.threadId); + // A delayed failure event scoped to some OTHER turn (a stale abort + // from a superseded turn surviving in a provider queue) says + // nothing about the tracked turn — session-scoped events carry no + // turnId and pass. + const eventTurnMatchesTracked = + tracked !== undefined && + (event.turnId === undefined || event.turnId === tracked.turnId); + if (tracked?.resumedProbation === true && eventTurnMatchesTracked) { + // A commanded interrupt (the user's Stop, or this watchdog's + // own stall handling) projects the turn as interrupted before + // the provider's terminal event arrives — deliberate stops must + // never be auto-restarted. A pause-broken stream dies with no + // command, leaving the projection running (or errored once the + // failure is ingested). Unknown projection state errs toward + // respecting the stop. + const thread = yield* projectionSnapshotQuery.getThreadShellById(event.threadId).pipe( + Effect.map(Option.getOrUndefined), + Effect.orElseSucceed(() => undefined), + ); + const latestTurn = thread?.latestTurn; + const uncommanded = + latestTurn != null && + latestTurn.turnId === tracked.turnId && + latestTurn.state !== "interrupted"; + const retryState = currentRetryState(event.threadId, nowMs); + if ( + uncommanded && + retryState.attempts < maxAutoRetries && + retryState.scheduled === null + ) { + // A bare abort with no session-level follow-up never settles + // the projected turn on its own (ingestion has no turn.aborted + // lifecycle case), and the retry dispatcher only accepts + // settled turns. Interrupt it explicitly so the scheduled + // retry can land instead of being silently rejected. + if (event.type === "turn.aborted" && latestTurn.state === "running") { + yield* Effect.gen(function* () { + const createdAt = DateTime.formatIso(yield* DateTime.now); + const interruptCommandId = yield* watchdogCommandId("suspend-interrupt"); + yield* orchestrationEngine.dispatch({ + type: "thread.turn.interrupt", + commandId: interruptCommandId, + threadId: event.threadId, + turnId: tracked.turnId, + createdAt, + }); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("turn.watchdog.suspend-interrupt-failed", { + threadId: event.threadId, + turnId: tracked.turnId, + cause, + }), + ), + ); + } + retryState.scheduled = { turnId: tracked.turnId, notBeforeMs: nowMs + retryDelayMs }; + retryStateByThread.set(event.threadId, retryState); + yield* Effect.logInfo("turn.watchdog.retry-after-suspend-failure", { + threadId: event.threadId, + turnId: tracked.turnId, + eventType: event.type, + }); + } + } + } const next = applyRuntimeEvent(entries.get(event.threadId), event, nowMs); if (next === null) { entries.delete(event.threadId); diff --git a/apps/server/src/orchestration/turnLiveness.test.ts b/apps/server/src/orchestration/turnLiveness.test.ts index 2b783fe90de1..dae52341e155 100644 --- a/apps/server/src/orchestration/turnLiveness.test.ts +++ b/apps/server/src/orchestration/turnLiveness.test.ts @@ -10,6 +10,7 @@ import { describe, expect, it } from "vite-plus/test"; import { applyRuntimeEvent, classifyWait, + rebaseAfterSuspend, seededTurnLiveness, stalledTurns, type TurnLiveness, @@ -60,6 +61,18 @@ describe("turnLiveness", () => { expect(applyRuntimeEvent(started, event({ type: "session.exited" }), 2_000)).toBeNull(); }); + it("keeps tracking the current turn through stale terminal events", () => { + const started = runningTurn(1_000); + // A delayed abort or completion flushed for a superseded turn says + // nothing about the tracked one. + expect( + applyRuntimeEvent(started, event({ type: "turn.aborted", turnId: "turn-old" }), 2_000), + ).toBe(started); + expect( + applyRuntimeEvent(started, event({ type: "turn.completed", turnId: "turn-old" }), 2_000), + ).toBe(started); + }); + it("ignores events for threads with no tracked turn", () => { expect(applyRuntimeEvent(undefined, event({ type: "content.delta" }), 1_000)).toBeNull(); }); @@ -202,4 +215,77 @@ describe("turnLiveness", () => { )!; expect(classifyWait(liveness)).toBe("model"); }); + + it("rebases model waits into resume probation after a suspend", () => { + const entries = new Map([[threadId, runningTurn(0)]]); + // The VM was paused; the wall clock jumped far past the silence budget. + const resumedAt = 3_600_000; + rebaseAfterSuspend(entries, resumedAt); + + // The paused hour does not read as silence... + expect(stalledTurns(entries, resumedAt + 1_000, thresholds)).toEqual([]); + // ...but the dead model stream only gets the short recovery grace, and + // the stall is retryable suspend-silence, not a restart orphan. + const stalled = stalledTurns(entries, resumedAt + thresholds.recoveryGraceMs, thresholds); + expect(stalled).toHaveLength(1); + expect(stalled[0]!.reason).toBe("suspend-silence"); + }); + + it("clears resume probation when a live event proves the stream survived", () => { + const entries = new Map([[threadId, runningTurn(0)]]); + rebaseAfterSuspend(entries, 3_600_000); + const next = applyRuntimeEvent( + entries.get(threadId), + event({ type: "content.delta" }), + 3_601_000, + )!; + expect(next.resumedProbation).toBe(false); + // Back on the full silence budget. + expect( + stalledTurns(new Map([[threadId, next]]), 3_601_000 + thresholds.recoveryGraceMs, thresholds), + ).toEqual([]); + }); + + it("keeps resume probation through events unrelated to the tracked turn", () => { + const entries = new Map([[threadId, runningTurn(0)]]); + rebaseAfterSuspend(entries, 3_600_000); + // A thread-scoped side event (or one for a superseded turn) proves + // nothing about the tracked turn's stream. + const next = applyRuntimeEvent( + entries.get(threadId), + event({ type: "account.updated", turnId: "turn-other" }), + 3_601_000, + )!; + expect(next.resumedProbation).toBe(true); + // The deadline is not extended either: repeated side events must not + // keep a dead stream from stalling. + expect(next.lastEventAtMs).toBe(3_600_000); + const stalled = stalledTurns( + new Map([[threadId, next]]), + 3_600_000 + thresholds.recoveryGraceMs, + thresholds, + ); + expect(stalled.map((turn) => turn.reason)).toEqual(["suspend-silence"]); + }); + + it("does not put tool waits or restart-seeded turns on resume probation", () => { + let toolWait = runningTurn(0); + toolWait = applyRuntimeEvent( + toolWait, + event({ type: "item.started", itemId: "item-3", payload: { itemType: "command_execution" } }), + 1_000, + )!; + const seeded = seededTurnLiveness(TurnId.make("turn-seeded"), 0); + const entries = new Map([ + [threadId, toolWait], + [ThreadId.make("thread-2"), seeded], + ]); + rebaseAfterSuspend(entries, 3_600_000); + + expect(entries.get(threadId)!.resumedProbation).toBe(false); + // A local tool keeps running through a pause; no stall however long it takes. + expect(stalledTurns(entries, 7_200_000, thresholds).map((turn) => turn.reason)).toEqual([ + "orphaned-after-restart", + ]); + }); }); diff --git a/apps/server/src/orchestration/turnLiveness.ts b/apps/server/src/orchestration/turnLiveness.ts index 0465e0ba1b04..808a8de11003 100644 --- a/apps/server/src/orchestration/turnLiveness.ts +++ b/apps/server/src/orchestration/turnLiveness.ts @@ -33,6 +33,15 @@ export interface TurnLiveness { * arrives for the thread shortly after boot, the turn is an orphan. */ readonly seededFromSnapshot: boolean; + /** + * True after the VM this server runs on was suspended and resumed while + * the turn was waiting on the model. The process survived the suspend but + * a model-wait stream almost never does, so the turn is judged against the + * short recovery grace instead of the full silence budget — and unlike a + * restart orphan it is safe to auto-retry, because the retry budget + * bookkeeping survived with the process. + */ + readonly resumedProbation?: boolean; } export type TurnWaitClass = "human" | "tool" | "model"; @@ -79,6 +88,13 @@ export function applyRuntimeEvent( } case "turn.completed": case "turn.aborted": + // A delayed terminal event for a superseded turn must not end tracking + // of the current one — providers can flush stale queue entries after a + // new turn started. Unscoped terminal events settle the tracked turn. + if (current !== undefined && event.turnId !== undefined && event.turnId !== current.turnId) { + return current; + } + return null; case "session.exited": return null; default: @@ -91,10 +107,21 @@ export function applyRuntimeEvent( if (current === undefined) return null; // A live event for a thread seeded from the snapshot means the turn - // outlived the restart after all; from here on track it as live. - const base: TurnLiveness = current.seededFromSnapshot - ? { ...current, seededFromSnapshot: false } - : current; + // outlived the restart after all. Resume probation demands more: only an + // event tied to the tracked turn proves the model stream survived — a + // thread-scoped side event (an account update, a config warning) proves + // nothing about the stream and must not soften the probation. + const provesTrackedTurn = event.turnId !== undefined && event.turnId === current.turnId; + const base: TurnLiveness = + current.seededFromSnapshot || (current.resumedProbation === true && provesTrackedTurn) + ? { ...current, seededFromSnapshot: false, resumedProbation: false } + : current; + // While on probation, an unrelated event must not extend the recovery + // deadline either — a periodic side event (rate limits, config warnings) + // could otherwise keep a dead model stream from ever stalling. + if (base.resumedProbation === true && !provesTrackedTurn) { + return base; + } switch (event.type) { case "item.started": @@ -151,7 +178,32 @@ export interface StalledTurn { readonly threadId: ThreadId; readonly turnId: TurnId; readonly silentForMs: number; - readonly reason: "model-silence" | "orphaned-after-restart"; + readonly reason: "model-silence" | "orphaned-after-restart" | "suspend-silence"; +} + +/** + * Rebase liveness across a detected VM suspend (the host paused this machine + * and resumed it later; the wall clock jumped while nothing actually ran). + * + * The paused interval must not count toward any silence budget, so every + * entry's clock restarts at the resume. Turns that were waiting on the model + * additionally enter resume probation: their provider stream almost certainly + * died with the suspended TCP connections, so instead of the full silence + * budget they get the short recovery grace to prove themselves with a live + * event. Tool and human waits carry no such suspicion — local subprocesses + * survive a suspend, and human waits are unbounded by design. + */ +export function rebaseAfterSuspend( + entries: Map, + resumedAtMs: number, +): void { + for (const [threadId, liveness] of entries) { + entries.set(threadId, { + ...liveness, + lastEventAtMs: resumedAtMs, + resumedProbation: !liveness.seededFromSnapshot && classifyWait(liveness) === "model", + }); + } } /** @@ -171,15 +223,18 @@ export function stalledTurns( for (const [threadId, liveness] of entries) { if (classifyWait(liveness) !== "model") continue; const silentForMs = nowMs - liveness.lastEventAtMs; - const threshold = liveness.seededFromSnapshot - ? thresholds.recoveryGraceMs - : thresholds.modelSilenceMs; + const probation = liveness.seededFromSnapshot || liveness.resumedProbation === true; + const threshold = probation ? thresholds.recoveryGraceMs : thresholds.modelSilenceMs; if (silentForMs < threshold) continue; stalled.push({ threadId, turnId: liveness.turnId, silentForMs, - reason: liveness.seededFromSnapshot ? "orphaned-after-restart" : "model-silence", + reason: liveness.seededFromSnapshot + ? "orphaned-after-restart" + : liveness.resumedProbation === true + ? "suspend-silence" + : "model-silence", }); } return stalled; diff --git a/apps/web/src/managedDevPc.test.ts b/apps/web/src/managedDevPc.test.ts index 1b36bb9050c8..225ecdc04e03 100644 --- a/apps/web/src/managedDevPc.test.ts +++ b/apps/web/src/managedDevPc.test.ts @@ -695,12 +695,15 @@ describe("managed DevPC WebSocket authorization", () => { Response.json({ ticket: "gateway-ticket-that-is-long-enough" }), ); vi.stubGlobal("fetch", fetchMock); + const timeoutSpy = vi.spyOn(AbortSignal, "timeout"); const { prepareManagedWebSocketUrl } = await import("./managedDevPc"); const resolved = await prepareManagedWebSocketUrl( "wss://app.example.test/ws?wsTicket=local-one-time-ticket", ); + expect(timeoutSpy).toHaveBeenCalledWith(10_000); + expect(resolved).toBe( "wss://app.example.test/ws?gatewayTicket=gateway-ticket-that-is-long-enough", ); @@ -713,6 +716,7 @@ describe("managed DevPC WebSocket authorization", () => { "content-type": "application/json", }, body: "{}", + signal: expect.any(AbortSignal), }); }); diff --git a/apps/web/src/managedDevPc.ts b/apps/web/src/managedDevPc.ts index ba402576b37a..ac698fd3b5a2 100644 --- a/apps/web/src/managedDevPc.ts +++ b/apps/web/src/managedDevPc.ts @@ -89,6 +89,26 @@ export function isAmbiguousLifecycleResponse(status: number): boolean { return status === 408 || status >= 500; } +/** A non-ok bootstrap response, carrying its status for retryability triage. */ +class ManagedBootstrapHttpError extends Error { + constructor(readonly status: number) { + super(`Workspace bootstrap returned ${status}.`); + } +} + +/** + * Only failures the gateway may heal on its own deserve the extended retry + * budget: network errors, timeouts, rate limiting, and 5xx. A definitive 4xx + * (an expired session, a missing workspace) will not change in eighteen + * seconds of polling. + */ +function isTransientBootstrapFailure(error: unknown): boolean { + if (error instanceof ManagedBootstrapHttpError) { + return isAmbiguousLifecycleResponse(error.status) || error.status === 429; + } + return true; +} + export function requiresManagedResume(bootstrap: ManagedDevPcBootstrap): boolean { const resumable = new Set(["paused", "stopped"]); if (bootstrap.status) return resumable.has(bootstrap.status); @@ -658,7 +678,7 @@ export async function prepareManagedDevPc(): Promise { headers: { accept: "application/json" }, }); if (!response.ok) { - throw new Error(`Workspace bootstrap returned ${response.status}.`); + throw new ManagedBootstrapHttpError(response.status); } const bootstrap = (await response.json()) as ManagedDevPcBootstrap; window.__DEVPC_MANAGED_BOOTSTRAP__ = bootstrap; @@ -736,13 +756,22 @@ export async function prepareManagedDevPc(): Promise { await new Promise((resolve) => window.setTimeout(resolve, 1_500)); } catch (error) { failures += 1; - if (failures >= 4) { + // The gateway answers transient control-plane trouble with 503 + + // retry-after while it recovers, which can span a relay handoff or a + // control-plane deploy (~15–20 s). Twelve polls at 1.5 s gives ~18 s + // of tolerance before the fatal card, instead of giving up during a + // blip the platform is already healing. Definitive failures skip the + // budget entirely. + if (!isTransientBootstrapFailure(error) || failures >= 12) { updateBootstrapMessage( error instanceof Error ? error.message : "The workspace could not be reached.", true, ); throw error; } + if (failures >= 2) { + showWakeProgress("Reconnecting to your workspace…", "connection"); + } await new Promise((resolve) => window.setTimeout(resolve, 1_500)); } } @@ -760,6 +789,10 @@ export async function prepareManagedWebSocketUrl(socketUrl: string): Promise { }), ); + it.effect("heartbeats an idle connection and recycles it when the probe fails", () => + Effect.gen(function* () { + const probeCount = yield* Ref.make(0); + const harness = yield* makeHarness({ + probe: (attempt) => + Ref.update(probeCount, (count) => count + 1).pipe( + Effect.andThen( + attempt === 1 ? Effect.fail(transient("The socket is half-open.")) : Effect.void, + ), + ), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + // No traffic and no wakeups: the idle heartbeat alone must find the + // dead socket and recycle the connection through the normal retry path. + yield* TestClock.adjust("30 seconds"); + yield* awaitState(supervisor.state, (state) => state.phase === "backoff"); + expect(yield* Ref.get(probeCount)).toBe(1); + + yield* TestClock.adjust("1 second"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + + // A healthy connection just keeps heartbeating. + yield* TestClock.adjust("30 seconds"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(yield* Ref.get(probeCount)).toBe(2); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("keeps blocked failures idle until an external signal requests another attempt", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d9efcd4263a0..cd61531b40bd 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -33,6 +33,13 @@ const RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const; const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds"; const CONNECTION_PROBE_TIMEOUT = "15 seconds"; const BACKOFF_RESET_AFTER_MS = 30_000; +/** + * Idle heartbeat cadence on an established connection. A socket that dies + * without a close frame (a dropped relay, a NATed path timing out) otherwise + * stays "connected" indefinitely while every subscription silently starves — + * the probe turns that into a detected failure and a normal reconnect. + */ +const CONNECTION_HEARTBEAT_INTERVAL = "30 seconds"; interface SupervisorIntent { readonly desired: boolean; @@ -386,65 +393,95 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const monitorConnectedLease = Effect.fnUntraced(function* ( lease: ConnectionDriver.EnvironmentConnectionLease, ) { + /** + * Probe the live session, draining supervisor signals while the probe is + * in flight. Returns "detach" when a signal ended the lease, "ok" when + * the probe succeeded; a failed or timed-out probe fails the monitor, + * which recycles the connection through the normal retry path. + */ + const probeLease = Effect.gen(function* () { + const probe = yield* lease.session.probe.pipe( + Effect.timeoutOrElse({ + duration: CONNECTION_PROBE_TIMEOUT, + orElse: () => + Effect.fail( + new ConnectionTransientError({ + reason: "timeout", + detail: `${target.label} did not respond to a connection health check.`, + }), + ), + }), + Effect.forkChild, + ); + for (;;) { + const probeEvent = yield* Effect.raceFirst( + Fiber.await(probe).pipe( + Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })), + ), + Queue.take(signals).pipe(Effect.map((signal) => ({ _tag: "Signal" as const, signal }))), + ); + if (probeEvent._tag === "ProbeCompleted") { + yield* probeEvent.exit; + return "ok" as const; + } + switch (probeEvent.signal._tag) { + case "DisconnectRequested": + case "RetryRequested": + yield* Fiber.interrupt(probe); + return "detach" as const; + case "NetworkChanged": + if (probeEvent.signal.network === "offline") { + yield* Fiber.interrupt(probe); + return "detach" as const; + } + break; + case "Wakeup": + // The same relay account-change handling as the main monitor + // loop — a probe in flight must not swallow it. + if ( + probeEvent.signal.reason === "credentials-changed" && + target._tag === "RelayConnectionTarget" + ) { + yield* logManagedRelayAccountChange; + yield* Fiber.interrupt(probe); + return "detach" as const; + } + break; + case "ConnectRequested": + break; + } + } + }); + for (;;) { - const next = yield* Queue.take(signals); - switch (next._tag) { + const next = yield* Queue.take(signals).pipe( + Effect.map((signal) => ({ _tag: "Signal" as const, signal })), + Effect.timeoutOrElse({ + duration: CONNECTION_HEARTBEAT_INTERVAL, + orElse: () => Effect.succeed({ _tag: "HeartbeatDue" as const }), + }), + ); + if (next._tag === "HeartbeatDue") { + if ((yield* probeLease) === "detach") return; + continue; + } + const signal = next.signal; + switch (signal._tag) { case "DisconnectRequested": case "RetryRequested": return; case "NetworkChanged": - if (next.network === "offline") { + if (signal.network === "offline") { return; } break; case "Wakeup": - if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") { + if (signal.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") { yield* logManagedRelayAccountChange; return; } - if (next.reason === "application-active") { - const probe = yield* lease.session.probe.pipe( - Effect.timeoutOrElse({ - duration: CONNECTION_PROBE_TIMEOUT, - orElse: () => - Effect.fail( - new ConnectionTransientError({ - reason: "timeout", - detail: `${target.label} did not respond to a connection health check.`, - }), - ), - }), - Effect.forkChild, - ); - for (;;) { - const probeEvent = yield* Effect.raceFirst( - Fiber.await(probe).pipe( - Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })), - ), - Queue.take(signals).pipe( - Effect.map((signal) => ({ _tag: "Signal" as const, signal })), - ), - ); - if (probeEvent._tag === "ProbeCompleted") { - yield* probeEvent.exit; - break; - } - switch (probeEvent.signal._tag) { - case "DisconnectRequested": - case "RetryRequested": - yield* Fiber.interrupt(probe); - return; - case "NetworkChanged": - if (probeEvent.signal.network === "offline") { - yield* Fiber.interrupt(probe); - return; - } - break; - case "ConnectRequested": - case "Wakeup": - break; - } - } + if (signal.reason === "application-active") { + if ((yield* probeLease) === "detach") return; } break; case "ConnectRequested":