authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-11 01:46:46-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-11 01:46:46-03:00
log57aee3e19f1910f3323f06384bb2a1d02b79e369
tree3af36d5c62845b2fddfae401442d5b2262478700
parentef6fa118749572da1905eca5d0543d1e98f0b1b2
signature Signed by PGP key B5690EEEBB952194

fix(server): queue messages during context compaction (#11107)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

2 files changed, 374 insertions(+), 93 deletions(-)

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+231-79
...@@ -175,6 +175,8 @@ describe("ProviderCommandReactor", () => {...@@ -175,6 +175,8 @@ describe("ProviderCommandReactor", () => {
175 readonly titleRegenerationBeforeStart?: "one" | "two";175 readonly titleRegenerationBeforeStart?: "one" | "two";
176 readonly serverActivation?: Effect.Effect<void>;176 readonly serverActivation?: Effect.Effect<void>;
177 readonly beforeReadySessionDispatch?: () => Effect.Effect<void>;177 readonly beforeReadySessionDispatch?: () => Effect.Effect<void>;
178 readonly beforeTurnStartDispatch?: () => Effect.Effect<void>;
179 readonly afterTurnStartDispatch?: () => Effect.Effect<void>;
178 readonly compactThreadEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;180 readonly compactThreadEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
179 readonly interruptTurnEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;181 readonly interruptTurnEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
180 readonly stopSessionEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;182 readonly stopSessionEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;
...@@ -431,11 +433,21 @@ describe("ProviderCommandReactor", () => {...@@ -431,11 +433,21 @@ describe("ProviderCommandReactor", () => {
431 return Effect.die(new Error("Injected title regeneration completion failure"));433 return Effect.die(new Error("Injected title regeneration completion failure"));
432 }434 }
433 }435 }
434 return (436 const isReplay =
437 command.type === "thread.turn.start" &&
438 command.commandId.startsWith("server:after-compaction:");
439 const before =
435 command.type === "thread.session.set" && command.session.status === "ready"440 command.type === "thread.session.set" && command.session.status === "ready"
436 ? (input?.beforeReadySessionDispatch?.() ?? Effect.void)441 ? input?.beforeReadySessionDispatch
437 : Effect.void442 : isReplay
438 ).pipe(Effect.andThen(engine.dispatch(command)));443 ? input?.beforeTurnStartDispatch
444 : undefined;
445 return (before?.() ?? Effect.void).pipe(
446 Effect.andThen(engine.dispatch(command)),
447 Effect.tap(() =>
448 isReplay ? (input?.afterTurnStartDispatch?.() ?? Effect.void) : Effect.void,
449 ),
450 );
439 },451 },
440 get streamDomainEvents() {452 get streamDomainEvents() {
441 return engine.streamDomainEvents;453 return engine.streamDomainEvents;
...@@ -973,86 +985,203 @@ describe("ProviderCommandReactor", () => {...@@ -973,86 +985,203 @@ describe("ProviderCommandReactor", () => {
973 }),985 }),
974 );986 );
975987
976 effectIt.effect("keeps turns blocked until compaction restores the session", () =>988 effectIt.effect.each(["resume", "stop before resume", "stop after send"])(
977 Effect.gen(function* () {989 "queues messages until compaction restores the session (%s)",
978 const readyDispatchStarted = yield* Deferred.make<void>();990 (scenario) =>
979 const releaseReadyDispatch = yield* Deferred.make<void>();991 Effect.gen(function* () {
980 let blockReadyDispatch = false;992 const stopBeforeResume = scenario === "stop before resume";
981 const harness = yield* Effect.promise(() =>993 const readyDispatchStarted = yield* Deferred.make<void>();
982 createHarness({994 const releaseReadyDispatch = yield* Deferred.make<void>();
983 beforeReadySessionDispatch: () =>995 const firstSent = yield* Deferred.make<void>();
984 blockReadyDispatch996 const queuedSent = yield* Deferred.make<void>();
985 ? Deferred.succeed(readyDispatchStarted, undefined).pipe(997 const resumeStarted = yield* Deferred.make<void>();
986 Effect.andThen(Deferred.await(releaseReadyDispatch)),998 const releaseResume = yield* Deferred.make<void>();
987 )999 const resumeDispatched = yield* Deferred.make<void>();
988 : Effect.void,1000 const queuedSendStarted = yield* Deferred.make<void>();
989 }),1001 const releaseQueuedSend = yield* Deferred.make<void>();
990 );1002 let blockReadyDispatch = false;
991 const threadId = ThreadId.make("thread-1");1003 const harness = yield* Effect.promise(() =>
992 const now = "2026-01-01T00:00:00.000Z";1004 createHarness({
993 const dispatchTurn = (id: string, text: string, createdAt: string) =>1005 beforeTurnStartDispatch: () =>
994 harness.engine.dispatch({1006 stopBeforeResume
995 type: "thread.turn.start",1007 ? Deferred.succeed(resumeStarted, undefined).pipe(
996 commandId: CommandId.make(`cmd-${id}`),1008 Effect.andThen(Deferred.await(releaseResume)),
1009 )
1010 : Effect.void,
1011 afterTurnStartDispatch: () => Deferred.succeed(resumeDispatched, undefined),
1012 beforeReadySessionDispatch: () =>
1013 blockReadyDispatch
1014 ? Deferred.succeed(readyDispatchStarted, undefined).pipe(
1015 Effect.andThen(Deferred.await(releaseReadyDispatch)),
1016 )
1017 : Effect.void,
1018 }),
1019 );
1020 const threadId = ThreadId.make("thread-1");
1021 let sentCount = 0;
1022 harness.sendTurn.mockImplementation(() =>
1023 Effect.succeed({ threadId, turnId: asTurnId("turn-1") }).pipe(
1024 Effect.tap(() => {
1025 sentCount++;
1026 return sentCount === 1
1027 ? Deferred.succeed(firstSent, undefined)
1028 : sentCount === 2 && scenario === "stop after send"
1029 ? Deferred.succeed(queuedSendStarted, undefined).pipe(
1030 Effect.andThen(Deferred.await(releaseQueuedSend)),
1031 )
1032 : sentCount === 3
1033 ? Deferred.succeed(queuedSent, undefined)
1034 : Effect.void;
1035 }),
1036 ),
1037 );
1038 const now = "2026-01-01T00:00:00.000Z";
1039 const dispatchTurn = (id: string, text: string, createdAt: string) =>
1040 harness.engine.dispatch({
1041 type: "thread.turn.start",
1042 commandId: CommandId.make(`cmd-${id}`),
1043 threadId,
1044 message: {
1045 messageId: asMessageId(`user-message-${id}`),
1046 role: "user",
1047 text,
1048 attachments: [],
1049 },
1050 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1051 runtimeMode: "approval-required",
1052 createdAt,
1053 });
1054
1055 yield* dispatchTurn("before-blocked-compact", "hello", now);
1056 yield* Deferred.await(firstSent);
1057 yield* harness.engine.dispatch({
1058 type: "thread.session.set",
1059 commandId: CommandId.make("cmd-session-ready-before-blocked-compact"),
997 threadId,1060 threadId,
998 message: {1061 session: {
999 messageId: asMessageId(`user-message-${id}`),1062 threadId,
1000 role: "user",1063 status: "ready",
1001 text,1064 providerName: "codex",
1002 attachments: [],1065 providerInstanceId: ProviderInstanceId.make("codex"),
1066 runtimeMode: "approval-required",
1067 activeTurnId: null,
1068 lastError: null,
1069 updatedAt: now,
1003 },1070 },
1004 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,1071 createdAt: now,
1005 runtimeMode: "approval-required",
1006 createdAt,
1007 });1072 });
10081073
1009 yield* dispatchTurn("before-blocked-compact", "hello", now);1074 blockReadyDispatch = true;
1010 yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 1));1075 yield* dispatchTurn("blocked-compact", "/compact", "2026-01-01T00:00:01.000Z");
1011 yield* harness.engine.dispatch({1076 yield* Deferred.await(readyDispatchStarted);
1012 type: "thread.session.set",
1013 commandId: CommandId.make("cmd-session-ready-before-blocked-compact"),
1014 threadId,
1015 session: {
1016 threadId,
1017 status: "ready",
1018 providerName: "codex",
1019 providerInstanceId: ProviderInstanceId.make("codex"),
1020 runtimeMode: "approval-required",
1021 activeTurnId: null,
1022 lastError: null,
1023 updatedAt: now,
1024 },
1025 createdAt: now,
1026 });
1027
1028 blockReadyDispatch = true;
1029 yield* dispatchTurn("blocked-compact", "/compact", "2026-01-01T00:00:01.000Z");
1030 yield* Deferred.await(readyDispatchStarted);
10311077
1032 yield* dispatchTurn("during-compact-recovery", "too soon", "2026-01-01T00:00:02.000Z");1078 yield* harness.engine.dispatch({
1033 yield* Effect.promise(() =>1079 type: "thread.interaction-mode.set",
1034 waitFor(async () => {1080 commandId: CommandId.make("cmd-queued-mode-plan"),
1035 const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId);1081 threadId,
1036 return (1082 interactionMode: "plan",
1037 thread?.activities.some(1083 createdAt: now,
1038 (activity) => activity.kind === "provider.turn.start.failed",1084 });
1039 ) === true1085 yield* dispatchTurn("during-compact-recovery", "first queued", "2026-01-01T00:00:02.000Z");
1086 yield* harness.engine.dispatch({
1087 type: "thread.interaction-mode.set",
1088 commandId: CommandId.make("cmd-queued-mode-default"),
1089 threadId,
1090 interactionMode: "default",
1091 createdAt: now,
1092 });
1093 yield* dispatchTurn(
1094 "during-compact-recovery-2",
1095 "second queued",
1096 "2026-01-01T00:00:03.000Z",
1097 );
1098 yield* Effect.promise(() => harness.drain());
1099 expect(harness.sendTurn).toHaveBeenCalledTimes(1);
1100 const beforeRestore = (yield* Effect.promise(() => harness.readModel())).threads.find(
1101 (entry) => entry.id === threadId,
1102 );
1103 expect(
1104 beforeRestore?.activities.filter(
1105 (activity) => activity.kind === "provider.turn.start.failed",
1106 ),
1107 ).toEqual([]);
1108 expect(yield* Effect.promise(() => harness.readPendingTurnStarts())).toEqual([
1109 { threadId: "thread-1" },
1110 ]);
1111
1112 yield* Deferred.succeed(releaseReadyDispatch, undefined);
1113 if (scenario === "stop after send") {
1114 yield* Deferred.await(queuedSendStarted);
1115 yield* harness.engine.dispatch({
1116 type: "thread.session.stop",
1117 commandId: CommandId.make("cmd-stop-after-queued-send"),
1118 threadId,
1119 createdAt: "2026-01-01T00:00:04.000Z",
1120 });
1121 yield* Effect.promise(() => harness.drain());
1122 const stoppedThread = (yield* Effect.promise(() => harness.readModel())).threads.find(
1123 (entry) => entry.id === threadId,
1040 );1124 );
1041 }),1125 expect(stoppedThread?.session?.status).toBe("stopped");
1042 );1126 expect(
1043 expect(harness.sendTurn).toHaveBeenCalledTimes(1);1127 stoppedThread?.activities.filter(
1044 expect(yield* Effect.promise(() => harness.readPendingTurnStarts())).toEqual([1128 (activity) => activity.summary === "Queued message was not sent",
1045 { threadId: "thread-1" },1129 ),
1046 ]);1130 ).toEqual([
10471131 expect.objectContaining({
1048 yield* Deferred.succeed(releaseReadyDispatch, undefined);1132 payload: {
1049 yield* Effect.promise(() =>1133 requestId: "user-message-during-compact-recovery-2",
1050 waitFor(async () => {1134 detail: expect.any(String),
1051 const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId);1135 },
1052 return thread?.session?.status === "ready";1136 }),
1053 }),1137 ]);
1054 );1138 expect(harness.sendTurn).toHaveBeenCalledTimes(2);
1055 }),1139 yield* Deferred.succeed(releaseQueuedSend, undefined);
1140 return;
1141 }
1142 if (stopBeforeResume) {
1143 yield* Deferred.await(resumeStarted);
1144 yield* dispatchTurn("compact-during-resume", "/compact", "2026-01-01T00:00:04.000Z");
1145 yield* Effect.promise(() => harness.drain());
1146 expect(harness.compactThread).toHaveBeenCalledTimes(1);
1147 yield* harness.engine.dispatch({
1148 type: "thread.session.stop",
1149 commandId: CommandId.make("cmd-stop-before-queued-resume"),
1150 threadId,
1151 createdAt: "2026-01-01T00:00:04.000Z",
1152 });
1153 yield* Effect.promise(() => harness.drain());
1154 yield* Deferred.succeed(releaseResume, undefined);
1155 yield* Deferred.await(resumeDispatched);
1156 yield* Effect.promise(() => harness.drain());
1157 expect(harness.sendTurn).toHaveBeenCalledTimes(1);
1158 const stoppedThread = (yield* Effect.promise(() => harness.readModel())).threads.find(
1159 (entry) => entry.id === threadId,
1160 );
1161 expect(stoppedThread?.session?.status).toBe("stopped");
1162 expect(yield* Effect.promise(() => harness.readPendingTurnStarts())).toEqual([]);
1163 expect(
1164 stoppedThread?.activities.filter(
1165 (activity) => activity.summary === "Queued message was not sent",
1166 ),
1167 ).toHaveLength(2);
1168 return;
1169 }
1170 yield* Deferred.await(queuedSent);
1171 expect(harness.sendTurn.mock.calls.slice(1).map(([request]) => request)).toEqual([
1172 expect.objectContaining({ input: "first queued", interactionMode: "plan" }),
1173 expect.objectContaining({ input: "second queued", interactionMode: "default" }),
1174 ]);
1175 const afterRestore = (yield* Effect.promise(() => harness.readModel())).threads.find(
1176 (entry) => entry.id === threadId,
1177 );
1178 expect(
1179 afterRestore?.messages.filter((message) => message.text === "first queued"),
1180 ).toHaveLength(1);
1181 expect(
1182 afterRestore?.messages.filter((message) => message.text === "second queued"),
1183 ).toHaveLength(1);
1184 }),
1056 );1185 );
10571186
1058 effectIt.effect("does not overwrite concurrent session state after compaction failure", () =>1187 effectIt.effect("does not overwrite concurrent session state after compaction failure", () =>
...@@ -1136,6 +1265,21 @@ describe("ProviderCommandReactor", () => {...@@ -1136,6 +1265,21 @@ describe("ProviderCommandReactor", () => {
1136 (entry) => entry.id === threadId,1265 (entry) => entry.id === threadId,
1137 );1266 );
1138 expect(compactingThread?.session?.status).toBe("starting");1267 expect(compactingThread?.session?.status).toBe("starting");
1268 yield* harness.engine.dispatch({
1269 type: "thread.turn.start",
1270 commandId: CommandId.make("cmd-queued-before-stop"),
1271 threadId,
1272 message: {
1273 messageId: asMessageId("user-message-queued-before-stop"),
1274 role: "user",
1275 text: "do not restart after stopping",
1276 attachments: [],
1277 },
1278 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1279 runtimeMode: "approval-required",
1280 createdAt: now,
1281 });
1282 yield* Effect.promise(() => harness.drain());
1139 yield* harness.engine.dispatch({1283 yield* harness.engine.dispatch({
1140 type: "thread.session.stop",1284 type: "thread.session.stop",
1141 commandId: CommandId.make("cmd-stop-during-compact"),1285 commandId: CommandId.make("cmd-stop-during-compact"),
...@@ -1151,7 +1295,7 @@ describe("ProviderCommandReactor", () => {...@@ -1151,7 +1295,7 @@ describe("ProviderCommandReactor", () => {
1151 );1295 );
1152 return (1296 return (
1153 compactingThread?.activities.some(1297 compactingThread?.activities.some(
1154 (activity) => activity.kind === "provider.turn.start.failed",1298 (activity) => activity.summary === "Context compaction failed",
1155 ) === true1299 ) === true
1156 );1300 );
1157 }),1301 }),
...@@ -1167,6 +1311,14 @@ describe("ProviderCommandReactor", () => {...@@ -1167,6 +1311,14 @@ describe("ProviderCommandReactor", () => {
1167 (entry) => entry.id === threadId,1311 (entry) => entry.id === threadId,
1168 );1312 );
1169 expect(recoveredThread?.session?.status).toBe("ready");1313 expect(recoveredThread?.session?.status).toBe("ready");
1314 expect(harness.sendTurn).toHaveBeenCalledTimes(1);
1315 expect(
1316 recoveredThread?.activities.find(
1317 (activity) => activity.summary === "Queued message was not sent",
1318 ),
1319 ).toMatchObject({
1320 payload: { requestId: "user-message-queued-before-stop" },
1321 });
1170 expect(1322 expect(
1171 recoveredThread?.activities.find(1323 recoveredThread?.activities.find(
1172 (activity) => activity.kind === "provider.session.stop.failed",1324 (activity) => activity.kind === "provider.session.stop.failed",
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+143-14
...@@ -18,6 +18,7 @@ import * as Cache from "effect/Cache";...@@ -18,6 +18,7 @@ import * as Cache from "effect/Cache";
18import * as Cause from "effect/Cause";18import * as Cause from "effect/Cause";
19import * as Crypto from "effect/Crypto";19import * as Crypto from "effect/Crypto";
20import * as DateTime from "effect/DateTime";20import * as DateTime from "effect/DateTime";
21import * as Deferred from "effect/Deferred";
21import * as Duration from "effect/Duration";22import * as Duration from "effect/Duration";
22import * as Effect from "effect/Effect";23import * as Effect from "effect/Effect";
23import * as Equal from "effect/Equal";24import * as Equal from "effect/Equal";
...@@ -346,6 +347,19 @@ const make = Effect.gen(function* () {...@@ -346,6 +347,19 @@ const make = Effect.gen(function* () {
346347
347 const threadModelSelections = new Map<string, ModelSelection>();348 const threadModelSelections = new Map<string, ModelSelection>();
348 const compactingThreadIds = new Set<ThreadId>();349 const compactingThreadIds = new Set<ThreadId>();
350 type QueuedTurnStart = Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>;
351 // Turn starts received while a thread compacts, replayed in order once its session is restored.
352 const turnsAfterCompaction = new Map<ThreadId, Array<QueuedTurnStart>>();
353 // Replay command id → the queued turn start it re-requests. `sent` settles once the replay's
354 // provider send finishes, which is what lets the next queued turn follow it in order.
355 const resumedTurnStarts = new Map<
356 CommandId,
357 {
358 readonly event: QueuedTurnStart;
359 readonly queued: Array<QueuedTurnStart>;
360 readonly sent: Deferred.Deferred<void>;
361 }
362 >();
349 const stoppingThreadIds = new Set<ThreadId>();363 const stoppingThreadIds = new Set<ThreadId>();
350364
351 const appendProviderFailureActivity = (input: {365 const appendProviderFailureActivity = (input: {
...@@ -388,6 +402,71 @@ const make = Effect.gen(function* () {...@@ -388,6 +402,71 @@ const make = Effect.gen(function* () {
388 ),402 ),
389 );403 );
390404
405 const cancelTurnsAfterCompaction = Effect.fn("cancelTurnsAfterCompaction")(function* (
406 threadId: ThreadId,
407 detail: string,
408 ) {
409 const queued = turnsAfterCompaction.get(threadId) ?? [];
410 turnsAfterCompaction.delete(threadId);
411 for (const event of queued) {
412 yield* appendProviderFailureActivity({
413 threadId,
414 kind: "provider.turn.start.failed",
415 summary: "Queued message was not sent",
416 detail,
417 turnId: null,
418 createdAt: DateTime.formatIso(yield* DateTime.now),
419 requestId: event.payload.messageId,
420 }).pipe(Effect.ignore({ log: true, message: "failed to report canceled queued message" }));
421 }
422 });
423
424 const resumeTurnsAfterCompaction = Effect.fn("resumeTurnsAfterCompaction")(function* (
425 threadId: ThreadId,
426 ) {
427 const queued = turnsAfterCompaction.get(threadId) ?? [];
428 while (queued.length > 0 && turnsAfterCompaction.get(threadId) === queued) {
429 const event = queued[0]!;
430 const turnStart = yield* projectionSnapshotQuery.getTurnStartMessage({
431 threadId,
432 messageId: event.payload.messageId,
433 });
434 if (turnsAfterCompaction.get(threadId) !== queued) return;
435 // In flight from here on: a cancellation reports it when the replay runs, not from the queue.
436 queued.shift();
437 if (Option.isNone(turnStart)) continue;
438 // Reissue the durable request after restoration clears compaction's
439 // pending slot. Reusing the message id preserves a single user bubble.
440 const commandId = yield* serverCommandId("after-compaction");
441 const sent = yield* Deferred.make<void>();
442 resumedTurnStarts.set(commandId, { event, queued, sent });
443 const { messageId, ...request } = event.payload;
444 yield* orchestrationEngine
445 .dispatch({
446 type: "thread.turn.start",
447 commandId,
448 ...request,
449 message: {
450 messageId,
451 role: "user",
452 text: turnStart.value.message.text,
453 attachments: turnStart.value.message.attachments ?? [],
454 },
455 })
456 .pipe(
457 Effect.onError(() =>
458 Effect.sync(() => {
459 resumedTurnStarts.delete(commandId);
460 queued.unshift(event);
461 }),
462 ),
463 );
464 yield* Deferred.await(sent);
465 resumedTurnStarts.delete(commandId);
466 }
467 if (turnsAfterCompaction.get(threadId) === queued) turnsAfterCompaction.delete(threadId);
468 });
469
391 const formatFailureDetail = (cause: Cause.Cause<unknown>): string => {470 const formatFailureDetail = (cause: Cause.Cause<unknown>): string => {
392 const failReason = cause.reasons.find(Cause.isFailReason);471 const failReason = cause.reasons.find(Cause.isFailReason);
393 if (isProviderAdapterRequestError(failReason?.error)) {472 if (isProviderAdapterRequestError(failReason?.error)) {
...@@ -1181,8 +1260,11 @@ const make = Effect.gen(function* () {...@@ -1181,8 +1260,11 @@ const make = Effect.gen(function* () {
1181 );1260 );
11821261
1183 const processTurnStartRequested = Effect.fn("processTurnStartRequested")(function* (1262 const processTurnStartRequested = Effect.fn("processTurnStartRequested")(function* (
1184 event: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,1263 receivedEvent: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,
1185 ) {1264 ) {
1265 const resumed =
1266 receivedEvent.commandId !== null ? resumedTurnStarts.get(receivedEvent.commandId) : undefined;
1267 const event = resumed ? { ...receivedEvent, payload: resumed.event.payload } : receivedEvent;
1186 const key = turnStartKeyForEvent(event);1268 const key = turnStartKeyForEvent(event);
1187 if (yield* hasHandledTurnStartRecently(key)) {1269 if (yield* hasHandledTurnStartRecently(key)) {
1188 return;1270 return;
...@@ -1219,6 +1301,12 @@ const make = Effect.gen(function* () {...@@ -1219,6 +1301,12 @@ const make = Effect.gen(function* () {
1219 createdAt: event.payload.createdAt,1301 createdAt: event.payload.createdAt,
1220 requestId: event.payload.messageId,1302 requestId: event.payload.messageId,
1221 });1303 });
1304 if (resumed && turnsAfterCompaction.get(event.payload.threadId) !== resumed.queued) {
1305 return yield* appendTurnStartFailure(
1306 "Queued message was not sent",
1307 "The queued message was canceled before it could resume. Send it again to continue.",
1308 );
1309 }
12221310
1223 const handleTurnStartFailure = (cause: Cause.Cause<unknown>) => {1311 const handleTurnStartFailure = (cause: Cause.Cause<unknown>) => {
1224 if (Cause.hasInterruptsOnly(cause)) {1312 if (Cause.hasInterruptsOnly(cause)) {
...@@ -1381,6 +1469,7 @@ const make = Effect.gen(function* () {...@@ -1381,6 +1469,7 @@ const make = Effect.gen(function* () {
1381 const latestThread = yield* resolveThreadShell(event.payload.threadId);1469 const latestThread = yield* resolveThreadShell(event.payload.threadId);
1382 if (1470 if (
1383 compactingThreadIds.has(event.payload.threadId) ||1471 compactingThreadIds.has(event.payload.threadId) ||
1472 turnsAfterCompaction.has(event.payload.threadId) ||
1384 latestThread?.session?.status === "starting" ||1473 latestThread?.session?.status === "starting" ||
1385 latestThread?.session?.status === "running"1474 latestThread?.session?.status === "running"
1386 ) {1475 ) {
...@@ -1391,6 +1480,9 @@ const make = Effect.gen(function* () {...@@ -1391,6 +1480,9 @@ const make = Effect.gen(function* () {
1391 return;1480 return;
1392 }1481 }
1393 compactingThreadIds.add(event.payload.threadId);1482 compactingThreadIds.add(event.payload.threadId);
1483 const clearCompacting = Effect.sync(
1484 () => void compactingThreadIds.delete(event.payload.threadId),
1485 );
1394 yield* Effect.gen(function* () {1486 yield* Effect.gen(function* () {
1395 yield* ensureSessionForThread(1487 yield* ensureSessionForThread(
1396 event.payload.threadId,1488 event.payload.threadId,
...@@ -1410,17 +1502,32 @@ const make = Effect.gen(function* () {...@@ -1410,17 +1502,32 @@ const make = Effect.gen(function* () {
1410 );1502 );
1411 }).pipe(1503 }).pipe(
1412 Effect.andThen(restoreCompaction(event.payload.threadId, true)),1504 Effect.andThen(restoreCompaction(event.payload.threadId, true)),
1413 Effect.catchCause(recoverCompactionFailure),1505 Effect.andThen(clearCompacting),
1414 Effect.ensuring(Effect.sync(() => void compactingThreadIds.delete(event.payload.threadId))),1506 Effect.andThen(resumeTurnsAfterCompaction(event.payload.threadId)),
1507 Effect.catchCause((cause) =>
1508 recoverCompactionFailure(cause).pipe(
1509 Effect.ensuring(clearCompacting),
1510 Effect.andThen(
1511 cancelTurnsAfterCompaction(
1512 event.payload.threadId,
1513 "Context compaction failed. Send this message again to continue.",
1514 ),
1515 ),
1516 ),
1517 ),
1415 Effect.forkScoped,1518 Effect.forkScoped,
1416 );1519 );
1417 return;1520 return;
1418 }1521 }
1419 if (compactingThreadIds.has(event.payload.threadId)) {1522 if (
1420 return yield* appendTurnStartFailure(1523 !resumed &&
1421 "Provider turn start failed",1524 (compactingThreadIds.has(event.payload.threadId) ||
1422 "Wait for context compaction to finish before sending another message.",1525 turnsAfterCompaction.has(event.payload.threadId))
1423 );1526 ) {
1527 const queued = turnsAfterCompaction.get(event.payload.threadId) ?? [];
1528 queued.push(event);
1529 turnsAfterCompaction.set(event.payload.threadId, queued);
1530 return;
1424 }1531 }
1425 const sendTurnRequest = yield* buildSendTurnRequestForThread({1532 const sendTurnRequest = yield* buildSendTurnRequestForThread({
1426 threadId: event.payload.threadId,1533 threadId: event.payload.threadId,
...@@ -1440,14 +1547,24 @@ const make = Effect.gen(function* () {...@@ -1440,14 +1547,24 @@ const make = Effect.gen(function* () {
1440 return;1547 return;
1441 }1548 }
14421549
1443 yield* providerService1550 const send = providerService
1444 .sendTurn(sendTurnRequest.value)1551 .sendTurn(sendTurnRequest.value)
1445 .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped);1552 .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure));
1553 // The forked send settles `sent` from here on, so drop the entry the post-processing hook uses.
1554 if (resumed && event.commandId !== null) resumedTurnStarts.delete(event.commandId);
1555 yield* send.pipe(
1556 Effect.ensuring(resumed ? Deferred.succeed(resumed.sent, undefined) : Effect.void),
1557 Effect.forkScoped,
1558 );
1446 });1559 });
14471560
1448 const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* (1561 const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* (
1449 event: Extract<ProviderIntentEvent, { type: "thread.turn-interrupt-requested" }>,1562 event: Extract<ProviderIntentEvent, { type: "thread.turn-interrupt-requested" }>,
1450 ) {1563 ) {
1564 yield* cancelTurnsAfterCompaction(
1565 event.payload.threadId,
1566 "Context compaction was interrupted. Send this message again to continue.",
1567 );
1451 const thread = yield* resolveThreadShell(event.payload.threadId);1568 const thread = yield* resolveThreadShell(event.payload.threadId);
1452 if (!thread) {1569 if (!thread) {
1453 return;1570 return;
...@@ -1643,11 +1760,15 @@ const make = Effect.gen(function* () {...@@ -1643,11 +1760,15 @@ const make = Effect.gen(function* () {
1643 const wasCompacting = compactingThreadIds.has(thread.id);1760 const wasCompacting = compactingThreadIds.has(thread.id);
1644 stoppingThreadIds.add(thread.id);1761 stoppingThreadIds.add(thread.id);
1645 const clearStopping = Effect.sync(() => void stoppingThreadIds.delete(thread.id));1762 const clearStopping = Effect.sync(() => void stoppingThreadIds.delete(thread.id));
1646 yield* (1763 yield* cancelTurnsAfterCompaction(
1647 thread.session && thread.session.status !== "stopped"1764 thread.id,
1648 ? providerService.stopSession({ threadId: thread.id })1765 "The session was stopped during context compaction. Send this message again to continue.",
1649 : Effect.void
1650 ).pipe(1766 ).pipe(
1767 Effect.andThen(
1768 thread.session && thread.session.status !== "stopped"
1769 ? providerService.stopSession({ threadId: thread.id })
1770 : Effect.void,
1771 ),
1651 Effect.matchCauseEffect({1772 Effect.matchCauseEffect({
1652 onFailure: (cause) => {1773 onFailure: (cause) => {
1653 if (Cause.hasInterruptsOnly(cause)) {1774 if (Cause.hasInterruptsOnly(cause)) {
...@@ -1761,6 +1882,14 @@ const make = Effect.gen(function* () {...@@ -1761,6 +1882,14 @@ const make = Effect.gen(function* () {
17611882
1762 const processDomainEventSafely = (event: ProviderIntentEvent) =>1883 const processDomainEventSafely = (event: ProviderIntentEvent) =>
1763 processDomainEvent(event).pipe(1884 processDomainEvent(event).pipe(
1885 // A replay that returned before forking its send still holds its entry; settle it so
1886 // the compaction queue moves on. Forked sends drop the entry first and settle it themselves.
1887 Effect.ensuring(
1888 Effect.suspend(() => {
1889 const resumed = event.commandId !== null && resumedTurnStarts.get(event.commandId);
1890 return resumed ? Deferred.succeed(resumed.sent, undefined) : Effect.void;
1891 }),
1892 ),
1764 Effect.catchCause((cause) => {1893 Effect.catchCause((cause) => {
1765 if (Cause.hasInterruptsOnly(cause)) {1894 if (Cause.hasInterruptsOnly(cause)) {
1766 return Effect.interrupt;1895 return Effect.interrupt;