import type { ProviderRuntimeEvent } from "@t3tools/contracts";import {  CommandId,  EventId,  MessageId,  ProjectId,  ProviderItemId,  ProviderSessionId,  ProviderThreadId,  ProviderTurnId,  ThreadId,} 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 { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts";import {  OrchestrationEngineService,  type OrchestrationEngineShape,} from "../Services/OrchestrationEngine.ts";import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);const asProviderThreadId = (value: string): ProviderThreadId =>  ProviderThreadId.makeUnsafe(value);const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);const asEventId = (value: string): EventId => EventId.makeUnsafe(value);const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); function createProviderServiceHarness() {  const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());   const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;  const service: ProviderServiceShape = {    startSession: () => unsupported(),    sendTurn: () => unsupported(),    interruptTurn: () => unsupported(),    respondToRequest: () => unsupported(),    stopSession: () => unsupported(),    listSessions: () => Effect.succeed([]),    rollbackConversation: () => unsupported(),    stopAll: () => Effect.void,    streamEvents: Stream.fromPubSub(runtimeEventPubSub),  };   const emit = (event: ProviderRuntimeEvent): void => {    Effect.runSync(PubSub.publish(runtimeEventPubSub, event));  };   return {    service,    emit,  };} async function waitForThread(  engine: OrchestrationEngineShape,  predicate: (thread: {    session: { status: string; activeTurnId: string | null; lastError: string | null } | null;    messages: ReadonlyArray<{ id: string; text: string; streaming: boolean }>;    activities: ReadonlyArray<{ kind: string }>;  }) => boolean,  timeoutMs = 2000,) {  const deadline = Date.now() + timeoutMs;  const poll = async (): Promise<{    session: { status: string; activeTurnId: string | null; lastError: string | null } | null;    messages: ReadonlyArray<{ id: string; text: string; streaming: boolean }>;    activities: ReadonlyArray<{ kind: string }>;  }> => {    const readModel = await Effect.runPromise(engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    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,    unknown  > | null = null;  let scope: Scope.Closeable | null = null;   afterEach(async () => {    if (scope) {      await Effect.runPromise(Scope.close(scope, Exit.void));    }    scope = null;    if (runtime) {      await runtime.dispose();    }    runtime = null;  });   async function createHarness() {    const provider = createProviderServiceHarness();    const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(SqlitePersistenceMemory),    );    const layer = ProviderRuntimeIngestionLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),    );    runtime = ManagedRuntime.make(layer);    const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService));    scope = await Effect.runPromise(Scope.make("sequential"));    await Effect.runPromise(ingestion.start.pipe(Scope.provide(scope)));    await Effect.runPromise(Effect.sleep("10 millis"));     const createdAt = new Date().toISOString();    await Effect.runPromise(      engine.dispatch({        type: "project.create",        commandId: CommandId.makeUnsafe("cmd-provider-project-create"),        projectId: asProjectId("project-1"),        title: "Provider Project",        workspaceRoot: "/tmp/provider-project",        defaultModel: "gpt-5-codex",        createdAt,      }),    );    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",        branch: null,        worktreePath: null,        createdAt,      }),    );    await Effect.runPromise(      engine.dispatch({        type: "thread.session.set",        commandId: CommandId.makeUnsafe("cmd-session-seed"),        threadId: ThreadId.makeUnsafe("thread-1"),        session: {          threadId: ThreadId.makeUnsafe("thread-1"),          status: "ready",          providerName: "codex",          providerSessionId: asSessionId("sess-1"),          providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),          approvalPolicy: "on-request",          sandboxMode: "workspace-write",          activeTurnId: null,          updatedAt: createdAt,          lastError: null,        },        createdAt,      }),    );     return {      engine,      emit: provider.emit,    };  }   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: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-1"),    });     await waitForThread(      harness.engine,      (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-1",    );     harness.emit({      type: "turn.completed",      eventId: asEventId("evt-turn-completed"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      turnId: asProviderTurnId("turn-1"),      status: "failed",      errorMessage: "turn failed",    });     const thread = await waitForThread(      harness.engine,      (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("ignores auxiliary turn completions from a different provider thread", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     harness.emit({      type: "turn.started",      eventId: asEventId("evt-turn-started-primary"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      threadId: asProviderThreadId("provider-thread-1"),      turnId: asProviderTurnId("turn-primary"),    });     await waitForThread(      harness.engine,      (thread) =>        thread.session?.status === "running" && thread.session?.activeTurnId === "turn-primary",    );     harness.emit({      type: "turn.completed",      eventId: asEventId("evt-turn-completed-aux"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      threadId: asProviderThreadId("provider-thread-aux"),      turnId: asProviderTurnId("turn-aux"),      status: "completed",    });     await Effect.runPromise(Effect.sleep("40 millis"));    const midReadModel = await Effect.runPromise(harness.engine.getReadModel());    const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(midThread?.session?.status).toBe("running");    expect(midThread?.session?.activeTurnId).toBe("turn-primary");     harness.emit({      type: "turn.completed",      eventId: asEventId("evt-turn-completed-primary"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      threadId: asProviderThreadId("provider-thread-1"),      turnId: asProviderTurnId("turn-primary"),      status: "completed",    });     await waitForThread(      harness.engine,      (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,    );  });   it("ignores non-active turn completion when runtime omits thread id", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     harness.emit({      type: "turn.started",      eventId: asEventId("evt-turn-started-guarded"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-guarded-main"),    });     await waitForThread(      harness.engine,      (thread) =>        thread.session?.status === "running" &&        thread.session?.activeTurnId === "turn-guarded-main",    );     harness.emit({      type: "turn.completed",      eventId: asEventId("evt-turn-completed-guarded-other"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      turnId: asProviderTurnId("turn-guarded-other"),      status: "completed",    });     await Effect.runPromise(Effect.sleep("40 millis"));    const midReadModel = await Effect.runPromise(harness.engine.getReadModel());    const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(midThread?.session?.status).toBe("running");    expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main");     harness.emit({      type: "turn.completed",      eventId: asEventId("evt-turn-completed-guarded-main"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: new Date().toISOString(),      turnId: asProviderTurnId("turn-guarded-main"),      status: "completed",    });     await waitForThread(      harness.engine,      (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,    );  });   it("maps message delta/completed into finalized assistant messages", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     harness.emit({      type: "message.delta",      eventId: asEventId("evt-message-delta-1"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-2"),      itemId: asItemId("item-1"),      delta: "hello",    });    harness.emit({      type: "message.delta",      eventId: asEventId("evt-message-delta-2"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-2"),      itemId: asItemId("item-1"),      delta: " world",    });    harness.emit({      type: "message.completed",      eventId: asEventId("evt-message-completed"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-2"),      itemId: asItemId("item-1"),    });     const thread = await waitForThread(harness.engine, (entry) =>      entry.messages.some((message) => message.id === "assistant:item-1" && !message.streaming),    );    const message = thread.messages.find((entry) => entry.id === "assistant:item-1");    expect(message?.text).toBe("hello world");    expect(message?.streaming).toBe(false);  });   it("buffers assistant deltas by default until completion", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     harness.emit({      type: "turn.started",      eventId: asEventId("evt-turn-started-buffered"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-buffered"),    });    await waitForThread(      harness.engine,      (thread) =>        thread.session?.status === "running" && thread.session?.activeTurnId === "turn-buffered",    );     harness.emit({      type: "message.delta",      eventId: asEventId("evt-message-delta-buffered"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-buffered"),      itemId: asItemId("item-buffered"),      delta: "buffer me",    });     await Effect.runPromise(Effect.sleep("30 millis"));    const midReadModel = await Effect.runPromise(harness.engine.getReadModel());    const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(midThread?.messages.some((message) => message.id === "assistant:item-buffered")).toBe(      false,    );     harness.emit({      type: "message.completed",      eventId: asEventId("evt-message-completed-buffered"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-buffered"),      itemId: asItemId("item-buffered"),    });     const thread = await waitForThread(      harness.engine,      (entry) =>        entry.messages.some((message) => message.id === "assistant:item-buffered" && !message.streaming),    );    const message = thread.messages.find((entry) => entry.id === "assistant:item-buffered");    expect(message?.text).toBe("buffer me");    expect(message?.streaming).toBe(false);  });   it("streams assistant deltas when thread.turn.start requests streaming mode", 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-streaming-mode"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("message-streaming-mode"),          role: "user",          text: "stream please",          attachments: [],        },        assistantDeliveryMode: "streaming",        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );    await Effect.runPromise(Effect.sleep("30 millis"));     harness.emit({      type: "turn.started",      eventId: asEventId("evt-turn-started-streaming-mode"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-streaming-mode"),    });    await waitForThread(      harness.engine,      (thread) =>        thread.session?.status === "running" &&        thread.session?.activeTurnId === "turn-streaming-mode",    );     harness.emit({      type: "message.delta",      eventId: asEventId("evt-message-delta-streaming-mode"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-streaming-mode"),      itemId: asItemId("item-streaming-mode"),      delta: "hello live",    });     const liveThread = await waitForThread(      harness.engine,      (entry) =>        entry.messages.some(          (message) =>            message.id === "assistant:item-streaming-mode" &&            message.streaming &&            message.text === "hello live",        ),    );    const liveMessage = liveThread.messages.find(      (entry) => entry.id === "assistant:item-streaming-mode",    );    expect(liveMessage?.streaming).toBe(true);     harness.emit({      type: "message.completed",      eventId: asEventId("evt-message-completed-streaming-mode"),      provider: "codex",      sessionId: asSessionId("sess-1"),      createdAt: now,      turnId: asProviderTurnId("turn-streaming-mode"),      itemId: asItemId("item-streaming-mode"),

Lines 1–500 of 751. Load more