diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..3e12a2a3e7c53b1a4e220a68d005393a3a4ef91f --- /dev/null +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -0,0 +1,202 @@ +import type { + ProviderApprovalDecision, + ProviderEvent, + ProviderSendTurnInput, + ProviderSession, + ProviderSessionStartInput, + ProviderTurnStartResult, +} from "@t3tools/contracts"; +import { afterAll, assert, it, vi } from "@effect/vitest"; +import { assertFailure } from "@effect/vitest/utils"; + +import { Effect } from "effect"; + +import { CodexAppServerManager } from "../../codexAppServerManager.ts"; +import { ProviderAdapterValidationError } from "../Errors.ts"; +import { CodexAdapter } from "../Services/CodexAdapter.ts"; +import { makeCodexAdapterLive } from "./CodexAdapter.ts"; + +class FakeCodexManager extends CodexAppServerManager { + public startSessionImpl = vi.fn( + async (input: ProviderSessionStartInput): Promise => { + const now = new Date().toISOString(); + return { + sessionId: "sess-1", + provider: "codex", + status: "ready", + threadId: "thread-1", + cwd: input.cwd, + createdAt: now, + updatedAt: now, + }; + }, + ); + + public sendTurnImpl = vi.fn( + async (_input: ProviderSendTurnInput): Promise => ({ + threadId: "thread-1", + turnId: "turn-1", + }), + ); + + public interruptTurnImpl = vi.fn( + async (_sessionId: string, _turnId?: string): Promise => undefined, + ); + + public readThreadImpl = vi.fn(async (_sessionId: string) => ({ + threadId: "thread-1", + turns: [], + })); + + public rollbackThreadImpl = vi.fn(async (_sessionId: string, _numTurns: number) => ({ + threadId: "thread-1", + turns: [], + })); + + public respondToRequestImpl = vi.fn( + async ( + _sessionId: string, + _requestId: string, + _decision: ProviderApprovalDecision, + ): Promise => undefined, + ); + + public stopAllImpl = vi.fn(() => undefined); + + override startSession(input: ProviderSessionStartInput): Promise { + return this.startSessionImpl(input); + } + + override sendTurn(input: ProviderSendTurnInput): Promise { + return this.sendTurnImpl(input); + } + + override interruptTurn(sessionId: string, turnId?: string): Promise { + return this.interruptTurnImpl(sessionId, turnId); + } + + override readThread(sessionId: string) { + return this.readThreadImpl(sessionId); + } + + override rollbackThread(sessionId: string, numTurns: number) { + return this.rollbackThreadImpl(sessionId, numTurns); + } + + override respondToRequest( + sessionId: string, + requestId: string, + decision: ProviderApprovalDecision, + ): Promise { + return this.respondToRequestImpl(sessionId, requestId, decision); + } + + override stopSession(_sessionId: string): void {} + + override listSessions(): ProviderSession[] { + return []; + } + + override hasSession(_sessionId: string): boolean { + return false; + } + + override stopAll(): void { + this.stopAllImpl(); + } +} + +const validationManager = new FakeCodexManager(); +const validationLayer = it.layer(makeCodexAdapterLive({ manager: validationManager })); + +validationLayer("CodexAdapterLive validation", (it) => { + it.effect("returns validation error for non-codex provider on startSession", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const result = yield* adapter + .startSession({ + provider: "claudeCode", + }) + .pipe(Effect.result); + + assertFailure( + result, + new ProviderAdapterValidationError({ + provider: "codex", + operation: "startSession", + issue: "Expected provider 'codex' but received 'claudeCode'.", + }), + ); + assert.equal(validationManager.startSessionImpl.mock.calls.length, 0); + }), + ); +}); + +const sessionErrorManager = new FakeCodexManager(); +sessionErrorManager.sendTurnImpl.mockImplementation(async () => { + throw new Error("Unknown session: sess-missing"); +}); +const sessionErrorLayer = it.layer(makeCodexAdapterLive({ manager: sessionErrorManager })); + +sessionErrorLayer("CodexAdapterLive session errors", (it) => { + it.effect("maps unknown-session sendTurn errors to ProviderAdapterSessionNotFoundError", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const result = yield* adapter + .sendTurn({ + sessionId: "sess-missing", + input: "hello", + attachments: [], + }) + .pipe(Effect.result); + + assert.equal(result._tag, "Failure"); + if (result._tag !== "Failure") { + return; + } + + assert.equal(result.failure._tag, "ProviderAdapterSessionNotFoundError"); + if (result.failure._tag !== "ProviderAdapterSessionNotFoundError") { + return; + } + assert.equal(result.failure.provider, "codex"); + assert.equal(result.failure.sessionId, "sess-missing"); + assert.instanceOf(result.failure.cause, Error); + }), + ); +}); + +const lifecycleManager = new FakeCodexManager(); +const lifecycleLayer = it.layer(makeCodexAdapterLive({ manager: lifecycleManager })); + +lifecycleLayer("CodexAdapterLive lifecycle", (it) => { + it.effect("subscribes and unsubscribes from manager events", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const onEvent = vi.fn((_event: ProviderEvent) => undefined); + const unsubscribe = yield* adapter.subscribeToEvents(onEvent); + + const event: ProviderEvent = { + id: "evt-1", + kind: "notification", + provider: "codex", + sessionId: "sess-1", + createdAt: new Date().toISOString(), + method: "turn/started", + }; + + lifecycleManager.emit("event", event); + assert.equal(onEvent.mock.calls.length, 1); + assert.equal(onEvent.mock.calls[0]?.[0]?.method, "turn/started"); + + unsubscribe(); + lifecycleManager.emit("event", event); + assert.equal(onEvent.mock.calls.length, 1); + assert.equal(lifecycleManager.stopAllImpl.mock.calls.length, 0); + }), + ); +}); + +afterAll(() => { + assert.equal(lifecycleManager.stopAllImpl.mock.calls.length, 1); +}); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts new file mode 100644 index 0000000000000000000000000000000000000000..2a3307a9626a9e91afc636f949a49a9ec65c9880 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -0,0 +1,213 @@ +/** + * CodexAdapterLive - Scoped live implementation for the Codex provider adapter. + * + * Wraps `CodexAppServerManager` behind the `CodexAdapter` service contract and + * maps manager failures into the shared `ProviderAdapterError` algebra. + * + * @module CodexAdapterLive + */ +import type { ProviderEvent } from "@t3tools/contracts"; +import { Effect, Layer } from "effect"; + +import { + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterSessionClosedError, + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, + type ProviderAdapterError, +} from "../Errors.ts"; +import { CodexAdapter, type CodexAdapterShape } from "../Services/CodexAdapter.ts"; +import { CodexAppServerManager } from "../../codexAppServerManager.ts"; + +const PROVIDER = "codex" as const; + +export interface CodexAdapterLiveOptions { + readonly manager?: CodexAppServerManager; + readonly makeManager?: () => CodexAppServerManager; +} + +function toMessage(cause: unknown, fallback: string): string { + if (cause instanceof Error && cause.message.length > 0) { + return cause.message; + } + return fallback; +} + +function toSessionError( + sessionId: string, + cause: unknown, +): ProviderAdapterSessionNotFoundError | ProviderAdapterSessionClosedError | undefined { + const normalized = toMessage(cause, "").toLowerCase(); + if (normalized.includes("unknown session") || normalized.includes("unknown provider session")) { + return new ProviderAdapterSessionNotFoundError({ + provider: PROVIDER, + sessionId, + cause, + }); + } + if (normalized.includes("session is closed")) { + return new ProviderAdapterSessionClosedError({ + provider: PROVIDER, + sessionId, + cause, + }); + } + return undefined; +} + +function toRequestError(sessionId: string, method: string, cause: unknown): ProviderAdapterError { + const sessionError = toSessionError(sessionId, cause); + if (sessionError) { + return sessionError; + } + return new ProviderAdapterRequestError({ + provider: PROVIDER, + method, + detail: toMessage(cause, `${method} failed`), + cause, + }); +} + +const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => + Effect.gen(function* () { + const manager = yield* Effect.acquireRelease( + Effect.sync(() => { + if (options?.manager) { + return options.manager; + } + if (options?.makeManager) { + return options.makeManager(); + } + return new CodexAppServerManager(); + }), + (manager) => + Effect.sync(() => { + try { + manager.stopAll(); + } catch { + // Finalizers should never fail and block shutdown. + } + }), + ); + + const startSession: CodexAdapterShape["startSession"] = (input) => { + if (input.provider !== undefined && input.provider !== PROVIDER) { + return Effect.fail( + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, + }), + ); + } + + return Effect.tryPromise({ + try: () => manager.startSession(input), + catch: (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + sessionId: "pending", + detail: toMessage(cause, "Failed to start Codex adapter session."), + cause, + }), + }); + }; + + const sendTurn: CodexAdapterShape["sendTurn"] = (input) => + Effect.tryPromise({ + try: () => manager.sendTurn(input), + catch: (cause) => toRequestError(input.sessionId, "turn/start", cause), + }); + + const interruptTurn: CodexAdapterShape["interruptTurn"] = (sessionId, turnId) => + Effect.tryPromise({ + try: () => manager.interruptTurn(sessionId, turnId), + catch: (cause) => toRequestError(sessionId, "turn/interrupt", cause), + }); + + const readThread: CodexAdapterShape["readThread"] = (sessionId) => + Effect.tryPromise({ + try: () => manager.readThread(sessionId), + catch: (cause) => toRequestError(sessionId, "thread/read", cause), + }); + + const rollbackThread: CodexAdapterShape["rollbackThread"] = (sessionId, numTurns) => { + if (!Number.isInteger(numTurns) || numTurns < 1) { + return Effect.fail( + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must be an integer >= 1.", + }), + ); + } + + return Effect.tryPromise({ + try: () => manager.rollbackThread(sessionId, numTurns), + catch: (cause) => toRequestError(sessionId, "thread/rollback", cause), + }); + }; + + const respondToRequest: CodexAdapterShape["respondToRequest"] = ( + sessionId, + requestId, + decision, + ) => + Effect.tryPromise({ + try: () => manager.respondToRequest(sessionId, requestId, decision), + catch: (cause) => toRequestError(sessionId, "item/requestApproval/decision", cause), + }); + + const stopSession: CodexAdapterShape["stopSession"] = (sessionId) => + Effect.sync(() => { + manager.stopSession(sessionId); + }); + + const listSessions: CodexAdapterShape["listSessions"] = () => + Effect.sync(() => manager.listSessions()); + + const hasSession: CodexAdapterShape["hasSession"] = (sessionId) => + Effect.sync(() => manager.hasSession(sessionId)); + + const stopAll: CodexAdapterShape["stopAll"] = () => + Effect.sync(() => { + manager.stopAll(); + }); + + const subscribeToEvents: CodexAdapterShape["subscribeToEvents"] = (callback) => + Effect.sync(() => { + const listener = (event: ProviderEvent) => { + try { + callback(event); + } catch { + // Adapter subscribers must not destabilize provider event processing. + } + }; + manager.on("event", listener); + return () => { + manager.off("event", listener); + }; + }); + + return { + provider: PROVIDER, + startSession, + sendTurn, + interruptTurn, + readThread, + rollbackThread, + respondToRequest, + stopSession, + listSessions, + hasSession, + stopAll, + subscribeToEvents, + } satisfies CodexAdapterShape; + }); + +export const CodexAdapterLive = Layer.effect(CodexAdapter, makeCodexAdapter()); + +export function makeCodexAdapterLive(options?: CodexAdapterLiveOptions) { + return Layer.effect(CodexAdapter, makeCodexAdapter(options)); +} diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..8384a212823f0e9e84b32cb22f61a4a4e68461c2 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts @@ -0,0 +1,50 @@ +import type { ProviderKind } from "@t3tools/contracts"; +import { it, assert, vi } from "@effect/vitest"; +import { assertFailure } from "@effect/vitest/utils"; + +import { Effect, Layer } from "effect"; + +import { CodexAdapter, CodexAdapterShape } from "../Services/CodexAdapter.ts"; +import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; +import { makeProviderAdapterRegistryLive } from "./ProviderAdapterRegistry.ts"; +import { ProviderUnsupportedError } from "../Errors.ts"; + +const fakeCodexAdapter: CodexAdapterShape = { + provider: "codex", + startSession: vi.fn(), + sendTurn: vi.fn(), + interruptTurn: vi.fn(), + respondToRequest: vi.fn(), + stopSession: vi.fn(), + listSessions: vi.fn(), + hasSession: vi.fn(), + readThread: vi.fn(), + rollbackThread: vi.fn(), + stopAll: vi.fn(), + subscribeToEvents: vi.fn(), +}; + +const layer = it.layer( + makeProviderAdapterRegistryLive().pipe( + Layer.provide(Layer.succeed(CodexAdapter, fakeCodexAdapter)), + ), +); + +layer("ProviderAdapterRegistryLive", (it) => { + it("resolves a registered provider adapter", () => + Effect.gen(function* () { + const registry = yield* ProviderAdapterRegistry; + const adapter = yield* registry.getByProvider("codex"); + assert.equal(adapter, fakeCodexAdapter); + + const providers = yield* registry.listProviders(); + assert.deepEqual(providers, ["codex"]); + })); + + it("fails with ProviderUnsupportedError for unknown providers", () => + Effect.gen(function* () { + const registry = yield* ProviderAdapterRegistry; + const adapter = yield* registry.getByProvider("unknown" as ProviderKind).pipe(Effect.result); + assertFailure(adapter, new ProviderUnsupportedError({ provider: "unknown" })); + })); +}); diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts new file mode 100644 index 0000000000000000000000000000000000000000..ace64737ca35bb22ff8117f167b7e632c0ee98d3 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts @@ -0,0 +1,57 @@ +/** + * ProviderAdapterRegistryLive - In-memory provider adapter lookup layer. + * + * Binds provider kinds (codex/claudeCode/...) to concrete adapter services. + * This layer only performs adapter lookup; it does not route session-scoped + * calls or own provider lifecycle workflows. + * + * @module ProviderAdapterRegistryLive + */ +import type { ProviderKind } from "@t3tools/contracts"; +import { Effect, Layer } from "effect"; + +import { ProviderUnsupportedError, type ProviderAdapterError } from "../Errors.ts"; +import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; +import { + ProviderAdapterRegistry, + type ProviderAdapterRegistryShape, +} from "../Services/ProviderAdapterRegistry.ts"; +import { CodexAdapter } from "../Services/CodexAdapter.ts"; + +export interface ProviderAdapterRegistryLiveOptions { + readonly adapters?: ReadonlyArray>; +} + +const makeProviderAdapterRegistry = (options?: ProviderAdapterRegistryLiveOptions) => + Effect.gen(function* () { + const adapters = options?.adapters !== undefined ? options.adapters : [yield* CodexAdapter]; + const byProvider = new Map>(); + for (const adapter of adapters) { + byProvider.set(adapter.provider, adapter); + } + + const getByProvider: ProviderAdapterRegistryShape["getByProvider"] = (provider) => { + const adapter = byProvider.get(provider); + if (!adapter) { + return Effect.fail(new ProviderUnsupportedError({ provider })); + } + return Effect.succeed(adapter); + }; + + const listProviders: ProviderAdapterRegistryShape["listProviders"] = () => + Effect.sync(() => Array.from(byProvider.keys())); + + return { + getByProvider, + listProviders, + } satisfies ProviderAdapterRegistryShape; + }); + +export const ProviderAdapterRegistryLive = Layer.effect( + ProviderAdapterRegistry, + makeProviderAdapterRegistry(), +); + +export function makeProviderAdapterRegistryLive(options?: ProviderAdapterRegistryLiveOptions) { + return Layer.effect(ProviderAdapterRegistry, makeProviderAdapterRegistry(options)); +} diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..726c190620e11308be179313380191e898437e49 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -0,0 +1,427 @@ +import type { + ProviderApprovalDecision, + ProviderEvent, + ProviderGetCheckpointDiffInput, + ProviderGetCheckpointDiffResult, + ProviderListCheckpointsInput, + ProviderListCheckpointsResult, + ProviderRevertToCheckpointInput, + ProviderRevertToCheckpointResult, + ProviderSendTurnInput, + ProviderSession, + ProviderSessionStartInput, + ProviderTurnStartResult, +} from "@t3tools/contracts"; +import { providerSessionStartInputSchema } from "@t3tools/contracts"; +import { it, assert, vi } from "@effect/vitest"; +import { assertFailure } from "@effect/vitest/utils"; + +import { Effect, Layer } from "effect"; + +import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts"; +import { + ProviderAdapterSessionNotFoundError, + ProviderSessionNotFoundError, + ProviderValidationError, + ProviderUnsupportedError, + type ProviderAdapterError, +} from "../Errors.ts"; +import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; +import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; +import { ProviderService } from "../Services/ProviderService.ts"; +import { ProviderServiceLive } from "./ProviderService.ts"; +import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; + +function makeFakeCodexAdapter() { + const sessions = new Map(); + const eventListeners = new Set<(event: ProviderEvent) => void>(); + let nextSession = 1; + + const startSession = vi.fn((input: ProviderSessionStartInput) => + Effect.sync(() => { + const now = new Date().toISOString(); + const session: ProviderSession = { + sessionId: `sess-${nextSession}`, + provider: "codex", + status: "ready", + threadId: `thread-${nextSession}`, + cwd: input.cwd ?? process.cwd(), + createdAt: now, + updatedAt: now, + }; + nextSession += 1; + sessions.set(session.sessionId, session); + return session; + }), + ); + + const sendTurn = vi.fn( + ( + input: ProviderSendTurnInput, + ): Effect.Effect => { + if (!sessions.has(input.sessionId)) { + return Effect.fail( + new ProviderAdapterSessionNotFoundError({ + provider: "codex", + sessionId: input.sessionId, + }), + ); + } + + return Effect.succeed({ + threadId: "thread-1", + turnId: "turn-1", + }); + }, + ); + + const interruptTurn = vi.fn( + (_sessionId: string, _turnId?: string): Effect.Effect => + Effect.void, + ); + + const respondToRequest = vi.fn( + ( + _sessionId: string, + _requestId: string, + _decision: ProviderApprovalDecision, + ): Effect.Effect => Effect.void, + ); + + const stopSession = vi.fn( + (sessionId: string): Effect.Effect => + Effect.sync(() => { + sessions.delete(sessionId); + }), + ); + + const listSessions = vi.fn( + (): Effect.Effect> => + Effect.sync(() => Array.from(sessions.values())), + ); + + const hasSession = vi.fn( + (sessionId: string): Effect.Effect => Effect.succeed(sessions.has(sessionId)), + ); + + const readThread = vi.fn( + ( + _sessionId: string, + ): Effect.Effect<{ threadId: string; turns: readonly [] }, ProviderAdapterError> => + Effect.succeed({ threadId: "thread-1", turns: [] }), + ); + + const rollbackThread = vi.fn( + ( + _sessionId: string, + _numTurns: number, + ): Effect.Effect<{ threadId: string; turns: readonly [] }, ProviderAdapterError> => + Effect.succeed({ threadId: "thread-1", turns: [] }), + ); + + const stopAll = vi.fn( + (): Effect.Effect => + Effect.sync(() => { + sessions.clear(); + }), + ); + + const subscribeToEvents = vi.fn( + (callback: (event: ProviderEvent) => void): Effect.Effect<() => void, ProviderAdapterError> => + Effect.sync(() => { + eventListeners.add(callback); + return () => { + eventListeners.delete(callback); + }; + }), + ); + + const adapter: ProviderAdapterShape = { + provider: "codex", + startSession, + sendTurn, + interruptTurn, + respondToRequest, + stopSession, + listSessions, + hasSession, + readThread, + rollbackThread, + stopAll, + subscribeToEvents, + }; + + const emit = (event: ProviderEvent): void => { + for (const listener of eventListeners) { + listener(event); + } + }; + + return { + adapter, + emit, + startSession, + sendTurn, + interruptTurn, + respondToRequest, + stopSession, + listSessions, + hasSession, + readThread, + rollbackThread, + stopAll, + subscribeToEvents, + }; +} + +function makeCheckpointServiceDouble() { + const initializeForSession = vi.fn( + (_: { providerSessionId: string; cwd: string }) => Effect.void, + ); + const captureCurrentTurn = vi.fn( + (_: { providerSessionId: string; turnId?: string; status?: string }) => Effect.void, + ); + const listCheckpoints = vi.fn( + (_input: ProviderListCheckpointsInput): Effect.Effect => + Effect.succeed({ + threadId: "thread-1", + checkpoints: [ + { + id: "root", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + isCurrent: true, + }, + ], + }), + ); + const getCheckpointDiff = vi.fn( + (input: ProviderGetCheckpointDiffInput): Effect.Effect => + Effect.succeed({ + threadId: "thread-1", + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff: "diff --git a/a.ts b/a.ts", + }), + ); + const revertToCheckpoint = vi.fn( + (input: ProviderRevertToCheckpointInput): Effect.Effect => + Effect.succeed({ + threadId: "thread-1", + turnCount: input.turnCount, + messageCount: 0, + rolledBackTurns: 0, + checkpoints: [ + { + id: "root", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + isCurrent: input.turnCount === 0, + }, + ], + }), + ); + + const service: typeof CheckpointService.Service = { + initializeForSession, + captureCurrentTurn, + listCheckpoints, + getCheckpointDiff, + revertToCheckpoint, + }; + + return { + service, + initializeForSession, + captureCurrentTurn, + listCheckpoints, + getCheckpointDiff, + revertToCheckpoint, + }; +} + +function makeProviderServiceLayer() { + const codex = makeFakeCodexAdapter(); + const checkpoint = makeCheckpointServiceDouble(); + const registry: typeof ProviderAdapterRegistry.Service = { + getByProvider: (provider) => + provider === "codex" + ? Effect.succeed(codex.adapter) + : Effect.fail(new ProviderUnsupportedError({ provider })), + listProviders: () => Effect.succeed(["codex"]), + }; + + const layer = it.layer( + ProviderServiceLive.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(ProviderAdapterRegistry, registry), + ProviderSessionDirectoryLive, + Layer.succeed(CheckpointService, checkpoint.service), + ), + ), + ), + ); + + return { + codex, + checkpoint, + layer, + }; +} + +const routing = makeProviderServiceLayer(); +routing.layer("ProviderServiceLive routing", (it) => { + it.effect("routes provider operations and delegates checkpoint workflows", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + + const session = yield* provider.startSession({ + provider: "codex", + cwd: "/tmp/project", + }); + assert.equal(session.provider, "codex"); + assert.deepEqual(routing.checkpoint.initializeForSession.mock.calls, [ + [{ providerSessionId: session.sessionId, cwd: "/tmp/project" }], + ]); + + const sessions = yield* provider.listSessions(); + assert.equal(sessions.length, 1); + + yield* provider.sendTurn({ + sessionId: session.sessionId, + input: "hello", + attachments: [], + }); + assert.equal(routing.codex.sendTurn.mock.calls.length, 1); + + yield* provider.interruptTurn({ sessionId: session.sessionId }); + assert.deepEqual(routing.codex.interruptTurn.mock.calls, [[session.sessionId, undefined]]); + + yield* provider.respondToRequest({ + sessionId: session.sessionId, + requestId: "req-1", + decision: "accept", + }); + assert.deepEqual(routing.codex.respondToRequest.mock.calls, [ + [session.sessionId, "req-1", "accept"], + ]); + + yield* provider.listCheckpoints({ sessionId: session.sessionId }); + yield* provider.getCheckpointDiff({ + sessionId: session.sessionId, + fromTurnCount: 0, + toTurnCount: 0, + }); + yield* provider.revertToCheckpoint({ + sessionId: session.sessionId, + turnCount: 0, + }); + + assert.deepEqual(routing.checkpoint.listCheckpoints.mock.calls, [ + [{ sessionId: session.sessionId }], + ]); + assert.deepEqual(routing.checkpoint.getCheckpointDiff.mock.calls, [ + [{ sessionId: session.sessionId, fromTurnCount: 0, toTurnCount: 0 }], + ]); + assert.deepEqual(routing.checkpoint.revertToCheckpoint.mock.calls, [ + [{ sessionId: session.sessionId, turnCount: 0 }], + ]); + + yield* provider.stopSession({ sessionId: session.sessionId }); + const sendAfterStop = yield* Effect.result( + provider.sendTurn({ + sessionId: session.sessionId, + input: "after-stop", + attachments: [], + }), + ); + assertFailure( + sendAfterStop, + new ProviderSessionNotFoundError({ sessionId: session.sessionId }), + ); + }), + ); +}); + +const fanout = makeProviderServiceLayer(); +fanout.layer("ProviderServiceLive fanout", (it) => { + it.effect("fans out adapter events and captures turn completion checkpoints", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession({ + provider: "codex", + }); + + const onEvent = vi.fn((_event: ProviderEvent) => undefined); + const unsubscribe = yield* provider.subscribeToEvents(onEvent); + + const completedEvent: ProviderEvent = { + id: "evt-1", + kind: "notification", + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + method: "turn/completed", + threadId: "thread-1", + turnId: "turn-1", + payload: { + turn: { + status: "completed", + }, + }, + }; + + fanout.codex.emit(completedEvent); + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); + + assert.equal(onEvent.mock.calls.length, 1); + assert.equal(onEvent.mock.calls[0]?.[0]?.method, "turn/completed"); + assert.deepEqual(fanout.checkpoint.captureCurrentTurn.mock.calls, [ + [{ providerSessionId: session.sessionId, turnId: "turn-1", status: "completed" }], + ]); + + unsubscribe(); + fanout.codex.emit({ + ...completedEvent, + id: "evt-2", + }); + assert.equal(onEvent.mock.calls.length, 1); + }), + ); +}); + +const validation = makeProviderServiceLayer(); +validation.layer("ProviderServiceLive validation", (it) => { + it.effect("returns ProviderValidationError for invalid input payloads", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + + const parse = providerSessionStartInputSchema.safeParse({ + provider: "invalid-provider", + }); + assert.equal(parse.success, false); + if (parse.success) { + return; + } + + const failure = yield* Effect.result( + provider.startSession({ + provider: "invalid-provider", + } as never), + ); + + const cause = (parse.error as { cause?: unknown }).cause; + assertFailure( + failure, + new ProviderValidationError({ + operation: "ProviderService.startSession", + issue: parse.error.message, + ...(cause !== undefined ? { cause } : {}), + }), + ); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts new file mode 100644 index 0000000000000000000000000000000000000000..77d1d0f9afc0e07f46a0a2bded5b13e18d419cea --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -0,0 +1,339 @@ +/** + * ProviderServiceLive - Cross-provider orchestration layer. + * + * Routes validated transport/API calls to provider adapters through + * `ProviderAdapterRegistry` and `ProviderSessionDirectory`, and exposes a + * unified provider event stream for subscribers. + * + * It does not implement provider protocol details (adapter concern) and does + * not implement checkpoint persistence mechanics (checkpointing concern). + * + * @module ProviderServiceLive + */ +import { randomUUID } from "node:crypto"; + +import type { ProviderEvent } from "@t3tools/contracts"; +import { + providerGetCheckpointDiffInputSchema, + providerInterruptTurnInputSchema, + providerListCheckpointsInputSchema, + providerRespondToRequestInputSchema, + providerRevertToCheckpointInputSchema, + providerSendTurnInputSchema, + providerSessionStartInputSchema, + providerStopSessionInputSchema, +} from "@t3tools/contracts"; +import { Effect, Layer } from "effect"; + +import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts"; +import { ProviderValidationError } from "../Errors.ts"; +import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; +import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; +import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; + +function asObject(value: unknown): Record | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + return value as Record; +} + +function asString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function readTurnCompletionStatus(event: ProviderEvent): string | undefined { + const payload = asObject(event.payload); + const turn = asObject(payload?.turn); + return asString(turn?.status); +} + +function toCheckpointCaptureErrorEvent( + event: ProviderEvent, + error: { readonly message: string }, +): ProviderEvent { + return { + id: randomUUID(), + kind: "error", + provider: event.provider, + sessionId: event.sessionId, + createdAt: new Date().toISOString(), + method: "checkpoint/captureError", + message: error.message, + ...(event.threadId !== undefined ? { threadId: event.threadId } : {}), + ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), + }; +} + +function toValidationError( + operation: string, + issue: string, + cause?: unknown, +): ProviderValidationError { + return new ProviderValidationError({ + operation, + issue, + ...(cause !== undefined ? { cause } : {}), + }); +} + +const makeProviderService = Effect.gen(function* () { + const registry = yield* ProviderAdapterRegistry; + const directory = yield* ProviderSessionDirectory; + const checkpointService = yield* CheckpointService; + + const eventSubscribers = new Set<(event: ProviderEvent) => void>(); + const notifySubscribers = (event: ProviderEvent): void => { + for (const callback of eventSubscribers) { + try { + callback(event); + } catch { + // Subscriber callbacks must not destabilize provider event fanout. + } + } + }; + + const providers = yield* registry.listProviders(); + const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider)); + + yield* Effect.acquireRelease( + Effect.forEach(adapters, (adapter) => + adapter.subscribeToEvents((event) => { + notifySubscribers(event); + if (event.kind !== "notification" || event.method !== "turn/completed") { + return; + } + + const status = readTurnCompletionStatus(event); + const captureInput = { + providerSessionId: event.sessionId, + ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), + ...(status !== undefined ? { status } : {}), + }; + + Effect.runFork( + checkpointService.captureCurrentTurn(captureInput).pipe( + Effect.catch((error) => + Effect.sync(() => { + notifySubscribers(toCheckpointCaptureErrorEvent(event, error)); + }), + ), + ), + ); + }), + ), + (unsubscribeAll) => + Effect.forEach(unsubscribeAll, (unsubscribe) => + Effect.sync(() => { + unsubscribe(); + }), + ).pipe(Effect.asVoid), + ); + + const adapterForSession = (sessionId: string) => + directory + .getProvider(sessionId) + .pipe(Effect.flatMap((provider) => registry.getByProvider(provider))); + + const startSession: ProviderServiceShape["startSession"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerSessionStartInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.startSession", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + const adapter = yield* registry.getByProvider(input.provider); + const session = yield* adapter.startSession(input); + + if (session.provider !== adapter.provider) { + return yield* Effect.fail( + toValidationError( + "ProviderService.startSession", + `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`, + ), + ); + } + + yield* directory.upsert({ + sessionId: session.sessionId, + provider: session.provider, + ...(session.threadId !== undefined ? { threadId: session.threadId } : {}), + }); + + const checkpointCwd = session.cwd ?? input.cwd ?? process.cwd(); + yield* checkpointService.initializeForSession({ + providerSessionId: session.sessionId, + cwd: checkpointCwd, + }); + + return session; + }); + + const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerSendTurnInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError("ProviderService.sendTurn", parsed.error.message, parsed.error.cause), + ); + } + + const input = parsed.data; + const adapter = yield* adapterForSession(input.sessionId); + return yield* adapter.sendTurn(input); + }); + + const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerInterruptTurnInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.interruptTurn", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + const adapter = yield* adapterForSession(input.sessionId); + yield* adapter.interruptTurn(input.sessionId, input.turnId); + }); + + const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerRespondToRequestInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.respondToRequest", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + const adapter = yield* adapterForSession(input.sessionId); + yield* adapter.respondToRequest(input.sessionId, input.requestId, input.decision); + }); + + const stopSession: ProviderServiceShape["stopSession"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerStopSessionInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.stopSession", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + const adapter = yield* adapterForSession(input.sessionId); + yield* adapter.stopSession(input.sessionId); + yield* directory.remove(input.sessionId); + }); + + const listSessions: ProviderServiceShape["listSessions"] = () => + Effect.forEach(adapters, (adapter) => adapter.listSessions()).pipe( + Effect.map((sessionsByProvider) => sessionsByProvider.flatMap((sessions) => sessions)), + ); + + const listCheckpoints: ProviderServiceShape["listCheckpoints"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerListCheckpointsInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.listCheckpoints", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + yield* directory.getProvider(input.sessionId); + return yield* checkpointService.listCheckpoints(input); + }); + + const getCheckpointDiff: ProviderServiceShape["getCheckpointDiff"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerGetCheckpointDiffInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.getCheckpointDiff", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + yield* directory.getProvider(input.sessionId); + return yield* checkpointService.getCheckpointDiff(input); + }); + + const revertToCheckpoint: ProviderServiceShape["revertToCheckpoint"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerRevertToCheckpointInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* Effect.fail( + toValidationError( + "ProviderService.revertToCheckpoint", + parsed.error.message, + parsed.error.cause, + ), + ); + } + + const input = parsed.data; + yield* directory.getProvider(input.sessionId); + return yield* checkpointService.revertToCheckpoint(input); + }); + + const stopAll: ProviderServiceShape["stopAll"] = () => + Effect.gen(function* () { + yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid); + const sessionIds = yield* directory.listSessionIds(); + yield* Effect.forEach(sessionIds, (sessionId) => directory.remove(sessionId)).pipe( + Effect.asVoid, + ); + }); + + const subscribeToEvents: ProviderServiceShape["subscribeToEvents"] = (callback) => + Effect.sync(() => { + eventSubscribers.add(callback); + return () => { + eventSubscribers.delete(callback); + }; + }); + + return { + startSession, + sendTurn, + interruptTurn, + respondToRequest, + stopSession, + listSessions, + listCheckpoints, + getCheckpointDiff, + revertToCheckpoint, + stopAll, + subscribeToEvents, + } satisfies ProviderServiceShape; +}); + +export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..1780609d37054c4c4130772f4322b653fb798902 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -0,0 +1,60 @@ +import { it, assert } from "@effect/vitest"; +import { assertFailure, assertSome } from "@effect/vitest/utils"; +import { Effect } from "effect"; + +import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; +import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; + +const layer = it.layer(ProviderSessionDirectoryLive); + +layer("ProviderSessionDirectoryLive", (it) => { + it("upserts, reads, and removes session bindings", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + + yield* directory.upsert({ + sessionId: "sess-1", + provider: "codex", + threadId: "thread-1", + }); + + const provider = yield* directory.getProvider("sess-1"); + assert.equal(provider, "codex"); + const threadId = yield* directory.getThreadId("sess-1"); + assertSome(threadId, "thread-1"); + + yield* directory.upsert({ + sessionId: "sess-1", + provider: "codex", + threadId: "thread-2", + }); + const updatedThreadId = yield* directory.getThreadId("sess-1"); + assertSome(updatedThreadId, "thread-2"); + + const sessionIds = yield* directory.listSessionIds(); + assert.deepEqual(sessionIds, ["sess-1"]); + + yield* directory.remove("sess-1"); + const missingProvider = yield* directory.getProvider("sess-1").pipe(Effect.result); + assertFailure(missingProvider, new ProviderSessionNotFoundError({ sessionId: "sess-1" })); + })); + + it("fails upsert for empty session id", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const result = yield* Effect.result( + directory.upsert({ + sessionId: " ", + provider: "codex", + }), + ); + assertFailure( + result, + new ProviderValidationError({ + operation: "ProviderSessionDirectory.upsert", + issue: "sessionId must be a non-empty string.", + }), + ); + })); +}); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts new file mode 100644 index 0000000000000000000000000000000000000000..4ff2f04955906f61d037354cc846ef50044e0757 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -0,0 +1,93 @@ +/** + * ProviderSessionDirectoryLive - In-memory session ownership index layer. + * + * Tracks provider ownership and optional thread id by `sessionId` so + * `ProviderService` can route session-scoped API calls to the right adapter. + * This layer stores metadata only and does not call provider adapters directly. + * + * @module ProviderSessionDirectoryLive + */ +import { Effect, Layer, Option, Ref } from "effect"; + +import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; +import { + ProviderSessionDirectory, + type ProviderSessionBinding, + type ProviderSessionDirectoryShape, +} from "../Services/ProviderSessionDirectory.ts"; + +const makeProviderSessionDirectory = Effect.gen(function* () { + const bindingsRef = yield* Ref.make(new Map()); + + const upsert: ProviderSessionDirectoryShape["upsert"] = (binding) => { + const sessionId = binding.sessionId.trim(); + if (sessionId.length === 0) { + return Effect.fail( + new ProviderValidationError({ + operation: "ProviderSessionDirectory.upsert", + issue: "sessionId must be a non-empty string.", + }), + ); + } + + const normalized: ProviderSessionBinding = { + sessionId, + provider: binding.provider, + ...(binding.threadId !== undefined ? { threadId: binding.threadId } : {}), + }; + + return Ref.update(bindingsRef, (current) => { + const next = new Map(current); + next.set(sessionId, normalized); + return next; + }); + }; + + const getProvider: ProviderSessionDirectoryShape["getProvider"] = (sessionId) => + Ref.get(bindingsRef).pipe( + Effect.flatMap((bindings) => { + const binding = bindings.get(sessionId); + if (!binding) { + return Effect.fail(new ProviderSessionNotFoundError({ sessionId })); + } + return Effect.succeed(binding.provider); + }), + ); + + const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) => + Ref.get(bindingsRef).pipe( + Effect.flatMap((bindings) => { + const binding = bindings.get(sessionId); + if (!binding) { + return Effect.fail(new ProviderSessionNotFoundError({ sessionId })); + } + return Effect.succeed(Option.fromNullishOr(binding.threadId)); + }), + ); + + const remove: ProviderSessionDirectoryShape["remove"] = (sessionId) => + Ref.update(bindingsRef, (current) => { + if (!current.has(sessionId)) { + return current; + } + const next = new Map(current); + next.delete(sessionId); + return next; + }); + + const listSessionIds: ProviderSessionDirectoryShape["listSessionIds"] = () => + Ref.get(bindingsRef).pipe(Effect.map((bindings) => Array.from(bindings.keys()))); + + return { + upsert, + getProvider, + getThreadId, + remove, + listSessionIds, + } satisfies ProviderSessionDirectoryShape; +}); + +export const ProviderSessionDirectoryLive = Layer.effect( + ProviderSessionDirectory, + makeProviderSessionDirectory, +);