import fs from "node:fs";import os from "node:os";import path from "node:path"; import type { ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts";import { ApprovalRequestId, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ThreadId, TurnId,} from "@t3tools/contracts";import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";import { afterEach, describe, expect, it, vi } from "vitest"; import { ServerConfig } from "../../config.ts";import { TextGenerationError } from "../../git/Errors.ts";import { ProviderAdapterRequestError } from "../../provider/Errors.ts";import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";import { ProviderService, type ProviderServiceShape,} from "../../provider/Services/ProviderService.ts";import { GitCore, type GitCoreShape } from "../../git/Services/GitCore.ts";import { TextGeneration, type TextGenerationShape } from "../../git/Services/TextGeneration.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";import * as NodeServices from "@effect/platform-node/NodeServices"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); async function waitFor( predicate: () => boolean | Promise<boolean>, timeoutMs = 2000,): Promise<void> { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<void> => { if (await predicate()) { return; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for expectation."); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll();} describe("ProviderCommandReactor", () => { let runtime: ManagedRuntime.ManagedRuntime< OrchestrationEngineService | ProviderCommandReactor, unknown > | null = null; let scope: Scope.Closeable | null = null; const createdStateDirs = new Set<string>(); afterEach(async () => { if (scope) { await Effect.runPromise(Scope.close(scope, Exit.void)); } scope = null; if (runtime) { await runtime.dispose(); } runtime = null; for (const stateDir of createdStateDirs) { fs.rmSync(stateDir, { recursive: true, force: true }); } createdStateDirs.clear(); }); async function createHarness(input?: { readonly stateDir?: string }) { const now = new Date().toISOString(); const stateDir = input?.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-")); createdStateDirs.add(stateDir); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); let nextSessionIndex = 1; const runtimeSessions: Array<ProviderSession> = []; const startSession = vi.fn((_: unknown, input: unknown) => { const sessionIndex = nextSessionIndex++; const provider = typeof input === "object" && input !== null && "provider" in input && input.provider === "codex" ? input.provider : "codex"; const resumeCursor = typeof input === "object" && input !== null && "resumeCursor" in input ? input.resumeCursor : undefined; const model = typeof input === "object" && input !== null && "model" in input && typeof input.model === "string" ? input.model : undefined; const threadId = typeof input === "object" && input !== null && "threadId" in input && typeof input.threadId === "string" ? ThreadId.makeUnsafe(input.threadId) : ThreadId.makeUnsafe(`thread-${sessionIndex}`); const session: ProviderSession = { provider, status: "ready" as const, runtimeMode: typeof input === "object" && input !== null && "runtimeMode" in input && (input.runtimeMode === "approval-required" || input.runtimeMode === "full-access") ? input.runtimeMode : "full-access", ...(model !== undefined ? { model } : {}), threadId, resumeCursor: resumeCursor ?? { opaque: `cursor-${sessionIndex}` }, createdAt: now, updatedAt: now, }; runtimeSessions.push(session); return Effect.succeed(session); }); const sendTurn = vi.fn((_: unknown) => Effect.succeed({ threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), }), ); const interruptTurn = vi.fn((_: unknown) => Effect.void); const respondToRequest = vi.fn<ProviderServiceShape["respondToRequest"]>(() => Effect.void); const respondToUserInput = vi.fn<ProviderServiceShape["respondToUserInput"]>(() => Effect.void); const stopSession = vi.fn((input: unknown) => Effect.sync(() => { const threadId = typeof input === "object" && input !== null && "threadId" in input ? (input as { threadId?: ThreadId }).threadId : undefined; if (!threadId) { return; } const index = runtimeSessions.findIndex((session) => session.threadId === threadId); if (index >= 0) { runtimeSessions.splice(index, 1); } }), ); const renameBranch = vi.fn((input: unknown) => Effect.succeed({ branch: typeof input === "object" && input !== null && "newBranch" in input && typeof input.newBranch === "string" ? input.newBranch : "renamed-branch", }), ); const generateBranchName = vi.fn(() => Effect.fail( new TextGenerationError({ operation: "generateBranchName", detail: "disabled in test harness", }), ), ); const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; const service: ProviderServiceShape = { startSession: startSession as ProviderServiceShape["startSession"], sendTurn: sendTurn as ProviderServiceShape["sendTurn"], interruptTurn: interruptTurn as ProviderServiceShape["interruptTurn"], respondToRequest: respondToRequest as ProviderServiceShape["respondToRequest"], respondToUserInput: respondToUserInput as ProviderServiceShape["respondToUserInput"], stopSession: stopSession as ProviderServiceShape["stopSession"], listSessions: () => Effect.succeed(runtimeSessions), getCapabilities: (provider) => Effect.succeed({ sessionModelSwitch: provider === "codex" ? "in-session" : "in-session", }), rollbackConversation: () => unsupported(), streamEvents: Stream.fromPubSub(runtimeEventPubSub), }; const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), ); const layer = ProviderCommandReactorLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(Layer.succeed(ProviderService, service)), Layer.provideMerge(Layer.succeed(GitCore, { renameBranch } as unknown as GitCoreShape)), Layer.provideMerge( Layer.succeed(TextGeneration, { generateBranchName } as unknown as TextGenerationShape), ), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)), Layer.provideMerge(NodeServices.layer), ); const runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start.pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(reactor.drain); await Effect.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: asProjectId("project-1"), title: "Provider Project", workspaceRoot: "/tmp/provider-project", defaultModel: "gpt-5-codex", createdAt: now, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.makeUnsafe("cmd-thread-create"), threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt: now, }), ); return { engine, startSession, sendTurn, interruptTurn, respondToRequest, respondToUserInput, stopSession, renameBranch, generateBranchName, stateDir, drain, }; } it("reacts to thread.turn.start by ensuring session and sending provider turn", 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-1"), threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-1"), role: "user", text: "hello reactor", attachments: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); expect(harness.startSession.mock.calls[0]?.[0]).toEqual(ThreadId.makeUnsafe("thread-1")); expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({ cwd: "/tmp/provider-project", model: "gpt-5-codex", runtimeMode: "approval-required", }); const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.session?.threadId).toBe("thread-1"); expect(thread?.session?.runtimeMode).toBe("approval-required"); }); it("forwards codex model options through session start and turn send", 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-fast"), threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-fast"), role: "user", text: "hello fast mode", attachments: [], }, provider: "codex", model: "gpt-5.3-codex", modelOptions: { codex: { reasoningEffort: "high", fastMode: true, }, }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({ model: "gpt-5.3-codex", modelOptions: { codex: { reasoningEffort: "high", fastMode: true, }, }, }); expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ threadId: ThreadId.makeUnsafe("thread-1"), model: "gpt-5.3-codex", modelOptions: { codex: { reasoningEffort: "high", fastMode: true, }, }, }); }); it("forwards plan interaction mode to the provider turn request", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.interaction-mode.set", commandId: CommandId.makeUnsafe("cmd-interaction-mode-set-plan"), threadId: ThreadId.makeUnsafe("thread-1"), interactionMode: "plan", createdAt: now, }), ); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", commandId: CommandId.makeUnsafe("cmd-turn-start-plan"), threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-plan"), role: "user", text: "plan this change", attachments: [], }, interactionMode: "plan", runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(() => harness.sendTurn.mock.calls.length === 1); expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ threadId: ThreadId.makeUnsafe("thread-1"), interactionMode: "plan", }); }); 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: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", 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: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", 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 is updated on the thread", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.runtime-mode.set", commandId: CommandId.makeUnsafe("cmd-runtime-mode-set-initial-full-access"), threadId: ThreadId.makeUnsafe("thread-1"), runtimeMode: "full-access", createdAt: now, }), ); 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: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "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.runtime-mode.set", commandId: CommandId.makeUnsafe("cmd-runtime-mode-set-1"), threadId: ThreadId.makeUnsafe("thread-1"), runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(async () => { const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), ); return thread?.runtimeMode === "approval-required"; }); await waitFor(() => harness.startSession.mock.calls.length === 2); 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: [], },Lines 1–500 of 851. Load more
import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import type { ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts"; import { ApprovalRequestId, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ThreadId, TurnId, } from "@t3tools/contracts"; import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; import { ServerConfig } from "../../config.ts"; import { TextGenerationError } from "../../git/Errors.ts"; import { ProviderAdapterRequestError } from "../../provider/Errors.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { ProviderService, type ProviderServiceShape, } from "../../provider/Services/ProviderService.ts"; import { GitCore, type GitCoreShape } from "../../git/Services/GitCore.ts"; import { TextGeneration, type TextGenerationShape } from "../../git/Services/TextGeneration.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); async function waitFor( predicate: () => boolean | Promise<boolean>, timeoutMs = 2000, ): Promise<void> { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<void> => { if (await predicate()) { return; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for expectation."); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll(); } describe("ProviderCommandReactor", () => { let runtime: ManagedRuntime.ManagedRuntime< OrchestrationEngineService | ProviderCommandReactor, unknown > | null = null; let scope: Scope.Closeable | null = null; const createdStateDirs = new Set<string>(); afterEach(async () => { if (scope) { await Effect.runPromise(Scope.close(scope, Exit.void)); } scope = null; if (runtime) { await runtime.dispose(); } runtime = null; for (const stateDir of createdStateDirs) { fs.rmSync(stateDir, { recursive: true, force: true }); } createdStateDirs.clear(); }); async function createHarness(input?: { readonly stateDir?: string }) { const now = new Date().toISOString(); const stateDir = input?.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-")); createdStateDirs.add(stateDir); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); let nextSessionIndex = 1; const runtimeSessions: Array<ProviderSession> = []; const startSession = vi.fn((_: unknown, input: unknown) => { const sessionIndex = nextSessionIndex++; const provider = typeof input === "object" && input !== null && "provider" in input && input.provider === "codex" ? input.provider : "codex"; const resumeCursor = typeof input === "object" && input !== null && "resumeCursor" in input ? input.resumeCursor : undefined; const model = typeof input === "object" && input !== null && "model" in input && typeof input.model === "string" ? input.model : undefined; const threadId = typeof input === "object" && input !== null && "threadId" in input && typeof input.threadId === "string" ? ThreadId.makeUnsafe(input.threadId) : ThreadId.makeUnsafe(`thread-${sessionIndex}`); const session: ProviderSession = { provider, status: "ready" as const, runtimeMode: typeof input === "object" && input !== null && "runtimeMode" in input && (input.runtimeMode === "approval-required" || input.runtimeMode === "full-access") ? input.runtimeMode : "full-access", ...(model !== undefined ? { model } : {}), threadId, resumeCursor: resumeCursor ?? { opaque: `cursor-${sessionIndex}` }, createdAt: now, updatedAt: now, }; runtimeSessions.push(session); return Effect.succeed(session); }); const sendTurn = vi.fn((_: unknown) => Effect.succeed({ threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), }), ); const interruptTurn = vi.fn((_: unknown) => Effect.void); const respondToRequest = vi.fn<ProviderServiceShape["respondToRequest"]>(() => Effect.void); const respondToUserInput = vi.fn<ProviderServiceShape["respondToUserInput"]>(() => Effect.void); const stopSession = vi.fn((input: unknown) => Effect.sync(() => { const threadId = typeof input === "object" && input !== null && "threadId" in input ? (input as { threadId?: ThreadId }).threadId : undefined; if (!threadId) { return; } const index = runtimeSessions.findIndex((session) => session.threadId === threadId); if (index >= 0) { runtimeSessions.splice(index, 1); } }), ); const renameBranch = vi.fn((input: unknown) => Effect.succeed({ branch: typeof input === "object" && input !== null && "newBranch" in input && typeof input.newBranch === "string" ? input.newBranch : "renamed-branch", }), ); const generateBranchName = vi.fn(() => Effect.fail( new TextGenerationError({ operation: "generateBranchName", detail: "disabled in test harness", }), ), ); const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; const service: ProviderServiceShape = { startSession: startSession as ProviderServiceShape["startSession"], sendTurn: sendTurn as ProviderServiceShape["sendTurn"], interruptTurn: interruptTurn as ProviderServiceShape["interruptTurn"], respondToRequest: respondToRequest as ProviderServiceShape["respondToRequest"], respondToUserInput: respondToUserInput as ProviderServiceShape["respondToUserInput"], stopSession: stopSession as ProviderServiceShape["stopSession"], listSessions: () => Effect.succeed(runtimeSessions), getCapabilities: (provider) => Effect.succeed({ sessionModelSwitch: provider === "codex" ? "in-session" : "in-session", }), rollbackConversation: () => unsupported(), streamEvents: Stream.fromPubSub(runtimeEventPubSub), }; const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), ); const layer = ProviderCommandReactorLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(Layer.succeed(ProviderService, service)), Layer.provideMerge(Layer.succeed(GitCore, { renameBranch } as unknown as GitCoreShape)), Layer.provideMerge( Layer.succeed(TextGeneration, { generateBranchName } as unknown as TextGenerationShape), ), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)), Layer.provideMerge(NodeServices.layer), ); const runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start.pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(reactor.drain); await Effect.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: asProjectId("project-1"), title: "Provider Project", workspaceRoot: "/tmp/provider-project", defaultModel: "gpt-5-codex", createdAt: now, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.makeUnsafe("cmd-thread-create"), threadId: ThreadId.makeUnsafe("thread-1"), projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt: now, }), ); return { engine, startSession, sendTurn, interruptTurn, respondToRequest, respondToUserInput, stopSession, renameBranch, generateBranchName, stateDir, drain, }; } it("reacts to thread.turn.start by ensuring session and sending provider turn", 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-1"), threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-1"), role: "user", text: "hello reactor", attachments: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); expect(harness.startSession.mock.calls[0]?.[0]).toEqual(ThreadId.makeUnsafe("thread-1")); expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({ cwd: "/tmp/provider-project", model: "gpt-5-codex", runtimeMode: "approval-required", }); const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.session?.threadId).toBe("thread-1"); expect(thread?.session?.runtimeMode).toBe("approval-required"); }); it("forwards codex model options through session start and turn send", 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-fast"), threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-fast"), role: "user", text: "hello fast mode", attachments: [], }, provider: "codex", model: "gpt-5.3-codex", modelOptions: { codex: { reasoningEffort: "high", fastMode: true, }, }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({ model: "gpt-5.3-codex", modelOptions: { codex: { reasoningEffort: "high", fastMode: true, }, }, }); expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ threadId: ThreadId.makeUnsafe("thread-1"), model: "gpt-5.3-codex", modelOptions: { codex: { reasoningEffort: "high", fastMode: true, }, }, }); }); it("forwards plan interaction mode to the provider turn request", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.interaction-mode.set", commandId: CommandId.makeUnsafe("cmd-interaction-mode-set-plan"), threadId: ThreadId.makeUnsafe("thread-1"), interactionMode: "plan", createdAt: now, }), ); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", commandId: CommandId.makeUnsafe("cmd-turn-start-plan"), threadId: ThreadId.makeUnsafe("thread-1"), message: { messageId: asMessageId("user-message-plan"), role: "user", text: "plan this change", attachments: [], }, interactionMode: "plan", runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(() => harness.sendTurn.mock.calls.length === 1); expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ threadId: ThreadId.makeUnsafe("thread-1"), interactionMode: "plan", }); }); 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: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", 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: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", 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 is updated on the thread", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.runtime-mode.set", commandId: CommandId.makeUnsafe("cmd-runtime-mode-set-initial-full-access"), threadId: ThreadId.makeUnsafe("thread-1"), runtimeMode: "full-access", createdAt: now, }), ); 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: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "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.runtime-mode.set", commandId: CommandId.makeUnsafe("cmd-runtime-mode-set-1"), threadId: ThreadId.makeUnsafe("thread-1"), runtimeMode: "approval-required", createdAt: now, }), ); await waitFor(async () => { const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), ); return thread?.runtimeMode === "approval-required"; }); await waitFor(() => harness.startSession.mock.calls.length === 2); 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: [], },