From 59b3207f604e369e807dee48dafd1393ce7afffa Mon Sep 17 00:00:00 2001 From: Raj D <25481060+radroid@users.noreply.github.com> Date: Sat, 25 Jul 2026 10:05:14 -0400 Subject: [PATCH] fix(server): reconcile crash-frozen `running` turns to interrupted on boot Fixes #5. When the backend exits ungracefully (SIGKILL/crash/OOM, no SIGTERM), the Effect finalizer a clean quit runs (ProviderService.runStopAll -> adapter.stopAll() -> a session-set that settles running turns to "interrupted") never fires. In-flight turns are left frozen at projection_turns.state="running" with active_turn_id still set; on a later resume the provider reports the cut-off turn as cancelled/aborted, and auto-resume never recovers them. At fresh boot no turn is actually live (adapter session maps start empty; runtime rows carry no pid/epoch), so every leftover in-flight turn is a crash remnant. Add a one-shot boot reconciler that mirrors the graceful path: for each such thread it dispatches a stopped `thread.session.set` command (through the event log, not a direct projection write), settling the running turn to the resumable "interrupted" state and clearing active_turn_id. Wired as a startup phase before reactors start and before commands are accepted. Scope: reconciling to a resumable interrupted state satisfies the issue's primary acceptance. Auto-resume RE-ARMING (a crash marker + boot producer to auto-continue the turn) is left as a documented follow-up. Adds CrashRecoveryReconciler.test.ts (precondition, reconcile, idempotency, multi-thread selectivity, event-log backing) and extends serverRuntimeStartup and autoResume/guards tests. Co-Authored-By: Claude Opus 4.8 --- .../Layers/CrashRecoveryReconciler.test.ts | 296 ++++++++++++++++++ .../Layers/CrashRecoveryReconciler.ts | 141 +++++++++ apps/server/src/serverRuntimeStartup.test.ts | 150 ++++++++- apps/server/src/serverRuntimeStartup.ts | 29 ++ apps/server/src/t3x/autoResume/guards.test.ts | 10 + 5 files changed, 625 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/orchestration/Layers/CrashRecoveryReconciler.test.ts create mode 100644 apps/server/src/orchestration/Layers/CrashRecoveryReconciler.ts diff --git a/apps/server/src/orchestration/Layers/CrashRecoveryReconciler.test.ts b/apps/server/src/orchestration/Layers/CrashRecoveryReconciler.test.ts new file mode 100644 index 00000000000..a38b2d6cebb --- /dev/null +++ b/apps/server/src/orchestration/Layers/CrashRecoveryReconciler.test.ts @@ -0,0 +1,296 @@ +import { + CommandId, + DEFAULT_PROVIDER_INTERACTION_MODE, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationCommand, + type OrchestrationEvent, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as ManagedRuntime from "effect/ManagedRuntime"; +import * as Stream from "effect/Stream"; +import { describe, expect, it } from "vite-plus/test"; + +import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; +import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; +import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; +import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts"; +import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; +import { ServerConfig } from "../../config.ts"; +import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; +import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; +import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { reconcileInterruptedTurnsOnBoot } from "./CrashRecoveryReconciler.ts"; + +const MODEL_SELECTION = { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", +}; + +async function createReconcilerSystem() { + const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { + prefix: "t3-crash-recovery-reconciler-test-", + }); + const orchestrationLayer = Layer.mergeAll( + OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(OrchestrationProjectionPipelineLive), + ), + OrchestrationProjectionSnapshotQueryLive, + ProjectionTurnRepositoryLive, + ).pipe( + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(RepositoryIdentityResolver.layer), + Layer.provide(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfigLayer), + Layer.provideMerge(NodeServices.layer), + ); + const runtime = ManagedRuntime.make(orchestrationLayer); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery)); + const turnRepository = await runtime.runPromise(Effect.service(ProjectionTurnRepository)); + + return { + dispatch: (command: OrchestrationCommand) => runtime.runPromise(engine.dispatch(command)), + readModel: () => runtime.runPromise(snapshotQuery.getSnapshot()), + reconcile: () => runtime.runPromise(reconcileInterruptedTurnsOnBoot), + getTurn: (threadId: string, turnId: string) => + runtime.runPromise( + turnRepository.getByTurnId({ + threadId: ThreadId.make(threadId), + turnId: TurnId.make(turnId), + }), + ), + readEvents: () => + runtime.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ), + dispose: () => runtime.dispose(), + }; +} + +type ReconcilerSystem = Awaited>; + +async function seedProject(system: ReconcilerSystem, projectId: string) { + await system.dispatch({ + type: "project.create", + commandId: CommandId.make(`cmd-project-${projectId}`), + projectId: ProjectId.make(projectId), + title: `Project ${projectId}`, + workspaceRoot: `/tmp/${projectId}`, + defaultModelSelection: MODEL_SELECTION, + createdAt: "2026-01-01T00:00:00.000Z", + }); +} + +async function seedThread(system: ReconcilerSystem, projectId: string, threadId: string) { + await system.dispatch({ + type: "thread.create", + commandId: CommandId.make(`cmd-thread-${threadId}`), + threadId: ThreadId.make(threadId), + projectId: ProjectId.make(projectId), + title: `Thread ${threadId}`, + modelSelection: MODEL_SELECTION, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:01.000Z", + }); +} + +async function setSession( + system: ReconcilerSystem, + threadId: string, + status: "running" | "stopped", + activeTurnId: string | null, + updatedAt: string, +) { + await system.dispatch({ + type: "thread.session.set", + commandId: CommandId.make(`cmd-session-${threadId}-${status}-${updatedAt}`), + threadId: ThreadId.make(threadId), + session: { + threadId: ThreadId.make(threadId), + status, + providerName: "claudeAgent", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access", + activeTurnId: activeTurnId === null ? null : TurnId.make(activeTurnId), + lastError: null, + updatedAt, + }, + createdAt: updatedAt, + }); +} + +/** Seed a thread frozen mid-turn the way an ungraceful crash leaves it. */ +async function seedCrashedThread( + system: ReconcilerSystem, + projectId: string, + threadId: string, + turnId: string, +) { + await seedThread(system, projectId, threadId); + await setSession(system, threadId, "running", turnId, "2026-01-01T00:00:02.000Z"); +} + +describe("reconcileInterruptedTurnsOnBoot", () => { + it("leaves a crashed thread with a running turn / active session (precondition)", async () => { + const system = await createReconcilerSystem(); + try { + await seedProject(system, "project-1"); + await seedCrashedThread(system, "project-1", "thread-1", "turn-1"); + + const snapshot = await system.readModel(); + const thread = snapshot.threads.find((entry) => entry.id === "thread-1"); + expect(thread?.latestTurn?.state).toBe("running"); + expect(thread?.session?.status).toBe("running"); + expect(thread?.session?.activeTurnId).toBe("turn-1"); + + const turn = await system.getTurn("thread-1", "turn-1"); + expect(turn._tag).toBe("Some"); + if (turn._tag === "Some") { + expect(turn.value.state).toBe("running"); + expect(turn.value.completedAt).toBeNull(); + } + } finally { + await system.dispose(); + } + }); + + it("settles a crashed thread to interrupted and stops the session", async () => { + const system = await createReconcilerSystem(); + try { + await seedProject(system, "project-1"); + await seedCrashedThread(system, "project-1", "thread-1", "turn-1"); + + const result = await system.reconcile(); + expect(result.reconciledCount).toBe(1); + + const snapshot = await system.readModel(); + const thread = snapshot.threads.find((entry) => entry.id === "thread-1"); + expect(thread?.session?.status).toBe("stopped"); + expect(thread?.session?.activeTurnId).toBeNull(); + + const turn = await system.getTurn("thread-1", "turn-1"); + expect(turn._tag).toBe("Some"); + if (turn._tag === "Some") { + expect(turn.value.state).toBe("interrupted"); + expect(turn.value.completedAt).not.toBeNull(); + } + } finally { + await system.dispose(); + } + }); + + it("is idempotent: a second boot reconciles nothing and leaves projections unchanged", async () => { + const system = await createReconcilerSystem(); + try { + await seedProject(system, "project-1"); + await seedCrashedThread(system, "project-1", "thread-1", "turn-1"); + + const first = await system.reconcile(); + expect(first.reconciledCount).toBe(1); + + const eventsAfterFirst = await system.readEvents(); + const snapshotAfterFirst = await system.readModel(); + + const second = await system.reconcile(); + expect(second.reconciledCount).toBe(0); + + const eventsAfterSecond = await system.readEvents(); + expect(eventsAfterSecond.length).toBe(eventsAfterFirst.length); + + const snapshotAfterSecond = await system.readModel(); + const threadFirst = snapshotAfterFirst.threads.find((entry) => entry.id === "thread-1"); + const threadSecond = snapshotAfterSecond.threads.find((entry) => entry.id === "thread-1"); + expect(threadSecond?.session?.status).toBe(threadFirst?.session?.status); + expect(threadSecond?.session?.activeTurnId).toBe(threadFirst?.session?.activeTurnId ?? null); + + const turn = await system.getTurn("thread-1", "turn-1"); + if (turn._tag === "Some") { + expect(turn.value.state).toBe("interrupted"); + } + } finally { + await system.dispose(); + } + }); + + it("only reconciles crashed threads, not already-settled ones (multi-thread selectivity)", async () => { + const system = await createReconcilerSystem(); + try { + await seedProject(system, "project-1"); + await seedCrashedThread(system, "project-1", "thread-a", "turn-a"); + await seedCrashedThread(system, "project-1", "thread-b", "turn-b"); + + // An already-settled thread: it ran a turn and the session was cleanly stopped. + await seedThread(system, "project-1", "thread-settled"); + await setSession(system, "thread-settled", "running", "turn-s", "2026-01-01T00:00:03.000Z"); + await setSession(system, "thread-settled", "stopped", null, "2026-01-01T00:00:04.000Z"); + + const settledTurnBefore = await system.getTurn("thread-settled", "turn-s"); + expect(settledTurnBefore._tag).toBe("Some"); + + const result = await system.reconcile(); + expect(result.reconciledCount).toBe(2); + + const crashedTurns: ReadonlyArray = [ + ["thread-a", "turn-a"], + ["thread-b", "turn-b"], + ]; + for (const [threadId, turnId] of crashedTurns) { + const turn = await system.getTurn(threadId, turnId); + if (turn._tag === "Some") { + expect(turn.value.state).toBe("interrupted"); + expect(turn.value.completedAt).not.toBeNull(); + } + } + + // The already-settled thread must be untouched (same completedAt). + const settledTurnAfter = await system.getTurn("thread-settled", "turn-s"); + if (settledTurnBefore._tag === "Some" && settledTurnAfter._tag === "Some") { + expect(settledTurnAfter.value.state).toBe("interrupted"); + expect(settledTurnAfter.value.completedAt).toBe(settledTurnBefore.value.completedAt); + } + } finally { + await system.dispose(); + } + }); + + it("appends a durable thread.session-set event that survives a projection rebuild", async () => { + const system = await createReconcilerSystem(); + try { + await seedProject(system, "project-1"); + await seedCrashedThread(system, "project-1", "thread-1", "turn-1"); + + await system.reconcile(); + + const events = await system.readEvents(); + const stopEvents = events.filter( + (event): event is Extract => + event.type === "thread.session-set" && + event.payload.threadId === "thread-1" && + event.payload.session.status === "stopped", + ); + expect(stopEvents.length).toBe(1); + const stopEvent = stopEvents[0]; + if (stopEvent) { + expect(stopEvent.payload.session.activeTurnId).toBeNull(); + expect(String(stopEvent.commandId).startsWith("t3-crash-recovery:")).toBe(true); + } + } finally { + await system.dispose(); + } + }); +}); diff --git a/apps/server/src/orchestration/Layers/CrashRecoveryReconciler.ts b/apps/server/src/orchestration/Layers/CrashRecoveryReconciler.ts new file mode 100644 index 00000000000..299734e3c74 --- /dev/null +++ b/apps/server/src/orchestration/Layers/CrashRecoveryReconciler.ts @@ -0,0 +1,141 @@ +/** + * CrashRecoveryReconciler - Boot-time reconciliation of leftover in-flight turns. + * + * When the desktop backend exits UNGRACEFULLY (SIGKILL / crash / OOM — no + * SIGTERM), the Effect finalizer that a clean quit runs + * (`ProviderService.runStopAll` -> `adapter.stopAll()` -> a session-set that + * settles running turns to `interrupted`) never fires. In-flight turns are left + * frozen at `projection_turns.state = "running"` with + * `projection_thread_sessions.active_turn_id` still set, and on a later resume + * the provider reports the cut-off turn as `cancelled`/`aborted`. + * + * At fresh boot NO turn is actually live (adapter session maps start empty and + * runtime rows carry no pid/epoch), so EVERY `running` turn / non-null + * `active_turn_id` in the read model is by definition a leftover from a previous + * process. This module reconciles them by MIRRORING the graceful path: it + * dispatches a `thread.session.set` command (status `stopped`, `activeTurnId` + * null) per crashed thread — exactly what `ProviderRuntimeIngestion` dispatches + * on `session.exited`. It never mutates projections directly (projections are a + * pure function of the event log and would be reverted on rebuild). + * + * FOLLOW-UP (deliberately out of scope): this only settles turns to a resumable + * `interrupted` state. Auto-resume RE-ARMING (a crash marker + boot producer + * that makes the turn auto-continue) is a separate follow-up. + * + * @module CrashRecoveryReconciler + */ +import { CommandId, type OrchestrationSession, type OrchestrationThread } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; + +/** + * A thread carries a leftover in-flight turn when the read model still reports + * an active turn — a running latest turn, a live (`running`/`starting`) session, + * or a non-null `activeTurnId`. At fresh boot none of these can reflect a truly + * live turn, so each is a crash leftover to settle. + */ +export function threadHasLeftoverInFlightTurn(thread: OrchestrationThread): boolean { + const status = thread.session?.status; + return ( + thread.latestTurn?.state === "running" || + status === "running" || + status === "starting" || + (thread.session?.activeTurnId ?? null) !== null + ); +} + +/** + * Settle one crashed thread by mirroring the `session.exited` session-set: + * force `status: "stopped"` and `activeTurnId: null` while preserving the other + * session fields, which drives the thread-turns projection to settle the still- + * running turn to `interrupted`. Returns 1 when a command was dispatched. + */ +const settleCrashedThread = (thread: OrchestrationThread) => + Effect.gen(function* () { + const engine = yield* OrchestrationEngineService; + const crypto = yield* Crypto.Crypto; + const now = DateTime.formatIso(yield* DateTime.now); + const uuid = yield* crypto.randomUUIDv4; + const existing = thread.session; + + const session: OrchestrationSession = { + threadId: thread.id, + status: "stopped", + providerName: existing?.providerName ?? null, + ...(existing?.providerInstanceId !== undefined + ? { providerInstanceId: existing.providerInstanceId } + : {}), + runtimeMode: existing?.runtimeMode ?? "full-access", + activeTurnId: null, + lastError: existing?.lastError ?? null, + updatedAt: now, + }; + + yield* engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make(`t3-crash-recovery:${uuid}`), + threadId: thread.id, + session, + createdAt: now, + }); + return 1; + }); + +/** + * Reconcile every leftover in-flight turn found in the read model at boot. + * + * Never fails: a snapshot read failure or a single per-thread dispatch failure + * is logged and swallowed so one bad thread can never abort the loop or block + * startup. Returns the number of threads actually reconciled. + */ +export const reconcileInterruptedTurnsOnBoot: Effect.Effect< + { readonly reconciledCount: number }, + never, + OrchestrationEngineService | ProjectionSnapshotQuery | Crypto.Crypto +> = Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + + const snapshot = yield* snapshotQuery.getSnapshot().pipe( + Effect.catchCause((cause) => + Effect.logError("crash-recovery reconcile: failed to read projection snapshot", { + cause: Cause.pretty(cause), + }).pipe(Effect.as(null)), + ), + ); + if (snapshot === null) { + return { reconciledCount: 0 }; + } + + const candidates = snapshot.threads.filter(threadHasLeftoverInFlightTurn); + + const settledCounts = yield* Effect.forEach( + candidates, + (thread) => + settleCrashedThread(thread).pipe( + Effect.catchCause((cause) => + Effect.logWarning( + "crash-recovery reconcile: failed to settle a crashed thread; continuing", + { + threadId: thread.id, + cause: Cause.pretty(cause), + }, + ).pipe(Effect.as(0)), + ), + ), + { concurrency: 1 }, + ); + + const reconciledCount = settledCounts.reduce((total, count) => total + count, 0); + if (reconciledCount > 0) { + yield* Effect.logInfo("crash-recovery reconcile: settled leftover in-flight turns", { + reconciledCount, + candidateCount: candidates.length, + }); + } + return { reconciledCount }; +}); diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index b8102bda9ad..97ae9fedb95 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -1,5 +1,13 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; -import { DEFAULT_MODEL, ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import { + DEFAULT_MODEL, + type OrchestrationCommand, + type OrchestrationReadModel, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, +} from "@t3tools/contracts"; import { assert, it } from "@effect/vitest"; import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; @@ -10,9 +18,20 @@ import * as PlatformError from "effect/PlatformError"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import { HttpServer } from "effect/unstable/http"; + import * as ServerConfig from "./config.ts"; +import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; +import * as ExternalLauncher from "./process/externalLauncher.ts"; +import * as Keybindings from "./keybindings.ts"; +import { OrchestrationCommandInvariantError } from "./orchestration/Errors.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; +import * as OrchestrationReactor from "./orchestration/Services/OrchestrationReactor.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; +import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts"; +import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; +import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; +import * as ServerSettings from "./serverSettings.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; @@ -275,3 +294,132 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa assert.deepStrictEqual(yield* Ref.get(dispatchCalls), []); }).pipe(Effect.provide(NodeServices.layer)), ); + +// A read model with one thread frozen mid-turn the way an ungraceful crash leaves it: +// running latest turn, live session, non-null activeTurnId. +const crashedReadModel = (): OrchestrationReadModel => + ({ + snapshotSequence: 1, + updatedAt: "2026-01-01T00:00:00.000Z", + projects: [], + threads: [ + { + id: ThreadId.make("thread-crashed"), + latestTurn: { turnId: TurnId.make("turn-1"), state: "running" }, + session: { + threadId: ThreadId.make("thread-crashed"), + status: "running", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: TurnId.make("turn-1"), + lastError: null, + updatedAt: "2026-01-01T00:00:00.000Z", + }, + }, + ], + }) as unknown as OrchestrationReadModel; + +/** + * Drive the real `ServerRuntimeStartup.make` orchestration with lightweight service + * doubles, seeding one crashed thread. The engine `dispatch` double appends "reconcile" + * and the orchestration reactor `start` double appends "reactor", so the returned order + * proves the reconcile phase runs before reactors start and (because command readiness is + * only signalled after the whole startup sequence completes) before command readiness. + */ +const driveStartupMake = (dispatchFails: boolean) => + Effect.gen(function* () { + const order = yield* Ref.make>([]); + + const engineDouble = { + readEvents: () => Stream.empty, + dispatch: (command: OrchestrationCommand) => + Ref.update(order, (entries) => [...entries, "reconcile"]).pipe( + Effect.flatMap(() => + dispatchFails + ? Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "simulated crash-recovery dispatch failure", + }), + ) + : Effect.succeed({ sequence: 1 }), + ), + ), + streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), + }; + + yield* Effect.scoped( + Effect.gen(function* () { + const startup = yield* ServerRuntimeStartup.make.pipe( + Effect.provideService(ServerConfig.ServerConfig, { + cwd: "/tmp/startup-crash-recovery", + mode: "web", + port: 3773, + host: "127.0.0.1", + autoBootstrapProjectFromCwd: false, + } as never), + Effect.provideService(Keybindings.Keybindings, { start: Effect.void } as never), + Effect.provideService(ServerSettings.ServerSettingsService, { + start: Effect.void, + } as never), + Effect.provideService(OrchestrationReactor.OrchestrationReactor, { + start: () => + Ref.update(order, (entries) => [...entries, "reactor"]).pipe(Effect.asVoid), + } as never), + Effect.provideService(ProviderSessionReaper.ProviderSessionReaper, { + start: () => Effect.void, + } as never), + Effect.provideService(ServerLifecycleEvents.ServerLifecycleEvents, { + publish: () => Effect.void, + } as never), + Effect.provideService(ServerEnvironment.ServerEnvironment, { + getDescriptor: Effect.succeed({ environmentId: "env-test" }), + } as never), + Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { + getSnapshot: () => Effect.succeed(crashedReadModel()), + } as never), + Effect.provideService( + OrchestrationEngine.OrchestrationEngineService, + engineDouble as never, + ), + // Only referenced by the post-`command-ready` startup tail (heartbeat / browser / + // headless access), which this test never reaches because it never marks the HTTP + // listener ready — but they are still part of `make`'s static requirement set. + Effect.provideService(AnalyticsService.AnalyticsService, { + record: () => Effect.void, + flush: Effect.void, + } as never), + Effect.provideService(EnvironmentAuth.EnvironmentAuth, { + issueStartupPairingUrl: () => Effect.succeed("http://localhost"), + } as never), + Effect.provideService(ExternalLauncher.ExternalLauncher, { + launchBrowser: () => Effect.void, + } as never), + Effect.provideService(HttpServer.HttpServer, {} as never), + ); + yield* startup.awaitCommandReady; + }), + ); + + return yield* Ref.get(order); + }).pipe(Effect.provide(NodeServices.layer)); + +it.effect("runs crash-recovery reconcile before reactors start and before command readiness", () => + Effect.gen(function* () { + const order = yield* driveStartupMake(false); + // Reconcile dispatched (settling the crashed thread) strictly before reactors started, + // and command readiness only resolves after the whole sequence, so reconcile precedes it. + assert.deepStrictEqual(order, ["reconcile", "reactor"]); + }), +); + +it.effect("swallows a reconcile dispatch failure so startup still signals command readiness", () => + Effect.gen(function* () { + // If the failure were not swallowed, startup would fail and `awaitCommandReady` would + // reject, failing this test. A resolved readiness with reactors still started proves the + // per-thread dispatch failure was contained. + const order = yield* driveStartupMake(true); + assert.deepStrictEqual(order, ["reconcile", "reactor"]); + }), +); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index b52b577c5b5..feff8ba5e5d 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -27,6 +27,7 @@ import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; +import { reconcileInterruptedTurnsOnBoot } from "./orchestration/Layers/CrashRecoveryReconciler.ts"; import * as OrchestrationReactor from "./orchestration/Services/OrchestrationReactor.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerSettings from "./serverSettings.ts"; @@ -291,6 +292,8 @@ const runStartupPhase = (phase: string, effect: Effect.Effect) export const make = Effect.gen(function* () { const serverConfig = yield* ServerConfig.ServerConfig; const keybindings = yield* Keybindings.Keybindings; + const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const orchestrationReactor = yield* OrchestrationReactor.OrchestrationReactor; const providerSessionReaper = yield* ProviderSessionReaper.ProviderSessionReaper; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; @@ -337,6 +340,32 @@ export const make = Effect.gen(function* () { ), ); + // Settle turns left frozen `running` by a prior UNGRACEFUL exit (crash / OOM / + // SIGKILL) before any reactor observes them and before commands are accepted. + // At fresh boot no turn is truly live, so every leftover in-flight turn is a + // crash remnant; this mirrors the graceful `session.exited` path by dispatching + // a stopped `thread.session.set`, settling the turn to a resumable `interrupted` + // state. FOLLOW-UP: auto-resume RE-ARMING (a crash marker + boot producer) is a + // deliberate separate change; this only reconciles to the settled state. + yield* Effect.logDebug("startup phase: reconciling interrupted turns from a prior crash"); + yield* runStartupPhase( + "reconcile.interrupted-turns", + reconcileInterruptedTurnsOnBoot.pipe( + Effect.provideService(OrchestrationEngine.OrchestrationEngineService, orchestrationEngine), + Effect.provideService( + ProjectionSnapshotQuery.ProjectionSnapshotQuery, + projectionSnapshotQuery, + ), + Effect.tap(({ reconciledCount }) => + reconciledCount > 0 + ? Effect.logInfo("startup phase: reconciled interrupted turns from a prior crash", { + reconciledCount, + }) + : Effect.void, + ), + ), + ); + yield* Effect.logDebug("startup phase: starting orchestration reactors"); yield* runStartupPhase( "reactors.start", diff --git a/apps/server/src/t3x/autoResume/guards.test.ts b/apps/server/src/t3x/autoResume/guards.test.ts index ec61da0f68b..39e80123a32 100644 --- a/apps/server/src/t3x/autoResume/guards.test.ts +++ b/apps/server/src/t3x/autoResume/guards.test.ts @@ -134,6 +134,16 @@ describe("simple predicates", () => { false, ); }); + + // Crash-recovery reconciliation settles leftover in-flight turns to a resumable + // `interrupted` state with a `stopped` session (see CrashRecoveryReconciler). Such a + // thread must NOT read as progressing, otherwise auto-resume would treat the crashed + // turn as still-working and never resume it. + it("threadIsProgressing is false for a reconciled interrupted/stopped thread", () => { + expect( + threadIsProgressing(makeThread({ status: "stopped", latestTurnState: "interrupted" })), + ).toBe(false); + }); }); describe("cancelReason", () => {