authorgravatar for hauke@schnau-lilienthal.deHauke Schnau <hauke@schnau-lilienthal.de> 2026-04-17 20:27:12+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-17 11:27:12-07:00
log721b6b4c6f596bd1d1b3af585ed9c62c4b6f72ab
tree8e1ca6264b4dd9d492d019b52a500a06bc953e82
parent40009735eea88b2266cf7d112217e85a688b3e63
signature Signed by PGP key B5690EEEBB952194

Preserve provider bindings when stopping sessions (#2125)

Co-authored-by: Julius Marminge <julius0216@outlook.com> Co-authored-by: codex <codex@users.noreply.github.com>

8 files changed, 198 insertions(+), 160 deletions(-)

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+108-108
...@@ -750,6 +750,57 @@ describe("ProviderCommandReactor", () => {...@@ -750,6 +750,57 @@ describe("ProviderCommandReactor", () => {
750 });750 });
751 });751 });
752752
753 it("preserves the active session model when in-session model switching is unsupported", async () => {
754 const harness = await createHarness({ sessionModelSwitch: "unsupported" });
755 const now = new Date().toISOString();
756
757 await Effect.runPromise(
758 harness.engine.dispatch({
759 type: "thread.turn.start",
760 commandId: CommandId.make("cmd-turn-start-unsupported-1"),
761 threadId: ThreadId.make("thread-1"),
762 message: {
763 messageId: asMessageId("user-message-unsupported-1"),
764 role: "user",
765 text: "first",
766 attachments: [],
767 },
768 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
769 runtimeMode: "approval-required",
770 createdAt: now,
771 }),
772 );
773
774 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
775
776 await Effect.runPromise(
777 harness.engine.dispatch({
778 type: "thread.turn.start",
779 commandId: CommandId.make("cmd-turn-start-unsupported-2"),
780 threadId: ThreadId.make("thread-1"),
781 message: {
782 messageId: asMessageId("user-message-unsupported-2"),
783 role: "user",
784 text: "second",
785 attachments: [],
786 },
787 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
788 runtimeMode: "approval-required",
789 createdAt: now,
790 }),
791 );
792
793 await waitFor(() => harness.sendTurn.mock.calls.length === 2);
794
795 expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({
796 threadId: ThreadId.make("thread-1"),
797 modelSelection: {
798 provider: "codex",
799 model: "gpt-5-codex",
800 },
801 });
802 });
803
753 it("rejects a first turn when requested provider conflicts with the thread model", async () => {804 it("rejects a first turn when requested provider conflicts with the thread model", async () => {
754 const harness = await createHarness({805 const harness = await createHarness({
755 threadModelSelection: { provider: "codex", model: "gpt-5-codex" },806 threadModelSelection: { provider: "codex", model: "gpt-5-codex" },
...@@ -802,57 +853,6 @@ describe("ProviderCommandReactor", () => {...@@ -802,57 +853,6 @@ describe("ProviderCommandReactor", () => {
802 });853 });
803 });854 });
804855
805 it("preserves the active session model when in-session model switching is unsupported", async () => {
806 const harness = await createHarness({ sessionModelSwitch: "unsupported" });
807 const now = new Date().toISOString();
808
809 await Effect.runPromise(
810 harness.engine.dispatch({
811 type: "thread.turn.start",
812 commandId: CommandId.make("cmd-turn-start-unsupported-1"),
813 threadId: ThreadId.make("thread-1"),
814 message: {
815 messageId: asMessageId("user-message-unsupported-1"),
816 role: "user",
817 text: "first",
818 attachments: [],
819 },
820 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
821 runtimeMode: "approval-required",
822 createdAt: now,
823 }),
824 );
825
826 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
827
828 await Effect.runPromise(
829 harness.engine.dispatch({
830 type: "thread.turn.start",
831 commandId: CommandId.make("cmd-turn-start-unsupported-2"),
832 threadId: ThreadId.make("thread-1"),
833 message: {
834 messageId: asMessageId("user-message-unsupported-2"),
835 role: "user",
836 text: "second",
837 attachments: [],
838 },
839 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
840 runtimeMode: "approval-required",
841 createdAt: now,
842 }),
843 );
844
845 await waitFor(() => harness.sendTurn.mock.calls.length === 2);
846
847 expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({
848 threadId: ThreadId.make("thread-1"),
849 modelSelection: {
850 provider: "codex",
851 model: "gpt-5-codex",
852 },
853 });
854 });
855
856 it("reuses the same provider session when runtime mode is unchanged", async () => {856 it("reuses the same provider session when runtime mode is unchanged", async () => {
857 const harness = await createHarness();857 const harness = await createHarness();
858 const now = new Date().toISOString();858 const now = new Date().toISOString();
...@@ -1100,23 +1100,33 @@ describe("ProviderCommandReactor", () => {...@@ -1100,23 +1100,33 @@ describe("ProviderCommandReactor", () => {
1100 });1100 });
1101 });1101 });
11021102
1103 it("rejects provider changes after a thread is already bound to a session provider", async () => {1103 it("does not stop the active session when restart fails before rebind", async () => {
1104 const harness = await createHarness();1104 const harness = await createHarness();
1105 const now = new Date().toISOString();1105 const now = new Date().toISOString();
11061106
1107 await Effect.runPromise(
1108 harness.engine.dispatch({
1109 type: "thread.runtime-mode.set",
1110 commandId: CommandId.make("cmd-runtime-mode-set-initial-full-access-2"),
1111 threadId: ThreadId.make("thread-1"),
1112 runtimeMode: "full-access",
1113 createdAt: now,
1114 }),
1115 );
1116
1107 await Effect.runPromise(1117 await Effect.runPromise(
1108 harness.engine.dispatch({1118 harness.engine.dispatch({
1109 type: "thread.turn.start",1119 type: "thread.turn.start",
1110 commandId: CommandId.make("cmd-turn-start-provider-switch-1"),1120 commandId: CommandId.make("cmd-turn-start-restart-failure-1"),
1111 threadId: ThreadId.make("thread-1"),1121 threadId: ThreadId.make("thread-1"),
1112 message: {1122 message: {
1113 messageId: asMessageId("user-message-provider-switch-1"),1123 messageId: asMessageId("user-message-restart-failure-1"),
1114 role: "user",1124 role: "user",
1115 text: "first",1125 text: "first",
1116 attachments: [],1126 attachments: [],
1117 },1127 },
1118 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,1128 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1119 runtimeMode: "approval-required",1129 runtimeMode: "full-access",
1120 createdAt: now,1130 createdAt: now,
1121 }),1131 }),
1122 );1132 );
...@@ -1124,22 +1134,15 @@ describe("ProviderCommandReactor", () => {...@@ -1124,22 +1134,15 @@ describe("ProviderCommandReactor", () => {
1124 await waitFor(() => harness.startSession.mock.calls.length === 1);1134 await waitFor(() => harness.startSession.mock.calls.length === 1);
1125 await waitFor(() => harness.sendTurn.mock.calls.length === 1);1135 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
11261136
1137 harness.startSession.mockImplementationOnce(
1138 (_: unknown, __: unknown) => Effect.fail(new Error("simulated restart failure")) as never,
1139 );
1140
1127 await Effect.runPromise(1141 await Effect.runPromise(
1128 harness.engine.dispatch({1142 harness.engine.dispatch({
1129 type: "thread.turn.start",1143 type: "thread.runtime-mode.set",
1130 commandId: CommandId.make("cmd-turn-start-provider-switch-2"),1144 commandId: CommandId.make("cmd-runtime-mode-set-restart-failure"),
1131 threadId: ThreadId.make("thread-1"),1145 threadId: ThreadId.make("thread-1"),
1132 message: {
1133 messageId: asMessageId("user-message-provider-switch-2"),
1134 role: "user",
1135 text: "second",
1136 attachments: [],
1137 },
1138 modelSelection: {
1139 provider: "claudeAgent",
1140 model: "claude-opus-4-6",
1141 },
1142 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1143 runtimeMode: "approval-required",1146 runtimeMode: "approval-required",
1144 createdAt: now,1147 createdAt: now,
1145 }),1148 }),
...@@ -1148,57 +1151,37 @@ describe("ProviderCommandReactor", () => {...@@ -1148,57 +1151,37 @@ describe("ProviderCommandReactor", () => {
1148 await waitFor(async () => {1151 await waitFor(async () => {
1149 const readModel = await Effect.runPromise(harness.engine.getReadModel());1152 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1150 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));1153 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
1151 return (1154 return thread?.runtimeMode === "approval-required";
1152 thread?.activities.some((activity) => activity.kind === "provider.turn.start.failed") ??
1153 false
1154 );
1155 });1155 });
1156 await waitFor(() => harness.startSession.mock.calls.length === 2);
1157 await harness.drain();
11561158
1157 expect(harness.startSession.mock.calls.length).toBe(1);
1158 expect(harness.sendTurn.mock.calls.length).toBe(1);
1159 expect(harness.stopSession.mock.calls.length).toBe(0);1159 expect(harness.stopSession.mock.calls.length).toBe(0);
1160 expect(harness.sendTurn.mock.calls.length).toBe(1);
11601161
1161 const readModel = await Effect.runPromise(harness.engine.getReadModel());1162 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1162 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));1163 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
1163 expect(thread?.session?.threadId).toBe("thread-1");1164 expect(thread?.session?.threadId).toBe("thread-1");
1164 expect(thread?.session?.providerName).toBe("codex");1165 expect(thread?.session?.runtimeMode).toBe("full-access");
1165 expect(thread?.session?.runtimeMode).toBe("approval-required");
1166 expect(
1167 thread?.activities.find((activity) => activity.kind === "provider.turn.start.failed"),
1168 ).toMatchObject({
1169 payload: {
1170 detail: expect.stringContaining("cannot switch to 'claudeAgent'"),
1171 },
1172 });
1173 });1166 });
11741167
1175 it("does not stop the active session when restart fails before rebind", async () => {1168 it("rejects provider changes after a thread is already bound to a session provider", async () => {
1176 const harness = await createHarness();1169 const harness = await createHarness();
1177 const now = new Date().toISOString();1170 const now = new Date().toISOString();
11781171
1179 await Effect.runPromise(
1180 harness.engine.dispatch({
1181 type: "thread.runtime-mode.set",
1182 commandId: CommandId.make("cmd-runtime-mode-set-initial-full-access-2"),
1183 threadId: ThreadId.make("thread-1"),
1184 runtimeMode: "full-access",
1185 createdAt: now,
1186 }),
1187 );
1188
1189 await Effect.runPromise(1172 await Effect.runPromise(
1190 harness.engine.dispatch({1173 harness.engine.dispatch({
1191 type: "thread.turn.start",1174 type: "thread.turn.start",
1192 commandId: CommandId.make("cmd-turn-start-restart-failure-1"),1175 commandId: CommandId.make("cmd-turn-start-provider-switch-1"),
1193 threadId: ThreadId.make("thread-1"),1176 threadId: ThreadId.make("thread-1"),
1194 message: {1177 message: {
1195 messageId: asMessageId("user-message-restart-failure-1"),1178 messageId: asMessageId("user-message-provider-switch-1"),
1196 role: "user",1179 role: "user",
1197 text: "first",1180 text: "first",
1198 attachments: [],1181 attachments: [],
1199 },1182 },
1200 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,1183 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1201 runtimeMode: "full-access",1184 runtimeMode: "approval-required",
1202 createdAt: now,1185 createdAt: now,
1203 }),1186 }),
1204 );1187 );
...@@ -1206,15 +1189,22 @@ describe("ProviderCommandReactor", () => {...@@ -1206,15 +1189,22 @@ describe("ProviderCommandReactor", () => {
1206 await waitFor(() => harness.startSession.mock.calls.length === 1);1189 await waitFor(() => harness.startSession.mock.calls.length === 1);
1207 await waitFor(() => harness.sendTurn.mock.calls.length === 1);1190 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
12081191
1209 harness.startSession.mockImplementationOnce(
1210 (_: unknown, __: unknown) => Effect.fail(new Error("simulated restart failure")) as never,
1211 );
1212
1213 await Effect.runPromise(1192 await Effect.runPromise(
1214 harness.engine.dispatch({1193 harness.engine.dispatch({
1215 type: "thread.runtime-mode.set",1194 type: "thread.turn.start",
1216 commandId: CommandId.make("cmd-runtime-mode-set-restart-failure"),1195 commandId: CommandId.make("cmd-turn-start-provider-switch-2"),
1217 threadId: ThreadId.make("thread-1"),1196 threadId: ThreadId.make("thread-1"),
1197 message: {
1198 messageId: asMessageId("user-message-provider-switch-2"),
1199 role: "user",
1200 text: "second",
1201 attachments: [],
1202 },
1203 modelSelection: {
1204 provider: "claudeAgent",
1205 model: "claude-opus-4-6",
1206 },
1207 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1218 runtimeMode: "approval-required",1208 runtimeMode: "approval-required",
1219 createdAt: now,1209 createdAt: now,
1220 }),1210 }),
...@@ -1223,18 +1213,28 @@ describe("ProviderCommandReactor", () => {...@@ -1223,18 +1213,28 @@ describe("ProviderCommandReactor", () => {
1223 await waitFor(async () => {1213 await waitFor(async () => {
1224 const readModel = await Effect.runPromise(harness.engine.getReadModel());1214 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1225 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));1215 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
1226 return thread?.runtimeMode === "approval-required";1216 return (
1217 thread?.activities.some((activity) => activity.kind === "provider.turn.start.failed") ??
1218 false
1219 );
1227 });1220 });
1228 await waitFor(() => harness.startSession.mock.calls.length === 2);
1229 await harness.drain();
12301221
1231 expect(harness.stopSession.mock.calls.length).toBe(0);1222 expect(harness.startSession.mock.calls.length).toBe(1);
1232 expect(harness.sendTurn.mock.calls.length).toBe(1);1223 expect(harness.sendTurn.mock.calls.length).toBe(1);
1224 expect(harness.stopSession.mock.calls.length).toBe(0);
12331225
1234 const readModel = await Effect.runPromise(harness.engine.getReadModel());1226 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1235 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));1227 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
1236 expect(thread?.session?.threadId).toBe("thread-1");1228 expect(thread?.session?.threadId).toBe("thread-1");
1237 expect(thread?.session?.runtimeMode).toBe("full-access");1229 expect(thread?.session?.providerName).toBe("codex");
1230 expect(thread?.session?.runtimeMode).toBe("approval-required");
1231 expect(
1232 thread?.activities.find((activity) => activity.kind === "provider.turn.start.failed"),
1233 ).toMatchObject({
1234 payload: {
1235 detail: expect.stringContaining("cannot switch to 'claudeAgent'"),
1236 },
1237 });
1238 });1238 });
12391239
1240 it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => {1240 it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => {
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+4-10
...@@ -247,7 +247,7 @@ const make = Effect.gen(function* () {...@@ -247,7 +247,7 @@ const make = Effect.gen(function* () {
247 detail: `Thread '${threadId}' is bound to provider '${threadProvider}' and cannot switch to '${requestedModelSelection.provider}'.`,247 detail: `Thread '${threadId}' is bound to provider '${threadProvider}' and cannot switch to '${requestedModelSelection.provider}'.`,
248 });248 });
249 }249 }
250 const preferredProvider: ProviderKind = currentProvider ?? threadProvider;250 const preferredProvider: ProviderKind = threadProvider;
251 const desiredModelSelection = requestedModelSelection ?? thread.modelSelection;251 const desiredModelSelection = requestedModelSelection ?? thread.modelSelection;
252 const effectiveCwd = resolveThreadWorkspaceCwd({252 const effectiveCwd = resolveThreadWorkspaceCwd({
253 thread,253 thread,
...@@ -293,9 +293,6 @@ const make = Effect.gen(function* () {...@@ -293,9 +293,6 @@ const make = Effect.gen(function* () {
293 thread.session && thread.session.status !== "stopped" && activeSession ? thread.id : null;293 thread.session && thread.session.status !== "stopped" && activeSession ? thread.id : null;
294 if (existingSessionThreadId) {294 if (existingSessionThreadId) {
295 const runtimeModeChanged = thread.runtimeMode !== thread.session?.runtimeMode;295 const runtimeModeChanged = thread.runtimeMode !== thread.session?.runtimeMode;
296 const providerChanged =
297 requestedModelSelection !== undefined &&
298 requestedModelSelection.provider !== currentProvider;
299 const sessionModelSwitch =296 const sessionModelSwitch =
300 currentProvider === undefined297 currentProvider === undefined
301 ? "in-session"298 ? "in-session"
...@@ -312,17 +309,15 @@ const make = Effect.gen(function* () {...@@ -312,17 +309,15 @@ const make = Effect.gen(function* () {
312309
313 if (310 if (
314 !runtimeModeChanged &&311 !runtimeModeChanged &&
315 !providerChanged &&
316 !shouldRestartForModelChange &&312 !shouldRestartForModelChange &&
317 !shouldRestartForModelSelectionChange313 !shouldRestartForModelSelectionChange
318 ) {314 ) {
319 return existingSessionThreadId;315 return existingSessionThreadId;
320 }316 }
321317
322 const resumeCursor =318 const resumeCursor = shouldRestartForModelChange
323 providerChanged || shouldRestartForModelChange319 ? undefined
324 ? undefined320 : (activeSession?.resumeCursor ?? undefined);
325 : (activeSession?.resumeCursor ?? undefined);
326 yield* Effect.logInfo("provider command reactor restarting provider session", {321 yield* Effect.logInfo("provider command reactor restarting provider session", {
327 threadId,322 threadId,
328 existingSessionThreadId,323 existingSessionThreadId,
...@@ -331,7 +326,6 @@ const make = Effect.gen(function* () {...@@ -331,7 +326,6 @@ const make = Effect.gen(function* () {
331 currentRuntimeMode: thread.session?.runtimeMode,326 currentRuntimeMode: thread.session?.runtimeMode,
332 desiredRuntimeMode: thread.runtimeMode,327 desiredRuntimeMode: thread.runtimeMode,
333 runtimeModeChanged,328 runtimeModeChanged,
334 providerChanged,
335 modelChanged,329 modelChanged,
336 shouldRestartForModelChange,330 shouldRestartForModelChange,
337 shouldRestartForModelSelectionChange,331 shouldRestartForModelSelectionChange,
apps/server/src/provider/Layers/CodexAdapter.test.ts-1
...@@ -144,7 +144,6 @@ const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory...@@ -144,7 +144,6 @@ const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory
144 getProvider: () =>144 getProvider: () =>
145 Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")),145 Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")),
146 getBinding: () => Effect.succeed(Option.none()),146 getBinding: () => Effect.succeed(Option.none()),
147 remove: () => Effect.void,
148 listThreadIds: () => Effect.succeed([]),147 listThreadIds: () => Effect.succeed([]),
149 listBindings: () => Effect.succeed([]),148 listBindings: () => Effect.succeed([]),
150});149});
apps/server/src/provider/Layers/ProviderService.test.ts+76-15
...@@ -18,7 +18,6 @@ import {...@@ -18,7 +18,6 @@ import {
18 TurnId,18 TurnId,
19} from "@t3tools/contracts";19} from "@t3tools/contracts";
20import { it, assert, vi } from "@effect/vitest";20import { it, assert, vi } from "@effect/vitest";
21import { assertFailure } from "@effect/vitest/utils";
2221
23import { Effect, Fiber, Layer, Metric, Option, PubSub, Ref, Stream } from "effect";22import { Effect, Fiber, Layer, Metric, Option, PubSub, Ref, Stream } from "effect";
24import * as SqlClient from "effect/unstable/sql/SqlClient";23import * as SqlClient from "effect/unstable/sql/SqlClient";
...@@ -574,20 +573,31 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -574,20 +573,31 @@ routing.layer("ProviderServiceLive routing", (it) => {
574 });573 });
575574
576 yield* provider.stopSession({ threadId: session.threadId });575 yield* provider.stopSession({ threadId: session.threadId });
577 const sendAfterStop = yield* Effect.result(576 routing.codex.startSession.mockClear();
578 provider.sendTurn({577 routing.codex.sendTurn.mockClear();
579 threadId: session.threadId,578
580 input: "after-stop",579 yield* provider.sendTurn({
581 attachments: [],580 threadId: session.threadId,
582 }),581 input: "after-stop",
583 );582 attachments: [],
584 assertFailure(583 });
585 sendAfterStop,584
586 new ProviderValidationError({585 assert.equal(routing.codex.startSession.mock.calls.length, 1);
587 operation: "ProviderService.sendTurn",586 const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0];
588 issue: `Cannot route thread '${session.threadId}' because no persisted provider binding exists.`,587 assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true);
589 }),588 if (resumedStartInput && typeof resumedStartInput === "object") {
590 );589 const startPayload = resumedStartInput as {
590 provider?: string;
591 cwd?: string;
592 resumeCursor?: unknown;
593 threadId?: string;
594 };
595 assert.equal(startPayload.provider, "codex");
596 assert.equal(startPayload.cwd, "/tmp/project");
597 assert.deepEqual(startPayload.resumeCursor, session.resumeCursor);
598 assert.equal(startPayload.threadId, session.threadId);
599 }
600 assert.equal(routing.codex.sendTurn.mock.calls.length, 1);
591 }),601 }),
592 );602 );
593603
...@@ -631,6 +641,57 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -631,6 +641,57 @@ routing.layer("ProviderServiceLive routing", (it) => {
631 }),641 }),
632 );642 );
633643
644 it.effect("preserves the persisted binding when stopping a session", () =>
645 Effect.gen(function* () {
646 const provider = yield* ProviderService;
647 const runtimeRepository = yield* ProviderSessionRuntimeRepository;
648
649 const initial = yield* provider.startSession(asThreadId("thread-reap-preserve"), {
650 provider: "codex",
651 threadId: asThreadId("thread-reap-preserve"),
652 cwd: "/tmp/project-reap-preserve",
653 runtimeMode: "full-access",
654 });
655
656 yield* provider.stopSession({ threadId: initial.threadId });
657
658 const persistedAfterStop = yield* runtimeRepository.getByThreadId({
659 threadId: initial.threadId,
660 });
661 assert.equal(Option.isSome(persistedAfterStop), true);
662 if (Option.isSome(persistedAfterStop)) {
663 assert.equal(persistedAfterStop.value.status, "stopped");
664 assert.deepEqual(persistedAfterStop.value.resumeCursor, initial.resumeCursor);
665 }
666
667 routing.codex.startSession.mockClear();
668 routing.codex.sendTurn.mockClear();
669
670 yield* provider.sendTurn({
671 threadId: initial.threadId,
672 input: "resume after reap",
673 attachments: [],
674 });
675
676 assert.equal(routing.codex.startSession.mock.calls.length, 1);
677 const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0];
678 assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true);
679 if (resumedStartInput && typeof resumedStartInput === "object") {
680 const startPayload = resumedStartInput as {
681 provider?: string;
682 cwd?: string;
683 resumeCursor?: unknown;
684 threadId?: string;
685 };
686 assert.equal(startPayload.provider, "codex");
687 assert.equal(startPayload.cwd, "/tmp/project-reap-preserve");
688 assert.deepEqual(startPayload.resumeCursor, initial.resumeCursor);
689 assert.equal(startPayload.threadId, initial.threadId);
690 }
691 assert.equal(routing.codex.sendTurn.mock.calls.length, 1);
692 }),
693 );
694
634 it.effect("routes explicit claudeAgent provider session starts to the claude adapter", () =>695 it.effect("routes explicit claudeAgent provider session starts to the claude adapter", () =>
635 Effect.gen(function* () {696 Effect.gen(function* () {
636 const provider = yield* ProviderService;697 const provider = yield* ProviderService;
apps/server/src/provider/Layers/ProviderService.ts+8-1
...@@ -620,7 +620,14 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -620,7 +620,14 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
620 if (routed.isActive) {620 if (routed.isActive) {
621 yield* routed.adapter.stopSession(routed.threadId);621 yield* routed.adapter.stopSession(routed.threadId);
622 }622 }
623 yield* directory.remove(input.threadId);623 yield* directory.upsert({
624 threadId: input.threadId,
625 provider: routed.adapter.provider,
626 status: "stopped",
627 runtimePayload: {
628 activeTurnId: null,
629 },
630 });
624 yield* analytics.record("provider.session.stopped", {631 yield* analytics.record("provider.session.stopped", {
625 provider: routed.adapter.provider,632 provider: routed.adapter.provider,
626 });633 });
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+2-13
...@@ -5,7 +5,7 @@ import path from "node:path";...@@ -5,7 +5,7 @@ import path from "node:path";
5import * as NodeServices from "@effect/platform-node/NodeServices";5import * as NodeServices from "@effect/platform-node/NodeServices";
6import { ThreadId } from "@t3tools/contracts";6import { ThreadId } from "@t3tools/contracts";
7import { it, assert } from "@effect/vitest";7import { it, assert } from "@effect/vitest";
8import { assertFailure, assertSome } from "@effect/vitest/utils";8import { assertSome } from "@effect/vitest/utils";
9import { Effect, Layer, Option } from "effect";9import { Effect, Layer, Option } from "effect";
10import * as SqlClient from "effect/unstable/sql/SqlClient";10import * as SqlClient from "effect/unstable/sql/SqlClient";
1111
...@@ -15,7 +15,6 @@ import {...@@ -15,7 +15,6 @@ import {
15} from "../../persistence/Layers/Sqlite.ts";15} from "../../persistence/Layers/Sqlite.ts";
16import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts";16import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts";
17import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts";17import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts";
18import { ProviderSessionDirectoryPersistenceError } from "../Errors.ts";
19import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";18import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
20import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";19import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
2120
...@@ -31,7 +30,7 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli...@@ -31,7 +30,7 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli
31}30}
3231
33it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryLive", (it) => {32it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryLive", (it) => {
34 it("upserts, reads, and removes thread bindings", () =>33 it("upserts and reads thread bindings", () =>
35 Effect.gen(function* () {34 Effect.gen(function* () {
36 const directory = yield* ProviderSessionDirectory;35 const directory = yield* ProviderSessionDirectory;
37 const runtimeRepository = yield* ProviderSessionRuntimeRepository;36 const runtimeRepository = yield* ProviderSessionRuntimeRepository;
...@@ -76,16 +75,6 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL...@@ -76,16 +75,6 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL
7675
77 const threadIds = yield* directory.listThreadIds();76 const threadIds = yield* directory.listThreadIds();
78 assert.deepEqual(threadIds, [nextThreadId]);77 assert.deepEqual(threadIds, [nextThreadId]);
79
80 yield* directory.remove(nextThreadId);
81 const missingProvider = yield* directory.getProvider(nextThreadId).pipe(Effect.result);
82 assertFailure(
83 missingProvider,
84 new ProviderSessionDirectoryPersistenceError({
85 operation: "ProviderSessionDirectory.getProvider",
86 detail: `No persisted provider binding found for thread '${nextThreadId}'.`,
87 }),
88 );
89 }));78 }));
9079
91 it("persists runtime fields and merges payload updates", () =>80 it("persists runtime fields and merges payload updates", () =>
apps/server/src/provider/Layers/ProviderSessionDirectory.ts-8
...@@ -145,13 +145,6 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -145,13 +145,6 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
145 ),145 ),
146 );146 );
147147
148 const remove: ProviderSessionDirectoryShape["remove"] = (threadId) =>
149 repository
150 .deleteByThreadId({ threadId })
151 .pipe(
152 Effect.mapError(toPersistenceError("ProviderSessionDirectory.remove:deleteByThreadId")),
153 );
154
155 const listThreadIds: ProviderSessionDirectoryShape["listThreadIds"] = () =>148 const listThreadIds: ProviderSessionDirectoryShape["listThreadIds"] = () =>
156 repository.list().pipe(149 repository.list().pipe(
157 Effect.mapError(toPersistenceError("ProviderSessionDirectory.listThreadIds:list")),150 Effect.mapError(toPersistenceError("ProviderSessionDirectory.listThreadIds:list")),
...@@ -174,7 +167,6 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -174,7 +167,6 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
174 upsert,167 upsert,
175 getProvider,168 getProvider,
176 getBinding,169 getBinding,
177 remove,
178 listThreadIds,170 listThreadIds,
179 listBindings,171 listBindings,
180 } satisfies ProviderSessionDirectoryShape;172 } satisfies ProviderSessionDirectoryShape;
apps/server/src/provider/Services/ProviderSessionDirectory.ts-4
...@@ -45,10 +45,6 @@ export interface ProviderSessionDirectoryShape {...@@ -45,10 +45,6 @@ export interface ProviderSessionDirectoryShape {
45 threadId: ThreadId,45 threadId: ThreadId,
46 ) => Effect.Effect<Option.Option<ProviderRuntimeBinding>, ProviderSessionDirectoryReadError>;46 ) => Effect.Effect<Option.Option<ProviderRuntimeBinding>, ProviderSessionDirectoryReadError>;
4747
48 readonly remove: (
49 threadId: ThreadId,
50 ) => Effect.Effect<void, ProviderSessionDirectoryPersistenceError>;
51
52 readonly listThreadIds: () => Effect.Effect<48 readonly listThreadIds: () => Effect.Effect<
53 ReadonlyArray<ThreadId>,49 ReadonlyArray<ThreadId>,
54 ProviderSessionDirectoryPersistenceError50 ProviderSessionDirectoryPersistenceError