import type { ProviderRuntimeEvent } from "@t3tools/contracts";import { ApprovalRequestId, CommandId, MessageId, ProjectId, ProviderSessionId, ProviderThreadId, ProviderTurnId, ThreadId, TurnId,} from "@t3tools/contracts";import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";import { afterEach, describe, expect, it, vi } 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 { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.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, timeoutMs = 2000): Promise<void> { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<void> => { if (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; 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 now = new Date().toISOString(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); const startSession = vi.fn((_: unknown, __: unknown) => Effect.succeed({ sessionId: asSessionId("sess-1"), provider: "codex" as const, status: "ready" as const, threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), createdAt: now, updatedAt: now, }), ); const sendTurn = vi.fn((_: unknown) => Effect.succeed({ threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asProviderTurnId("provider-turn-1"), }), ); const interruptTurn = vi.fn((_: unknown) => Effect.void); const respondToRequest = vi.fn((_: unknown) => Effect.void); const stopSession = vi.fn((_: unknown) => Effect.void); 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"], stopSession: stopSession as ProviderServiceShape["stopSession"], listSessions: () => Effect.succeed([]), rollbackConversation: () => unsupported(), stopAll: () => Effect.void, 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)), ); 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))); await Effect.runPromise(Effect.sleep("10 millis")); 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", branch: null, worktreePath: null, createdAt: now, }), ); return { engine, startSession, sendTurn, interruptTurn, respondToRequest, stopSession, }; } 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: [], }, 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", }); 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"); }); it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set"), threadId: ThreadId.makeUnsafe("thread-1"), session: { threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", providerSessionId: asSessionId("sess-1"), providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), activeTurnId: asTurnId("turn-1"), lastError: null, updatedAt: now, }, createdAt: now, }), ); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.interrupt", commandId: CommandId.makeUnsafe("cmd-turn-interrupt"), threadId: ThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), createdAt: now, }), ); await waitFor(() => harness.interruptTurn.mock.calls.length === 1); expect(harness.interruptTurn.mock.calls[0]?.[0]).toEqual({ sessionId: "sess-1", }); }); it("reacts to thread.approval.respond by forwarding provider approval response", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set-for-approval"), threadId: ThreadId.makeUnsafe("thread-1"), session: { threadId: ThreadId.makeUnsafe("thread-1"), status: "running", providerName: "codex", providerSessionId: asSessionId("sess-1"), providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), activeTurnId: null, lastError: null, updatedAt: now, }, createdAt: now, }), ); await Effect.runPromise( harness.engine.dispatch({ type: "thread.approval.respond", commandId: CommandId.makeUnsafe("cmd-approval-respond"), threadId: ThreadId.makeUnsafe("thread-1"), requestId: asApprovalRequestId("approval-request-1"), decision: "accept", createdAt: now, }), ); await waitFor(() => harness.respondToRequest.mock.calls.length === 1); expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ sessionId: "sess-1", requestId: "approval-request-1", decision: "accept", }); }); it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => { const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.set", commandId: CommandId.makeUnsafe("cmd-session-set-for-stop"), 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"), activeTurnId: null, lastError: null, updatedAt: now, }, createdAt: now, }), ); await Effect.runPromise( harness.engine.dispatch({ type: "thread.session.stop", commandId: CommandId.makeUnsafe("cmd-session-stop"), threadId: ThreadId.makeUnsafe("thread-1"), createdAt: now, }), ); await waitFor(() => harness.stopSession.mock.calls.length === 1); const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.session).not.toBeNull(); expect(thread?.session?.status).toBe("stopped"); expect(thread?.session?.providerSessionId).toBeNull(); expect(thread?.session?.providerThreadId).toBeNull(); expect(thread?.session?.activeTurnId).toBeNull(); });});
import type { ProviderRuntimeEvent } from "@t3tools/contracts";
import {
ApprovalRequestId,
CommandId,
MessageId,
ProjectId,
ProviderSessionId,
ProviderThreadId,
ProviderTurnId,
ThreadId,
TurnId,
} from "@t3tools/contracts";
import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";
import { afterEach, describe, expect, it, vi } 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 { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";
import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";
const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.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, timeoutMs = 2000): Promise<void> {
const deadline = Date.now() + timeoutMs;
const poll = async (): Promise<void> => {
if (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;
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 now = new Date().toISOString();
const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
const startSession = vi.fn((_: unknown, __: unknown) =>
Effect.succeed({
sessionId: asSessionId("sess-1"),
provider: "codex" as const,
status: "ready" as const,
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
createdAt: now,
updatedAt: now,
}),
);
const sendTurn = vi.fn((_: unknown) =>
Effect.succeed({
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asProviderTurnId("provider-turn-1"),
}),
);
const interruptTurn = vi.fn((_: unknown) => Effect.void);
const respondToRequest = vi.fn((_: unknown) => Effect.void);
const stopSession = vi.fn((_: unknown) => Effect.void);
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"],
stopSession: stopSession as ProviderServiceShape["stopSession"],
listSessions: () => Effect.succeed([]),
rollbackConversation: () => unsupported(),
stopAll: () => Effect.void,
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)),
);
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)));
await Effect.runPromise(Effect.sleep("10 millis"));
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",
branch: null,
worktreePath: null,
createdAt: now,
}),
);
return {
engine,
startSession,
sendTurn,
interruptTurn,
respondToRequest,
stopSession,
};
}
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: [],
},
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",
});
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");
});
it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => {
const harness = await createHarness();
const now = new Date().toISOString();
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.makeUnsafe("cmd-session-set"),
threadId: ThreadId.makeUnsafe("thread-1"),
session: {
threadId: ThreadId.makeUnsafe("thread-1"),
status: "running",
providerName: "codex",
providerSessionId: asSessionId("sess-1"),
providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
activeTurnId: asTurnId("turn-1"),
lastError: null,
updatedAt: now,
},
createdAt: now,
}),
);
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.interrupt",
commandId: CommandId.makeUnsafe("cmd-turn-interrupt"),
threadId: ThreadId.makeUnsafe("thread-1"),
turnId: asTurnId("turn-1"),
createdAt: now,
}),
);
await waitFor(() => harness.interruptTurn.mock.calls.length === 1);
expect(harness.interruptTurn.mock.calls[0]?.[0]).toEqual({
sessionId: "sess-1",
});
});
it("reacts to thread.approval.respond by forwarding provider approval response", async () => {
const harness = await createHarness();
const now = new Date().toISOString();
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.makeUnsafe("cmd-session-set-for-approval"),
threadId: ThreadId.makeUnsafe("thread-1"),
session: {
threadId: ThreadId.makeUnsafe("thread-1"),
status: "running",
providerName: "codex",
providerSessionId: asSessionId("sess-1"),
providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
activeTurnId: null,
lastError: null,
updatedAt: now,
},
createdAt: now,
}),
);
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.approval.respond",
commandId: CommandId.makeUnsafe("cmd-approval-respond"),
threadId: ThreadId.makeUnsafe("thread-1"),
requestId: asApprovalRequestId("approval-request-1"),
decision: "accept",
createdAt: now,
}),
);
await waitFor(() => harness.respondToRequest.mock.calls.length === 1);
expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({
sessionId: "sess-1",
requestId: "approval-request-1",
decision: "accept",
});
});
it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => {
const harness = await createHarness();
const now = new Date().toISOString();
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.makeUnsafe("cmd-session-set-for-stop"),
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"),
activeTurnId: null,
lastError: null,
updatedAt: now,
},
createdAt: now,
}),
);
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.stop",
commandId: CommandId.makeUnsafe("cmd-session-stop"),
threadId: ThreadId.makeUnsafe("thread-1"),
createdAt: now,
}),
);
await waitFor(() => harness.stopSession.mock.calls.length === 1);
const readModel = await Effect.runPromise(harness.engine.getReadModel());
const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
expect(thread?.session).not.toBeNull();
expect(thread?.session?.status).toBe("stopped");
expect(thread?.session?.providerSessionId).toBeNull();
expect(thread?.session?.providerThreadId).toBeNull();
expect(thread?.session?.activeTurnId).toBeNull();
});
});