From d4b6536a502e34e473cfa4f5d6b54be0f483cc1e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 23 Feb 2026 11:14:56 -0800 Subject: [PATCH] runtime ingestion --- .../TestProviderAdapter.integration.ts | 11 +- .../Layers/ProviderRuntimeIngestion.test.ts | 264 ++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 326 ++++++++++++ .../Services/ProviderRuntimeIngestion.ts | 17 + apps/server/src/wsServer.test.ts | 123 ++++- apps/server/src/wsServer.ts | 474 +++--------------- 6 files changed, 814 insertions(+), 401 deletions(-) create mode 100644 apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts create mode 100644 apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts create mode 100644 apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts index b9828f881c1344882ba7248fb4978d2928bb1324..e121aed10679537fbcfbf9dde07bf97fb154bf5d 100644 --- a/apps/server/integration/TestProviderAdapter.integration.ts +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -131,7 +131,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { } const assistantDeltas: string[] = []; - let hasCompletedEvent = false; + const deferredTurnCompletedEvents: ProviderRuntimeEvent[] = []; for (const fixtureEvent of response.events) { const rawEvent: Record = { ...(fixtureEvent as Record), @@ -152,7 +152,8 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { assistantDeltas.push(runtimeEvent.delta); } if (runtimeEvent.type === "turn.completed") { - hasCompletedEvent = true; + deferredTurnCompletedEvents.push(runtimeEvent); + continue; } yield* emit(runtimeEvent); @@ -182,7 +183,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { turns: [...state.snapshot.turns, nextTurn], }; - if (!hasCompletedEvent) { + if (deferredTurnCompletedEvents.length === 0) { yield* emit({ type: "turn.completed", eventId: randomUUID(), @@ -193,6 +194,10 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { turnId, status: "completed", }); + } else { + for (const completedEvent of deferredTurnCompletedEvents) { + yield* emit(completedEvent); + } } return { diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..1bbd471a13f8e47706069ee1f96b12489ca425d2 --- /dev/null +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -0,0 +1,264 @@ +import type { ProviderRuntimeEvent } 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 { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { + ProviderService, + type ProviderServiceShape, +} from "../../provider/Services/ProviderService.ts"; +import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; +import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts"; +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "../Services/OrchestrationEngine.ts"; +import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; + +function createProviderServiceHarness() { + const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); + const getCheckpointDiff = vi.fn( + (input: { sessionId: string; fromTurnCount: number; toTurnCount: number }) => + Effect.succeed({ + threadId: "thread-1", + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff: "", + }), + ); + + 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([]), + listCheckpoints: () => unsupported(), + getCheckpointDiff, + revertToCheckpoint: () => unsupported(), + stopAll: () => Effect.void, + streamEvents: Stream.fromPubSub(runtimeEventPubSub), + }; + + const emit = (event: ProviderRuntimeEvent): void => { + Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); + }; + + return { + service, + emit, + getCheckpointDiff, + }; +} + +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 }>; + }) => 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 }>; + }> => { + const readModel = await Effect.runPromise(engine.getReadModel()); + const thread = readModel.threads.find((entry) => entry.id === "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(OrchestrationEventStoreLive), + 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: "thread.create", + commandId: "cmd-thread-create", + threadId: "thread-1", + projectId: "project-1", + title: "Thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + await Effect.runPromise( + engine.dispatch({ + type: "thread.session", + commandId: "cmd-session-seed", + threadId: "thread-1", + session: { + sessionId: "sess-1", + provider: "codex", + status: "ready", + threadId: "thread-1", + activeTurnId: null, + createdAt, + 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: "evt-turn-started", + provider: "codex", + sessionId: "sess-1", + createdAt: now, + turnId: "turn-1", + }); + + await waitForThread( + harness.engine, + (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-1", + ); + + harness.emit({ + type: "turn.completed", + eventId: "evt-turn-completed", + provider: "codex", + sessionId: "sess-1", + createdAt: new Date().toISOString(), + turnId: "turn-1", + status: "failed", + errorMessage: "turn failed", + }); + + const thread = await waitForThread( + harness.engine, + (entry) => + entry.session?.status === "ready" && + entry.session?.activeTurnId === null && + entry.session?.lastError === "turn failed", + ); + expect(thread.session?.status).toBe("ready"); + expect(thread.session?.lastError).toBe("turn failed"); + }); + + 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: "evt-message-delta-1", + provider: "codex", + sessionId: "sess-1", + createdAt: now, + turnId: "turn-2", + itemId: "item-1", + delta: "hello", + }); + harness.emit({ + type: "message.delta", + eventId: "evt-message-delta-2", + provider: "codex", + sessionId: "sess-1", + createdAt: now, + turnId: "turn-2", + itemId: "item-1", + delta: " world", + }); + harness.emit({ + type: "message.completed", + eventId: "evt-message-completed", + provider: "codex", + sessionId: "sess-1", + createdAt: now, + turnId: "turn-2", + itemId: "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("maps runtime.error into errored session state", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + harness.emit({ + type: "runtime.error", + eventId: "evt-runtime-error", + provider: "codex", + sessionId: "sess-1", + createdAt: now, + turnId: "turn-3", + message: "runtime exploded", + }); + + const thread = await waitForThread( + harness.engine, + (entry) => + entry.session?.status === "error" && + entry.session?.activeTurnId === "turn-3" && + entry.session?.lastError === "runtime exploded", + ); + expect(thread.session?.status).toBe("error"); + expect(thread.session?.lastError).toBe("runtime exploded"); + }); +}); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts new file mode 100644 index 0000000000000000000000000000000000000000..e33fba3f31b1413bbbc42bf3a4d00d58d1f77658 --- /dev/null +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -0,0 +1,326 @@ +import type { ProviderRuntimeEvent } from "@t3tools/contracts"; +import { Effect, Layer, Queue, Stream } from "effect"; + +import { + parseTurnDiffFilesFromUnifiedDiff, + type TurnDiffFileSummary, +} from "../../checkpointing/Diffs.ts"; +import { createLogger } from "../../logger.ts"; +import { ProviderService } from "../../provider/Services/ProviderService.ts"; +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { + ProviderRuntimeIngestionService, + type ProviderRuntimeIngestionShape, +} from "../Services/ProviderRuntimeIngestion.ts"; + +const providerTurnKey = (sessionId: string, turnId: string) => `${sessionId}:${turnId}`; + +type ActivityTone = "thinking" | "tool" | "info" | "error"; + +interface ThreadActivityInput { + readonly id: string; + readonly createdAt: string; + readonly label: string; + readonly detail?: string; + readonly tone: ActivityTone; + readonly turnId?: string; + readonly requestId?: string; + readonly requestKind?: "command" | "file-change"; +} + +function truncateDetail(value: string, limit = 180): string { + return value.length > limit ? `${value.slice(0, limit - 1)}...` : value; +} + +function runtimeEventToActivities(event: ProviderRuntimeEvent): ReadonlyArray { + switch (event.type) { + case "approval.requested": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: + event.requestKind === "command" + ? "Command approval requested" + : "File-change approval requested", + ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), + tone: "tool", + ...(event.turnId ? { turnId: event.turnId } : {}), + requestId: event.requestId, + requestKind: event.requestKind, + }, + ]; + case "approval.resolved": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: "Approval resolved", + tone: "info", + ...(event.turnId ? { turnId: event.turnId } : {}), + requestId: event.requestId, + ...(event.requestKind ? { requestKind: event.requestKind } : {}), + }, + ]; + case "runtime.error": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: "Runtime error", + detail: truncateDetail(event.message), + tone: "error", + ...(event.turnId ? { turnId: event.turnId } : {}), + }, + ]; + case "checkpoint.captured": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: "Checkpoint captured", + tone: "info", + ...(event.turnId ? { turnId: event.turnId } : {}), + }, + ]; + case "tool.completed": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: `${event.title} complete`, + ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), + tone: "tool", + ...(event.turnId ? { turnId: event.turnId } : {}), + }, + ]; + default: + return []; + } +} + +const make = Effect.gen(function* () { + const orchestrationEngine = yield* OrchestrationEngineService; + const providerService = yield* ProviderService; + const logger = createLogger("provider-runtime-ingestion"); + + const turnMessageIdsByTurnKey = new Map>(); + const latestMessageIdByTurnKey = new Map(); + + const rememberAssistantMessageId = (sessionId: string, turnId: string, messageId: string) => { + const key = providerTurnKey(sessionId, turnId); + const existingIds = turnMessageIdsByTurnKey.get(key); + if (existingIds) { + existingIds.add(messageId); + } else { + turnMessageIdsByTurnKey.set(key, new Set([messageId])); + } + latestMessageIdByTurnKey.set(key, messageId); + }; + + const getAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => { + return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set(); + }; + + const clearAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => { + turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId)); + }; + + const getLatestAssistantMessageIdForTurn = (sessionId: string, turnId: string) => { + return latestMessageIdByTurnKey.get(providerTurnKey(sessionId, turnId)); + }; + + const clearTurnStateForSession = (sessionId: string) => { + const prefix = `${sessionId}:`; + for (const key of turnMessageIdsByTurnKey.keys()) { + if (key.startsWith(prefix)) { + turnMessageIdsByTurnKey.delete(key); + } + } + for (const key of latestMessageIdByTurnKey.keys()) { + if (key.startsWith(prefix)) { + latestMessageIdByTurnKey.delete(key); + } + } + }; + + const processEvent = (event: ProviderRuntimeEvent) => + Effect.gen(function* () { + const readModel = yield* orchestrationEngine.getReadModel(); + const thread = readModel.threads.find( + (entry) => entry.session?.sessionId === event.sessionId, + ); + if (!thread) return; + + const now = event.createdAt; + if (event.type === "turn.started" || event.type === "turn.completed") { + yield* orchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), + threadId: thread.id, + session: { + sessionId: event.sessionId, + provider: event.provider, + status: event.type === "turn.started" ? "running" : "ready", + threadId: thread.id, + activeTurnId: event.type === "turn.started" ? event.turnId : null, + createdAt: thread.session?.createdAt ?? now, + updatedAt: now, + lastError: + event.type === "turn.completed" && event.status === "failed" + ? (event.errorMessage ?? null) + : null, + }, + createdAt: now, + }); + } + + if (event.type === "message.delta" && event.delta.length > 0) { + const assistantMessageId = `assistant:${event.itemId ?? event.turnId ?? event.sessionId}`; + if (event.turnId) { + rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId); + } + yield* orchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: event.delta, + streaming: true, + createdAt: now, + }); + } + + if (event.type === "message.completed") { + const assistantMessageId = `assistant:${event.itemId}`; + if (event.turnId) { + rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId); + } + yield* orchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: "", + streaming: false, + createdAt: now, + }); + } + + if (event.type === "turn.completed" && event.turnId) { + const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, event.turnId); + yield* Effect.forEach(assistantMessageIds, (assistantMessageId) => + orchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: "", + streaming: false, + createdAt: now, + }), + ).pipe(Effect.asVoid); + clearAssistantMessageIdsForTurn(event.sessionId, event.turnId); + } + + if (event.type === "session.exited") { + clearTurnStateForSession(event.sessionId); + } + + if (event.type === "checkpoint.captured" && event.turnId) { + const files: ReadonlyArray = yield* providerService + .getCheckpointDiff({ + sessionId: event.sessionId, + fromTurnCount: Math.max(0, event.turnCount - 1), + toTurnCount: event.turnCount, + }) + .pipe( + Effect.map((result) => parseTurnDiffFilesFromUnifiedDiff(result.diff)), + Effect.catch((error) => + Effect.sync(() => { + logger.warn("failed to derive checkpoint file summary", { + sessionId: event.sessionId, + turnId: event.turnId, + turnCount: event.turnCount, + error: + error instanceof Error + ? error.message + : typeof error === "string" + ? error + : "unknown error", + }); + }).pipe(Effect.as([] as ReadonlyArray)), + ), + ); + const assistantMessageId = + getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ?? + `assistant:${event.turnId}`; + yield* orchestrationEngine.dispatch({ + type: "thread.turnDiff.complete", + commandId: crypto.randomUUID(), + threadId: thread.id, + turnId: event.turnId, + completedAt: now, + ...(event.status ? { status: event.status } : {}), + files, + assistantMessageId, + checkpointTurnCount: event.turnCount, + createdAt: now, + }); + } + + if (event.type === "runtime.error") { + yield* orchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), + threadId: thread.id, + session: { + sessionId: event.sessionId, + provider: event.provider, + status: "error", + threadId: thread.id, + activeTurnId: event.turnId ?? null, + createdAt: thread.session?.createdAt ?? now, + updatedAt: now, + lastError: event.message, + }, + createdAt: now, + }); + } + + const activities = runtimeEventToActivities(event); + yield* Effect.forEach(activities, (activity) => + orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: crypto.randomUUID(), + threadId: thread.id, + activity, + createdAt: activity.createdAt, + }), + ).pipe(Effect.asVoid); + }); + + const start: ProviderRuntimeIngestionShape["start"] = Effect.gen(function* () { + const providerEventQueue = yield* Queue.unbounded(); + + yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid)); + + yield* Effect.forkScoped( + Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEvent))), + ); + yield* Effect.forkScoped( + Stream.runForEach(providerService.streamEvents, (event) => + Queue.offer(providerEventQueue, event).pipe(Effect.asVoid), + ), + ); + }); + + return { + start, + } satisfies ProviderRuntimeIngestionShape; +}); + +export const ProviderRuntimeIngestionLive = Layer.effect(ProviderRuntimeIngestionService, make); diff --git a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts new file mode 100644 index 0000000000000000000000000000000000000000..ba45bac8578ab272051ca87501633037ef707544 --- /dev/null +++ b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts @@ -0,0 +1,17 @@ +import { ServiceMap } from "effect"; +import type { Effect, Scope } from "effect"; + +export interface ProviderRuntimeIngestionShape { + /** + * Start ingesting provider runtime events into orchestration commands. + * + * The returned effect must be run in a scope so all worker fibers can be + * finalized on shutdown. + */ + readonly start: Effect.Effect; +} + +export class ProviderRuntimeIngestionService extends ServiceMap.Service< + ProviderRuntimeIngestionService, + ProviderRuntimeIngestionShape +>()("orchestration/ProviderRuntimeIngestion") {} diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index b00dbbe9e95d80ac73a72a93fe4d856f3dc9f6ed..23599b77016e9c5dcf7c1b514dc8888aa6a03987 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -3,15 +3,18 @@ import os from "node:os"; import path from "node:path"; import { EventEmitter } from "node:events"; -import { Effect, Layer, ManagedRuntime } from "effect"; +import { Effect, Layer, ManagedRuntime, PubSub, Stream } from "effect"; import { describe, expect, it, afterEach, vi } from "vitest"; import { createServer } from "./wsServer"; import WebSocket from "ws"; import { DEFAULT_TERMINAL_ID, + ORCHESTRATION_WS_CHANNELS, + ORCHESTRATION_WS_METHODS, WS_CHANNELS, WS_METHODS, + type ProviderRuntimeEvent, type KeybindingsConfig, type ResolvedKeybindingsConfig, type WsPush, @@ -33,6 +36,7 @@ import { ProjectRepositoryLive } from "./persistence/Layers/Projects"; import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./persistence/Layers/Sqlite"; import { NodeServices } from "@effect/platform-node"; import { SqlClient } from "effect/unstable/sql"; +import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; interface PendingMessages { queue: unknown[]; @@ -200,6 +204,28 @@ async function sendRequest(ws: WebSocket, method: string, params?: unknown): Pro } } +async function waitForPush( + ws: WebSocket, + channel: string, + predicate?: (push: WsPush) => boolean, + maxMessages = 120, +): Promise { + const take = async (remaining: number): Promise => { + if (remaining <= 0) { + throw new Error(`Timed out waiting for push on ${channel}`); + } + const message = (await waitForMessage(ws)) as WsPush; + if (message.type !== "push" || message.channel !== channel) { + return take(remaining - 1); + } + if (!predicate || predicate(message)) { + return message; + } + return take(remaining - 1); + }; + return take(maxMessages); +} + function compileKeybindings(bindings: KeybindingsConfig): ResolvedKeybindingsConfig { const resolved: ResolvedKeybindingsConfig = []; for (const binding of bindings) { @@ -244,6 +270,7 @@ describe("WebSocket Server", () => { devUrl?: string; authToken?: string; stateDir?: string; + providerLayer?: Layer.Layer; gitManager?: { status: (input: { cwd: string }) => Promise; runStackedAction: (input: { cwd: string; action: string }) => Promise; @@ -259,6 +286,7 @@ describe("WebSocket Server", () => { persistenceLayer: options.persistenceLayer ?? SqlitePersistenceMemory, ...(options.devUrl ? { devUrl: options.devUrl } : {}), ...(options.authToken ? { authToken: options.authToken } : {}), + ...(options.providerLayer ? { providerLayer: options.providerLayer } : {}), ...(options.gitManager ? { gitManager: options.gitManager as never } : {}), ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}), }); @@ -668,6 +696,99 @@ describe("WebSocket Server", () => { expect(diffResponse.error?.message).toContain("Unknown provider session"); }); + it("keeps orchestration domain push behavior for provider runtime events", async () => { + const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); + const emitRuntimeEvent = (event: ProviderRuntimeEvent) => { + Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); + }; + const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; + const providerService: ProviderServiceShape = { + startSession: () => unsupported(), + sendTurn: () => unsupported(), + interruptTurn: () => unsupported(), + respondToRequest: () => unsupported(), + stopSession: () => unsupported(), + listSessions: () => Effect.succeed([]), + listCheckpoints: () => unsupported(), + getCheckpointDiff: (input) => + Effect.succeed({ + threadId: "thread-1", + fromTurnCount: input.fromTurnCount, + toTurnCount: input.toTurnCount, + diff: "", + }), + revertToCheckpoint: () => unsupported(), + stopAll: () => Effect.void, + streamEvents: Stream.fromPubSub(runtimeEventPubSub), + }; + const providerLayer = Layer.succeed(ProviderService, providerService); + + server = createTestServer({ + cwd: "/test", + providerLayer, + }); + await server.start(); + const addr = server.httpServer.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + + const ws = await connectWs(port); + connections.push(ws); + await waitForMessage(ws); + + const createdAt = new Date().toISOString(); + const createThreadResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, { + type: "thread.create", + commandId: "cmd-ws-runtime-thread-create", + threadId: "thread-1", + projectId: "project-1", + title: "Thread 1", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }); + expect(createThreadResponse.error).toBeUndefined(); + + const setSessionResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, { + type: "thread.session", + commandId: "cmd-ws-runtime-session-set", + threadId: "thread-1", + session: { + sessionId: "sess-test", + provider: "codex", + status: "ready", + threadId: "thread-1", + activeTurnId: null, + createdAt, + updatedAt: createdAt, + lastError: null, + }, + createdAt, + }); + expect(setSessionResponse.error).toBeUndefined(); + + emitRuntimeEvent({ + type: "message.delta", + eventId: "evt-ws-runtime-message-delta", + provider: "codex", + sessionId: "sess-test", + createdAt: new Date().toISOString(), + turnId: "turn-1", + itemId: "item-1", + delta: "hello from runtime", + }); + + const domainPush = await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => { + const event = push.data as { type?: string; payload?: { id?: string; text?: string } }; + return event.type === "message.sent" && event.payload?.id === "assistant:item-1"; + }); + + const domainEvent = domainPush.data as { type: string; payload: { id: string; text: string } }; + expect(domainEvent.type).toBe("message.sent"); + expect(domainEvent.payload.id).toBe("assistant:item-1"); + expect(domainEvent.payload.text).toBe("hello from runtime"); + }); + it("routes terminal RPC methods and broadcasts terminal events", async () => { const cwd = makeTempDir("t3code-ws-terminal-cwd-"); const terminalManager = new MockTerminalManager(); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index 7fb9ab4f03a78a9925e278538661cd65f0764589..d600f960b28e7a473a3f571385954d7bb023c52e 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -11,7 +11,6 @@ import { ORCHESTRATION_WS_METHODS, WS_CHANNELS, WS_METHODS, - type ProviderRuntimeEvent, type TerminalEvent, type WsPush, type WsRequest, @@ -19,7 +18,7 @@ import { wsRequestSchema, } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; -import { Effect, Exit, Layer, ManagedRuntime, Queue, Scope, Stream } from "effect"; +import { Effect, Exit, Layer, ManagedRuntime, Scope, Stream } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; @@ -37,10 +36,12 @@ import { TerminalManager } from "./terminalManager"; import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings"; import { searchWorkspaceEntries } from "./workspaceEntries"; import { OrchestrationEngineLive } from "./orchestration/Layers/OrchestrationEngine"; +import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRuntimeIngestion"; import { OrchestrationEngineService, type OrchestrationEngineShape, } from "./orchestration/Services/OrchestrationEngine"; +import { ProviderRuntimeIngestionService } from "./orchestration/Services/ProviderRuntimeIngestion"; import { ProjectRepositoryLive } from "./persistence/Layers/Projects"; import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects"; @@ -53,7 +54,6 @@ import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionD import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService"; -import { parseTurnDiffFilesFromUnifiedDiff, type TurnDiffFileSummary } from "./checkpointing/Diffs"; import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints"; import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -88,6 +88,7 @@ export interface ServerOptions { logWebSocketEvents?: boolean | undefined; gitManager?: GitManager | undefined; terminalManager?: TerminalManager | undefined; + providerLayer?: Layer.Layer | undefined; authToken?: string | undefined; } @@ -119,93 +120,13 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined { return undefined; } -const providerTurnKey = (sessionId: string, turnId: string) => `${sessionId}:${turnId}`; - -type ActivityTone = "thinking" | "tool" | "info" | "error"; - -interface ThreadActivityInput { - readonly id: string; - readonly createdAt: string; - readonly label: string; - readonly detail?: string; - readonly tone: ActivityTone; - readonly turnId?: string; - readonly requestId?: string; - readonly requestKind?: "command" | "file-change"; -} - -function truncateDetail(value: string, limit = 180): string { - return value.length > limit ? `${value.slice(0, limit - 1)}...` : value; -} - -function runtimeEventToActivities(event: ProviderRuntimeEvent): ReadonlyArray { - switch (event.type) { - case "approval.requested": - return [ - { - id: event.eventId, - createdAt: event.createdAt, - label: - event.requestKind === "command" - ? "Command approval requested" - : "File-change approval requested", - ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), - tone: "tool", - ...(event.turnId ? { turnId: event.turnId } : {}), - requestId: event.requestId, - requestKind: event.requestKind, - }, - ]; - case "approval.resolved": - return [ - { - id: event.eventId, - createdAt: event.createdAt, - label: "Approval resolved", - tone: "info", - ...(event.turnId ? { turnId: event.turnId } : {}), - requestId: event.requestId, - ...(event.requestKind ? { requestKind: event.requestKind } : {}), - }, - ]; - case "runtime.error": - return [ - { - id: event.eventId, - createdAt: event.createdAt, - label: "Runtime error", - detail: truncateDetail(event.message), - tone: "error", - ...(event.turnId ? { turnId: event.turnId } : {}), - }, - ]; - case "checkpoint.captured": - return [ - { - id: event.eventId, - createdAt: event.createdAt, - label: "Checkpoint captured", - tone: "info", - ...(event.turnId ? { turnId: event.turnId } : {}), - }, - ]; - case "tool.completed": - return [ - { - id: event.eventId, - createdAt: event.createdAt, - label: `${event.title} complete`, - ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), - tone: "tool", - ...(event.turnId ? { turnId: event.turnId } : {}), - }, - ]; - default: - return []; - } -} - -const noop = () => {}; +type EffectRuntime = ManagedRuntime.ManagedRuntime< + | ProjectRepository + | OrchestrationEngineService + | ProviderService + | ProviderRuntimeIngestionService, + unknown +>; export function createServer(options: ServerOptions) { const { @@ -219,14 +140,12 @@ export function createServer(options: ServerOptions) { logWebSocketEvents: explicitLogWsEvents, gitManager = new GitManager(), terminalManager = new TerminalManager(), + providerLayer: customProviderLayer, authToken, } = options; const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata"); - let effectRuntime: ManagedRuntime.ManagedRuntime< - ProjectRepository | OrchestrationEngineService | ProviderService, - unknown - > | null = null; + let effectRuntime: EffectRuntime | null = null; let projectRepository: ProjectRepositoryShape | null = null; let orchestrationEngine: OrchestrationEngineShape | null = null; let providerService: ProviderServiceShape | null = null; @@ -237,8 +156,7 @@ export function createServer(options: ServerOptions) { let keybindingsConfig = loadResolvedKeybindingsConfig(logger); let orchestrationDomainEventLogger: EventNdjsonLogger | undefined; let orchestrationCommandLogger: EventNdjsonLogger | undefined; - let unsubscribeDomainEvents = noop; - let unsubscribeProviderEvents = noop; + let subscriptionsScope: Scope.Closeable | null = null; function logOutgoingPush(push: WsPush, recipients: number) { if (!logWebSocketEvents) return; @@ -269,282 +187,36 @@ export function createServer(options: ServerOptions) { return providerService; }; - async function attachOrchestrationSubscriptions( - runtime: ManagedRuntime.ManagedRuntime< - ProjectRepository | OrchestrationEngineService | ProviderService, - unknown - >, - orchestrationEngine: OrchestrationEngineShape, - ) { - const subscriptionScope = await runtime.runPromise(Scope.make("sequential")); - await runtime.runPromise( - Effect.gen(function* () { - yield* Effect.forkScoped( - Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => - Effect.sync(() => { - orchestrationDomainEventLogger?.write({ - observedAt: new Date().toISOString(), - event, - }); - const push: WsPush = { - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.domainEvent, - data: event, - }; - const message = JSON.stringify(push); - let recipients = 0; - for (const client of clients) { - if (client.readyState === client.OPEN) { - client.send(message); - recipients += 1; - } - } - logOutgoingPush(push, recipients); - }), - ), - ); - }).pipe(Scope.provide(subscriptionScope)), - ); - - unsubscribeDomainEvents = () => { - runtime.runFork(Scope.close(subscriptionScope, Exit.void)); - }; - } - - async function attachProviderSubscriptions( - runtime: ManagedRuntime.ManagedRuntime< - ProjectRepository | OrchestrationEngineService | ProviderService, - unknown - >, - providerService: ProviderServiceShape, - orchestrationEngine: OrchestrationEngineShape, - ) { - const turnMessageIdsByTurnKey = new Map>(); - const latestMessageIdByTurnKey = new Map(); - - const rememberAssistantMessageId = (sessionId: string, turnId: string, messageId: string) => { - const key = providerTurnKey(sessionId, turnId); - const existingIds = turnMessageIdsByTurnKey.get(key); - if (existingIds) { - existingIds.add(messageId); - } else { - turnMessageIdsByTurnKey.set(key, new Set([messageId])); - } - latestMessageIdByTurnKey.set(key, messageId); - }; - - const getAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => { - return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set(); - }; - - const clearAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => { - turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId)); - }; - - const getLatestAssistantMessageIdForTurn = (sessionId: string, turnId: string) => { - return latestMessageIdByTurnKey.get(providerTurnKey(sessionId, turnId)); - }; - - const clearTurnStateForSession = (sessionId: string) => { - const prefix = `${sessionId}:`; - for (const key of turnMessageIdsByTurnKey.keys()) { - if (key.startsWith(prefix)) { - turnMessageIdsByTurnKey.delete(key); - } - } - for (const key of latestMessageIdByTurnKey.keys()) { - if (key.startsWith(prefix)) { - latestMessageIdByTurnKey.delete(key); - } - } - }; - - const processEvent = (event: ProviderRuntimeEvent) => - Effect.gen(function* () { - const readModel = yield* orchestrationEngine.getReadModel(); - const thread = readModel.threads.find( - (entry) => entry.session?.sessionId === event.sessionId, - ); - if (!thread) return; - - const now = event.createdAt; - if (event.type === "turn.started" || event.type === "turn.completed") { - yield* orchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - sessionId: event.sessionId, - provider: event.provider, - status: event.type === "turn.started" ? "running" : "ready", - threadId: thread.id, - activeTurnId: event.type === "turn.started" ? event.turnId : null, - createdAt: thread.session?.createdAt ?? now, - updatedAt: now, - lastError: - event.type === "turn.completed" && event.status === "failed" - ? (event.errorMessage ?? null) - : null, - }, - createdAt: now, - }); - } - - if (event.type === "message.delta" && event.delta.length > 0) { - const assistantMessageId = `assistant:${event.itemId ?? event.turnId ?? event.sessionId}`; - if (event.turnId) { - rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId); + const attachSubscriptionss = Effect.gen(function* () { + const orchestrationEngine = yield* OrchestrationEngineService; + const runtimeIngestion = yield* ProviderRuntimeIngestionService; + subscriptionsScope = yield* Scope.make("sequential"); + + yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => + Effect.sync(() => { + orchestrationDomainEventLogger?.write({ + observedAt: new Date().toISOString(), + event, + }); + const push: WsPush = { + type: "push", + channel: ORCHESTRATION_WS_CHANNELS.domainEvent, + data: event, + }; + const message = JSON.stringify(push); + let recipients = 0; + for (const client of clients) { + if (client.readyState === client.OPEN) { + client.send(message); + recipients += 1; } - yield* orchestrationEngine.dispatch({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: thread.id, - messageId: assistantMessageId, - role: "assistant", - text: event.delta, - streaming: true, - createdAt: now, - }); } + logOutgoingPush(push, recipients); + }), + ).pipe(Effect.forkIn(subscriptionsScope)); - if (event.type === "message.completed") { - const assistantMessageId = `assistant:${event.itemId}`; - if (event.turnId) { - rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId); - } - yield* orchestrationEngine.dispatch({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: thread.id, - messageId: assistantMessageId, - role: "assistant", - text: "", - streaming: false, - createdAt: now, - }); - } - - if (event.type === "turn.completed") { - if (event.turnId) { - const assistantMessageIds = getAssistantMessageIdsForTurn( - event.sessionId, - event.turnId, - ); - yield* Effect.forEach(assistantMessageIds, (assistantMessageId) => - orchestrationEngine.dispatch({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: thread.id, - messageId: assistantMessageId, - role: "assistant", - text: "", - streaming: false, - createdAt: now, - }), - ).pipe(Effect.asVoid); - clearAssistantMessageIdsForTurn(event.sessionId, event.turnId); - } - } - - if (event.type === "session.exited") { - clearTurnStateForSession(event.sessionId); - } - - if (event.type === "checkpoint.captured" && event.turnId) { - const files: ReadonlyArray = yield* providerService - .getCheckpointDiff({ - sessionId: event.sessionId, - fromTurnCount: Math.max(0, event.turnCount - 1), - toTurnCount: event.turnCount, - }) - .pipe( - Effect.map((result) => parseTurnDiffFilesFromUnifiedDiff(result.diff)), - Effect.catch((error) => - Effect.sync(() => { - logger.warn("failed to derive checkpoint file summary", { - sessionId: event.sessionId, - turnId: event.turnId, - turnCount: event.turnCount, - error: - error instanceof Error - ? error.message - : typeof error === "string" - ? error - : "unknown error", - }); - }).pipe(Effect.as([] as ReadonlyArray)), - ), - ); - const assistantMessageId = - getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ?? - `assistant:${event.turnId}`; - yield* orchestrationEngine.dispatch({ - type: "thread.turnDiff.complete", - commandId: crypto.randomUUID(), - threadId: thread.id, - turnId: event.turnId, - completedAt: now, - ...(event.status ? { status: event.status } : {}), - files, - assistantMessageId, - checkpointTurnCount: event.turnCount, - createdAt: now, - }); - } - - if (event.type === "runtime.error") { - yield* orchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - sessionId: event.sessionId, - provider: event.provider, - status: "error", - threadId: thread.id, - activeTurnId: event.turnId ?? null, - createdAt: thread.session?.createdAt ?? now, - updatedAt: now, - lastError: event.message, - }, - createdAt: now, - }); - } - - const activities = runtimeEventToActivities(event); - yield* Effect.forEach(activities, (activity) => - orchestrationEngine.dispatch({ - type: "thread.activity.append", - commandId: crypto.randomUUID(), - threadId: thread.id, - activity, - createdAt: activity.createdAt, - }), - ).pipe(Effect.asVoid); - }); - - const subscriptionScope = await runtime.runPromise(Scope.make("sequential")); - await runtime.runPromise( - Effect.gen(function* () { - const providerEventQueue = yield* Queue.unbounded(); - - yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid)); - - yield* Effect.forkScoped( - Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEvent))), - ); - yield* Effect.forkScoped( - Stream.runForEach(providerService.streamEvents, (event) => - Queue.offer(providerEventQueue, event).pipe(Effect.asVoid), - ), - ); - }).pipe(Scope.provide(subscriptionScope)), - ); - - unsubscribeProviderEvents = () => { - runtime.runFork(Scope.close(subscriptionScope, Exit.void)); - }; - } + yield* runtimeIngestion.start.pipe(Scope.provide(subscriptionsScope)); + }); const onTerminalEvent = (event: TerminalEvent) => { const push: WsPush = { @@ -955,29 +627,38 @@ export function createServer(options: ServerOptions) { path.join(orchestrationLogsDir, "orchestration-command.ndjson"), ); const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath); - const orchestrationLayer = Layer.provide( - OrchestrationEngineLive, - OrchestrationEventStoreLive, - ); - const codexAdapterLayer = makeCodexAdapterLive({ - nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), - }); - const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(codexAdapterLayer)); - const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); - const checkpointDependenciesLayer = Layer.mergeAll( - checkpointStoreLayer, - CheckpointRepositoryLive, - adapterRegistryLayer, - ProviderSessionDirectoryLive, - ); - const checkpointServiceLayer = CheckpointServiceLive.pipe( - Layer.provideMerge(checkpointDependenciesLayer), - ); - const providerLayer = makeProviderServiceLive({ - canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"), - }).pipe(Layer.provideMerge(checkpointServiceLayer)); + const orchestrationLayer = Layer.provide(OrchestrationEngineLive, OrchestrationEventStoreLive); + const providerLayer = + customProviderLayer ?? + (() => { + const codexAdapterLayer = makeCodexAdapterLive({ + nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), + }); + const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe( + Layer.provide(codexAdapterLayer), + ); + const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); + const checkpointDependenciesLayer = Layer.mergeAll( + checkpointStoreLayer, + CheckpointRepositoryLive, + adapterRegistryLayer, + ProviderSessionDirectoryLive, + ); + const checkpointServiceLayer = CheckpointServiceLive.pipe( + Layer.provideMerge(checkpointDependenciesLayer), + ); + return makeProviderServiceLive({ + canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"), + }).pipe(Layer.provideMerge(checkpointServiceLayer)); + })(); - const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive, providerLayer).pipe( + const runtimeServicesLayer = Layer.mergeAll( + orchestrationLayer, + ProjectRepositoryLive, + providerLayer, + ); + const layer = ProviderRuntimeIngestionLive.pipe( + Layer.provideMerge(runtimeServicesLayer), Layer.provide(persistenceLayer), Layer.provideMerge(NodeServices.layer), ); @@ -993,8 +674,7 @@ export function createServer(options: ServerOptions) { projectRepository = repository; providerService = nextProviderService; await runtime.runPromise(repository.pruneMissing()); - await attachOrchestrationSubscriptions(runtime, nextOrchestrationEngine); - await attachProviderSubscriptions(runtime, nextProviderService, nextOrchestrationEngine); + await runtime.runPromise(attachSubscriptionss); } catch (error) { await runtime.dispose().catch(() => undefined); throw error; @@ -1004,10 +684,10 @@ export function createServer(options: ServerOptions) { } async function disposeEffectRuntime() { - unsubscribeDomainEvents(); - unsubscribeProviderEvents(); - unsubscribeDomainEvents = noop; - unsubscribeProviderEvents = noop; + if (subscriptionsScope && effectRuntime) { + await effectRuntime.runPromise(Scope.close(subscriptionsScope, Exit.void)); + subscriptionsScope = null; + } orchestrationDomainEventLogger?.close(); orchestrationDomainEventLogger = undefined; orchestrationCommandLogger?.close(); -- 2.54.0