diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index c41ce614aea8c97d3433d1c1e22b610d835d8b0e..bacb78607b45b0c27eb67bf59a8c9c94e9071721 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -71,16 +71,18 @@ describe("ProviderCommandReactor", () => { async function createHarness() { const now = new Date().toISOString(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); - const startSession = vi.fn((_: unknown, __: unknown) => - Effect.succeed({ - sessionId: asSessionId("sess-1"), + let nextSessionIndex = 1; + const startSession = vi.fn((_: unknown, __: unknown) => { + const sessionIndex = nextSessionIndex++; + return Effect.succeed({ + sessionId: asSessionId(`sess-${sessionIndex}`), provider: "codex" as const, status: "ready" as const, - threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`), createdAt: now, updatedAt: now, - }), - ); + }); + }); const sendTurn = vi.fn((_: unknown) => Effect.succeed({ threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), @@ -190,6 +192,116 @@ describe("ProviderCommandReactor", () => { const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.session?.providerSessionId).toBe("sess-1"); + expect(thread?.session?.approvalPolicy).toBe("on-request"); + expect(thread?.session?.sandboxMode).toBe("workspace-write"); + }); + + it("reuses the same provider session when runtime mode is unchanged", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-1"), + threadId: ThreadId.makeUnsafe("thread-1"), + message: { + messageId: asMessageId("user-message-unchanged-1"), + role: "user", + text: "first", + attachments: [], + }, + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + createdAt: now, + }), + ); + + await waitFor(() => harness.startSession.mock.calls.length === 1); + await waitFor(() => harness.sendTurn.mock.calls.length === 1); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-2"), + threadId: ThreadId.makeUnsafe("thread-1"), + message: { + messageId: asMessageId("user-message-unchanged-2"), + role: "user", + text: "second", + attachments: [], + }, + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + createdAt: now, + }), + ); + + await waitFor(() => harness.sendTurn.mock.calls.length === 2); + expect(harness.startSession.mock.calls.length).toBe(1); + expect(harness.stopSession.mock.calls.length).toBe(0); + }); + + it("restarts the provider session when runtime mode changes", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-1"), + threadId: ThreadId.makeUnsafe("thread-1"), + message: { + messageId: asMessageId("user-message-runtime-mode-1"), + role: "user", + text: "first", + attachments: [], + }, + approvalPolicy: "never", + sandboxMode: "danger-full-access", + createdAt: now, + }), + ); + + await waitFor(() => harness.startSession.mock.calls.length === 1); + await waitFor(() => harness.sendTurn.mock.calls.length === 1); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-2"), + threadId: ThreadId.makeUnsafe("thread-1"), + message: { + messageId: asMessageId("user-message-runtime-mode-2"), + role: "user", + text: "second", + attachments: [], + }, + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + createdAt: now, + }), + ); + + await waitFor(() => harness.stopSession.mock.calls.length === 1); + await waitFor(() => harness.startSession.mock.calls.length === 2); + await waitFor(() => harness.sendTurn.mock.calls.length === 2); + + expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") }); + expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({ + resumeThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + }); + expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({ + sessionId: asSessionId("sess-2"), + }); + + const readModel = await Effect.runPromise(harness.engine.getReadModel()); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(thread?.session?.providerSessionId).toBe("sess-2"); + expect(thread?.session?.approvalPolicy).toBe("on-request"); + expect(thread?.session?.sandboxMode).toBe("workspace-write"); }); it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => { diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index ecba45185183d300da5f959556c0261d59ac90f2..bee328e479381269b4ede087ab4c1c71595f573e 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -8,6 +8,8 @@ import { type ProviderSandboxMode, type OrchestrationSession, type ThreadId, + type ProviderSession, + type ProviderThreadId, type TurnId, } from "@t3tools/contracts"; import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect"; @@ -141,11 +143,8 @@ const make = Effect.gen(function* () { return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`)); } - const existingSessionId = thread.session?.providerSessionId; - if (existingSessionId) { - return existingSessionId; - } - + const desiredApprovalPolicy = options?.approvalPolicy ?? thread.session?.approvalPolicy; + const desiredSandboxMode = options?.sandboxMode ?? thread.session?.sandboxMode; const preferredProvider: ProviderKind | undefined = thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode" ? thread.session.providerName @@ -155,30 +154,54 @@ const make = Effect.gen(function* () { projects: readModel.projects, }); - const startedSession = yield* providerService.startSession(threadId, { - ...(preferredProvider ? { provider: preferredProvider } : {}), - ...(effectiveCwd ? { cwd: effectiveCwd } : {}), - ...(thread.model ? { model: thread.model } : {}), - ...(options?.approvalPolicy !== undefined ? { approvalPolicy: options.approvalPolicy } : {}), - ...(options?.sandboxMode !== undefined ? { sandboxMode: options.sandboxMode } : {}), - }); + const startProviderSession = (resumeThreadId?: ProviderThreadId | null) => + providerService.startSession(threadId, { + ...(preferredProvider ? { provider: preferredProvider } : {}), + ...(effectiveCwd ? { cwd: effectiveCwd } : {}), + ...(thread.model ? { model: thread.model } : {}), + ...(resumeThreadId ? { resumeThreadId } : {}), + ...(desiredApprovalPolicy !== undefined ? { approvalPolicy: desiredApprovalPolicy } : {}), + ...(desiredSandboxMode !== undefined ? { sandboxMode: desiredSandboxMode } : {}), + }); - yield* setThreadSession({ - threadId, - session: { + const bindSessionToThread = (session: ProviderSession) => + setThreadSession({ threadId, - status: mapProviderSessionStatusToOrchestrationStatus(startedSession.status), - providerName: startedSession.provider, - providerSessionId: startedSession.sessionId, - providerThreadId: startedSession.threadId ?? null, - // Provider turn ids are not orchestration turn ids. - activeTurnId: null, - lastError: startedSession.lastError ?? null, - updatedAt: startedSession.updatedAt, - }, - createdAt, - }); + session: { + threadId, + status: mapProviderSessionStatusToOrchestrationStatus(session.status), + providerName: session.provider, + providerSessionId: session.sessionId, + providerThreadId: session.threadId ?? null, + ...(desiredApprovalPolicy !== undefined ? { approvalPolicy: desiredApprovalPolicy } : {}), + ...(desiredSandboxMode !== undefined ? { sandboxMode: desiredSandboxMode } : {}), + // Provider turn ids are not orchestration turn ids. + activeTurnId: null, + lastError: session.lastError ?? null, + updatedAt: session.updatedAt, + }, + createdAt, + }); + const existingSessionId = thread.session?.providerSessionId; + if (existingSessionId) { + const approvalPolicyChanged = + options?.approvalPolicy !== undefined && options.approvalPolicy !== thread.session?.approvalPolicy; + const sandboxModeChanged = + options?.sandboxMode !== undefined && options.sandboxMode !== thread.session?.sandboxMode; + + if (!approvalPolicyChanged && !sandboxModeChanged) { + return existingSessionId; + } + + yield* providerService.stopSession({ sessionId: existingSessionId }); + const restartedSession = yield* startProviderSession(thread.session?.providerThreadId ?? null); + yield* bindSessionToThread(restartedSession); + return restartedSession.sessionId; + } + + const startedSession = yield* startProviderSession(); + yield* bindSessionToThread(startedSession); return startedSession.sessionId; }); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 84e2a936dbf3d6030bb6f09ad78f8a44495133ee..889a8903e0a612ed439de2bebb5ea5da906c0606 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -371,6 +371,12 @@ const make = Effect.gen(function* () { providerName: event.provider, providerSessionId: event.sessionId, providerThreadId, + ...(thread.session?.approvalPolicy !== undefined + ? { approvalPolicy: thread.session.approvalPolicy } + : {}), + ...(thread.session?.sandboxMode !== undefined + ? { sandboxMode: thread.session.sandboxMode } + : {}), activeTurnId, lastError, updatedAt: now, @@ -479,6 +485,12 @@ const make = Effect.gen(function* () { providerName: event.provider, providerSessionId: event.sessionId, providerThreadId, + ...(thread.session?.approvalPolicy !== undefined + ? { approvalPolicy: thread.session.approvalPolicy } + : {}), + ...(thread.session?.sandboxMode !== undefined + ? { sandboxMode: thread.session.sandboxMode } + : {}), activeTurnId: toTurnId(event.turnId) ?? null, lastError: event.message, updatedAt: now, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index cf69a8db844f31e7644b85347e20376afd411bad..cdf32a02f15741ea4c54f38dcf7e185a169e9406 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -142,6 +142,8 @@ export const OrchestrationSession = Schema.Struct({ providerName: Schema.NullOr(TrimmedNonEmptyString), providerSessionId: Schema.NullOr(ProviderSessionId), providerThreadId: Schema.NullOr(ProviderThreadId), + approvalPolicy: Schema.optional(ProviderApprovalPolicy), + sandboxMode: Schema.optional(ProviderSandboxMode), activeTurnId: Schema.NullOr(TurnId), lastError: Schema.NullOr(TrimmedNonEmptyString), updatedAt: IsoDateTime,