History

Handle stale pending requests and reused Claude text block indices

- Map unknown pending approval/user-input errors to explicit stale-request details - Clear stale pending approvals and user-input prompts in session derivation logic - Treat reused Claude text block indexes as new assistant messages to avoid item ID reuse - Persist latest runtime metadata before provider stop-all and expand regression coverage

Julius Marminge committed 7 months ago

Commit 754efdb1 parent ef843d48 files · +473 −36Browse files

Unified · Split

8 files
  1. apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +112 −0
  2. apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +21 −2
  3. apps/server/src/provider/Layers/ClaudeAdapter.test.ts +166 −0
  4. apps/server/src/provider/Layers/ClaudeAdapter.ts +18 −23
  5. apps/server/src/provider/Layers/ProviderService.test.ts +32 −7
  6. apps/server/src/provider/Layers/ProviderService.ts +27 −3
  7. apps/web/src/session-logic.test.ts +72 −0
  8. apps/web/src/session-logic.ts +25 −1
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+112−0
@@ -1135,6 +1135,7 @@ describe("ProviderCommandReactor", () => { · 1134 unmodified lines
1135 expect(failureActivity).toBeDefined();1135 expect(failureActivity).toBeDefined();
1136 expect(failureActivity?.payload).toMatchObject({1136 expect(failureActivity?.payload).toMatchObject({
1137 requestId: "approval-request-1",1137 requestId: "approval-request-1",
1138 detail: expect.stringContaining("Stale pending approval request: approval-request-1"),
1138 });1139 });
1139 1140
1140 const resolvedActivity = thread?.activities.find(1141 const resolvedActivity = thread?.activities.find(
@@ -1147,6 +1148,117 @@ describe("ProviderCommandReactor", () => { · 6 unmodified lines
1147 expect(resolvedActivity).toBeUndefined();1148 expect(resolvedActivity).toBeUndefined();
1148 });1149 });
1149 1150
1151 it("surfaces stale provider user-input failures without faking user-input resolution", async () => {
1152 const harness = await createHarness();
1153 const now = new Date().toISOString();
1154 harness.respondToUserInput.mockImplementation(() =>
1155 Effect.fail(
1156 new ProviderAdapterRequestError({
1157 provider: "claudeAgent",
1158 method: "item/tool/respondToUserInput",
1159 detail: "Unknown pending user-input request: user-input-request-1",
1160 }),
1161 ),
1162 );
1163
1164 await Effect.runPromise(
1165 harness.engine.dispatch({
1166 type: "thread.session.set",
1167 commandId: CommandId.makeUnsafe("cmd-session-set-for-user-input-error"),
1168 threadId: ThreadId.makeUnsafe("thread-1"),
1169 session: {
1170 threadId: ThreadId.makeUnsafe("thread-1"),
1171 status: "running",
1172 providerName: "claudeAgent",
1173 runtimeMode: "approval-required",
1174 activeTurnId: null,
1175 lastError: null,
1176 updatedAt: now,
1177 },
1178 createdAt: now,
1179 }),
1180 );
1181
1182 await Effect.runPromise(
1183 harness.engine.dispatch({
1184 type: "thread.activity.append",
1185 commandId: CommandId.makeUnsafe("cmd-user-input-requested"),
1186 threadId: ThreadId.makeUnsafe("thread-1"),
1187 activity: {
1188 id: EventId.makeUnsafe("activity-user-input-requested"),
1189 tone: "info",
1190 kind: "user-input.requested",
1191 summary: "User input requested",
1192 payload: {
1193 requestId: "user-input-request-1",
1194 questions: [
1195 {
1196 id: "sandbox_mode",
1197 header: "Sandbox",
1198 question: "Which mode should be used?",
1199 options: [
1200 {
1201 label: "workspace-write",
1202 description: "Allow workspace writes only",
1203 },
1204 ],
1205 },
1206 ],
1207 },
1208 turnId: null,
1209 createdAt: now,
1210 },
1211 createdAt: now,
1212 }),
1213 );
1214
1215 await Effect.runPromise(
1216 harness.engine.dispatch({
1217 type: "thread.user-input.respond",
1218 commandId: CommandId.makeUnsafe("cmd-user-input-respond-stale"),
1219 threadId: ThreadId.makeUnsafe("thread-1"),
1220 requestId: asApprovalRequestId("user-input-request-1"),
1221 answers: {
1222 sandbox_mode: "workspace-write",
1223 },
1224 createdAt: now,
1225 }),
1226 );
1227
1228 await waitFor(async () => {
1229 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1230 const thread = readModel.threads.find(
1231 (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),
1232 );
1233 if (!thread) return false;
1234 return thread.activities.some(
1235 (activity) => activity.kind === "provider.user-input.respond.failed",
1236 );
1237 });
1238
1239 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1240 const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
1241 expect(thread).toBeDefined();
1242
1243 const failureActivity = thread?.activities.find(
1244 (activity) => activity.kind === "provider.user-input.respond.failed",
1245 );
1246 expect(failureActivity).toBeDefined();
1247 expect(failureActivity?.payload).toMatchObject({
1248 requestId: "user-input-request-1",
1249 detail: expect.stringContaining("Stale pending user-input request: user-input-request-1"),
1250 });
1251
1252 const resolvedActivity = thread?.activities.find(
1253 (activity) =>
1254 activity.kind === "user-input.resolved" &&
1255 typeof activity.payload === "object" &&
1256 activity.payload !== null &&
1257 (activity.payload as Record<string, unknown>).requestId === "user-input-request-1",
1258 );
1259 expect(resolvedActivity).toBeUndefined();
1260 });
1261
1150 it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => {1262 it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => {
1151 const harness = await createHarness();1263 const harness = await createHarness();
1152 const now = new Date().toISOString();1264 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+21−2
@@ -97,6 +97,21 @@ function isUnknownPendingApprovalRequestError(cause: Cause.Cause<ProviderService · 96 unmodified lines
97 );97 );
98}98}
99 99
100function isUnknownPendingUserInputRequestError(cause: Cause.Cause<ProviderServiceError>): boolean {
101 const error = Cause.squash(cause);
102 if (Schema.is(ProviderAdapterRequestError)(error)) {
103 return error.detail.toLowerCase().includes("unknown pending user-input request");
104 }
105 return Cause.pretty(cause).toLowerCase().includes("unknown pending user-input request");
106}
107
108function stalePendingRequestDetail(
109 requestKind: "approval" | "user-input",
110 requestId: string,
111): string {
112 return `Stale pending ${requestKind} request: ${requestId}. Provider callback state does not survive app restarts or recovered sessions. Restart the turn to continue.`;
113}
114
100function isTemporaryWorktreeBranch(branch: string): boolean {115function isTemporaryWorktreeBranch(branch: string): boolean {
101 return TEMP_WORKTREE_BRANCH_PATTERN.test(branch.trim().toLowerCase());116 return TEMP_WORKTREE_BRANCH_PATTERN.test(branch.trim().toLowerCase());
102}117}
@@ -589,7 +604,9 @@ const make = Effect.gen(function* () { · 486 unmodified lines
589 threadId: event.payload.threadId,604 threadId: event.payload.threadId,
590 kind: "provider.approval.respond.failed",605 kind: "provider.approval.respond.failed",
591 summary: "Provider approval response failed",606 summary: "Provider approval response failed",
592 detail: Cause.pretty(cause),607 detail: isUnknownPendingApprovalRequestError(cause)
608 ? stalePendingRequestDetail("approval", event.payload.requestId)
609 : Cause.pretty(cause),
593 turnId: null,610 turnId: null,
594 createdAt: event.payload.createdAt,611 createdAt: event.payload.createdAt,
595 requestId: event.payload.requestId,612 requestId: event.payload.requestId,
@@ -633,7 +650,9 @@ const make = Effect.gen(function* () { · 37 unmodified lines
633 threadId: event.payload.threadId,650 threadId: event.payload.threadId,
634 kind: "provider.user-input.respond.failed",651 kind: "provider.user-input.respond.failed",
635 summary: "Provider user input response failed",652 summary: "Provider user input response failed",
636 detail: Cause.pretty(cause),653 detail: isUnknownPendingUserInputRequestError(cause)
654 ? stalePendingRequestDetail("user-input", event.payload.requestId)
655 : Cause.pretty(cause),
637 turnId: null,656 turnId: null,
638 createdAt: event.payload.createdAt,657 createdAt: event.payload.createdAt,
639 requestId: event.payload.requestId,658 requestId: event.payload.requestId,
apps/server/src/provider/Layers/ClaudeAdapter.test.ts+166−0
@@ -1089,6 +1089,172 @@ describe("ClaudeAdapterLive", () => { · 1088 unmodified lines
1089 },1089 },
1090 );1090 );
1091 1091
1092 it.effect("creates a fresh assistant message when Claude reuses a text block index", () => {
1093 const harness = makeHarness();
1094 return Effect.gen(function* () {
1095 const adapter = yield* ClaudeAdapter;
1096
1097 const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 9).pipe(
1098 Stream.runCollect,
1099 Effect.forkChild,
1100 );
1101
1102 const session = yield* adapter.startSession({
1103 threadId: THREAD_ID,
1104 provider: "claudeAgent",
1105 runtimeMode: "full-access",
1106 });
1107
1108 yield* adapter.sendTurn({
1109 threadId: session.threadId,
1110 input: "hello",
1111 attachments: [],
1112 });
1113
1114 harness.query.emit({
1115 type: "stream_event",
1116 session_id: "sdk-session-reused-text-index",
1117 uuid: "stream-reused-start-1",
1118 parent_tool_use_id: null,
1119 event: {
1120 type: "content_block_start",
1121 index: 0,
1122 content_block: {
1123 type: "text",
1124 text: "",
1125 },
1126 },
1127 } as unknown as SDKMessage);
1128
1129 harness.query.emit({
1130 type: "stream_event",
1131 session_id: "sdk-session-reused-text-index",
1132 uuid: "stream-reused-delta-1",
1133 parent_tool_use_id: null,
1134 event: {
1135 type: "content_block_delta",
1136 index: 0,
1137 delta: {
1138 type: "text_delta",
1139 text: "First",
1140 },
1141 },
1142 } as unknown as SDKMessage);
1143
1144 harness.query.emit({
1145 type: "stream_event",
1146 session_id: "sdk-session-reused-text-index",
1147 uuid: "stream-reused-stop-1",
1148 parent_tool_use_id: null,
1149 event: {
1150 type: "content_block_stop",
1151 index: 0,
1152 },
1153 } as unknown as SDKMessage);
1154
1155 harness.query.emit({
1156 type: "stream_event",
1157 session_id: "sdk-session-reused-text-index",
1158 uuid: "stream-reused-start-2",
1159 parent_tool_use_id: null,
1160 event: {
1161 type: "content_block_start",
1162 index: 0,
1163 content_block: {
1164 type: "text",
1165 text: "",
1166 },
1167 },
1168 } as unknown as SDKMessage);
1169
1170 harness.query.emit({
1171 type: "stream_event",
1172 session_id: "sdk-session-reused-text-index",
1173 uuid: "stream-reused-delta-2",
1174 parent_tool_use_id: null,
1175 event: {
1176 type: "content_block_delta",
1177 index: 0,
1178 delta: {
1179 type: "text_delta",
1180 text: "Second",
1181 },
1182 },
1183 } as unknown as SDKMessage);
1184
1185 harness.query.emit({
1186 type: "stream_event",
1187 session_id: "sdk-session-reused-text-index",
1188 uuid: "stream-reused-stop-2",
1189 parent_tool_use_id: null,
1190 event: {
1191 type: "content_block_stop",
1192 index: 0,
1193 },
1194 } as unknown as SDKMessage);
1195
1196 harness.query.emit({
1197 type: "result",
1198 subtype: "success",
1199 is_error: false,
1200 errors: [],
1201 session_id: "sdk-session-reused-text-index",
1202 uuid: "result-reused-text-index",
1203 } as unknown as SDKMessage);
1204
1205 const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber));
1206 assert.deepEqual(
1207 runtimeEvents.map((event) => event.type),
1208 [
1209 "session.started",
1210 "session.configured",
1211 "session.state.changed",
1212 "turn.started",
1213 "thread.started",
1214 "content.delta",
1215 "item.completed",
1216 "content.delta",
1217 "item.completed",
1218 ],
1219 );
1220
1221 const assistantDeltas = runtimeEvents.filter(
1222 (event) => event.type === "content.delta" && event.payload.streamKind === "assistant_text",
1223 );
1224 assert.equal(assistantDeltas.length, 2);
1225 if (assistantDeltas.length !== 2) {
1226 return;
1227 }
1228 const [firstAssistantDelta, secondAssistantDelta] = assistantDeltas;
1229 assert.equal(firstAssistantDelta?.type, "content.delta");
1230 assert.equal(secondAssistantDelta?.type, "content.delta");
1231 if (
1232 firstAssistantDelta?.type !== "content.delta" ||
1233 secondAssistantDelta?.type !== "content.delta"
1234 ) {
1235 return;
1236 }
1237 assert.equal(firstAssistantDelta.payload.delta, "First");
1238 assert.equal(secondAssistantDelta.payload.delta, "Second");
1239 assert.notEqual(firstAssistantDelta.itemId, secondAssistantDelta.itemId);
1240
1241 const assistantCompletions = runtimeEvents.filter(
1242 (event) =>
1243 event.type === "item.completed" && event.payload.itemType === "assistant_message",
1244 );
1245 assert.equal(assistantCompletions.length, 2);
1246 assert.equal(String(assistantCompletions[0]?.itemId), String(firstAssistantDelta.itemId));
1247 assert.equal(String(assistantCompletions[1]?.itemId), String(secondAssistantDelta.itemId));
1248 assert.notEqual(
1249 String(assistantCompletions[0]?.itemId),
1250 String(assistantCompletions[1]?.itemId),
1251 );
1252 }).pipe(
1253 Effect.provideService(Random.Random, makeDeterministicRandomService()),
1254 Effect.provide(harness.layer),
1255 );
1256 });
1257
1092 it.effect("falls back to assistant payload text when stream deltas are absent", () => {1258 it.effect("falls back to assistant payload text when stream deltas are absent", () => {
1093 const harness = makeHarness();1259 const harness = makeHarness();
1094 return Effect.gen(function* () {1260 return Effect.gen(function* () {
apps/server/src/provider/Layers/ClaudeAdapter.ts+18−23
@@ -87,13 +87,14 @@ interface ClaudeTurnState { · 86 unmodified lines
87 readonly startedAt: string;87 readonly startedAt: string;
88 readonly items: Array<unknown>;88 readonly items: Array<unknown>;
89 readonly assistantTextBlocks: Map<number, AssistantTextBlockState>;89 readonly assistantTextBlocks: Map<number, AssistantTextBlockState>;
90 readonly assistantTextBlockOrder: Array<number>;90 readonly assistantTextBlockOrder: Array<AssistantTextBlockState>;
91 readonly capturedProposedPlanKeys: Set<string>;91 readonly capturedProposedPlanKeys: Set<string>;
92 nextSyntheticAssistantBlockIndex: number;92 nextSyntheticAssistantBlockIndex: number;
93}93}
94 94
95interface AssistantTextBlockState {95interface AssistantTextBlockState {
96 readonly itemId: string;96 readonly itemId: string;
97 readonly blockIndex: number;
97 emittedTextDelta: boolean;98 emittedTextDelta: boolean;
98 fallbackText: string;99 fallbackText: string;
99 streamClosed: boolean;100 streamClosed: boolean;
@@ -829,7 +830,7 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 729 unmodified lines
829 }830 }
830 831
831 const existing = turnState.assistantTextBlocks.get(blockIndex);832 const existing = turnState.assistantTextBlocks.get(blockIndex);
832 if (existing) {833 if (existing && !existing.completionEmitted) {
833 if (existing.fallbackText.length === 0 && options?.fallbackText) {834 if (existing.fallbackText.length === 0 && options?.fallbackText) {
834 existing.fallbackText = options.fallbackText;835 existing.fallbackText = options.fallbackText;
835 }836 }
@@ -841,13 +842,14 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 5 unmodified lines
841 842
842 const block: AssistantTextBlockState = {843 const block: AssistantTextBlockState = {
843 itemId: yield* Random.nextUUIDv4,844 itemId: yield* Random.nextUUIDv4,
845 blockIndex,
844 emittedTextDelta: false,846 emittedTextDelta: false,
845 fallbackText: options?.fallbackText ?? "",847 fallbackText: options?.fallbackText ?? "",
846 streamClosed: options?.streamClosed ?? false,848 streamClosed: options?.streamClosed ?? false,
847 completionEmitted: false,849 completionEmitted: false,
848 };850 };
849 turnState.assistantTextBlocks.set(blockIndex, block);851 turnState.assistantTextBlocks.set(blockIndex, block);
850 turnState.assistantTextBlockOrder.push(blockIndex);852 turnState.assistantTextBlockOrder.push(block);
851 return { blockIndex, block };853 return { blockIndex, block };
852 });854 });
853 855
@@ -877,7 +879,7 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 23 unmodified lines
877 879
878 const completeAssistantTextBlock = (880 const completeAssistantTextBlock = (
879 context: ClaudeSessionContext,881 context: ClaudeSessionContext,
880 blockIndex: number,882 block: AssistantTextBlockState,
881 options?: {883 options?: {
882 readonly force?: boolean;884 readonly force?: boolean;
883 readonly rawMethod?: string;885 readonly rawMethod?: string;
@@ -886,8 +888,7 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 2 unmodified lines
886 ): Effect.Effect<void> =>888 ): Effect.Effect<void> =>
887 Effect.gen(function* () {889 Effect.gen(function* () {
888 const turnState = context.turnState;890 const turnState = context.turnState;
889 const block = turnState?.assistantTextBlocks.get(blockIndex);891 if (!turnState || block.completionEmitted) {
890 if (!turnState || !block || block.completionEmitted) {
891 return;892 return;
892 }893 }
893 894
@@ -923,6 +924,9 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 29 unmodified lines
923 }924 }
924 925
925 block.completionEmitted = true;926 block.completionEmitted = true;
927 if (turnState.assistantTextBlocks.get(block.blockIndex) === block) {
928 turnState.assistantTextBlocks.delete(block.blockIndex);
929 }
926 930
927 const stamp = yield* makeEventStamp();931 const stamp = yield* makeEventStamp();
928 yield* offerRuntimeEvent({932 yield* offerRuntimeEvent({
@@ -967,19 +971,10 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 38 unmodified lines
967 return;971 return;
968 }972 }
969 973
970 const orderedBlocks = turnState.assistantTextBlockOrder974 const orderedBlocks = turnState.assistantTextBlockOrder.map((block) => ({
971 .map((blockIndex) => ({975 blockIndex: block.blockIndex,
972 blockIndex,976 block,
973 block: turnState.assistantTextBlocks.get(blockIndex),977 }));
974 }))
975 .filter(
976 (
977 entry,
978 ): entry is {
979 readonly blockIndex: number;
980 readonly block: AssistantTextBlockState;
981 } => entry.block !== undefined,
982 );
983 978
984 for (const [position, text] of snapshotTextBlocks.entries()) {979 for (const [position, text] of snapshotTextBlocks.entries()) {
985 const existingEntry = orderedBlocks[position];980 const existingEntry = orderedBlocks[position];
@@ -1003,7 +998,7 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 17 unmodified lines
1003 }998 }
1004 999
1005 if (entry.block.streamClosed && !entry.block.completionEmitted) {1000 if (entry.block.streamClosed && !entry.block.completionEmitted) {
1006 yield* completeAssistantTextBlock(context, entry.blockIndex, {1001 yield* completeAssistantTextBlock(context, entry.block, {
1007 rawMethod: "claude/assistant",1002 rawMethod: "claude/assistant",
1008 rawPayload: message,1003 rawPayload: message,
1009 });1004 });
@@ -1208,8 +1203,8 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 198 unmodified lines
1208 // Clear any remaining stale entries (e.g. from interrupted content blocks)1203 // Clear any remaining stale entries (e.g. from interrupted content blocks)
1209 context.inFlightTools.clear();1204 context.inFlightTools.clear();
1210 1205
1211 for (const blockIndex of turnState.assistantTextBlockOrder) {1206 for (const block of turnState.assistantTextBlockOrder) {
1212 yield* completeAssistantTextBlock(context, blockIndex, {1207 yield* completeAssistantTextBlock(context, block, {
1213 force: true,1208 force: true,
1214 rawMethod: "claude/result",1209 rawMethod: "claude/result",
1215 rawPayload: result ?? { status },1210 rawPayload: result ?? { status },
@@ -1460,7 +1455,7 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { · 244 unmodified lines
1460 const assistantBlock = context.turnState?.assistantTextBlocks.get(index);1455 const assistantBlock = context.turnState?.assistantTextBlocks.get(index);
1461 if (assistantBlock) {1456 if (assistantBlock) {
1462 assistantBlock.streamClosed = true;1457 assistantBlock.streamClosed = true;
1463 yield* completeAssistantTextBlock(context, index, {1458 yield* completeAssistantTextBlock(context, assistantBlock, {
1464 rawMethod: "claude/stream_event/content_block_stop",1459 rawMethod: "claude/stream_event/content_block_stop",
1465 rawPayload: message,1460 rawPayload: message,
1466 });1461 });
apps/server/src/provider/Layers/ProviderService.test.ts+32−7
@@ -195,9 +195,21 @@ function makeFakeCodexAdapter(provider: ProviderKind = "codex") { · 194 unmodified lines
195 Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));195 Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));
196 };196 };
197 197
198 const updateSession = (
199 threadId: ThreadId,
200 update: (session: ProviderSession) => ProviderSession,
201 ): void => {
202 const existing = sessions.get(threadId);
203 if (!existing) {
204 return;
205 }
206 sessions.set(threadId, update(existing));
207 };
208
198 return {209 return {
199 adapter,210 adapter,
200 emit,211 emit,
212 updateSession,
201 startSession,213 startSession,
202 sendTurn,214 sendTurn,
203 interruptTurn,215 interruptTurn,
@@ -348,16 +360,29 @@ it.effect( · 144 unmodified lines
348 Layer.provide(firstDirectoryLayer),360 Layer.provide(firstDirectoryLayer),
349 Layer.provide(AnalyticsService.layerTest),361 Layer.provide(AnalyticsService.layerTest),
350 );362 );
363 const updatedResumeCursor = {
364 threadId: asThreadId("thread-1"),
365 resume: "resume-session-1",
366 resumeSessionAt: "assistant-message-1",
367 turnCount: 1,
368 };
351 369
352 const startedSession = yield* Effect.gen(function* () {370 const startedSession = yield* Effect.gen(function* () {
353 const provider = yield* ProviderService;371 const provider = yield* ProviderService;
354 const threadId = asThreadId("thread-1");372 const threadId = asThreadId("thread-1");
355 return yield* provider.startSession(threadId, {373 const session = yield* provider.startSession(threadId, {
356 provider: "codex",374 provider: "codex",
357 cwd: "/tmp/project",375 cwd: "/tmp/project",
358 runtimeMode: "full-access",376 runtimeMode: "full-access",
359 threadId,377 threadId,
360 });378 });
379 firstCodex.updateSession(threadId, (existing) => ({
380 ...existing,
381 status: "ready",
382 resumeCursor: updatedResumeCursor,
383 updatedAt: new Date(Date.now() + 1_000).toISOString(),
384 }));
385 return session;
361 }).pipe(Effect.provide(firstProviderLayer));386 }).pipe(Effect.provide(firstProviderLayer));
362 387
363 const persistedAfterStopAll = yield* Effect.gen(function* () {388 const persistedAfterStopAll = yield* Effect.gen(function* () {
@@ -367,7 +392,7 @@ it.effect( · 3 unmodified lines
367 assert.equal(Option.isSome(persistedAfterStopAll), true);392 assert.equal(Option.isSome(persistedAfterStopAll), true);
368 if (Option.isSome(persistedAfterStopAll)) {393 if (Option.isSome(persistedAfterStopAll)) {
369 assert.equal(persistedAfterStopAll.value.status, "stopped");394 assert.equal(persistedAfterStopAll.value.status, "stopped");
370 assert.deepEqual(persistedAfterStopAll.value.resumeCursor, startedSession.resumeCursor);395 assert.deepEqual(persistedAfterStopAll.value.resumeCursor, updatedResumeCursor);
371 }396 }
372 397
373 const secondCodex = makeFakeCodexAdapter();398 const secondCodex = makeFakeCodexAdapter();
@@ -410,7 +435,7 @@ it.effect( · 36 unmodified lines
410 };435 };
411 assert.equal(startPayload.provider, "codex");436 assert.equal(startPayload.provider, "codex");
412 assert.equal(startPayload.cwd, "/tmp/project");437 assert.equal(startPayload.cwd, "/tmp/project");
413 assert.deepEqual(startPayload.resumeCursor, startedSession.resumeCursor);438 assert.deepEqual(startPayload.resumeCursor, updatedResumeCursor);
414 assert.equal(startPayload.threadId, startedSession.threadId);439 assert.equal(startPayload.threadId, startedSession.threadId);
415 }440 }
416 assert.equal(secondCodex.rollbackThread.mock.calls.length, 1);441 assert.equal(secondCodex.rollbackThread.mock.calls.length, 1);
@@ -733,7 +758,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { · 316 unmodified lines
733 const consumer = yield* Stream.runForEach(provider.streamEvents, (event) =>758 const consumer = yield* Stream.runForEach(provider.streamEvents, (event) =>
734 Ref.update(eventsRef, (current) => [...current, event]),759 Ref.update(eventsRef, (current) => [...current, event]),
735 ).pipe(Effect.forkChild);760 ).pipe(Effect.forkChild);
736 yield* sleep(20);761 yield* sleep(50);
737 762
738 const completedEvent: LegacyProviderRuntimeEvent = {763 const completedEvent: LegacyProviderRuntimeEvent = {
739 type: "turn.completed",764 type: "turn.completed",
@@ -746,7 +771,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { · 6 unmodified lines
746 };771 };
747 772
748 fanout.codex.emit(completedEvent);773 fanout.codex.emit(completedEvent);
749 yield* sleep(20);774 yield* sleep(50);
750 775
751 const events = yield* Ref.get(eventsRef);776 const events = yield* Ref.get(eventsRef);
752 yield* Fiber.interrupt(consumer);777 yield* Fiber.interrupt(consumer);
@@ -772,7 +797,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { · 19 unmodified lines
772 Stream.runForEach((event) => Ref.update(receivedRef, (current) => [...current, event])),797 Stream.runForEach((event) => Ref.update(receivedRef, (current) => [...current, event])),
773 Effect.forkChild,798 Effect.forkChild,
774 );799 );
775 yield* sleep(20);800 yield* sleep(50);
776 801
777 fanout.codex.emit({802 fanout.codex.emit({
778 type: "tool.started",803 type: "tool.started",
@@ -836,7 +861,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { · 57 unmodified lines
836 Stream.runForEach(() => Effect.fail("listener crash")),861 Stream.runForEach(() => Effect.fail("listener crash")),
837 Effect.forkChild,862 Effect.forkChild,
838 );863 );
839 yield* sleep(20);864 yield* sleep(50);
840 865
841 const events: ReadonlyArray<LegacyProviderRuntimeEvent> = [866 const events: ReadonlyArray<LegacyProviderRuntimeEvent> = [
842 {867 {
apps/server/src/provider/Layers/ProviderService.ts+27−3
@@ -88,7 +88,12 @@ function toRuntimeStatus(session: ProviderSession): "starting" | "running" | "st · 87 unmodified lines
88 88
89function toRuntimePayloadFromSession(89function toRuntimePayloadFromSession(
90 session: ProviderSession,90 session: ProviderSession,
91 extra?: { readonly modelOptions?: unknown; readonly providerOptions?: unknown },91 extra?: {
92 readonly modelOptions?: unknown;
93 readonly providerOptions?: unknown;
94 readonly lastRuntimeEvent?: string;
95 readonly lastRuntimeEventAt?: string;
96 },
92): Record<string, unknown> {97): Record<string, unknown> {
93 return {98 return {
94 cwd: session.cwd ?? null,99 cwd: session.cwd ?? null,
@@ -97,6 +102,10 @@ function toRuntimePayloadFromSession( · 2 unmodified lines
97 lastError: session.lastError ?? null,102 lastError: session.lastError ?? null,
98 ...(extra?.modelOptions !== undefined ? { modelOptions: extra.modelOptions } : {}),103 ...(extra?.modelOptions !== undefined ? { modelOptions: extra.modelOptions } : {}),
99 ...(extra?.providerOptions !== undefined ? { providerOptions: extra.providerOptions } : {}),104 ...(extra?.providerOptions !== undefined ? { providerOptions: extra.providerOptions } : {}),
105 ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}),
106 ...(extra?.lastRuntimeEventAt !== undefined
107 ? { lastRuntimeEventAt: extra.lastRuntimeEventAt }
108 : {}),
100 };109 };
101}110}
102 111
@@ -162,7 +171,12 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => · 59 unmodified lines
162 const upsertSessionBinding = (171 const upsertSessionBinding = (
163 session: ProviderSession,172 session: ProviderSession,
164 threadId: ThreadId,173 threadId: ThreadId,
165 extra?: { readonly modelOptions?: unknown; readonly providerOptions?: unknown },174 extra?: {
175 readonly modelOptions?: unknown;
176 readonly providerOptions?: unknown;
177 readonly lastRuntimeEvent?: string;
178 readonly lastRuntimeEventAt?: string;
179 },
166 ) =>180 ) =>
167 directory.upsert({181 directory.upsert({
168 threadId,182 threadId,
@@ -177,7 +191,6 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => · 8 unmodified lines
177 const adapters = yield* Effect.forEach(providers, (provider) =>191 const adapters = yield* Effect.forEach(providers, (provider) =>
178 registry.getByProvider(provider),192 registry.getByProvider(provider),
179 );193 );
180
181 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>194 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
182 publishRuntimeEvent(event);195 publishRuntimeEvent(event);
183 196
@@ -513,6 +526,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => · 329 unmodified lines
513 const runStopAll = () =>526 const runStopAll = () =>
514 Effect.gen(function* () {527 Effect.gen(function* () {
515 const threadIds = yield* directory.listThreadIds();528 const threadIds = yield* directory.listThreadIds();
529 const activeSessions = yield* Effect.forEach(adapters, (adapter) =>
530 adapter.listSessions(),
531 ).pipe(
532 Effect.map((sessionsByAdapter) => sessionsByAdapter.flatMap((sessions) => sessions)),
533 );
534 yield* Effect.forEach(activeSessions, (session) =>
535 upsertSessionBinding(session, session.threadId, {
536 lastRuntimeEvent: "provider.stopAll",
537 lastRuntimeEventAt: new Date().toISOString(),
538 }),
539 ).pipe(Effect.asVoid);
516 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);540 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
517 yield* Effect.forEach(threadIds, (threadId) =>541 yield* Effect.forEach(threadIds, (threadId) =>
518 directory.getProvider(threadId).pipe(542 directory.getProvider(threadId).pipe(
apps/web/src/session-logic.test.ts+72−0
@@ -142,6 +142,36 @@ describe("derivePendingApprovals", () => { · 141 unmodified lines
142 142
143 expect(derivePendingApprovals(activities)).toEqual([]);143 expect(derivePendingApprovals(activities)).toEqual([]);
144 });144 });
145
146 it("clears stale pending approvals when the backend marks them stale after restart", () => {
147 const activities: OrchestrationThreadActivity[] = [
148 makeActivity({
149 id: "approval-open-stale-restart",
150 createdAt: "2026-02-23T00:00:01.000Z",
151 kind: "approval.requested",
152 summary: "Command approval requested",
153 tone: "approval",
154 payload: {
155 requestId: "req-stale-restart-1",
156 requestKind: "command",
157 },
158 }),
159 makeActivity({
160 id: "approval-failed-stale-restart",
161 createdAt: "2026-02-23T00:00:02.000Z",
162 kind: "provider.approval.respond.failed",
163 summary: "Provider approval response failed",
164 tone: "error",
165 payload: {
166 requestId: "req-stale-restart-1",
167 detail:
168 "Stale pending approval request: req-stale-restart-1. Provider callback state does not survive app restarts or recovered sessions. Restart the turn to continue.",
169 },
170 }),
171 ];
172
173 expect(derivePendingApprovals(activities)).toEqual([]);
174 });
145});175});
146 176
147describe("derivePendingUserInputs", () => {177describe("derivePendingUserInputs", () => {
@@ -228,6 +258,48 @@ describe("derivePendingUserInputs", () => { · 80 unmodified lines
228 },258 },
229 ]);259 ]);
230 });260 });
261
262 it("clears stale pending user-input prompts when the provider reports an orphaned request", () => {
263 const activities: OrchestrationThreadActivity[] = [
264 makeActivity({
265 id: "user-input-open-stale",
266 createdAt: "2026-02-23T00:00:01.000Z",
267 kind: "user-input.requested",
268 summary: "User input requested",
269 tone: "info",
270 payload: {
271 requestId: "req-user-input-stale-1",
272 questions: [
273 {
274 id: "sandbox_mode",
275 header: "Sandbox",
276 question: "Which mode should be used?",
277 options: [
278 {
279 label: "workspace-write",
280 description: "Allow workspace writes only",
281 },
282 ],
283 },
284 ],
285 },
286 }),
287 makeActivity({
288 id: "user-input-failed-stale",
289 createdAt: "2026-02-23T00:00:02.000Z",
290 kind: "provider.user-input.respond.failed",
291 summary: "Provider user input response failed",
292 tone: "error",
293 payload: {
294 requestId: "req-user-input-stale-1",
295 detail:
296 "Stale pending user-input request: req-user-input-stale-1. Provider callback state does not survive app restarts or recovered sessions. Restart the turn to continue.",
297 },
298 }),
299 ];
300
301 expect(derivePendingUserInputs(activities)).toEqual([]);
302 });
231});303});
232 304
233describe("deriveActivePlanState", () => {305describe("deriveActivePlanState", () => {
apps/web/src/session-logic.ts+25−1
@@ -160,6 +160,20 @@ function requestKindFromRequestType(requestType: unknown): PendingApproval["requ · 159 unmodified lines
160 }160 }
161}161}
162 162
163function isStalePendingRequestFailureDetail(detail: string | undefined): boolean {
164 const normalized = detail?.toLowerCase();
165 if (!normalized) {
166 return false;
167 }
168 return (
169 normalized.includes("stale pending approval request") ||
170 normalized.includes("stale pending user-input request") ||
171 normalized.includes("unknown pending approval request") ||
172 normalized.includes("unknown pending permission request") ||
173 normalized.includes("unknown pending user-input request")
174 );
175}
176
163export function derivePendingApprovals(177export function derivePendingApprovals(
164 activities: ReadonlyArray<OrchestrationThreadActivity>,178 activities: ReadonlyArray<OrchestrationThreadActivity>,
165): PendingApproval[] {179): PendingApproval[] {
@@ -204,7 +218,7 @@ export function derivePendingApprovals( · 38 unmodified lines
204 if (218 if (
205 activity.kind === "provider.approval.respond.failed" &&219 activity.kind === "provider.approval.respond.failed" &&
206 requestId &&220 requestId &&
207 detail?.includes("Unknown pending permission request")221 isStalePendingRequestFailureDetail(detail)
208 ) {222 ) {
209 openByRequestId.delete(requestId);223 openByRequestId.delete(requestId);
210 continue;224 continue;
@@ -280,6 +294,7 @@ export function derivePendingUserInputs( · 69 unmodified lines
280 payload && typeof payload.requestId === "string"294 payload && typeof payload.requestId === "string"
281 ? ApprovalRequestId.makeUnsafe(payload.requestId)295 ? ApprovalRequestId.makeUnsafe(payload.requestId)
282 : null;296 : null;
297 const detail = payload && typeof payload.detail === "string" ? payload.detail : undefined;
283 298
284 if (activity.kind === "user-input.requested" && requestId) {299 if (activity.kind === "user-input.requested" && requestId) {
285 const questions = parseUserInputQuestions(payload);300 const questions = parseUserInputQuestions(payload);
@@ -296,6 +311,15 @@ export function derivePendingUserInputs( · 10 unmodified lines
296 311
297 if (activity.kind === "user-input.resolved" && requestId) {312 if (activity.kind === "user-input.resolved" && requestId) {
298 openByRequestId.delete(requestId);313 openByRequestId.delete(requestId);
314 continue;
315 }
316
317 if (
318 activity.kind === "provider.user-input.respond.failed" &&
319 requestId &&
320 isStalePendingRequestFailureDetail(detail)
321 ) {
322 openByRequestId.delete(requestId);
299 }323 }
300 }324 }
301 325