authorgravatar for mrmg@deflexion.netMark Griffin <mrmg@deflexion.net> 2026-08-24 11:47:06+01:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-24 03:47:06-07:00
log17822fab708a90e82ac961e3b4895932667ce419
tree07213eef0a2c7f8bb5596e5d8965a6b3c7f4799e
parent3fd50643311d0b0e702628f906f7cb6bbe84e1ba
signature Signed by PGP key B5690EEEBB952194

fix(server): recover from provider interrupt failures (#7412)

2 files changed, 308 insertions(+), 18 deletions(-)

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+233-15
......@@ -150,6 +150,8 @@ describe("ProviderCommandReactor", () => {
150150 readonly requiresNewThreadForModelChange?: boolean;
151151 readonly titleRegenerationCompletionDispatchFailures?: number;
152152 readonly titleRegenerationBeforeStart?: "one" | "two";
153 readonly interruptTurnEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
154 readonly stopSessionEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
153155 readonly startSessionEffect?: (
154156 session: ProviderSession,
155157 ) => Effect.Effect<ProviderSession, ProviderAdapterRequestError>;
......@@ -235,23 +237,27 @@ describe("ProviderCommandReactor", () => {
235237 turnId: asTurnId("turn-1"),
236238 }),
237239 );
238 const interruptTurn = vi.fn((_: unknown) => Effect.void);
240 const interruptTurn = vi.fn((_: unknown) => input?.interruptTurnEffect?.() ?? Effect.void);
239241 const respondToRequest = vi.fn<ProviderServiceShape["respondToRequest"]>(() => Effect.void);
240242 const respondToUserInput = vi.fn<ProviderServiceShape["respondToUserInput"]>(() => Effect.void);
241 const stopSession = vi.fn((input: unknown) =>
242 Effect.sync(() => {
243 const threadId =
244 typeof input === "object" && input !== null && "threadId" in input
245 ? (input as { threadId?: ThreadId }).threadId
246 : undefined;
247 if (!threadId) {
248 return;
249 }
250 const index = runtimeSessions.findIndex((session) => session.threadId === threadId);
251 if (index >= 0) {
252 runtimeSessions.splice(index, 1);
253 }
254 }),
243 const stopSession = vi.fn((stopInput: unknown) =>
244 (input?.stopSessionEffect?.() ?? Effect.void).pipe(
245 Effect.tap(() =>
246 Effect.sync(() => {
247 const threadId =
248 typeof stopInput === "object" && stopInput !== null && "threadId" in stopInput
249 ? (stopInput as { threadId?: ThreadId }).threadId
250 : undefined;
251 if (!threadId) {
252 return;
253 }
254 const index = runtimeSessions.findIndex((session) => session.threadId === threadId);
255 if (index >= 0) {
256 runtimeSessions.splice(index, 1);
257 }
258 }),
259 ),
260 ),
255261 );
256262 const renameBranch = vi.fn((input: unknown) =>
257263 Effect.succeed({
......@@ -2487,6 +2493,218 @@ describe("ProviderCommandReactor", () => {
24872493 });
24882494 });
24892495
2496 effectIt.effect(
2497 "stops a running session and records the failure when provider interrupt fails",
2498 () =>
2499 Effect.gen(function* () {
2500 const harness = yield* Effect.promise(() =>
2501 createHarness({
2502 interruptTurnEffect: () =>
2503 Effect.fail(
2504 new ProviderAdapterRequestError({
2505 provider: "codex",
2506 method: "thread.interrupt",
2507 detail: "provider session disappeared",
2508 }),
2509 ),
2510 stopSessionEffect: () =>
2511 Effect.fail(
2512 new ProviderAdapterRequestError({
2513 provider: "codex",
2514 method: "session.stop",
2515 detail: "provider process already exited",
2516 }),
2517 ),
2518 }),
2519 );
2520 const now = "2026-01-01T00:00:00.000Z";
2521
2522 yield* harness.engine.dispatch({
2523 type: "thread.session.set",
2524 commandId: CommandId.make("cmd-session-set-interrupt-failure"),
2525 threadId: ThreadId.make("thread-1"),
2526 session: {
2527 threadId: ThreadId.make("thread-1"),
2528 status: "running",
2529 providerName: "codex",
2530 runtimeMode: "approval-required",
2531 activeTurnId: asTurnId("turn-1"),
2532 lastError: null,
2533 updatedAt: now,
2534 },
2535 createdAt: now,
2536 });
2537
2538 yield* harness.engine.dispatch({
2539 type: "thread.turn.interrupt",
2540 commandId: CommandId.make("cmd-turn-interrupt-provider-failure"),
2541 threadId: ThreadId.make("thread-1"),
2542 turnId: asTurnId("turn-1"),
2543 createdAt: now,
2544 });
2545
2546 yield* Effect.promise(() =>
2547 waitFor(async () => {
2548 const thread = (await harness.readModel()).threads.find(
2549 (entry) => entry.id === ThreadId.make("thread-1"),
2550 );
2551 return thread?.session?.status === "stopped";
2552 }),
2553 );
2554
2555 const thread = (yield* Effect.promise(() => harness.readModel())).threads.find(
2556 (entry) => entry.id === ThreadId.make("thread-1"),
2557 );
2558 expect(thread?.session).toMatchObject({
2559 status: "stopped",
2560 activeTurnId: null,
2561 lastError: "provider session disappeared",
2562 });
2563 expect(
2564 thread?.activities.find((activity) => activity.kind === "provider.turn.interrupt.failed"),
2565 ).toMatchObject({
2566 summary: "Provider turn interrupt failed",
2567 payload: { detail: "provider session disappeared" },
2568 });
2569 expect(harness.stopSession).toHaveBeenCalledWith({ threadId: ThreadId.make("thread-1") });
2570 }),
2571 );
2572
2573 effectIt.effect("stops a starting session without a bound turn when interrupt fails", () =>
2574 Effect.gen(function* () {
2575 const harness = yield* Effect.promise(() =>
2576 createHarness({
2577 interruptTurnEffect: () =>
2578 Effect.fail(
2579 new ProviderAdapterRequestError({
2580 provider: "codex",
2581 method: "thread.interrupt",
2582 detail: "provider session disappeared",
2583 }),
2584 ),
2585 }),
2586 );
2587 const now = "2026-01-01T00:00:00.000Z";
2588
2589 yield* harness.engine.dispatch({
2590 type: "thread.session.set",
2591 commandId: CommandId.make("cmd-session-set-interrupt-starting"),
2592 threadId: ThreadId.make("thread-1"),
2593 session: {
2594 threadId: ThreadId.make("thread-1"),
2595 status: "starting",
2596 providerName: "codex",
2597 runtimeMode: "approval-required",
2598 activeTurnId: null,
2599 lastError: null,
2600 updatedAt: now,
2601 },
2602 createdAt: now,
2603 });
2604
2605 yield* harness.engine.dispatch({
2606 type: "thread.turn.interrupt",
2607 commandId: CommandId.make("cmd-turn-interrupt-starting-provider-failure"),
2608 threadId: ThreadId.make("thread-1"),
2609 createdAt: now,
2610 });
2611
2612 yield* Effect.promise(() => harness.drain());
2613
2614 const thread = (yield* Effect.promise(() => harness.readModel())).threads.find(
2615 (entry) => entry.id === ThreadId.make("thread-1"),
2616 );
2617 expect(thread?.session).toMatchObject({
2618 status: "stopped",
2619 activeTurnId: null,
2620 lastError: "provider session disappeared",
2621 });
2622 expect(harness.stopSession).toHaveBeenCalledWith({ threadId: ThreadId.make("thread-1") });
2623 expect(
2624 thread?.activities.find((activity) => activity.kind === "provider.turn.interrupt.failed"),
2625 ).toMatchObject({ payload: { detail: "provider session disappeared" } });
2626 }),
2627 );
2628
2629 effectIt.effect("does not overwrite a session that became ready while an interrupt failed", () =>
2630 Effect.gen(function* () {
2631 const harness = yield* Effect.promise(() => createHarness());
2632 const now = "2026-01-01T00:00:00.000Z";
2633 const completedAt = "2026-01-01T00:00:01.000Z";
2634
2635 yield* harness.engine.dispatch({
2636 type: "thread.session.set",
2637 commandId: CommandId.make("cmd-session-set-interrupt-race"),
2638 threadId: ThreadId.make("thread-1"),
2639 session: {
2640 threadId: ThreadId.make("thread-1"),
2641 status: "running",
2642 providerName: "codex",
2643 runtimeMode: "approval-required",
2644 activeTurnId: asTurnId("turn-1"),
2645 lastError: null,
2646 updatedAt: now,
2647 },
2648 createdAt: now,
2649 });
2650
2651 harness.interruptTurn.mockImplementation(() =>
2652 harness.engine
2653 .dispatch({
2654 type: "thread.session.set",
2655 commandId: CommandId.make("cmd-session-set-natural-completion"),
2656 threadId: ThreadId.make("thread-1"),
2657 session: {
2658 threadId: ThreadId.make("thread-1"),
2659 status: "ready",
2660 providerName: "codex",
2661 runtimeMode: "approval-required",
2662 activeTurnId: null,
2663 lastError: null,
2664 updatedAt: completedAt,
2665 },
2666 createdAt: completedAt,
2667 })
2668 .pipe(
2669 Effect.catchCause((cause) => Effect.die(cause)),
2670 Effect.andThen(
2671 Effect.fail(
2672 new ProviderAdapterRequestError({
2673 provider: "codex",
2674 method: "thread.interrupt",
2675 detail: "provider session disappeared",
2676 }),
2677 ),
2678 ),
2679 ),
2680 );
2681
2682 yield* harness.engine.dispatch({
2683 type: "thread.turn.interrupt",
2684 commandId: CommandId.make("cmd-turn-interrupt-race"),
2685 threadId: ThreadId.make("thread-1"),
2686 turnId: asTurnId("turn-1"),
2687 createdAt: now,
2688 });
2689
2690 yield* Effect.promise(() => harness.drain());
2691
2692 const thread = (yield* Effect.promise(() => harness.readModel())).threads.find(
2693 (entry) => entry.id === ThreadId.make("thread-1"),
2694 );
2695 expect(thread?.session).toMatchObject({
2696 status: "ready",
2697 activeTurnId: null,
2698 lastError: null,
2699 updatedAt: completedAt,
2700 });
2701 expect(harness.stopSession).not.toHaveBeenCalled();
2702 expect(
2703 thread?.activities.some((activity) => activity.kind === "provider.turn.interrupt.failed"),
2704 ).toBe(false);
2705 }),
2706 );
2707
24902708 it("starts a fresh session when only projected session state exists", async () => {
24912709 const harness = await createHarness();
24922710 const now = "2026-01-01T00:00:00.000Z";
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+75-3
......@@ -1180,8 +1180,8 @@ const make = Effect.gen(function* () {
11801180 if (!thread) {
11811181 return;
11821182 }
1183 const hasSession = thread.session && thread.session.status !== "stopped";
1184 if (!hasSession) {
1183 const session = thread.session;
1184 if (!session || session.status === "stopped") {
11851185 return yield* appendProviderFailureActivity({
11861186 threadId: event.payload.threadId,
11871187 kind: "provider.turn.interrupt.failed",
......@@ -1192,8 +1192,80 @@ const make = Effect.gen(function* () {
11921192 });
11931193 }
11941194
1195 const recoverInterruptFailure = (cause: Cause.Cause<unknown>) => {
1196 if (Cause.hasInterruptsOnly(cause)) {
1197 return Effect.interrupt;
1198 }
1199
1200 const detail = formatFailureDetail(cause);
1201 return Effect.gen(function* () {
1202 const latestThread = yield* resolveThread(event.payload.threadId);
1203 const latestSession = latestThread?.session;
1204 if (
1205 !latestSession ||
1206 latestSession.status === "stopped" ||
1207 latestSession.status === "ready" ||
1208 (event.payload.turnId !== undefined &&
1209 latestSession.activeTurnId !== null &&
1210 latestSession.activeTurnId !== event.payload.turnId)
1211 ) {
1212 return;
1213 }
1214
1215 yield* providerService.stopSession({ threadId: event.payload.threadId }).pipe(
1216 Effect.catchCause((stopCause) => {
1217 if (Cause.hasInterruptsOnly(stopCause)) {
1218 return Effect.interrupt;
1219 }
1220 return Effect.logWarning(
1221 "provider command reactor failed to stop session after interrupt failure",
1222 {
1223 threadId: event.payload.threadId,
1224 cause: Cause.pretty(stopCause),
1225 originalCause: Cause.pretty(cause),
1226 },
1227 );
1228 }),
1229 );
1230 const stoppedThread = yield* resolveThread(event.payload.threadId);
1231 const stoppedSession = stoppedThread?.session;
1232 if (
1233 !stoppedSession ||
1234 stoppedSession.status === "stopped" ||
1235 stoppedSession.status === "ready" ||
1236 (event.payload.turnId !== undefined &&
1237 stoppedSession.activeTurnId !== null &&
1238 stoppedSession.activeTurnId !== event.payload.turnId)
1239 ) {
1240 return;
1241 }
1242
1243 yield* setThreadSession({
1244 threadId: event.payload.threadId,
1245 session: {
1246 ...stoppedSession,
1247 status: "stopped",
1248 activeTurnId: null,
1249 lastError: detail,
1250 updatedAt: event.payload.createdAt,
1251 },
1252 createdAt: event.payload.createdAt,
1253 });
1254 yield* appendProviderFailureActivity({
1255 threadId: event.payload.threadId,
1256 kind: "provider.turn.interrupt.failed",
1257 summary: "Provider turn interrupt failed",
1258 detail,
1259 turnId: event.payload.turnId ?? null,
1260 createdAt: event.payload.createdAt,
1261 });
1262 });
1263 };
1264
11951265 // Orchestration turn ids are not provider turn ids, so interrupt by session.
1196 yield* providerService.interruptTurn({ threadId: event.payload.threadId });
1266 yield* providerService
1267 .interruptTurn({ threadId: event.payload.threadId })
1268 .pipe(Effect.catchCause(recoverInterruptFailure));
11971269 });
11981270
11991271 const processApprovalResponseRequested = Effect.fn("processApprovalResponseRequested")(function* (