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 c590beb1 parent e9ced442 files · +93 −211Browse files

Unified · Split

2 files
  1. apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +70 −204
  2. apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +23 −7
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+70−204
@@ -87,13 +87,20 @@ describe("ProviderCommandReactor", () => { · 86 unmodified lines
8787 const runtimeSessions: Array<ProviderSession> = [];
8888 const startSession = vi.fn((_: unknown, input: unknown) => {
8989 const sessionIndex = nextSessionIndex++;
90+ const provider =
91+ typeof input === "object" &&
92+ input !== null &&
93+ "provider" in input &&
94+ (input.provider === "codex" || input.provider === "claudeCode")
95+ ? input.provider
96+ : "codex";
9097 const resumeCursor =
9198 typeof input === "object" && input !== null && "resumeCursor" in input
9299 ? input.resumeCursor
93100 : undefined;
94101 const session: ProviderSession = {
95102 sessionId: asSessionId(`sess-${sessionIndex}`),
96− provider: "codex" as const,
103+ provider,
97104 status: "ready" as const,
98105 threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`),
99106 resumeCursor: resumeCursor ?? { opaque: `cursor-${sessionIndex}` },
@@ -240,165 +247,55 @@ describe("ProviderCommandReactor", () => { · 140 unmodified lines
240247 expect(thread?.session?.sandboxMode).toBe("workspace-write");
241248 });
242249
243− it("generates and renames temporary worktree branch on first turn", async () => {
250+ it("starts first turn with requested provider when provider is specified", async () => {
244251 const harness = await createHarness();
245252 const now = new Date().toISOString();
246253
247− await Effect.runPromise(
248− harness.engine.dispatch({
249− type: "thread.meta.update",
250− commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch"),
251− threadId: ThreadId.makeUnsafe("thread-1"),
252− branch: "t3code/89abc123",
253− worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-89abc123",
254− }),
255− );
256−
257− await Effect.runPromise(
258− harness.engine.dispatch({
259− type: "thread.turn.start",
260− commandId: CommandId.makeUnsafe("cmd-turn-start-worktree-rename"),
261− threadId: ThreadId.makeUnsafe("thread-1"),
262− message: {
263− messageId: asMessageId("user-message-worktree-rename"),
264− role: "user",
265− text: "Fix visual bug from screenshot",
266− attachments: [
267− {
268− type: "image",
269− id: "thread-1-att-rename",
270− name: "bug.png",
271− mimeType: "image/png",
272− sizeBytes: 5,
273− },
274− ],
275− },
276− approvalPolicy: "on-request",
277− sandboxMode: "workspace-write",
278− createdAt: now,
279− }),
280− );
281−
282− await waitFor(() => harness.generateBranchName.mock.calls.length === 1);
283− await waitFor(() => harness.renameBranch.mock.calls.length === 1);
284−
285− expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({
286− cwd: "/tmp/provider-project/.t3/worktrees/t3code-89abc123",
287− message: "Fix visual bug from screenshot",
288− attachments: [
289− {
290− type: "image",
291− id: "thread-1-att-rename",
292− name: "bug.png",
293− mimeType: "image/png",
294− sizeBytes: 5,
295− },
296− ],
297− });
298− expect(harness.renameBranch.mock.calls[0]?.[0]).toEqual({
299− cwd: "/tmp/provider-project/.t3/worktrees/t3code-89abc123",
300− oldBranch: "t3code/89abc123",
301− newBranch: "t3code/generated-name",
302− });
303−
304− await waitFor(() => {
305− const readModel = Effect.runSync(harness.engine.getReadModel());
306− const thread = readModel.threads.find(
307− (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),
308− );
309− return thread?.branch === "t3code/generated-name";
310− });
311− });
312−
313− it("passes persisted attachment references to branch generation and turn start", async () => {
314− const harness = await createHarness();
315− const now = new Date().toISOString();
316−
317− await Effect.runPromise(
318− harness.engine.dispatch({
319− type: "thread.meta.update",
320− commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch-persisted"),
321− threadId: ThreadId.makeUnsafe("thread-1"),
322− branch: "t3code/abcdef12",
323− worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12",
324− }),
325− );
326−
327254 await Effect.runPromise(
328255 harness.engine.dispatch({
329256 type: "thread.turn.start",
330− commandId: CommandId.makeUnsafe("cmd-turn-start-persisted-attachments"),
257+ commandId: CommandId.makeUnsafe("cmd-turn-start-provider-first"),
331258 threadId: ThreadId.makeUnsafe("thread-1"),
332259 message: {
333− messageId: asMessageId("user-message-persisted-attachments"),
260+ messageId: asMessageId("user-message-provider-first"),
334261 role: "user",
335− text: "Fix visual bug from screenshot",
336− attachments: [
337− {
338− type: "image",
339− id: "thread-1-att-persisted",
340− name: "bug.png",
341− mimeType: "image/png",
342− sizeBytes: 5,
343− },
344− ],
262+ text: "hello claude",
263+ attachments: [],
345264 },
265+ provider: "claudeCode",
346266 approvalPolicy: "on-request",
347267 sandboxMode: "workspace-write",
348268 createdAt: now,
349269 }),
350270 );
351271
352− await waitFor(() => harness.generateBranchName.mock.calls.length === 1);
272+ await waitFor(() => harness.startSession.mock.calls.length === 1);
353273 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
354−
355− expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({
356− cwd: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12",
357− message: "Fix visual bug from screenshot",
358− attachments: [
359− {
360− type: "image",
361− id: "thread-1-att-persisted",
362− name: "bug.png",
363− mimeType: "image/png",
364− sizeBytes: 5,
365− },
366− ],
367− });
368− expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({
369− attachments: [
370− {
371− type: "image",
372− id: "thread-1-att-persisted",
373− name: "bug.png",
374− mimeType: "image/png",
375− sizeBytes: 5,
376− },
377− ],
274+ expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({
275+ provider: "claudeCode",
276+ cwd: "/tmp/provider-project",
277+ model: "gpt-5-codex",
278+ approvalPolicy: "on-request",
279+ sandboxMode: "workspace-write",
378280 });
281+
282+ const readModel = await Effect.runPromise(harness.engine.getReadModel());
283+ const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
284+ expect(thread?.session?.providerName).toBe("claudeCode");
285+ expect(thread?.session?.providerSessionId).toBe("sess-1");
379286 });
380287
381− it("skips worktree branch generation after the first user turn", async () => {
288+ it("reuses the same provider session when runtime mode is unchanged", async () => {
382289 const harness = await createHarness();
383290 const now = new Date().toISOString();
384291
385− await Effect.runPromise(
386− harness.engine.dispatch({
387− type: "thread.meta.update",
388− commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch-2"),
389− threadId: ThreadId.makeUnsafe("thread-1"),
390− branch: "t3code/1234abcd",
391− worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-1234abcd",
392− }),
393− );
394−
395292 await Effect.runPromise(
396293 harness.engine.dispatch({
397294 type: "thread.turn.start",
398− commandId: CommandId.makeUnsafe("cmd-turn-start-first"),
295+ commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-1"),
399296 threadId: ThreadId.makeUnsafe("thread-1"),
400297 message: {
401− messageId: asMessageId("user-message-first"),
298+ messageId: asMessageId("user-message-unchanged-1"),
402299 role: "user",
403300 text: "first",
404301 attachments: [],
@@ -409,15 +306,16 @@ describe("ProviderCommandReactor", () => { · 4 unmodified lines
409306 }),
410307 );
411308
412− await waitFor(() => harness.generateBranchName.mock.calls.length === 1);
309+ await waitFor(() => harness.startSession.mock.calls.length === 1);
310+ await waitFor(() => harness.sendTurn.mock.calls.length === 1);
413311
414312 await Effect.runPromise(
415313 harness.engine.dispatch({
416314 type: "thread.turn.start",
417− commandId: CommandId.makeUnsafe("cmd-turn-start-second"),
315+ commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-2"),
418316 threadId: ThreadId.makeUnsafe("thread-1"),
419317 message: {
420− messageId: asMessageId("user-message-second"),
318+ messageId: asMessageId("user-message-unchanged-2"),
421319 role: "user",
422320 text: "second",
423321 attachments: [],
@@ -429,75 +327,27 @@ describe("ProviderCommandReactor", () => { · 5 unmodified lines
429327 );
430328
431329 await waitFor(() => harness.sendTurn.mock.calls.length === 2);
432− expect(harness.generateBranchName.mock.calls.length).toBe(1);
433− expect(harness.renameBranch.mock.calls.length).toBe(1);
434− });
435−
436− it("skips worktree rename when branch-name generation fails", async () => {
437− const harness = await createHarness();
438− const now = new Date().toISOString();
439− harness.generateBranchName.mockImplementationOnce(() =>
440− Effect.fail(
441− new TextGenerationError({
442− operation: "generateBranchName",
443− detail: "model returned invalid payload",
444− }),
445− ),
446− );
447−
448− await Effect.runPromise(
449− harness.engine.dispatch({
450− type: "thread.meta.update",
451− commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch-null"),
452− threadId: ThreadId.makeUnsafe("thread-1"),
453− branch: "t3code/0000abcd",
454− worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-0000abcd",
455− }),
456− );
457−
458− await Effect.runPromise(
459− harness.engine.dispatch({
460− type: "thread.turn.start",
461− commandId: CommandId.makeUnsafe("cmd-turn-start-null-branch"),
462− threadId: ThreadId.makeUnsafe("thread-1"),
463− message: {
464− messageId: asMessageId("user-message-null-branch"),
465− role: "user",
466− text: "Fix visual regression",
467− attachments: [],
468− },
469− approvalPolicy: "on-request",
470− sandboxMode: "workspace-write",
471− createdAt: now,
472− }),
473− );
474−
475− await waitFor(() => harness.generateBranchName.mock.calls.length === 1);
476− await Effect.runPromise(Effect.sleep("20 millis"));
477− expect(harness.renameBranch.mock.calls.length).toBe(0);
478−
479− const readModel = await Effect.runPromise(harness.engine.getReadModel());
480− const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
481− expect(thread?.branch).toBe("t3code/0000abcd");
330+ expect(harness.startSession.mock.calls.length).toBe(1);
331+ expect(harness.stopSession.mock.calls.length).toBe(0);
482332 });
483333
484− it("reuses the same provider session when runtime mode is unchanged", async () => {
334+ it("restarts the provider session when runtime mode changes", async () => {
485335 const harness = await createHarness();
486336 const now = new Date().toISOString();
487337
488338 await Effect.runPromise(
489339 harness.engine.dispatch({
490340 type: "thread.turn.start",
491− commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-1"),
341+ commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-1"),
492342 threadId: ThreadId.makeUnsafe("thread-1"),
493343 message: {
494− messageId: asMessageId("user-message-unchanged-1"),
344+ messageId: asMessageId("user-message-runtime-mode-1"),
495345 role: "user",
496346 text: "first",
497347 attachments: [],
498348 },
499− approvalPolicy: "on-request",
500− sandboxMode: "workspace-write",
349+ approvalPolicy: "never",
350+ sandboxMode: "danger-full-access",
501351 createdAt: now,
502352 }),
503353 );
@@ -508,10 +358,10 @@ describe("ProviderCommandReactor", () => { · 4 unmodified lines
508358 await Effect.runPromise(
509359 harness.engine.dispatch({
510360 type: "thread.turn.start",
511− commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-2"),
361+ commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-2"),
512362 threadId: ThreadId.makeUnsafe("thread-1"),
513363 message: {
514− messageId: asMessageId("user-message-unchanged-2"),
364+ messageId: asMessageId("user-message-runtime-mode-2"),
515365 role: "user",
516366 text: "second",
517367 attachments: [],
@@ -522,28 +372,44 @@ describe("ProviderCommandReactor", () => { · 4 unmodified lines
522372 }),
523373 );
524374
375+ await waitFor(() => harness.stopSession.mock.calls.length === 1);
376+ await waitFor(() => harness.startSession.mock.calls.length === 2);
525377 await waitFor(() => harness.sendTurn.mock.calls.length === 2);
526− expect(harness.startSession.mock.calls.length).toBe(1);
527− expect(harness.stopSession.mock.calls.length).toBe(0);
378+
379+ expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") });
380+ expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({
381+ resumeCursor: { opaque: "cursor-1" },
382+ approvalPolicy: "on-request",
383+ sandboxMode: "workspace-write",
384+ });
385+ expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({
386+ sessionId: asSessionId("sess-2"),
387+ });
388+
389+ const readModel = await Effect.runPromise(harness.engine.getReadModel());
390+ const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
391+ expect(thread?.session?.providerSessionId).toBe("sess-2");
392+ expect(thread?.session?.approvalPolicy).toBe("on-request");
393+ expect(thread?.session?.sandboxMode).toBe("workspace-write");
528394 });
529395
530− it("restarts the provider session when runtime mode changes", async () => {
396+ it("switches provider by restarting the session when turn request provider changes", async () => {
531397 const harness = await createHarness();
532398 const now = new Date().toISOString();
533399
534400 await Effect.runPromise(
535401 harness.engine.dispatch({
536402 type: "thread.turn.start",
537− commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-1"),
403+ commandId: CommandId.makeUnsafe("cmd-turn-start-provider-switch-1"),
538404 threadId: ThreadId.makeUnsafe("thread-1"),
539405 message: {
540− messageId: asMessageId("user-message-runtime-mode-1"),
406+ messageId: asMessageId("user-message-provider-switch-1"),
541407 role: "user",
542408 text: "first",
543409 attachments: [],
544410 },
545− approvalPolicy: "never",
546− sandboxMode: "danger-full-access",
411+ approvalPolicy: "on-request",
412+ sandboxMode: "workspace-write",
547413 createdAt: now,
548414 }),
549415 );
@@ -554,14 +420,15 @@ describe("ProviderCommandReactor", () => { · 4 unmodified lines
554420 await Effect.runPromise(
555421 harness.engine.dispatch({
556422 type: "thread.turn.start",
557− commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-2"),
423+ commandId: CommandId.makeUnsafe("cmd-turn-start-provider-switch-2"),
558424 threadId: ThreadId.makeUnsafe("thread-1"),
559425 message: {
560− messageId: asMessageId("user-message-runtime-mode-2"),
426+ messageId: asMessageId("user-message-provider-switch-2"),
561427 role: "user",
562428 text: "second",
563429 attachments: [],
564430 },
431+ provider: "claudeCode",
565432 approvalPolicy: "on-request",
566433 sandboxMode: "workspace-write",
567434 createdAt: now,
@@ -574,17 +441,16 @@ describe("ProviderCommandReactor", () => { · 6 unmodified lines
574441
575442 expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") });
576443 expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({
444+ provider: "claudeCode",
577445 resumeCursor: { opaque: "cursor-1" },
578446 approvalPolicy: "on-request",
579447 sandboxMode: "workspace-write",
580448 });
581− expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({
582− sessionId: asSessionId("sess-2"),
583− });
584449
585450 const readModel = await Effect.runPromise(harness.engine.getReadModel());
586451 const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
587452 expect(thread?.session?.providerSessionId).toBe("sess-2");
453+ expect(thread?.session?.providerName).toBe("claudeCode");
588454 expect(thread?.session?.approvalPolicy).toBe("on-request");
589455 expect(thread?.session?.sandboxMode).toBe("workspace-write");
590456 });
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+23−7
@@ -168,6 +168,7 @@ const make = Effect.gen(function* () { · 167 unmodified lines
168168 threadId: ThreadId,
169169 createdAt: string,
170170 options?: {
171+ readonly provider?: ProviderKind;
171172 readonly approvalPolicy?: ProviderApprovalPolicy;
172173 readonly sandboxMode?: ProviderSandboxMode;
173174 },
@@ -182,10 +183,11 @@ const make = Effect.gen(function* () { · 8 unmodified lines
182183 options?.approvalPolicy ?? thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY;
183184 const desiredSandboxMode =
184185 options?.sandboxMode ?? thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE;
185− const preferredProvider: ProviderKind | undefined =
186+ const currentProvider: ProviderKind | undefined =
186187 thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode"
187188 ? thread.session.providerName
188189 : undefined;
190+ const preferredProvider: ProviderKind | undefined = options?.provider ?? currentProvider;
189191 const effectiveCwd = resolveThreadWorkspaceCwd({
190192 thread,
191193 projects: readModel.projects,
@@ -196,12 +198,17 @@ const make = Effect.gen(function* () { · 4 unmodified lines
196198 Effect.map((sessions) => sessions.find((session) => session.sessionId === sessionId)?.resumeCursor),
197199 );
198200
199− const startProviderSession = (resumeCursor?: unknown) =>
201+ const startProviderSession = (input?: {
202+ readonly resumeCursor?: unknown;
203+ readonly provider?: ProviderKind;
204+ }) =>
200205 providerService.startSession(threadId, {
201− ...(preferredProvider ? { provider: preferredProvider } : {}),
206+ ...(input?.provider ?? preferredProvider
207+ ? { provider: input?.provider ?? preferredProvider }
208+ : {}),
202209 ...(effectiveCwd ? { cwd: effectiveCwd } : {}),
203210 ...(thread.model ? { model: thread.model } : {}),
204− ...(resumeCursor !== undefined ? { resumeCursor } : {}),
211+ ...(input?.resumeCursor !== undefined ? { resumeCursor: input.resumeCursor } : {}),
205212 approvalPolicy: desiredApprovalPolicy,
206213 sandboxMode: desiredSandboxMode,
207214 });
@@ -232,13 +239,17 @@ const make = Effect.gen(function* () { · 24 unmodified lines
232239 options.approvalPolicy !== thread.session?.approvalPolicy;
233240 const sandboxModeChanged =
234241 options?.sandboxMode !== undefined && options.sandboxMode !== thread.session?.sandboxMode;
242+ const providerChanged = options?.provider !== undefined && options.provider !== currentProvider;
235243
236− if (!approvalPolicyChanged && !sandboxModeChanged) {
244+ if (!approvalPolicyChanged && !sandboxModeChanged && !providerChanged) {
237245 return existingSessionId;
238246 }
239247
240248 const resumeCursor = yield* resolveResumeCursorForSession(existingSessionId);
241− const restartedSession = yield* startProviderSession(resumeCursor);
249+ const restartedSession = yield* startProviderSession({
250+ ...(resumeCursor !== undefined ? { resumeCursor } : {}),
251+ ...(options?.provider !== undefined ? { provider: options.provider } : {}),
252+ });
242253 yield* bindSessionToThread(restartedSession);
243254 yield* providerService.stopSession({ sessionId: existingSessionId }).pipe(
244255 Effect.catchCause((cause) =>
@@ -252,7 +263,9 @@ const make = Effect.gen(function* () { · 7 unmodified lines
252263 return restartedSession.sessionId;
253264 }
254265
255− const startedSession = yield* startProviderSession();
266+ const startedSession = yield* startProviderSession({
267+ ...(options?.provider !== undefined ? { provider: options.provider } : {}),
268+ });
256269 yield* bindSessionToThread(startedSession);
257270 return startedSession.sessionId;
258271 });
@@ -261,6 +274,7 @@ const make = Effect.gen(function* () { · 2 unmodified lines
261274 readonly threadId: ThreadId;
262275 readonly messageText: string;
263276 readonly attachments?: ReadonlyArray<ChatAttachment>;
277+ readonly provider?: ProviderKind;
264278 readonly model?: string;
265279 readonly effort?: string;
266280 readonly approvalPolicy: ProviderApprovalPolicy;
@@ -272,6 +286,7 @@ const make = Effect.gen(function* () { · 5 unmodified lines
272286 return;
273287 }
274288 const sessionId = yield* ensureSessionForThread(input.threadId, input.createdAt, {
289+ ...(input.provider !== undefined ? { provider: input.provider } : {}),
275290 approvalPolicy: input.approvalPolicy,
276291 sandboxMode: input.sandboxMode,
277292 });
@@ -394,6 +409,7 @@ const make = Effect.gen(function* () { · 116 unmodified lines
394409 threadId: event.payload.threadId,
395410 messageText: message.text,
396411 ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
412+ ...(event.payload.provider !== undefined ? { provider: event.payload.provider } : {}),
397413 ...(event.payload.model !== undefined ? { model: event.payload.model } : {}),
398414 ...(event.payload.effort !== undefined ? { effort: event.payload.effort } : {}),
399415 approvalPolicy: event.payload.approvalPolicy,