fix(server): queued messages keep starting after a failed startup read

apps/server/src/orchestration-v2/Orchestrator.reactors.test.ts

@@ -0,0 +1,124 @@
+import { assert, it } from "@effect/vitest";
+import {
+  CommandId,
+  EventId,
+  MessageId,
+  ProjectId,
+  ProviderDriverKind,
+  ProviderInstanceId,
+  ThreadId,
+} from "@t3tools/contracts";
+import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
+import * as DateTime from "effect/DateTime";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
+
+import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
+import * as EventSink from "./EventSink.ts";
+import * as Orchestrator from "./Orchestrator.ts";
+import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
+import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
+
+const instanceId = ProviderInstanceId.make("codex");
+const adapter = {
+  instanceId,
+  driver: ProviderDriverKind.make("codex"),
+  getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
+  planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
+  openSession: () => Effect.die("Runs here never reach a provider"),
+} as ProviderAdapter.ProviderAdapterV2["Service"];
+
+it.effect("promotes the queue after the run-update subscription fails to start", () =>
+  Effect.gen(function* () {
+    const subscriptionFailed = yield* Deferred.make<void>();
+    let failNextSubscription = true;
+    const eventSink = (sink: EventSink.EventSinkV2Shape): EventSink.EventSinkV2Shape => ({
+      ...sink,
+      stream: (input) => {
+        if (input?.eventType !== "run.updated" || !failNextSubscription) return sink.stream(input);
+        failNextSubscription = false;
+        return Stream.fail(
+          new EventSink.EventSinkStreamError({ cause: "database is locked" }),
+        ).pipe(Stream.ensuring(Deferred.succeed(subscriptionFailed, undefined)));
+      },
+    });
+    yield* Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const sink = yield* EventSink.EventSinkV2;
+      const threadId = ThreadId.make("thread:reactor-resubscribe");
+      yield* Deferred.await(subscriptionFailed);
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make("create"),
+        threadId,
+        projectId: ProjectId.make("project:reactor-resubscribe"),
+        title: "Reactor resubscribe",
+        modelSelection: { instanceId, model: "gpt-5.1-codex" },
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        branch: null,
+        worktreePath: null,
+        createdBy: "user",
+        creationSource: "web",
+      });
+      for (const [text, type] of [
+        ["active", "start_immediately"],
+        ["queued", "queue_after_active"],
+      ] as const) {
+        yield* orchestrator.dispatch({
+          type: "message.dispatch",
+          commandId: CommandId.make(`send:${text}`),
+          threadId,
+          messageId: MessageId.make(`message:${text}`),
+          text,
+          attachments: [],
+          dispatchMode: { type },
+          createdBy: "user",
+          creationSource: "web",
+        });
+      }
+      const [active, queued] = (yield* orchestrator.getThreadProjection(threadId)).runs;
+      assert.equal(queued?.status, "queued");
+      const promoted = yield* orchestrator.streamDomainEvents.pipe(
+        Stream.filter(
+          (event) =>
+            event.type === "run.updated" &&
+            event.runId === queued!.id &&
+            event.payload.status !== "queued",
+        ),
+        Stream.take(1),
+        Stream.runDrain,
+        Effect.forkScoped,
+      );
+      const now = yield* DateTime.now;
+      yield* sink.write({
+        events: [
+          {
+            id: EventId.make("event:reactor-resubscribe:completed"),
+            type: "run.updated",
+            threadId,
+            runId: active!.id,
+            occurredAt: now,
+            payload: { ...active!, status: "completed", startedAt: now, completedAt: now },
+          },
+        ],
+      });
+      // The completion landed while no reactor was subscribed; the retry replays it.
+      yield* TestClock.adjust("1 second");
+      yield* Fiber.join(promoted);
+      const after = yield* orchestrator.getThreadProjection(threadId);
+      assert.notEqual(after.runs.find((run) => run.id === queued!.id)?.status, "queued");
+    }).pipe(
+      Effect.provide(
+        ProviderReplayHarness.layerWithRegistry(
+          { name: "reactor-resubscribe" },
+          ProviderAdapterRegistry.layerFromAdapters([adapter]),
+          { runEffectWorker: false, eventSink },
+        ),
+      ),
+    );
+  }).pipe(Effect.scoped),
+);

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -87,6 +87,7 @@ import {
 } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as FileSystem from "effect/FileSystem";
@@ -95,6 +96,7 @@ import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Ref from "effect/Ref";
 import * as Result from "effect/Result";
+import * as Schedule from "effect/Schedule";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 
@@ -14308,26 +14310,51 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
 
   // Projections already contain historical terminals; replaying them delays startup queue promotion with database size.
   const terminalEventsAfterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie);
