From a61acfdb7d8b1c80a856f66277f1fade7352d58a Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 08:52:04 -0500 Subject: [PATCH] fix(server): queued messages still start after a failed startup event read The orchestrator's startup reactors subscribe to run, provider-thread and subagent updates once and fork the stream detached. Subscribing reads the event table, so a failed read there ended the reactor for the life of the server: queued messages stopped promoting and delegated children stopped finalizing until a restart. Each reactor now resubscribes with a capped backoff, resuming after the last event it received. Co-Authored-By: Claude Opus 5.5 --- .../Orchestrator.reactors.test.ts | 124 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 72 ++++++---- .../testkit/ProviderReplayHarness.ts | 13 +- 3 files changed, 184 insertions(+), 25 deletions(-) create mode 100644 apps/server/src/orchestration-v2/Orchestrator.reactors.test.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.reactors.test.ts b/apps/server/src/orchestration-v2/Orchestrator.reactors.test.ts new file mode 100644 index 0000000000000..998211634a3ef --- /dev/null +++ b/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(); + 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), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 77533f593e9a7..c55a38278bda1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -71,6 +71,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"; @@ -78,6 +79,7 @@ import * as Path from "effect/Path"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -12000,27 +12002,52 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // below. Replaying the full event table on every server start delays live // queue promotion in proportion to the lifetime size of the database. const terminalEventsAfterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); - // Queue promotion can wait on a provider or a thread lock. Subscribe to run - // updates before buffering so that wait never retains 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.forkDetach, ); + }; + // Queue promotion can wait on a provider or a thread lock. Subscribe to run + // updates before buffering so that wait never retains 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.forkDetach, + ); // 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 @@ -12038,11 +12065,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) => diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 0342a3ccc9d94..715204ebd9a13 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -270,6 +270,7 @@ export function layerWithRegistry( readonly runContinuationWorker?: boolean; readonly recoverOnStartup?: boolean; readonly continueThreadsAfterServerUpdate?: boolean; + readonly eventSink?: (sink: EventSink.EventSinkV2Shape) => EventSink.EventSinkV2Shape; } = {}, ): Layer.Layer< | Orchestrator.OrchestratorV2 @@ -311,9 +312,19 @@ export function layerWithRegistry( 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 layerCommandReceiptStoreProvided = CommandReceiptStore.layer.pipe( Layer.provide(layerDatabase), );