From ce9cf9fe23c623f3c74248524e5a74edf5e5e4d7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:01:10 +0000 Subject: [PATCH 1/8] Make managed workspaces survive pauses and report richer activity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three managed-mode reliability changes for the DevPC platform: The activity endpoint now returns {active, working, pendingWork}. `active` keeps its exact meaning for older control planes; `working` widens it with imminent work — a queued user message no turn has adopted yet (the decider's two-minute grace) and booting provider sessions — so the platform's work claims cover the gap in which a workspace could be paused between "message sent" and "turn running". `pendingWork` surfaces human-blocked work (approvals, user input, actionable plans) as distinct from idle. Turn liveness is now suspend-aware. A sweep arriving far past its schedule means the VM was paused: the paused span no longer counts as provider silence, model-wait turns enter a short resume probation (their streams die with the VM's TCP state) and stall as retryable suspend-silence instead of waiting out the full ten-minute budget, and a turn that fails shortly after a detected resume is auto-retried within the normal budget instead of ending in a manual-retry error. The client detects half-open sockets with a 30-second idle heartbeat on established connections (probe failure recycles the connection through the normal retry path), bounds the managed ws-ticket fetch at ten seconds, and tolerates ~18 seconds of gateway 5xx during bootstrap — matching the gateway's retry-after guidance — instead of showing the fatal card after six. The release workflow's byte-exact activity assertion is updated in lockstep. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .github/workflows/devpc-managed-release.yml | 2 +- apps/server/src/managedDevPcActivity.test.ts | 94 +++++++++++++- apps/server/src/managedDevPcActivity.ts | 77 ++++++++++- .../Layers/TurnLivenessWatchdog.ts | 72 ++++++++++- .../src/orchestration/turnLiveness.test.ts | 52 ++++++++ apps/server/src/orchestration/turnLiveness.ts | 56 ++++++-- apps/web/src/managedDevPc.test.ts | 1 + apps/web/src/managedDevPc.ts | 14 +- .../src/connection/supervisor.test.ts | 38 ++++++ .../src/connection/supervisor.ts | 120 +++++++++++------- 10 files changed, 457 insertions(+), 69 deletions(-) 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..0ffba68c1d21 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 sweep arriving this much later than its schedule 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_THRESHOLD_MS = 2 * DEFAULT_SWEEP_INTERVAL_MS; +/** + * 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,8 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const entries = new Map(); const retryStateByThread = new Map(); + let lastSweepAtMs: number | null = null; + let resumedAtMs: number | null = null; const watchdogCommandId = (tag: string) => crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`watchdog:${tag}:${uuid}`))); @@ -136,7 +151,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 +196,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) { @@ -278,6 +296,22 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const sweep = Effect.gen(function* () { const nowMs = yield* Clock.currentTimeMillis; + // A sweep landing far past its schedule means the whole VM was + // suspended. The paused span must not read as provider silence, and + // model-wait turns whose streams died with the suspend get a short + // probation instead of the full silence budget. + if ( + lastSweepAtMs !== null && + nowMs - lastSweepAtMs - sweepIntervalMs >= SUSPEND_GAP_THRESHOLD_MS + ) { + resumedAtMs = nowMs; + rebaseAfterSuspend(entries, nowMs); + yield* Effect.logInfo("turn.watchdog.resumed-after-suspend", { + suspendedForMs: nowMs - lastSweepAtMs - sweepIntervalMs, + rebasedTurnCount: entries.size, + }); + } + lastSweepAtMs = nowMs; const stalled = stalledTurns(entries, nowMs, { modelSilenceMs, recoveryGraceMs }); for (const turn of stalled) { yield* interruptStalledTurn(turn).pipe( @@ -316,6 +350,32 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => retryStateByThread.delete(event.threadId); } const nowMs = yield* Clock.currentTimeMillis; + // A tracked 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. + if ( + (event.type === "turn.aborted" || + event.type === "session.exited" || + event.type === "runtime.error") && + resumedAtMs !== null && + nowMs - resumedAtMs <= POST_RESUME_FAILURE_WINDOW_MS && + maxAutoRetries > 0 + ) { + const tracked = entries.get(event.threadId); + if (tracked !== undefined) { + const retryState = currentRetryState(event.threadId, nowMs); + if (retryState.attempts < maxAutoRetries && retryState.scheduled === null) { + 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..edb354e4c325 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, @@ -202,4 +203,55 @@ 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("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..0372521f4954 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"; @@ -91,10 +100,11 @@ 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; the same proof ends resume probation. + const base: TurnLiveness = + current.seededFromSnapshot || current.resumedProbation === true + ? { ...current, seededFromSnapshot: false, resumedProbation: false } + : current; switch (event.type) { case "item.started": @@ -151,7 +161,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 +206,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..4e1a50499bc2 100644 --- a/apps/web/src/managedDevPc.test.ts +++ b/apps/web/src/managedDevPc.test.ts @@ -713,6 +713,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..e7299fd59b70 100644 --- a/apps/web/src/managedDevPc.ts +++ b/apps/web/src/managedDevPc.ts @@ -736,13 +736,21 @@ 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. + if (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 +768,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..282dd64f76ca 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,84 @@ 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 "ConnectRequested": + case "Wakeup": + 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": From d5d7c4f7bdda0043a8b39106378db5c747260ffe Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:10:17 +0000 Subject: [PATCH 2/8] Address review: event-path resume detection, retry gating, failure triage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The suspend detector now runs in the runtime-event path as well as the sweep — the first thing delivered after a resume is often the dying provider stream's own terminal event, up to a full sweep interval before the timer fires, and retry attribution must already know about the resume by then. The detection threshold derives from the configured sweep interval, and the post-resume retry only covers turns that were on resume probation, so fresh turns and deliberate aborts are never second-guessed. Client side: a credentials-changed relay wakeup arriving mid-probe now detaches like the main monitor path instead of being swallowed; bootstrap retries are reserved for transient failures (network, 408/429/5xx) while a definitive 4xx fails fast; and the ticket-timeout test asserts the actual ten-second bound. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .../Layers/TurnLivenessWatchdog.ts | 65 ++++++++++++------- apps/web/src/managedDevPc.test.ts | 3 + apps/web/src/managedDevPc.ts | 27 +++++++- .../src/connection/supervisor.ts | 13 +++- 4 files changed, 81 insertions(+), 27 deletions(-) diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index 0ffba68c1d21..a04b087f9b25 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -33,11 +33,11 @@ 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 sweep arriving this much later than its schedule 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. + * 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_THRESHOLD_MS = 2 * DEFAULT_SWEEP_INTERVAL_MS; +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 @@ -102,9 +102,36 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const entries = new Map(); const retryStateByThread = new Map(); - let lastSweepAtMs: number | null = null; + 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}`))); @@ -296,22 +323,7 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => const sweep = Effect.gen(function* () { const nowMs = yield* Clock.currentTimeMillis; - // A sweep landing far past its schedule means the whole VM was - // suspended. The paused span must not read as provider silence, and - // model-wait turns whose streams died with the suspend get a short - // probation instead of the full silence budget. - if ( - lastSweepAtMs !== null && - nowMs - lastSweepAtMs - sweepIntervalMs >= SUSPEND_GAP_THRESHOLD_MS - ) { - resumedAtMs = nowMs; - rebaseAfterSuspend(entries, nowMs); - yield* Effect.logInfo("turn.watchdog.resumed-after-suspend", { - suspendedForMs: nowMs - lastSweepAtMs - sweepIntervalMs, - rebasedTurnCount: entries.size, - }); - } - lastSweepAtMs = nowMs; + yield* observeClock(nowMs); const stalled = stalledTurns(entries, nowMs, { modelSilenceMs, recoveryGraceMs }); for (const turn of stalled) { yield* interruptStalledTurn(turn).pipe( @@ -350,10 +362,17 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => retryStateByThread.delete(event.threadId); } const nowMs = yield* Clock.currentTimeMillis; - // A tracked turn dying shortly after a resume-from-suspend is the + // 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, including deliberate user aborts, and are not + // second-guessed here. if ( (event.type === "turn.aborted" || event.type === "session.exited" || @@ -363,7 +382,7 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => maxAutoRetries > 0 ) { const tracked = entries.get(event.threadId); - if (tracked !== undefined) { + if (tracked?.resumedProbation === true) { const retryState = currentRetryState(event.threadId, nowMs); if (retryState.attempts < maxAutoRetries && retryState.scheduled === null) { retryState.scheduled = { turnId: tracked.turnId, notBeforeMs: nowMs + retryDelayMs }; diff --git a/apps/web/src/managedDevPc.test.ts b/apps/web/src/managedDevPc.test.ts index 4e1a50499bc2..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", ); diff --git a/apps/web/src/managedDevPc.ts b/apps/web/src/managedDevPc.ts index e7299fd59b70..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; @@ -740,8 +760,9 @@ export async function prepareManagedDevPc(): Promise { // 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. - if (failures >= 12) { + // 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, diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 282dd64f76ca..cd61531b40bd 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -435,8 +435,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( return "detach" as const; } break; - case "ConnectRequested": 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; } } From ec50429c0868eb198217573af1782726e5118768 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:19:41 +0000 Subject: [PATCH 3/8] Cover failed completions and commanded stops in post-resume retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two provider-event realities the post-resume retry missed: some providers report a pause-broken turn as turn.completed with a failed state (ahead of their own error or session-exit events), which both escaped the retry whitelist and wrongly refilled the retry budget; and a Stop the user issues right after a resume produces a turn.aborted that must never be auto-restarted. Failed completions now schedule the budgeted retry before the tracking entry is dropped, and a commanded interrupt — which projects the turn as interrupted before the provider's terminal event arrives — is never second-guessed, with unknown projection state erring toward respecting the stop. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .../Layers/TurnLivenessWatchdog.ts | 42 +++++++++++++++---- 1 file changed, 35 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index a04b087f9b25..d797351ed8f4 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -356,9 +356,17 @@ 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. + if (event.type === "turn.completed" && !failedCompletion) { retryStateByThread.delete(event.threadId); } const nowMs = yield* Clock.currentTimeMillis; @@ -371,20 +379,40 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => // 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, including deliberate user aborts, and are not - // second-guessed here. + // their own reasons and are not second-guessed here. if ( (event.type === "turn.aborted" || event.type === "session.exited" || - event.type === "runtime.error") && + event.type === "runtime.error" || + failedCompletion) && resumedAtMs !== null && nowMs - resumedAtMs <= POST_RESUME_FAILURE_WINDOW_MS && maxAutoRetries > 0 ) { const tracked = entries.get(event.threadId); if (tracked?.resumedProbation === true) { + // 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 (retryState.attempts < maxAutoRetries && retryState.scheduled === null) { + if ( + uncommanded && + retryState.attempts < maxAutoRetries && + retryState.scheduled === null + ) { retryState.scheduled = { turnId: tracked.turnId, notBeforeMs: nowMs + retryDelayMs }; retryStateByThread.set(event.threadId, retryState); yield* Effect.logInfo("turn.watchdog.retry-after-suspend-failure", { From be811a7cde7649223de6bca28b671f82f2b3b9d6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:26:57 +0000 Subject: [PATCH 4/8] Settle bare aborts before scheduling their post-resume retry An uncommanded turn.aborted with no session-level follow-up never settles the projected turn (ingestion has no turn.aborted lifecycle case), so the scheduled retry was rejected by the dispatcher's settled-state guard and silently cleared. The watchdog now interrupts the still-running projected turn explicitly before scheduling, the same settling its stall path performs, so the retry can land. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .../Layers/TurnLivenessWatchdog.ts | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index d797351ed8f4..3d6c1096158e 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -413,6 +413,32 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => 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", { From 4b26d9693a9f4be5e488fd6fa12930be5ba80e95 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:37:54 +0000 Subject: [PATCH 5/8] Scope post-resume recovery to the tracked turn and yield to newer work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review-found edges in the recovery path: resume probation is now cleared only by events tied to the tracked turn (a thread-scoped side event like an account update proves nothing about the stream); a delayed failure event scoped to a superseded turn no longer interrupts or retries the current one; and a scheduled retry yields when the human has moved on — a user message newer than the failed turn, or a session already booting, cancels the dispatch instead of re-driving the old message against it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .../Layers/TurnLivenessWatchdog.ts | 30 +++++++++++++++++-- .../src/orchestration/turnLiveness.test.ts | 19 ++++++++++++ apps/server/src/orchestration/turnLiveness.ts | 8 +++-- 3 files changed, 53 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index 3d6c1096158e..57cf2c9add1a 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -259,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)); @@ -269,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; @@ -390,7 +409,14 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => maxAutoRetries > 0 ) { const tracked = entries.get(event.threadId); - if (tracked?.resumedProbation === true) { + // 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 diff --git a/apps/server/src/orchestration/turnLiveness.test.ts b/apps/server/src/orchestration/turnLiveness.test.ts index edb354e4c325..9b14d6679eb3 100644 --- a/apps/server/src/orchestration/turnLiveness.test.ts +++ b/apps/server/src/orchestration/turnLiveness.test.ts @@ -234,6 +234,25 @@ describe("turnLiveness", () => { ).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); + const stalled = stalledTurns( + new Map([[threadId, next]]), + 3_601_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( diff --git a/apps/server/src/orchestration/turnLiveness.ts b/apps/server/src/orchestration/turnLiveness.ts index 0372521f4954..a3d86ee8fd38 100644 --- a/apps/server/src/orchestration/turnLiveness.ts +++ b/apps/server/src/orchestration/turnLiveness.ts @@ -100,9 +100,13 @@ 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; the same proof ends resume probation. + // 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 + current.seededFromSnapshot || (current.resumedProbation === true && provesTrackedTurn) ? { ...current, seededFromSnapshot: false, resumedProbation: false } : current; From 817aafda1e821289ad84740006a3875e58c7400f Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:43:33 +0000 Subject: [PATCH 6/8] Keep tracking the current turn through stale terminal events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The liveness fold ended tracking on any turn.completed or turn.aborted, including a delayed event flushed for a superseded turn — deleting the current turn's entry so it could no longer stall or recover. Terminal events scoped to a different turn now leave the tracked entry untouched; unscoped ones settle it as before. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- apps/server/src/orchestration/turnLiveness.test.ts | 12 ++++++++++++ apps/server/src/orchestration/turnLiveness.ts | 7 +++++++ 2 files changed, 19 insertions(+) diff --git a/apps/server/src/orchestration/turnLiveness.test.ts b/apps/server/src/orchestration/turnLiveness.test.ts index 9b14d6679eb3..e2cb50ead8f6 100644 --- a/apps/server/src/orchestration/turnLiveness.test.ts +++ b/apps/server/src/orchestration/turnLiveness.test.ts @@ -61,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(); }); diff --git a/apps/server/src/orchestration/turnLiveness.ts b/apps/server/src/orchestration/turnLiveness.ts index a3d86ee8fd38..43645279221d 100644 --- a/apps/server/src/orchestration/turnLiveness.ts +++ b/apps/server/src/orchestration/turnLiveness.ts @@ -88,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: From 2bb74560fe2ad32ab3b75af92b30d22e8bb06977 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:49:05 +0000 Subject: [PATCH 7/8] Refill the retry budget only for completions of the tracked turn A delayed successful completion flushed for a superseded turn was still deleting the thread's retry state, so repeated stale completions could defeat the advertised maxAutoRetries cap. The refill now requires the completion to match the tracked turn when one is tracked. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .../src/orchestration/Layers/TurnLivenessWatchdog.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index 57cf2c9add1a..4cf046af0a6f 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -384,8 +384,16 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => 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. - if (event.type === "turn.completed" && !failedCompletion) { + // toward the cap. A stale completion flushed for a superseded turn + // proves nothing and must not defeat the cap. + const trackedForRefill = entries.get(event.threadId); + if ( + event.type === "turn.completed" && + !failedCompletion && + (trackedForRefill === undefined || + event.turnId === undefined || + event.turnId === trackedForRefill.turnId) + ) { retryStateByThread.delete(event.threadId); } const nowMs = yield* Clock.currentTimeMillis; From a2954b74dfac8dfe966aeb73e9d4788a85e42154 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 03:54:58 +0000 Subject: [PATCH 8/8] Hold probation deadlines and scheduled retries against stale completions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unrelated events during resume probation no longer extend the recovery deadline (periodic side events could keep a dead stream from ever stalling), and a stale successful completion for a different turn no longer cancels a retry scheduled for a failed one — when nothing is tracked, the refill now matches against the scheduled retry's turn. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017eomHzc8Ese8TSm5WZBCX9 --- .../Layers/TurnLivenessWatchdog.ts | 19 +++++++++++-------- .../src/orchestration/turnLiveness.test.ts | 5 ++++- apps/server/src/orchestration/turnLiveness.ts | 6 ++++++ 3 files changed, 21 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts index 4cf046af0a6f..272ea66011e7 100644 --- a/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts +++ b/apps/server/src/orchestration/Layers/TurnLivenessWatchdog.ts @@ -385,15 +385,18 @@ const makeTurnLivenessWatchdog = (options?: TurnLivenessWatchdogLiveOptions) => // 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. + // 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); - if ( - event.type === "turn.completed" && - !failedCompletion && - (trackedForRefill === undefined || - event.turnId === undefined || - event.turnId === trackedForRefill.turnId) - ) { + 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; diff --git a/apps/server/src/orchestration/turnLiveness.test.ts b/apps/server/src/orchestration/turnLiveness.test.ts index e2cb50ead8f6..dae52341e155 100644 --- a/apps/server/src/orchestration/turnLiveness.test.ts +++ b/apps/server/src/orchestration/turnLiveness.test.ts @@ -257,9 +257,12 @@ describe("turnLiveness", () => { 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_601_000 + thresholds.recoveryGraceMs, + 3_600_000 + thresholds.recoveryGraceMs, thresholds, ); expect(stalled.map((turn) => turn.reason)).toEqual(["suspend-silence"]); diff --git a/apps/server/src/orchestration/turnLiveness.ts b/apps/server/src/orchestration/turnLiveness.ts index 43645279221d..808a8de11003 100644 --- a/apps/server/src/orchestration/turnLiveness.ts +++ b/apps/server/src/orchestration/turnLiveness.ts @@ -116,6 +116,12 @@ export function applyRuntimeEvent( 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":