-  // Filter run updates before buffering so blocked queue promotion does not retain unrelated tool bodies.
-  yield* eventSink
-    .stream({ afterSequence: terminalEventsAfterSequence, eventType: "run.updated" })
-    .pipe(
-      Stream.filter(
-        (stored) =>
-          stored.event.type === "run.updated" &&
-          !String(stored.commandId).startsWith("command:runtime-reconcile:") &&
-          ((stored.event.payload.delegatedTaskId !== undefined &&
-            (stored.event.payload.status === "queued" ||
-              stored.event.payload.status === "starting")) ||
-            stored.event.payload.status === "completed" ||
-            stored.event.payload.status === "interrupted" ||
-            stored.event.payload.status === "failed" ||
-            stored.event.payload.status === "cancelled" ||
-            stored.event.payload.status === "rolled_back"),
+  // The reactors below run for the life of the server. Subscribing reads the
+  // event table, and a failed read would end a reactor for good, so each one
+  // resubscribes after the last event it received.
+  const reactorEvents = (eventType: OrchestrationV2DomainEvent["type"]) => {
+    let afterSequence = terminalEventsAfterSequence;
+    return Stream.suspend(() => eventSink.stream({ afterSequence, eventType })).pipe(
+      Stream.tap((stored) =>
+        Effect.sync(() => {
+          afterSequence = stored.sequence;
+        }),
+      ),
+      Stream.tapError((cause) =>
+        Effect.logWarning("Resubscribing to V2 events after a failed read", {
+          eventType,
+          afterSequence,
+          cause,
+        }),
+      ),
+      Stream.retry(
+        Schedule.exponential("1 second").pipe(
+          Schedule.modifyDelay(({ duration }) =>
+            Effect.succeed(Duration.min(duration, Duration.seconds(30))),
+          ),
+        ),
       ),
-      Stream.runForEach(handleRunUpdate),
-      Effect.forkScoped,
     );
+  };
+  // Filter run updates before buffering so blocked queue promotion does not retain unrelated tool bodies.
+  yield* reactorEvents("run.updated").pipe(
+    Stream.filter(
+      (stored) =>
+        stored.event.type === "run.updated" &&
+        !String(stored.commandId).startsWith("command:runtime-reconcile:") &&
+        ((stored.event.payload.delegatedTaskId !== undefined &&
+          (stored.event.payload.status === "queued" ||
+            stored.event.payload.status === "starting")) ||
+          stored.event.payload.status === "completed" ||
+          stored.event.payload.status === "interrupted" ||
+          stored.event.payload.status === "failed" ||
+          stored.event.payload.status === "cancelled" ||
+          stored.event.payload.status === "rolled_back"),
+    ),
+    Stream.runForEach(handleRunUpdate),
+    Effect.forkScoped,
+  );
 
   // A child whose run ends while its background tasks or its own subagents still run has no
   // result yet. When that work settles there is no run update to finalize it, so these events
@@ -14345,11 +14372,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     return providerThreadsWithBackgroundWork.delete(event.payload.id);
   };
   yield* Stream.merge(
-    eventSink.stream({
-      afterSequence: terminalEventsAfterSequence,
-      eventType: "provider-thread.updated",
-    }),
-    eventSink.stream({ afterSequence: terminalEventsAfterSequence, eventType: "subagent.updated" }),
+    reactorEvents("provider-thread.updated"),
+    reactorEvents("subagent.updated"),
   ).pipe(
     Stream.filter(settlesChildWork),
     Stream.runForEach((stored) =>

apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts

@@ -282,6 +282,7 @@ export function layerWithRegistry<Error>(
     readonly runContinuationWorker?: boolean;
     readonly recoverOnStartup?: boolean;
     readonly continueThreadsAfterServerUpdate?: boolean;
+    readonly eventSink?: (sink: EventSink.EventSinkV2Shape) => EventSink.EventSinkV2Shape;
     readonly managedPolicy?: ManagedPolicyDocument;
   } = {},
 ): Layer.Layer<
@@ -330,9 +331,19 @@ export function layerWithRegistry<Error>(
     EffectOutbox.layer,
     TurnItemPositionStore.layer,
   ).pipe(Layer.provide(layerDatabase));
-  const layerEventSinkProvided = EventSink.layerFromStores.pipe(
+  const layerStoreEventSink = EventSink.layerFromStores.pipe(
     Layer.provide(Layer.mergeAll(layerStores, layerDatabase)),
   );
+  const wrapEventSink = options.eventSink;
+  const layerEventSinkProvided =
+    wrapEventSink === undefined
+      ? layerStoreEventSink
+      : Layer.effect(
+          EventSink.EventSinkV2,
+          Effect.gen(function* () {
+            return wrapEventSink(yield* EventSink.EventSinkV2);
+          }),
+        ).pipe(Layer.provide(layerStoreEventSink));
   const layerRuntime = RuntimePolicy.layerEnforcingCeiling.pipe(
     Layer.provide(Layer.merge(layerRuntimeBase, layerServerSettings)),
   );