History

Support provider-aware session selection in reactor

Add restart semantics when requested provider changes and cover with tests.

Julius Marminge committed 8 months ago

Commit 1fb35741 parent b8e3b462 files · +131 −8Browse files

Unified · Split

2 files
  1. apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +108 −1
  2. apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +23 −7
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+108−1
@@ -75,13 +75,20 @@ describe("ProviderCommandReactor", () => { · 74 unmodified lines
75 const runtimeSessions: Array<ProviderSession> = [];75 const runtimeSessions: Array<ProviderSession> = [];
76 const startSession = vi.fn((_: unknown, input: unknown) => {76 const startSession = vi.fn((_: unknown, input: unknown) => {
77 const sessionIndex = nextSessionIndex++;77 const sessionIndex = nextSessionIndex++;
78 const provider =
79 typeof input === "object" &&
80 input !== null &&
81 "provider" in input &&
82 (input.provider === "codex" || input.provider === "claudeCode")
83 ? input.provider
84 : "codex";
78 const resumeCursor =85 const resumeCursor =
79 typeof input === "object" && input !== null && "resumeCursor" in input86 typeof input === "object" && input !== null && "resumeCursor" in input
80 ? input.resumeCursor87 ? input.resumeCursor
81 : undefined;88 : undefined;
82 const session: ProviderSession = {89 const session: ProviderSession = {
83 sessionId: asSessionId(`sess-${sessionIndex}`),90 sessionId: asSessionId(`sess-${sessionIndex}`),
84 provider: "codex" as const,91 provider,
85 status: "ready" as const,92 status: "ready" as const,
86 threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`),93 threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`),
87 resumeCursor: resumeCursor ?? { opaque: `cursor-${sessionIndex}` },94 resumeCursor: resumeCursor ?? { opaque: `cursor-${sessionIndex}` },
@@ -218,6 +225,44 @@ describe("ProviderCommandReactor", () => { · 130 unmodified lines
218 expect(thread?.session?.sandboxMode).toBe("workspace-write");225 expect(thread?.session?.sandboxMode).toBe("workspace-write");
219 });226 });
220 227
228 it("starts first turn with requested provider when provider is specified", async () => {
229 const harness = await createHarness();
230 const now = new Date().toISOString();
231
232 await Effect.runPromise(
233 harness.engine.dispatch({
234 type: "thread.turn.start",
235 commandId: CommandId.makeUnsafe("cmd-turn-start-provider-first"),
236 threadId: ThreadId.makeUnsafe("thread-1"),
237 message: {
238 messageId: asMessageId("user-message-provider-first"),
239 role: "user",
240 text: "hello claude",
241 attachments: [],
242 },
243 provider: "claudeCode",
244 approvalPolicy: "on-request",
245 sandboxMode: "workspace-write",
246 createdAt: now,
247 }),
248 );
249
250 await waitFor(() => harness.startSession.mock.calls.length === 1);
251 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
252 expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({
253 provider: "claudeCode",
254 cwd: "/tmp/provider-project",
255 model: "gpt-5-codex",
256 approvalPolicy: "on-request",
257 sandboxMode: "workspace-write",
258 });
259
260 const readModel = await Effect.runPromise(harness.engine.getReadModel());
261 const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
262 expect(thread?.session?.providerName).toBe("claudeCode");
263 expect(thread?.session?.providerSessionId).toBe("sess-1");
264 });
265
221 it("reuses the same provider session when runtime mode is unchanged", async () => {266 it("reuses the same provider session when runtime mode is unchanged", async () => {
222 const harness = await createHarness();267 const harness = await createHarness();
223 const now = new Date().toISOString();268 const now = new Date().toISOString();
@@ -326,6 +371,68 @@ describe("ProviderCommandReactor", () => { · 102 unmodified lines
326 expect(thread?.session?.sandboxMode).toBe("workspace-write");371 expect(thread?.session?.sandboxMode).toBe("workspace-write");
327 });372 });
328 373
374 it("switches provider by restarting the session when turn request provider changes", async () => {
375 const harness = await createHarness();
376 const now = new Date().toISOString();
377
378 await Effect.runPromise(
379 harness.engine.dispatch({
380 type: "thread.turn.start",
381 commandId: CommandId.makeUnsafe("cmd-turn-start-provider-switch-1"),
382 threadId: ThreadId.makeUnsafe("thread-1"),
383 message: {
384 messageId: asMessageId("user-message-provider-switch-1"),
385 role: "user",
386 text: "first",
387 attachments: [],
388 },
389 approvalPolicy: "on-request",
390 sandboxMode: "workspace-write",
391 createdAt: now,
392 }),
393 );
394
395 await waitFor(() => harness.startSession.mock.calls.length === 1);
396 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
397
398 await Effect.runPromise(
399 harness.engine.dispatch({
400 type: "thread.turn.start",
401 commandId: CommandId.makeUnsafe("cmd-turn-start-provider-switch-2"),
402 threadId: ThreadId.makeUnsafe("thread-1"),
403 message: {
404 messageId: asMessageId("user-message-provider-switch-2"),
405 role: "user",
406 text: "second",
407 attachments: [],
408 },
409 provider: "claudeCode",
410 approvalPolicy: "on-request",
411 sandboxMode: "workspace-write",
412 createdAt: now,
413 }),
414 );
415
416 await waitFor(() => harness.stopSession.mock.calls.length === 1);
417 await waitFor(() => harness.startSession.mock.calls.length === 2);
418 await waitFor(() => harness.sendTurn.mock.calls.length === 2);
419
420 expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") });
421 expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({
422 provider: "claudeCode",
423 resumeCursor: { opaque: "cursor-1" },
424 approvalPolicy: "on-request",
425 sandboxMode: "workspace-write",
426 });
427
428 const readModel = await Effect.runPromise(harness.engine.getReadModel());
429 const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
430 expect(thread?.session?.providerSessionId).toBe("sess-2");
431 expect(thread?.session?.providerName).toBe("claudeCode");
432 expect(thread?.session?.approvalPolicy).toBe("on-request");
433 expect(thread?.session?.sandboxMode).toBe("workspace-write");
434 });
435
329 it("does not stop the active session when restart fails before rebind", async () => {436 it("does not stop the active session when restart fails before rebind", async () => {
330 const harness = await createHarness();437 const harness = await createHarness();
331 const now = new Date().toISOString();438 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+23−7
@@ -135,6 +135,7 @@ const make = Effect.gen(function* () { · 134 unmodified lines
135 threadId: ThreadId,135 threadId: ThreadId,
136 createdAt: string,136 createdAt: string,
137 options?: {137 options?: {
138 readonly provider?: ProviderKind;
138 readonly approvalPolicy?: ProviderApprovalPolicy;139 readonly approvalPolicy?: ProviderApprovalPolicy;
139 readonly sandboxMode?: ProviderSandboxMode;140 readonly sandboxMode?: ProviderSandboxMode;
140 },141 },
@@ -149,10 +150,11 @@ const make = Effect.gen(function* () { · 8 unmodified lines
149 options?.approvalPolicy ?? thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY;150 options?.approvalPolicy ?? thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY;
150 const desiredSandboxMode =151 const desiredSandboxMode =
151 options?.sandboxMode ?? thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE;152 options?.sandboxMode ?? thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE;
152 const preferredProvider: ProviderKind | undefined =153 const currentProvider: ProviderKind | undefined =
153 thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode"154 thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode"
154 ? thread.session.providerName155 ? thread.session.providerName
155 : undefined;156 : undefined;
157 const preferredProvider: ProviderKind | undefined = options?.provider ?? currentProvider;
156 const effectiveCwd = resolveThreadWorkspaceCwd({158 const effectiveCwd = resolveThreadWorkspaceCwd({
157 thread,159 thread,
158 projects: readModel.projects,160 projects: readModel.projects,
@@ -163,12 +165,17 @@ const make = Effect.gen(function* () { · 4 unmodified lines
163 Effect.map((sessions) => sessions.find((session) => session.sessionId === sessionId)?.resumeCursor),165 Effect.map((sessions) => sessions.find((session) => session.sessionId === sessionId)?.resumeCursor),
164 );166 );
165 167
166 const startProviderSession = (resumeCursor?: unknown) =>168 const startProviderSession = (input?: {
169 readonly resumeCursor?: unknown;
170 readonly provider?: ProviderKind;
171 }) =>
167 providerService.startSession(threadId, {172 providerService.startSession(threadId, {
168 ...(preferredProvider ? { provider: preferredProvider } : {}),173 ...(input?.provider ?? preferredProvider
174 ? { provider: input?.provider ?? preferredProvider }
175 : {}),
169 ...(effectiveCwd ? { cwd: effectiveCwd } : {}),176 ...(effectiveCwd ? { cwd: effectiveCwd } : {}),
170 ...(thread.model ? { model: thread.model } : {}),177 ...(thread.model ? { model: thread.model } : {}),
171 ...(resumeCursor !== undefined ? { resumeCursor } : {}),178 ...(input?.resumeCursor !== undefined ? { resumeCursor: input.resumeCursor } : {}),
172 approvalPolicy: desiredApprovalPolicy,179 approvalPolicy: desiredApprovalPolicy,
173 sandboxMode: desiredSandboxMode,180 sandboxMode: desiredSandboxMode,
174 });181 });
@@ -198,13 +205,17 @@ const make = Effect.gen(function* () { · 23 unmodified lines
198 options?.approvalPolicy !== undefined && options.approvalPolicy !== thread.session?.approvalPolicy;205 options?.approvalPolicy !== undefined && options.approvalPolicy !== thread.session?.approvalPolicy;
199 const sandboxModeChanged =206 const sandboxModeChanged =
200 options?.sandboxMode !== undefined && options.sandboxMode !== thread.session?.sandboxMode;207 options?.sandboxMode !== undefined && options.sandboxMode !== thread.session?.sandboxMode;
208 const providerChanged = options?.provider !== undefined && options.provider !== currentProvider;
201 209
202 if (!approvalPolicyChanged && !sandboxModeChanged) {210 if (!approvalPolicyChanged && !sandboxModeChanged && !providerChanged) {
203 return existingSessionId;211 return existingSessionId;
204 }212 }
205 213
206 const resumeCursor = yield* resolveResumeCursorForSession(existingSessionId);214 const resumeCursor = yield* resolveResumeCursorForSession(existingSessionId);
207 const restartedSession = yield* startProviderSession(resumeCursor);215 const restartedSession = yield* startProviderSession({
216 ...(resumeCursor !== undefined ? { resumeCursor } : {}),
217 ...(options?.provider !== undefined ? { provider: options.provider } : {}),
218 });
208 yield* bindSessionToThread(restartedSession);219 yield* bindSessionToThread(restartedSession);
209 yield* providerService.stopSession({ sessionId: existingSessionId }).pipe(220 yield* providerService.stopSession({ sessionId: existingSessionId }).pipe(
210 Effect.catchCause((cause) =>221 Effect.catchCause((cause) =>
@@ -218,7 +229,9 @@ const make = Effect.gen(function* () { · 7 unmodified lines
218 return restartedSession.sessionId;229 return restartedSession.sessionId;
219 }230 }
220 231
221 const startedSession = yield* startProviderSession();232 const startedSession = yield* startProviderSession({
233 ...(options?.provider !== undefined ? { provider: options.provider } : {}),
234 });
222 yield* bindSessionToThread(startedSession);235 yield* bindSessionToThread(startedSession);
223 return startedSession.sessionId;236 return startedSession.sessionId;
224 });237 });
@@ -227,6 +240,7 @@ const make = Effect.gen(function* () { · 2 unmodified lines
227 readonly threadId: ThreadId;240 readonly threadId: ThreadId;
228 readonly messageText: string;241 readonly messageText: string;
229 readonly attachments?: ReadonlyArray<ChatAttachment>;242 readonly attachments?: ReadonlyArray<ChatAttachment>;
243 readonly provider?: ProviderKind;
230 readonly model?: string;244 readonly model?: string;
231 readonly effort?: string;245 readonly effort?: string;
232 readonly approvalPolicy: ProviderApprovalPolicy;246 readonly approvalPolicy: ProviderApprovalPolicy;
@@ -238,6 +252,7 @@ const make = Effect.gen(function* () { · 5 unmodified lines
238 return;252 return;
239 }253 }
240 const sessionId = yield* ensureSessionForThread(input.threadId, input.createdAt, {254 const sessionId = yield* ensureSessionForThread(input.threadId, input.createdAt, {
255 ...(input.provider !== undefined ? { provider: input.provider } : {}),
241 approvalPolicy: input.approvalPolicy,256 approvalPolicy: input.approvalPolicy,
242 sandboxMode: input.sandboxMode,257 sandboxMode: input.sandboxMode,
243 });258 });
@@ -283,6 +298,7 @@ const make = Effect.gen(function* () { · 39 unmodified lines
283 threadId: event.payload.threadId,298 threadId: event.payload.threadId,
284 messageText: message.text,299 messageText: message.text,
285 ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),300 ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
301 ...(event.payload.provider !== undefined ? { provider: event.payload.provider } : {}),
286 ...(event.payload.model !== undefined ? { model: event.payload.model } : {}),302 ...(event.payload.model !== undefined ? { model: event.payload.model } : {}),
287 ...(event.payload.effort !== undefined ? { effort: event.payload.effort } : {}),303 ...(event.payload.effort !== undefined ? { effort: event.payload.effort } : {}),
288 approvalPolicy: event.payload.approvalPolicy,304 approvalPolicy: event.payload.approvalPolicy,