authorgravatar for cursoragent@cursor.comCursor Agent <cursoragent@cursor.com> 2026-02-26 18:49:09+00:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log79b8fd7b0096c3a75e69e877df1d5253daa7e1f8
tree6ee1c31768322d01ca0bf1abe7d4ae5ac0b8a174
parent61560bef235fb6a04c36b99190a762713141e35a

Fix ProviderCommandReactor: add error recovery and deduplicate CWD resolution

- Wrap processDomainEvent in catchCause to prevent the reactor fiber from terminating on handler errors, matching the pattern used by ProviderRuntimeIngestion and CheckpointReactor. - Replace duplicated resolveEffectiveThreadCwd with the shared resolveThreadWorkspaceCwd from checkpointing/Utils.ts.

1 files changed, 17 insertions(+), 21 deletions(-)

apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+17-21
......@@ -8,8 +8,9 @@ import {
88 type ThreadId,
99 type TurnId,
1010} from "@t3tools/contracts";
11import { Cache, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
11import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
1212
13import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts";
1314import { ProviderService } from "../../provider/Services/ProviderService.ts";
1415import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
1516import {
......@@ -33,24 +34,6 @@ function toNonEmptyProviderInput(value: string | undefined): string | undefined
3334 return normalized && normalized.length > 0 ? normalized : undefined;
3435}
3536
36function resolveEffectiveThreadCwd(input: {
37 readonly thread: {
38 readonly projectId: string;
39 readonly worktreePath: string | null;
40 };
41 readonly projects: ReadonlyArray<{
42 readonly id: string;
43 readonly workspaceRoot: string;
44 }>;
45}): string | undefined {
46 const worktreeCwd = input.thread.worktreePath ?? undefined;
47 if (worktreeCwd) {
48 return worktreeCwd;
49 }
50
51 return input.projects.find((project) => project.id === input.thread.projectId)?.workspaceRoot;
52}
53
5437function mapProviderSessionStatusToOrchestrationStatus(
5538 status: "connecting" | "ready" | "running" | "error" | "closed",
5639): OrchestrationSession["status"] {
......@@ -161,7 +144,7 @@ const make = Effect.gen(function* () {
161144 thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode"
162145 ? thread.session.providerName
163146 : undefined;
164 const effectiveCwd = resolveEffectiveThreadCwd({
147 const effectiveCwd = resolveThreadWorkspaceCwd({
165148 thread,
166149 projects: readModel.projects,
167150 });
......@@ -353,12 +336,25 @@ const make = Effect.gen(function* () {
353336 }
354337 });
355338
339 const processDomainEventSafely = (event: ProviderIntentEvent) =>
340 processDomainEvent(event).pipe(
341 Effect.catchCause((cause) => {
342 if (Cause.hasInterruptsOnly(cause)) {
343 return Effect.failCause(cause);
344 }
345 return Effect.logWarning("provider command reactor failed to process event", {
346 eventType: event.type,
347 cause: Cause.pretty(cause),
348 });
349 }),
350 );
351
356352 const start: ProviderCommandReactorShape["start"] = Effect.gen(function* () {
357353 const queue = yield* Queue.unbounded<ProviderIntentEvent>();
358354 yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid));
359355
360356 yield* Effect.forkScoped(
361 Effect.forever(Queue.take(queue).pipe(Effect.flatMap(processDomainEvent))),
357 Effect.forever(Queue.take(queue).pipe(Effect.flatMap(processDomainEventSafely))),
362358 );
363359
364360 yield* Effect.forkScoped(