feat(web): add configurable queue/steer follow-up behavior

apps/server/integration/OrchestrationEngineHarness.integration.ts

@@ -48,6 +48,7 @@ import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointRea
 import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts";
 import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts";
 import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts";
+import { QueuedFollowUpReactorLive } from "../src/orchestration/Layers/QueuedFollowUpReactor.ts";
 import { RuntimeReceiptBusLive } from "../src/orchestration/Layers/RuntimeReceiptBus.ts";
 import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";
 import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";
@@ -165,6 +166,7 @@ export interface OrchestrationIntegrationHarness {
   readonly dbPath: string;
   readonly adapterHarness: TestProviderAdapterHarness | null;
   readonly engine: OrchestrationEngineShape;
+  readonly startReactor: Effect.Effect<void, never>;
   readonly snapshotQuery: ProjectionSnapshotQuery["Service"];
   readonly providerService: ProviderService["Service"];
   readonly checkpointStore: CheckpointStore["Service"];
@@ -211,6 +213,8 @@ export interface OrchestrationIntegrationHarness {
 interface MakeOrchestrationIntegrationHarnessOptions {
   readonly provider?: ProviderKind;
   readonly realCodex?: boolean;
+  readonly rootDir?: string;
+  readonly autoStartReactor?: boolean;
 }
 
 export const makeOrchestrationIntegrationHarness = (
@@ -236,16 +240,25 @@ export const makeOrchestrationIntegrationHarness = (
           listProviders: () => Effect.succeed([adapterHarness.provider]),
         } as typeof ProviderAdapterRegistry.Service)
       : null;
-    const rootDir = yield* fileSystem.makeTempDirectoryScoped({
-      prefix: "t3-orchestration-integration-",
-    });
+    const rootDir =
+      options?.rootDir ??
+      (yield* fileSystem.makeTempDirectoryScoped({
+        prefix: "t3-orchestration-integration-",
+      }));
     const workspaceDir = path.join(rootDir, "workspace");
     const { stateDir, dbPath } = yield* deriveServerPaths(rootDir, undefined).pipe(
       Effect.provideService(Path.Path, path),
     );
+    yield* fileSystem.makeDirectory(rootDir, { recursive: true });
     yield* fileSystem.makeDirectory(workspaceDir, { recursive: true });
     yield* fileSystem.makeDirectory(stateDir, { recursive: true });
-    yield* initializeGitWorkspace(workspaceDir);
+    const workspaceGitDir = path.join(workspaceDir, ".git");
+    const gitDirExists = yield* fileSystem
+      .exists(workspaceGitDir)
+      .pipe(Effect.orElseSucceed(() => false));
+    if (!gitDirExists) {
+      yield* initializeGitWorkspace(workspaceDir);
+    }
 
     const persistenceLayer = makeSqlitePersistenceLive(dbPath);
     const orchestrationLayer = OrchestrationEngineLive.pipe(
@@ -318,10 +331,14 @@ export const makeOrchestrationIntegrationHarness = (
     const checkpointReactorLayer = CheckpointReactorLive.pipe(
       Layer.provideMerge(runtimeServicesLayer),
     );
+    const queuedFollowUpReactorLayer = QueuedFollowUpReactorLive.pipe(
+      Layer.provideMerge(runtimeServicesLayer),
+    );
     const orchestrationReactorLayer = OrchestrationReactorLive.pipe(
       Layer.provideMerge(runtimeIngestionLayer),
       Layer.provideMerge(providerCommandReactorLayer),
       Layer.provideMerge(checkpointReactorLayer),
+      Layer.provideMerge(queuedFollowUpReactorLayer),
     );
     const layer = orchestrationReactorLayer.pipe(
       Layer.provide(persistenceLayer),
@@ -359,9 +376,19 @@ export const makeOrchestrationIntegrationHarness = (
     ).pipe(Effect.orDie);
 
     const scope = yield* Scope.make("sequential");
-    yield* tryRuntimePromise("start OrchestrationReactor", () =>
-      runtime.runPromise(reactor.start.pipe(Scope.provide(scope))),
-    ).pipe(Effect.orDie);
+    let reactorStarted = false;
+    const startReactor = Effect.gen(function* () {
+      if (reactorStarted) {
+        return;
+      }
+      reactorStarted = true;
+      yield* tryRuntimePromise("start OrchestrationReactor", () =>
+        runtime.runPromise(reactor.start.pipe(Scope.provide(scope))),
+      ).pipe(Effect.orDie);
+    }).pipe(Effect.orDie);
+    if (options?.autoStartReactor !== false) {
+      yield* startReactor;
+    }
     const receiptHistory = yield* Ref.make<ReadonlyArray<OrchestrationRuntimeReceipt>>([]);
     yield* Stream.runForEach(runtimeReceiptBus.stream, (receipt) =>
       Ref.update(receiptHistory, (history) => [...history, receipt]).pipe(Effect.asVoid),
@@ -492,6 +519,7 @@ export const makeOrchestrationIntegrationHarness = (
       dbPath,
       adapterHarness,
       engine,
+      startReactor,
       snapshotQuery,
       providerService,
       checkpointStore,

apps/server/integration/orchestrationEngine.integration.test.ts

@@ -1,4 +1,5 @@
 import fs from "node:fs";
+import os from "node:os";
 import path from "node:path";
 
 import {
@@ -94,6 +95,17 @@ function withHarness<A, E>(
   ).pipe(Effect.provide(NodeServices.layer));
 }
 
+function withHarnessOptions<A, E>(
+  options: Parameters<typeof makeOrchestrationIntegrationHarness>[0],
+  use: (harness: OrchestrationIntegrationHarness) => Effect.Effect<A, E>,
+) {
+  return Effect.acquireUseRelease(
+    makeOrchestrationIntegrationHarness(options),
+    use,
+    (harness) => harness.dispose,
+  ).pipe(Effect.provide(NodeServices.layer));
+}
+
 function withRealCodexHarness<A, E>(
   use: (harness: OrchestrationIntegrationHarness) => Effect.Effect<A, E>,
 ) {
@@ -252,6 +264,89 @@ it.live("runs a single turn end-to-end and persists checkpoint state in sqlite +
   ),
 );
 
+it.live("replays queued follow-ups after orchestration restarts", () =>
+  Effect.gen(function* () {
+    const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-orchestration-queue-restart-"));
+
+    yield* withHarnessOptions({ rootDir }, (harness) =>
+      Effect.gen(function* () {
+        yield* seedProjectAndThread(harness);
+
+        yield* harness.engine.dispatch({
+          type: "thread.queued-follow-up.enqueue",
+          commandId: CommandId.makeUnsafe("cmd-queued-follow-up-restart-enqueue"),
+          threadId: THREAD_ID,
+          followUp: {
+            id: "follow-up-restart-1",
+            createdAt: nowIso(),
+            prompt: "Resume this after restart",
+            attachments: [],
+            terminalContexts: [],
+            modelSelection: {
+              provider: "codex",
+              model: DEFAULT_MODEL_BY_PROVIDER.codex,
+            },
+            runtimeMode: "approval-required",
+            interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+            lastSendError: null,
+          },
+          createdAt: nowIso(),
+        });
+
+        const queuedThread = yield* harness.waitForThread(
+          THREAD_ID,
+          (thread) => thread.queuedFollowUps.length === 1,
+        );
+        assert.equal(queuedThread.queuedFollowUps[0]?.prompt, "Resume this after restart");
+      }),
+    );
+
+    yield* withHarnessOptions({ rootDir, autoStartReactor: false }, (harness) =>
+      Effect.gen(function* () {
+        yield* harness.adapterHarness!.queueTurnResponseForNextSession({
+          events: [
+            {
+              type: "turn.started",
+              ...runtimeBase("evt-queued-restart-1", "2026-03-28T12:05:00.000Z"),
+              threadId: THREAD_ID,
+              turnId: FIXTURE_TURN_ID,
+            },
+            {
+              type: "message.delta",
+              ...runtimeBase("evt-queued-restart-2", "2026-03-28T12:05:00.050Z"),
+              threadId: THREAD_ID,
+              turnId: FIXTURE_TURN_ID,
+              delta: "Recovered queued follow-up output.\n",
+            },
+            {
+              type: "turn.completed",
+              ...runtimeBase("evt-queued-restart-3", "2026-03-28T12:05:00.100Z"),
+              threadId: THREAD_ID,
+              turnId: FIXTURE_TURN_ID,
+              status: "completed",
+            },
+          ],
+        });
+
+        yield* harness.startReactor;
+
+        const recoveredThread = yield* harness.waitForThread(
+          THREAD_ID,
+          (thread) =>
+            thread.queuedFollowUps.length === 0 &&
+            thread.messages.some(
+              (message) =>
+                message.role === "assistant" &&
+                message.text.includes("Recovered queued follow-up output."),
+            ),
+        );
+
+        assert.equal(recoveredThread.queuedFollowUps.length, 0);
+      }),
+    );
+  }).pipe(Effect.provide(NodeServices.layer)),
+);
+
 it.live.skipIf(!process.env.CODEX_BINARY_PATH)(
   "keeps the same Codex provider thread across runtime mode switches",
   () =>

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -64,6 +64,7 @@ function makeSnapshot(input: {
         archivedAt: null,
         deletedAt: null,
         messages: [],
+        queuedFollowUps: [],
         activities: [],
         proposedPlans: [],
         checkpoints: [

apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts

@@ -5,6 +5,7 @@ import { CheckpointReactor } from "../Services/CheckpointReactor.ts";
 import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";
 import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";
 import { OrchestrationReactor } from "../Services/OrchestrationReactor.ts";
+import { QueuedFollowUpReactor } from "../Services/QueuedFollowUpReactor.ts";
 import { makeOrchestrationReactor } from "./OrchestrationReactor.ts";
 
 describe("OrchestrationReactor", () => {
@@ -46,6 +47,14 @@ describe("OrchestrationReactor", () => {
             drain: Effect.void,
           }),
         ),
+        Layer.provideMerge(
+          Layer.succeed(QueuedFollowUpReactor, {
+            start: Effect.sync(() => {
+              started.push("queued-follow-up-reactor");
+            }),
+            drain: Effect.void,
+          }),
+        ),
       ),
     );
 
@@ -57,6 +66,7 @@ describe("OrchestrationReactor", () => {
       "provider-runtime-ingestion",
       "provider-command-reactor",
       "checkpoint-reactor",
+      "queued-follow-up-reactor",
     ]);
 
     await Effect.runPromise(Scope.close(scope, Exit.void));

apps/server/src/orchestration/Layers/OrchestrationReactor.ts

@@ -7,16 +7,19 @@ import {
 import { CheckpointReactor } from "../Services/CheckpointReactor.ts";
 import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";
 import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";
+import { QueuedFollowUpReactor } from "../Services/QueuedFollowUpReactor.ts";
 
 export const makeOrchestrationReactor = Effect.gen(function* () {
   const providerRuntimeIngestion = yield* ProviderRuntimeIngestionService;
   const providerCommandReactor = yield* ProviderCommandReactor;
   const checkpointReactor = yield* CheckpointReactor;
+  const queuedFollowUpReactor = yield* QueuedFollowUpReactor;
 
   const start: OrchestrationReactorShape["start"] = Effect.gen(function* () {
     yield* providerRuntimeIngestion.start;
     yield* providerCommandReactor.start;
     yield* checkpointReactor.start;
+    yield* queuedFollowUpReactor.start;
   });
 
   return {