import fs from "node:fs";import os from "node:os";import path from "node:path"; import { OrchestrationReadModel, ProviderDriverKind, ProviderRuntimeEvent, ProviderSession, ProviderInstanceId,} from "@t3tools/contracts";import { ApprovalRequestId, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ProviderItemId, type ServerSettings, ThreadId, TurnId,} from "@t3tools/contracts";import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";import { afterEach, describe, expect, it } from "vitest"; 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 { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import { ServerConfig } from "../../config.ts";import { ServerSettingsService } from "../../serverSettings.ts";import * as NodeServices from "@effect/platform-node/NodeServices"; function makeTestServerSettingsLayer(overrides: Partial<ServerSettings> = {}) { return ServerSettingsService.layerTest(overrides);} const asProjectId = (value: string): ProjectId => ProjectId.make(value);const asItemId = (value: string): ProviderItemId => ProviderItemId.make(value);const asEventId = (value: string): EventId => EventId.make(value);const asMessageId = (value: string): MessageId => MessageId.make(value);const asThreadId = (value: string): ThreadId => ThreadId.make(value);const asTurnId = (value: string): TurnId => TurnId.make(value); type LegacyProviderRuntimeEvent = { readonly type: string; readonly eventId: EventId; readonly provider: ProviderRuntimeEvent["provider"]; readonly createdAt: string; readonly threadId: ThreadId; readonly turnId?: string | undefined; readonly itemId?: string | undefined; readonly requestId?: string | undefined; readonly payload?: unknown | undefined; readonly [key: string]: unknown;}; type LegacyTurnCompletedEvent = LegacyProviderRuntimeEvent & { readonly type: "turn.completed"; readonly payload?: undefined; readonly status: "completed" | "failed" | "interrupted" | "cancelled"; readonly errorMessage?: string | undefined;}; function isLegacyTurnCompletedEvent( event: LegacyProviderRuntimeEvent,): event is LegacyTurnCompletedEvent { return ( event.type === "turn.completed" && event.payload === undefined && typeof event.status === "string" );} function createProviderServiceHarness() { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); const runtimeSessions: ProviderSession[] = []; const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; const service: ProviderServiceShape = { startSession: () => unsupported(), sendTurn: () => unsupported(), interruptTurn: () => unsupported(), respondToRequest: () => unsupported(), respondToUserInput: () => unsupported(), stopSession: () => unsupported(), listSessions: () => Effect.succeed([...runtimeSessions]), getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), getInstanceInfo: (instanceId) => { const driverKind = ProviderDriverKind.make(String(instanceId)); return Effect.succeed({ instanceId, driverKind, displayName: undefined, enabled: true, continuationIdentity: { driverKind, continuationKey: `${driverKind}:instance:${instanceId}`, }, }); }, rollbackConversation: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, }; const setSession = (session: ProviderSession): void => { const existingIndex = runtimeSessions.findIndex((entry) => entry.threadId === session.threadId); if (existingIndex >= 0) { runtimeSessions[existingIndex] = session; return; } runtimeSessions.push(session); }; const normalizeLegacyEvent = (event: LegacyProviderRuntimeEvent): ProviderRuntimeEvent => { if (isLegacyTurnCompletedEvent(event)) { const normalized: Extract<ProviderRuntimeEvent, { type: "turn.completed" }> = { ...(event as Omit<Extract<ProviderRuntimeEvent, { type: "turn.completed" }>, "payload">), payload: { state: event.status, ...(typeof event.errorMessage === "string" ? { errorMessage: event.errorMessage } : {}), }, }; return normalized; } return event as ProviderRuntimeEvent; }; const emit = (event: LegacyProviderRuntimeEvent): void => { Effect.runSync(PubSub.publish(runtimeEventPubSub, normalizeLegacyEvent(event))); }; return { service, emit, setSession, };} type ProviderRuntimeTestReadModel = OrchestrationReadModel;type ProviderRuntimeTestThread = ProviderRuntimeTestReadModel["threads"][number];type ProviderRuntimeTestMessage = ProviderRuntimeTestThread["messages"][number];type ProviderRuntimeTestProposedPlan = ProviderRuntimeTestThread["proposedPlans"][number];type ProviderRuntimeTestActivity = ProviderRuntimeTestThread["activities"][number];type ProviderRuntimeTestCheckpoint = ProviderRuntimeTestThread["checkpoints"][number]; async function waitForThread( readModel: () => Promise<ProviderRuntimeTestReadModel>, predicate: (thread: ProviderRuntimeTestThread) => boolean, timeoutMs = 2000, threadId: ThreadId = asThreadId("thread-1"),) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<ProviderRuntimeTestThread> => { const snapshot = await readModel(); const thread = snapshot.threads.find((entry) => entry.id === threadId); if (thread && predicate(thread)) { return thread; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for thread state"); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll();} describe("ProviderRuntimeIngestion", () => { let runtime: ManagedRuntime.ManagedRuntime< OrchestrationEngineService | ProviderRuntimeIngestionService | ProjectionSnapshotQuery, unknown > | null = null; let scope: Scope.Closeable | null = null; const tempDirs: string[] = []; function makeTempDir(prefix: string): string { const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); tempDirs.push(dir); return dir; } afterEach(async () => { if (scope) { await Effect.runPromise(Scope.close(scope, Exit.void)); } scope = null; if (runtime) { await runtime.dispose(); } runtime = null; for (const dir of tempDirs.splice(0)) { fs.rmSync(dir, { recursive: true, force: true }); } }); async function createHarness(options?: { serverSettings?: Partial<ServerSettings> }) { const workspaceRoot = makeTempDir("t3-provider-project-"); fs.mkdirSync(path.join(workspaceRoot, ".git")); const provider = createProviderServiceHarness(); const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolverLive), Layer.provide(SqlitePersistenceMemory), ); const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe( Layer.provide(RepositoryIdentityResolverLive), Layer.provide(SqlitePersistenceMemory), ); const layer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(projectionSnapshotLayer), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), Layer.provideMerge(NodeServices.layer), ); runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery)); const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(ingestion.drain); const createdAt = new Date().toISOString(); await Effect.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-provider-project-create"), projectId: asProjectId("project-1"), title: "Provider Project", workspaceRoot, defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-create"), threadId: ThreadId.make("thread-1"), projectId: asProjectId("project-1"), title: "Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("cmd-session-seed"), threadId: ThreadId.make("thread-1"), session: { threadId: ThreadId.make("thread-1"), status: "ready", providerName: "codex", runtimeMode: "approval-required", activeTurnId: null, updatedAt: createdAt, lastError: null, }, createdAt, }), ); provider.setSession({ provider: ProviderDriverKind.make("codex"), status: "ready", runtimeMode: "approval-required", threadId: ThreadId.make("thread-1"), createdAt, updatedAt: createdAt, }); return { engine, readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), emit: provider.emit, setProviderSession: provider.setSession, drain, }; } it("maps turn started/completed events into thread session updates", async () => { const harness = await createHarness(); const now = new Date().toISOString(); harness.emit({ type: "turn.started", eventId: asEventId("evt-turn-started"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: now, turnId: asTurnId("turn-1"), }); await waitForThread( harness.readModel, (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-1", ); harness.emit({ type: "turn.completed", eventId: asEventId("evt-turn-completed"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), turnId: asTurnId("turn-1"), payload: { state: "failed", errorMessage: "turn failed", }, }); const thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "error" && entry.session?.activeTurnId === null && entry.session?.lastError === "turn failed", ); expect(thread.session?.status).toBe("error"); expect(thread.session?.lastError).toBe("turn failed"); }); it("applies provider session.state.changed transitions directly", async () => { const harness = await createHarness(); const waitingAt = new Date().toISOString(); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-waiting"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: waitingAt, payload: { state: "waiting", reason: "awaiting approval", }, }); let thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === null, ); expect(thread.session?.status).toBe("running"); expect(thread.session?.lastError).toBeNull(); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-error"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), payload: { state: "error", reason: "provider crashed", }, }); thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "error" && entry.session?.activeTurnId === null && entry.session?.lastError === "provider crashed", ); expect(thread.session?.status).toBe("error"); expect(thread.session?.lastError).toBe("provider crashed"); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-stopped"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), payload: { state: "stopped", }, }); thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "stopped" && entry.session?.activeTurnId === null && entry.session?.lastError === "provider crashed", ); expect(thread.session?.status).toBe("stopped"); expect(thread.session?.lastError).toBe("provider crashed"); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-ready"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), payload: { state: "ready", }, }); thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "ready" && entry.session?.activeTurnId === null && entry.session?.lastError === null, ); expect(thread.session?.status).toBe("ready"); expect(thread.session?.lastError).toBeNull(); }); it("does not clear active turn when session/thread started arrives mid-turn", async () => { const harness = await createHarness(); const now = new Date().toISOString(); harness.emit({ type: "turn.started", eventId: asEventId("evt-turn-started-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: now, threadId: asThreadId("thread-1"), turnId: asTurnId("turn-midturn-lifecycle"), }); await waitForThread( harness.readModel, (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-midturn-lifecycle", ); harness.emit({ type: "thread.started", eventId: asEventId("evt-thread-started-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: new Date().toISOString(), threadId: asThreadId("thread-1"), }); harness.emit({ type: "session.started", eventId: asEventId("evt-session-started-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: new Date().toISOString(), threadId: asThreadId("thread-1"), }); await harness.drain(); const midReadModel = await harness.readModel(); const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); expect(midThread?.session?.status).toBe("running"); expect(midThread?.session?.activeTurnId).toBe("turn-midturn-lifecycle"); harness.emit({ type: "turn.completed", eventId: asEventId("evt-turn-completed-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: new Date().toISOString(), threadId: asThreadId("thread-1"), turnId: asTurnId("turn-midturn-lifecycle"), status: "completed", }); await waitForThread( harness.readModel, (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, ); }); it("accepts claude turn lifecycle when seeded thread id is a synthetic placeholder", async () => {Lines 1–500 of 2969. Load more
import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { OrchestrationReadModel, ProviderDriverKind, ProviderRuntimeEvent, ProviderSession, ProviderInstanceId, } from "@t3tools/contracts"; import { ApprovalRequestId, CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, ProjectId, ProviderItemId, type ServerSettings, ThreadId, TurnId, } from "@t3tools/contracts"; import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect"; import { afterEach, describe, expect, it } from "vitest"; 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 { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; function makeTestServerSettingsLayer(overrides: Partial<ServerSettings> = {}) { return ServerSettingsService.layerTest(overrides); } const asProjectId = (value: string): ProjectId => ProjectId.make(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.make(value); const asEventId = (value: string): EventId => EventId.make(value); const asMessageId = (value: string): MessageId => MessageId.make(value); const asThreadId = (value: string): ThreadId => ThreadId.make(value); const asTurnId = (value: string): TurnId => TurnId.make(value); type LegacyProviderRuntimeEvent = { readonly type: string; readonly eventId: EventId; readonly provider: ProviderRuntimeEvent["provider"]; readonly createdAt: string; readonly threadId: ThreadId; readonly turnId?: string | undefined; readonly itemId?: string | undefined; readonly requestId?: string | undefined; readonly payload?: unknown | undefined; readonly [key: string]: unknown; }; type LegacyTurnCompletedEvent = LegacyProviderRuntimeEvent & { readonly type: "turn.completed"; readonly payload?: undefined; readonly status: "completed" | "failed" | "interrupted" | "cancelled"; readonly errorMessage?: string | undefined; }; function isLegacyTurnCompletedEvent( event: LegacyProviderRuntimeEvent, ): event is LegacyTurnCompletedEvent { return ( event.type === "turn.completed" && event.payload === undefined && typeof event.status === "string" ); } function createProviderServiceHarness() { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); const runtimeSessions: ProviderSession[] = []; const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; const service: ProviderServiceShape = { startSession: () => unsupported(), sendTurn: () => unsupported(), interruptTurn: () => unsupported(), respondToRequest: () => unsupported(), respondToUserInput: () => unsupported(), stopSession: () => unsupported(), listSessions: () => Effect.succeed([...runtimeSessions]), getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), getInstanceInfo: (instanceId) => { const driverKind = ProviderDriverKind.make(String(instanceId)); return Effect.succeed({ instanceId, driverKind, displayName: undefined, enabled: true, continuationIdentity: { driverKind, continuationKey: `${driverKind}:instance:${instanceId}`, }, }); }, rollbackConversation: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, }; const setSession = (session: ProviderSession): void => { const existingIndex = runtimeSessions.findIndex((entry) => entry.threadId === session.threadId); if (existingIndex >= 0) { runtimeSessions[existingIndex] = session; return; } runtimeSessions.push(session); }; const normalizeLegacyEvent = (event: LegacyProviderRuntimeEvent): ProviderRuntimeEvent => { if (isLegacyTurnCompletedEvent(event)) { const normalized: Extract<ProviderRuntimeEvent, { type: "turn.completed" }> = { ...(event as Omit<Extract<ProviderRuntimeEvent, { type: "turn.completed" }>, "payload">), payload: { state: event.status, ...(typeof event.errorMessage === "string" ? { errorMessage: event.errorMessage } : {}), }, }; return normalized; } return event as ProviderRuntimeEvent; }; const emit = (event: LegacyProviderRuntimeEvent): void => { Effect.runSync(PubSub.publish(runtimeEventPubSub, normalizeLegacyEvent(event))); }; return { service, emit, setSession, }; } type ProviderRuntimeTestReadModel = OrchestrationReadModel; type ProviderRuntimeTestThread = ProviderRuntimeTestReadModel["threads"][number]; type ProviderRuntimeTestMessage = ProviderRuntimeTestThread["messages"][number]; type ProviderRuntimeTestProposedPlan = ProviderRuntimeTestThread["proposedPlans"][number]; type ProviderRuntimeTestActivity = ProviderRuntimeTestThread["activities"][number]; type ProviderRuntimeTestCheckpoint = ProviderRuntimeTestThread["checkpoints"][number]; async function waitForThread( readModel: () => Promise<ProviderRuntimeTestReadModel>, predicate: (thread: ProviderRuntimeTestThread) => boolean, timeoutMs = 2000, threadId: ThreadId = asThreadId("thread-1"), ) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<ProviderRuntimeTestThread> => { const snapshot = await readModel(); const thread = snapshot.threads.find((entry) => entry.id === threadId); if (thread && predicate(thread)) { return thread; } if (Date.now() >= deadline) { throw new Error("Timed out waiting for thread state"); } await new Promise((resolve) => setTimeout(resolve, 10)); return poll(); }; return poll(); } describe("ProviderRuntimeIngestion", () => { let runtime: ManagedRuntime.ManagedRuntime< OrchestrationEngineService | ProviderRuntimeIngestionService | ProjectionSnapshotQuery, unknown > | null = null; let scope: Scope.Closeable | null = null; const tempDirs: string[] = []; function makeTempDir(prefix: string): string { const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); tempDirs.push(dir); return dir; } afterEach(async () => { if (scope) { await Effect.runPromise(Scope.close(scope, Exit.void)); } scope = null; if (runtime) { await runtime.dispose(); } runtime = null; for (const dir of tempDirs.splice(0)) { fs.rmSync(dir, { recursive: true, force: true }); } }); async function createHarness(options?: { serverSettings?: Partial<ServerSettings> }) { const workspaceRoot = makeTempDir("t3-provider-project-"); fs.mkdirSync(path.join(workspaceRoot, ".git")); const provider = createProviderServiceHarness(); const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolverLive), Layer.provide(SqlitePersistenceMemory), ); const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe( Layer.provide(RepositoryIdentityResolverLive), Layer.provide(SqlitePersistenceMemory), ); const layer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(projectionSnapshotLayer), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), Layer.provideMerge(NodeServices.layer), ); runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery)); const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(ingestion.drain); const createdAt = new Date().toISOString(); await Effect.runPromise( engine.dispatch({ type: "project.create", commandId: CommandId.make("cmd-provider-project-create"), projectId: asProjectId("project-1"), title: "Provider Project", workspaceRoot, defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, createdAt, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.create", commandId: CommandId.make("cmd-thread-create"), threadId: ThreadId.make("thread-1"), projectId: asProjectId("project-1"), title: "Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: "approval-required", branch: null, worktreePath: null, createdAt, }), ); await Effect.runPromise( engine.dispatch({ type: "thread.session.set", commandId: CommandId.make("cmd-session-seed"), threadId: ThreadId.make("thread-1"), session: { threadId: ThreadId.make("thread-1"), status: "ready", providerName: "codex", runtimeMode: "approval-required", activeTurnId: null, updatedAt: createdAt, lastError: null, }, createdAt, }), ); provider.setSession({ provider: ProviderDriverKind.make("codex"), status: "ready", runtimeMode: "approval-required", threadId: ThreadId.make("thread-1"), createdAt, updatedAt: createdAt, }); return { engine, readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), emit: provider.emit, setProviderSession: provider.setSession, drain, }; } it("maps turn started/completed events into thread session updates", async () => { const harness = await createHarness(); const now = new Date().toISOString(); harness.emit({ type: "turn.started", eventId: asEventId("evt-turn-started"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: now, turnId: asTurnId("turn-1"), }); await waitForThread( harness.readModel, (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-1", ); harness.emit({ type: "turn.completed", eventId: asEventId("evt-turn-completed"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), turnId: asTurnId("turn-1"), payload: { state: "failed", errorMessage: "turn failed", }, }); const thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "error" && entry.session?.activeTurnId === null && entry.session?.lastError === "turn failed", ); expect(thread.session?.status).toBe("error"); expect(thread.session?.lastError).toBe("turn failed"); }); it("applies provider session.state.changed transitions directly", async () => { const harness = await createHarness(); const waitingAt = new Date().toISOString(); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-waiting"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: waitingAt, payload: { state: "waiting", reason: "awaiting approval", }, }); let thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === null, ); expect(thread.session?.status).toBe("running"); expect(thread.session?.lastError).toBeNull(); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-error"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), payload: { state: "error", reason: "provider crashed", }, }); thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "error" && entry.session?.activeTurnId === null && entry.session?.lastError === "provider crashed", ); expect(thread.session?.status).toBe("error"); expect(thread.session?.lastError).toBe("provider crashed"); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-stopped"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), payload: { state: "stopped", }, }); thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "stopped" && entry.session?.activeTurnId === null && entry.session?.lastError === "provider crashed", ); expect(thread.session?.status).toBe("stopped"); expect(thread.session?.lastError).toBe("provider crashed"); harness.emit({ type: "session.state.changed", eventId: asEventId("evt-session-state-ready"), provider: ProviderDriverKind.make("codex"), threadId: asThreadId("thread-1"), createdAt: new Date().toISOString(), payload: { state: "ready", }, }); thread = await waitForThread( harness.readModel, (entry) => entry.session?.status === "ready" && entry.session?.activeTurnId === null && entry.session?.lastError === null, ); expect(thread.session?.status).toBe("ready"); expect(thread.session?.lastError).toBeNull(); }); it("does not clear active turn when session/thread started arrives mid-turn", async () => { const harness = await createHarness(); const now = new Date().toISOString(); harness.emit({ type: "turn.started", eventId: asEventId("evt-turn-started-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: now, threadId: asThreadId("thread-1"), turnId: asTurnId("turn-midturn-lifecycle"), }); await waitForThread( harness.readModel, (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-midturn-lifecycle", ); harness.emit({ type: "thread.started", eventId: asEventId("evt-thread-started-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: new Date().toISOString(), threadId: asThreadId("thread-1"), }); harness.emit({ type: "session.started", eventId: asEventId("evt-session-started-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: new Date().toISOString(), threadId: asThreadId("thread-1"), }); await harness.drain(); const midReadModel = await harness.readModel(); const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); expect(midThread?.session?.status).toBe("running"); expect(midThread?.session?.activeTurnId).toBe("turn-midturn-lifecycle"); harness.emit({ type: "turn.completed", eventId: asEventId("evt-turn-completed-midturn-lifecycle"), provider: ProviderDriverKind.make("codex"), createdAt: new Date().toISOString(), threadId: asThreadId("thread-1"), turnId: asTurnId("turn-midturn-lifecycle"), status: "completed", }); await waitForThread( harness.readModel, (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, ); }); it("accepts claude turn lifecycle when seeded thread id is a synthetic placeholder", async () => {