diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 0255c80eb94..38f15cb8164 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -66,6 +66,7 @@ import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts" import * as NodeServices from "@effect/platform-node/NodeServices"; import * as Clock from "effect/Clock"; import { ServerSettingsService } from "../../serverSettings.ts"; +import { ServerActivation } from "../../serverActivation.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import * as GitWorkflowService from "../../git/GitWorkflowService.ts"; @@ -157,6 +158,8 @@ describe("ProviderCommandReactor", () => { readonly sessionModelSwitch?: "unsupported" | "in-session"; readonly requiresNewThreadForModelChange?: boolean; readonly deferReactorStart?: boolean; + /** When set, recovery continuations park until this effect succeeds (production shape). */ + readonly serverActivation?: Effect.Effect; readonly providerBindings?: ReadonlyArray; readonly providerBindingsMap?: Map; readonly listBindingsEffect?: () => Effect.Effect< @@ -454,7 +457,12 @@ describe("ProviderCommandReactor", () => { const startReactor = async () => { if (reactorStarted) return; reactorStarted = true; - await Effect.runPromise(reactor.start().pipe(Scope.provide(scope!))); + const start = reactor.start().pipe(Scope.provide(scope!)); + await runtime!.runPromise( + input?.serverActivation !== undefined + ? start.pipe(Effect.provideService(ServerActivation, input.serverActivation)) + : start, + ); }; if (input?.deferReactorStart !== true) { await startReactor(); @@ -750,7 +758,7 @@ describe("ProviderCommandReactor", () => { }); }); - it("does not finish reactor startup before restart reconciliation completes", async () => { + it("does not finish reactor startup before restart recovery claims complete", async () => { const reconciliationGate = Effect.runSync(Deferred.make()); const harness = await createHarness({ deferReactorStart: true, @@ -769,6 +777,83 @@ describe("ProviderCommandReactor", () => { expect(startupFinished).toBe(true); }); + it("finishes reactor startup while recovery provider session start is still pending", async () => { + const modelSelection: ModelSelection = { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-recovery", + }; + const threadId = ThreadId.make("thread-1"); + const activation = Effect.runSync(Deferred.make()); + const releaseStart = Effect.runSync(Deferred.make()); + const harness = await createHarness({ + deferReactorStart: true, + threadModelSelection: modelSelection, + // Production path: recovery continuations park on ServerActivation. + serverActivation: Deferred.await(activation), + startSessionEffect: (session) => Deferred.await(releaseStart).pipe(Effect.as(session)), + providerBindings: [ + { + threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "stopped", + resumeCursor: { threadId: "provider-thread-resume" }, + runtimePayload: { + cwd: "/tmp/persisted-recovery-cwd", + modelSelection, + interactionMode: "plan", + activeTurnId: null, + restartRecovery: makeProviderRestartRecoveryMarker({ + interruptedProviderTurnId: asTurnId("provider-turn-before-restart"), + shutdownAt: "2026-01-01T00:00:01.000Z", + }), + }, + lastSeenAt: "2026-01-01T00:00:01.000Z", + }, + ], + }); + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-running-before-restart-hang"), + threadId, + message: { + messageId: asMessageId("running-user-message-hang"), + role: "user", + text: "original request", + attachments: [], + }, + modelSelection, + interactionMode: "plan", + runtimeMode: "full-access", + createdAt: "2026-01-01T00:00:00.000Z", + }), + ); + + // Claim phase must complete without waiting on activation or provider start. + await harness.startReactor(); + expect(harness.startSession).not.toHaveBeenCalled(); + expect(harness.sendTurn).not.toHaveBeenCalled(); + expect(harness.providerBindings.get(threadId)?.runtimePayload).toMatchObject({ + restartRecovery: expect.objectContaining({ version: 1 }), + lastRuntimeEvent: "provider.restartRecovery.claimed", + }); + + Effect.runSync(Deferred.succeed(activation, undefined)); + await waitFor(() => harness.startSession.mock.calls.length === 1); + expect(harness.sendTurn).not.toHaveBeenCalled(); + + Effect.runSync(Deferred.succeed(releaseStart, undefined)); + await harness.drain(); + await waitFor(() => harness.sendTurn.mock.calls.length === 1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId, + input: RESTART_RECOVERY_CONTINUATION_INSTRUCTION, + }); + }); + it("does not resume settled turns from stale recovery bindings", async () => { const modelSelection: ModelSelection = { instanceId: ProviderInstanceId.make("codex"), diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 8fc41cce25e..6a7cb3380aa 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -39,6 +39,7 @@ import { import { ProviderAdapterRequestError } from "../../provider/Errors.ts"; import type { ProviderServiceError } from "../../provider/Errors.ts"; import { + makeProviderRestartRecoveryMarker, readPersistedProviderCwd, readPersistedProviderInteractionMode, readPersistedProviderModelSelection, @@ -172,10 +173,19 @@ function formatThreadTitleContext( }; } const STARTUP_RECOVERY_CONCURRENCY = 4; +/** Bound recovery provider calls so a hung agent cannot pin reactor start forever. */ +const STARTUP_RECOVERY_PROVIDER_TIMEOUT = Duration.seconds(45); export const RESTART_RECOVERY_CONTINUATION_INSTRUCTION = "The server restarted while you were working. Inspect the conversation and current workspace state, verify which side effects from the interrupted turn already happened, and continue the unfinished work safely. Do not repeat completed work or assume an earlier tool call failed merely because its response is absent."; +function runtimePayloadRecord(value: unknown): Record { + if (value !== null && typeof value === "object" && !Array.isArray(value)) { + return { ...(value as Record) }; + } + return {}; +} + export function providerErrorLabel(value: string | undefined): string { const normalized = value?.trim(); return normalized && normalized.length > 0 ? normalized : "unknown"; @@ -1425,7 +1435,26 @@ const make = Effect.gen(function* () { }); }); - const recoverInterruptedTurn = Effect.fn("recoverInterruptedTurn")(function* (input: { + type ClaimedInterruptedRecovery = { + readonly binding: ProviderRuntimeBindingWithMetadata; + readonly candidate: ProviderRestartRecoveryCandidate; + readonly thread: { + readonly id: ThreadId; + readonly projectId: ProjectId; + readonly worktreePath: string | null; + readonly runtimeMode: RuntimeMode; + readonly modelSelection: ModelSelection; + readonly interactionMode: ProviderInteractionMode; + readonly latestTurn?: { readonly turnId?: TurnId | null } | null; + }; + readonly createdAt: string; + }; + + /** + * Sync claim only: durable marker + interrupted projection so orphan settle + * cannot wipe recovery intent, without waiting on provider start/sendTurn. + */ + const claimInterruptedTurn = Effect.fn("claimInterruptedTurn")(function* (input: { readonly binding: ProviderRuntimeBindingWithMetadata; readonly candidate: ProviderRestartRecoveryCandidate; }) { @@ -1436,7 +1465,7 @@ const make = Effect.gen(function* () { reason: "duplicate-in-boot", provider: binding.provider, }); - return; + return undefined; } recoveredThreadIds.add(binding.threadId); @@ -1452,7 +1481,7 @@ const make = Effect.gen(function* () { reason: "inactive-thread", provider: binding.provider, }); - return; + return undefined; } const createdAt = DateTime.formatIso(yield* DateTime.now); @@ -1502,30 +1531,88 @@ const make = Effect.gen(function* () { reason: "projected-turn-settled", provider: binding.provider, }); - return; + return undefined; } + interruptedRecoveryThreadIds.add(thread.id); + const runtimeMode = binding.runtimeMode ?? thread.runtimeMode; + const recoveryMarker = makeProviderRestartRecoveryMarker({ + interruptedProviderTurnId: candidate.interruptedProviderTurnId, + shutdownAt: candidate.shutdownAt, + }); - const recover = Effect.gen(function* () { - const runtimeMode = binding.runtimeMode ?? thread.runtimeMode; - // This lifecycle transition settles the concrete old projection row as - // interrupted before validation or replacement work can proceed. - yield* setThreadSession({ - threadId: thread.id, - session: { - threadId: thread.id, - status: "interrupted", - providerName: binding.provider, - ...(binding.providerInstanceId !== undefined - ? { providerInstanceId: binding.providerInstanceId } - : {}), - runtimeMode, + // Persist a durable marker and demote live-claiming status before orphan + // audit. Orphan settle spreads payload fields, so the marker survives even + // if a late stop rewrites the binding; status "stopped" skips live claims. + yield* providerSessionDirectory + .upsert({ + threadId: binding.threadId, + provider: binding.provider, + ...(binding.providerInstanceId !== undefined + ? { providerInstanceId: binding.providerInstanceId } + : {}), + ...(binding.runtimeMode !== undefined ? { runtimeMode: binding.runtimeMode } : {}), + status: "stopped", + ...(binding.resumeCursor !== undefined && binding.resumeCursor !== null + ? { resumeCursor: binding.resumeCursor } + : {}), + runtimePayload: { + ...runtimePayloadRecord(binding.runtimePayload), activeTurnId: null, - lastError: null, - updatedAt: createdAt, + restartRecovery: recoveryMarker, + lastRuntimeEvent: "provider.restartRecovery.claimed", + lastRuntimeEventAt: createdAt, }, - createdAt, - }); + }) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("provider restart recovery claim was not persisted", { + threadId: binding.threadId, + cause: Cause.pretty(cause), + }), + ), + ); + + // Settle the concrete old projection row as interrupted before replacement. + yield* setThreadSession({ + threadId: thread.id, + session: { + threadId: thread.id, + status: "interrupted", + providerName: binding.provider, + ...(binding.providerInstanceId !== undefined + ? { providerInstanceId: binding.providerInstanceId } + : {}), + runtimeMode, + activeTurnId: null, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }); + + yield* Effect.logInfo("provider turn restart recovery claimed", { + threadId: thread.id, + provider: binding.provider, + recoverySource: candidate.source, + interruptedProviderTurnId: candidate.interruptedProviderTurnId, + }); + + return { + binding, + candidate, + thread, + createdAt, + } satisfies ClaimedInterruptedRecovery; + }); + + const continueInterruptedTurn = Effect.fn("continueInterruptedTurn")(function* ( + input: ClaimedInterruptedRecovery, + ) { + const { binding, candidate, thread, createdAt } = input; + + const recover = Effect.gen(function* () { + const runtimeMode = binding.runtimeMode ?? thread.runtimeMode; const providerInstanceId = binding.providerInstanceId; if (providerInstanceId === undefined) { @@ -1575,15 +1662,25 @@ const make = Effect.gen(function* () { readPersistedProviderCwd(binding.runtimePayload) ?? resolveThreadWorkspaceCwd({ thread, projects: project ? [project] : [] }); - const session = yield* providerService.startSession(thread.id, { - threadId: thread.id, - provider: binding.provider, - providerInstanceId, - ...(cwd !== undefined ? { cwd } : {}), - modelSelection, - resumeCursor: binding.resumeCursor, - runtimeMode, - }); + const sessionResult = yield* providerService + .startSession(thread.id, { + threadId: thread.id, + provider: binding.provider, + providerInstanceId, + ...(cwd !== undefined ? { cwd } : {}), + modelSelection, + resumeCursor: binding.resumeCursor, + runtimeMode, + }) + .pipe(Effect.interruptible, Effect.timeoutOption(STARTUP_RECOVERY_PROVIDER_TIMEOUT)); + if (Option.isNone(sessionResult)) { + return yield* new ProviderAdapterRequestError({ + provider: binding.provider, + method: "provider.turn.restart-recovery", + detail: `Provider session start timed out after ${Duration.format(STARTUP_RECOVERY_PROVIDER_TIMEOUT)} during restart recovery.`, + }); + } + const session = sessionResult.value; yield* setThreadSession({ threadId: thread.id, session: { @@ -1599,13 +1696,23 @@ const make = Effect.gen(function* () { createdAt, }); - const replacement = yield* providerService.sendTurn({ - threadId: thread.id, - input: RESTART_RECOVERY_CONTINUATION_INSTRUCTION, - attachments: [], - modelSelection, - interactionMode, - }); + const replacementResult = yield* providerService + .sendTurn({ + threadId: thread.id, + input: RESTART_RECOVERY_CONTINUATION_INSTRUCTION, + attachments: [], + modelSelection, + interactionMode, + }) + .pipe(Effect.interruptible, Effect.timeoutOption(STARTUP_RECOVERY_PROVIDER_TIMEOUT)); + if (Option.isNone(replacementResult)) { + return yield* new ProviderAdapterRequestError({ + provider: binding.provider, + method: "provider.turn.restart-recovery", + detail: `Provider recovery sendTurn timed out after ${Duration.format(STARTUP_RECOVERY_PROVIDER_TIMEOUT)} during restart recovery.`, + }); + } + const replacement = replacementResult.value; // ProviderService clears this in the accepted sendTurn transaction. The // explicit write keeps the reconciliation invariant local and obvious. @@ -1816,15 +1923,21 @@ const make = Effect.gen(function* () { ); } - yield* Effect.forEach(recoveryCandidates, recoverInterruptedTurn, { + // Claim phase is synchronous and must finish before orphan settle (next + // startup phase). Provider startSession/sendTurn runs after activation so a + // hung agent cannot block HTTP readiness / Discord oauth bootstrap. + const claimedRecoveries = yield* Effect.forEach(recoveryCandidates, claimInterruptedTurn, { concurrency: STARTUP_RECOVERY_CONCURRENCY, - discard: true, - }); + }).pipe( + Effect.map((claims) => claims.flatMap((claim) => (claim === undefined ? [] : [claim]))), + ); const pendingWithoutInterruptedRecovery = pendingTurnStarts.filter( (pending) => !interruptedRecoveryThreadIds.has(pending.threadId), ); - if (pendingWithoutInterruptedRecovery.length === 0) return; + if (pendingWithoutInterruptedRecovery.length === 0) { + return claimedRecoveries; + } const persistedEvents = yield* Stream.runCollect( orchestrationEngine.readEvents(0, Number.MAX_SAFE_INTEGER), @@ -1907,6 +2020,8 @@ const make = Effect.gen(function* () { ), { concurrency: STARTUP_RECOVERY_CONCURRENCY, discard: true }, ); + + return claimedRecoveries; }); const start: ProviderCommandReactorShape["start"] = Effect.fn("start")(function* () { @@ -1966,17 +2081,37 @@ const make = Effect.gen(function* () { yield* forkParked(clearInterrupted); } - // Startup recovery must finish before server startup settles orphaned - // sessions. Otherwise the orphan audit can clear the persisted recovery - // marker or stop the replacement process while it is being started. - yield* reconcileStartup().pipe( + // Claim interrupted recoveries (and enqueue pending turn starts) before this + // reactor returns. Server startup then runs orphan settle against claimed + // durable markers, not against live-claiming zombie runtimes. + // + // Provider startSession/sendTurn is intentionally parked until activation so + // a hung agent cannot block command readiness (HTTP/oauth). drain still + // waits for those continuations via startupReconciliationDone. + const claimedRecoveries = yield* reconcileStartup().pipe( Effect.catchCause((cause) => Effect.logWarning("provider restart reconciliation failed", { cause: Cause.pretty(cause), - }), + }).pipe(Effect.as([] as ReadonlyArray)), ), + ); + + const runClaimedRecoveries = Effect.forEach(claimedRecoveries, continueInterruptedTurn, { + concurrency: STARTUP_RECOVERY_CONCURRENCY, + discard: true, + }).pipe( Effect.ensuring(Deferred.succeed(startupReconciliationDone, undefined).pipe(Effect.ignore)), ); + + // Mirror clearInterrupted: unit tests omit ServerActivation and run + // recovery inline for determinism. Production installs activation so + // startSession/sendTurn wait until after orphan settle + the readiness + // boundary, and cannot block HTTP/oauth on a hung agent. + if (activation === undefined) { + yield* runClaimedRecoveries; + } else { + yield* forkParked(runClaimedRecoveries); + } }); return { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 4feb3aa166f..3d2ca136ff0 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -79,7 +79,7 @@ const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z"); import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; import * as ServerConfig from "./config.ts"; import * as HttpResponseCompression from "./httpCompression/HttpResponseCompression.ts"; -import { makeRoutesLayer } from "./server.ts"; +import { isCommandReadinessExemptPath, makeRoutesLayer } from "./server.ts"; import { resolveAvailableEditorsForConfig } from "./ws.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as GrokTranscriptResync from "./externalSessions/GrokTranscriptResync.ts"; @@ -1438,6 +1438,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it("exempts bootstrap paths from command readiness", () => { + assert.isTrue(isCommandReadinessExemptPath("/oauth/token")); + assert.isTrue(isCommandReadinessExemptPath("/oauth/token?grant_type=client_credentials")); + assert.isTrue(isCommandReadinessExemptPath("/.well-known/t3/environment")); + assert.isTrue(isCommandReadinessExemptPath("/api/t3-connect/health")); + assert.isTrue(isCommandReadinessExemptPath("/api/connect/health")); + assert.isFalse(isCommandReadinessExemptPath("/")); + assert.isFalse(isCommandReadinessExemptPath("/api/orchestration/snapshot")); + assert.isFalse(isCommandReadinessExemptPath("/api/auth/session")); + }); + it.effect("serves static index content for GET / when staticDir is configured", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 82674e84bee..47a0d5e5583 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -4,7 +4,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schedule from "effect/Schedule"; -import { FetchHttpClient, HttpRouter, HttpServer } from "effect/unstable/http"; +import { FetchHttpClient, HttpRouter, HttpServer, HttpServerRequest } from "effect/unstable/http"; import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import * as ServerConfig from "./config.ts"; @@ -494,11 +494,32 @@ const RuntimeDependenciesLive = RuntimeCoreWithIntegrationsLive.pipe( Layer.provide(NetService.layer), ); +/** + * Bootstrap / liveness surfaces that must not wait for provider recovery or the + * broader command-readiness gate. Discord bot token exchange and environment + * discovery hang the whole client fleet if they sit behind a stuck recovery turn. + */ +export function isCommandReadinessExemptPath(url: string): boolean { + const pathOnly = (url.split("?", 1)[0] ?? url).split("#", 1)[0] ?? url; + return ( + pathOnly === "/oauth/token" || + pathOnly === "/.well-known/t3/environment" || + pathOnly === "/api/t3-connect/health" || + pathOnly === "/api/connect/health" + ); +} + const commandReadinessLayer = HttpRouter.middleware( (httpEffect) => - Effect.flatMap(ServerRuntimeStartup.ServerRuntimeStartup, (startup) => - startup.awaitCommandReady.pipe(Effect.orDie, Effect.andThen(httpEffect)), - ), + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + if (isCommandReadinessExemptPath(request.url)) { + return yield* httpEffect; + } + const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; + yield* startup.awaitCommandReady.pipe(Effect.orDie); + return yield* httpEffect; + }), { global: true }, ); diff --git a/oxlint-plugin-t3code/rules/no-manual-effect-runtime-in-tests.ts b/oxlint-plugin-t3code/rules/no-manual-effect-runtime-in-tests.ts index 237135f7a74..1369371023d 100644 --- a/oxlint-plugin-t3code/rules/no-manual-effect-runtime-in-tests.ts +++ b/oxlint-plugin-t3code/rules/no-manual-effect-runtime-in-tests.ts @@ -29,7 +29,7 @@ const LEGACY_BASELINE = new Map([ ["apps/server/src/orchestration/Layers/CheckpointReactor.test.ts", 42], ["apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts", 5], ["apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts", 4], - ["apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts", 70], + ["apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts", 74], ["apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts", 31], ["apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts", 2], ["apps/server/src/orchestration/projector.test.ts", 21],