diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..87afdcea4c38d1385ed23e4d5d56275150eae1bf --- /dev/null +++ b/apps/server/src/orchestration/Layer.test.ts @@ -0,0 +1,439 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import type { OrchestrationEvent } from "@t3tools/contracts"; +import { Effect } from "effect"; +import { afterEach, describe, expect, it } from "vitest"; + +import { + OrchestrationEventRepository, + type OrchestrationEventRepositoryShape, +} from "../persistence/Services/OrchestrationEvents"; + +import { Layer, ManagedRuntime } from "effect"; + +import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite"; + +import { OrchestrationEngineLive } from "./Layer"; +import { OrchestrationEngineService } from "./Service"; +import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents"; + +const tempDirs: string[] = []; + +function makeTempDir(prefix: string): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0, tempDirs.length)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +export async function createOrchestrationSystem(stateDir: string) { + const dbPath = path.join(stateDir, "orchestration.sqlite"); + const orchestrationLayer = OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationEventRepositoryLive), + Layer.provide(makeSqlitePersistenceLive(dbPath)), + ); + const runtime = ManagedRuntime.make(orchestrationLayer); + const engine = await runtime.runPromise(OrchestrationEngineService); + return { + engine, + run: (effect: Effect.Effect) => runtime.runPromise(effect), + dispose: () => runtime.dispose(), + }; +} + +describe("OrchestrationEngine", () => { + it("replays to the same deterministic snapshot", async () => { + const stateDir = makeTempDir("t3code-orchestration-"); + const createdAt = new Date().toISOString(); + const projectId = "project-1"; + const threadId = "thread-1"; + + const firstSystem = await createOrchestrationSystem(stateDir); + const engineA = firstSystem.engine; + await firstSystem.run( + engineA.dispatch({ + type: "thread.create", + commandId: "cmd-1", + threadId, + projectId, + title: "Thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + await firstSystem.run( + engineA.dispatch({ + type: "message.send", + commandId: "cmd-3", + threadId, + messageId: "msg-1", + role: "user", + text: "hello", + streaming: false, + createdAt, + }), + ); + const snapshotA = await firstSystem.run(engineA.getSnapshot()); + await firstSystem.dispose(); + + const secondSystem = await createOrchestrationSystem(stateDir); + const engineB = secondSystem.engine; + const snapshotB = await secondSystem.run(engineB.getSnapshot()); + expect(snapshotB).toEqual(snapshotA); + await secondSystem.dispose(); + }); + + it("fans out read-model updates to subscribers", async () => { + const stateDir = makeTempDir("t3code-orchestration-fanout-"); + const system = await createOrchestrationSystem(stateDir); + const engine = system.engine; + const updates: number[] = []; + const unsubscribe = await system.run( + engine.subscribeToReadModel((snapshot) => { + updates.push(snapshot.sequence); + }), + ); + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-thread", + threadId: "thread-2", + projectId: "project-2", + title: "fanout", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: new Date().toISOString(), + }), + ); + unsubscribe(); + expect(updates.length).toBeGreaterThan(0); + await system.dispose(); + }); + + it("replays append-only events from sequence", async () => { + const stateDir = makeTempDir("t3code-orchestration-replay-"); + const system = await createOrchestrationSystem(stateDir); + const engine = system.engine; + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-a", + threadId: "thread-replay", + projectId: "project-replay", + title: "replay", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: new Date().toISOString(), + }), + ); + await system.run( + engine.dispatch({ + type: "thread.delete", + commandId: "cmd-b", + threadId: "thread-replay", + createdAt: new Date().toISOString(), + }), + ); + const events = await system.run(engine.replayEvents(0)); + expect(events.length).toBe(2); + expect(events[0]?.type).toBe("thread.created"); + expect(events[1]?.type).toBe("thread.deleted"); + await system.dispose(); + }); + + it("stores completed turn summaries even when no files changed", async () => { + const stateDir = makeTempDir("t3code-orchestration-turn-diff-"); + const firstSystem = await createOrchestrationSystem(stateDir); + const engine = firstSystem.engine; + const completedAt = new Date().toISOString(); + await firstSystem.run( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-thread-turn-diff", + threadId: "thread-turn-diff", + projectId: "project-turn-diff", + title: "Turn diff thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: completedAt, + }), + ); + await firstSystem.run( + engine.dispatch({ + type: "thread.turnDiff.complete", + commandId: "cmd-turn-diff-complete", + threadId: "thread-turn-diff", + turnId: "turn-1", + completedAt, + status: "completed", + files: [], + checkpointTurnCount: 1, + createdAt: completedAt, + }), + ); + + const firstThread = (await firstSystem.run(engine.getSnapshot())).threads.find( + (thread) => thread.id === "thread-turn-diff", + ); + expect(firstThread?.turnDiffSummaries).toEqual([ + { + turnId: "turn-1", + completedAt, + status: "completed", + files: [], + checkpointTurnCount: 1, + }, + ]); + await firstSystem.dispose(); + + const secondSystem = await createOrchestrationSystem(stateDir); + const restartedEngine = secondSystem.engine; + const restartedThread = (await secondSystem.run(restartedEngine.getSnapshot())).threads.find( + (thread) => thread.id === "thread-turn-diff", + ); + expect(restartedThread?.turnDiffSummaries).toEqual([ + { + turnId: "turn-1", + completedAt, + status: "completed", + files: [], + checkpointTurnCount: 1, + }, + ]); + await secondSystem.dispose(); + }); + + it("reverts thread messages and turn summaries to a checkpoint", async () => { + const stateDir = makeTempDir("t3code-orchestration-revert-"); + const firstSystem = await createOrchestrationSystem(stateDir); + const engine = firstSystem.engine; + const createdAt = new Date().toISOString(); + await firstSystem.run( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-thread-revert", + threadId: "thread-revert", + projectId: "project-revert", + title: "Revert thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + await firstSystem.run( + engine.dispatch({ + type: "message.send", + commandId: "cmd-msg-1-user", + threadId: "thread-revert", + messageId: "user-1", + role: "user", + text: "first", + streaming: false, + createdAt, + }), + ); + await firstSystem.run( + engine.dispatch({ + type: "message.send", + commandId: "cmd-msg-1-assistant", + threadId: "thread-revert", + messageId: "assistant:turn-1", + role: "assistant", + text: "first-response", + streaming: false, + createdAt, + }), + ); + await firstSystem.run( + engine.dispatch({ + type: "thread.turnDiff.complete", + commandId: "cmd-turn-1-complete", + threadId: "thread-revert", + turnId: "turn-1", + completedAt: createdAt, + status: "completed", + files: [], + assistantMessageId: "assistant:turn-1", + checkpointTurnCount: 1, + createdAt, + }), + ); + + const createdAtSecond = new Date(Date.now() + 1_000).toISOString(); + await firstSystem.run( + engine.dispatch({ + type: "message.send", + commandId: "cmd-msg-2-user", + threadId: "thread-revert", + messageId: "user-2", + role: "user", + text: "second", + streaming: false, + createdAt: createdAtSecond, + }), + ); + await firstSystem.run( + engine.dispatch({ + type: "message.send", + commandId: "cmd-msg-2-assistant", + threadId: "thread-revert", + messageId: "assistant:turn-2", + role: "assistant", + text: "second-response", + streaming: false, + createdAt: createdAtSecond, + }), + ); + await firstSystem.run( + engine.dispatch({ + type: "thread.turnDiff.complete", + commandId: "cmd-turn-2-complete", + threadId: "thread-revert", + turnId: "turn-2", + completedAt: createdAtSecond, + status: "completed", + files: [], + assistantMessageId: "assistant:turn-2", + checkpointTurnCount: 2, + createdAt: createdAtSecond, + }), + ); + + await firstSystem.run( + engine.dispatch({ + type: "thread.revert", + commandId: "cmd-thread-revert-apply", + threadId: "thread-revert", + turnCount: 1, + messageCount: 2, + createdAt: createdAtSecond, + }), + ); + + const thread = (await firstSystem.run(engine.getSnapshot())).threads.find( + (entry) => entry.id === "thread-revert", + ); + expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]); + expect(thread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]); + expect(thread?.latestTurnId).toBe("turn-1"); + await firstSystem.dispose(); + + const secondSystem = await createOrchestrationSystem(stateDir); + const restarted = secondSystem.engine; + const restartedThread = (await secondSystem.run(restarted.getSnapshot())).threads.find( + (entry) => entry.id === "thread-revert", + ); + expect(restartedThread?.messages.map((message) => message.id)).toEqual([ + "user-1", + "assistant:turn-1", + ]); + expect(restartedThread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]); + await secondSystem.dispose(); + }); + + it("allows stop to be called multiple times", async () => { + const inMemoryStore: OrchestrationEventRepositoryShape = { + append(event) { + return Effect.succeed({ ...event, sequence: 1 }); + }, + readFromSequence() { + return Effect.succeed([]); + }, + readAll() { + return Effect.succeed([]); + }, + }; + const runtime = ManagedRuntime.make( + OrchestrationEngineLive.pipe( + Layer.provide(Layer.succeed(OrchestrationEventRepository, inMemoryStore)), + ), + ); + await runtime.runPromise(OrchestrationEngineService); + await runtime.dispose(); + await expect(runtime.dispose()).resolves.toBeUndefined(); + }); + + it("keeps processing queued commands after a storage failure", async () => { + const events: OrchestrationEvent[] = []; + let nextSequence = 1; + let shouldFailFirstAppend = true; + + const flakyStore: OrchestrationEventRepositoryShape = { + append(event) { + if (shouldFailFirstAppend) { + shouldFailFirstAppend = false; + return Effect.die(new Error("append failed")); + } + const savedEvent = { + ...event, + sequence: nextSequence, + } satisfies OrchestrationEvent; + nextSequence += 1; + events.push(savedEvent); + return Effect.succeed(savedEvent); + }, + readFromSequence(sequenceExclusive) { + return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive)); + }, + readAll() { + return Effect.succeed(events); + }, + }; + + const runtime = ManagedRuntime.make( + OrchestrationEngineLive.pipe( + Layer.provide(Layer.succeed(OrchestrationEventRepository, flakyStore)), + ), + ); + const engine = await runtime.runPromise(OrchestrationEngineService); + const createdAt = new Date().toISOString(); + + await expect( + runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-flaky-1", + threadId: "thread-flaky-fail", + projectId: "project-flaky", + title: "flaky-fail", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ), + ).rejects.toThrow("append failed"); + + const result = await runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-flaky-2", + threadId: "thread-flaky-ok", + projectId: "project-flaky", + title: "flaky-ok", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + expect(result.sequence).toBe(1); + expect((await runtime.runPromise(engine.getSnapshot())).sequence).toBe(1); + + await runtime.dispose(); + }); +}); diff --git a/apps/server/src/orchestration/Layer.ts b/apps/server/src/orchestration/Layer.ts new file mode 100644 index 0000000000000000000000000000000000000000..9974132fee21b7aa2ceebba925f897e1eb5568f3 --- /dev/null +++ b/apps/server/src/orchestration/Layer.ts @@ -0,0 +1,297 @@ +import type { + OrchestrationCommand, + OrchestrationEvent, + OrchestrationReadModel, +} from "@t3tools/contracts"; +import { OrchestrationCommandSchema } from "@t3tools/contracts"; +import { + Cause, + Deferred, + Effect, + Either, + Fiber, + Layer, + PubSub, + Queue, + Runtime, + Schema, +} from "effect"; + +import { createLogger } from "../logger"; +import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents"; +import { createEmptyReadModel, reduceEvent } from "./reducer"; +import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service"; + +type CommandEnvelope = { + command: OrchestrationCommand; + result: Deferred.Deferred<{ sequence: number }, Error>; +}; + +function asError(error: unknown, fallbackMessage: string): Error { + return error instanceof Error ? error : new Error(fallbackMessage); +} + +function mapCommandToEvent(command: OrchestrationCommand): Omit { + const eventId = crypto.randomUUID(); + switch (command.type) { + case "thread.create": + return { + eventId, + type: "thread.created", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + id: command.threadId, + projectId: command.projectId, + title: command.title, + model: command.model, + branch: command.branch, + worktreePath: command.worktreePath, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }; + case "thread.delete": + return { + eventId, + type: "thread.deleted", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + id: command.threadId, + deletedAt: command.createdAt, + }, + }; + case "thread.meta.update": + return { + eventId, + type: "thread.meta-updated", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + ...(command.title !== undefined ? { title: command.title } : {}), + ...(command.model !== undefined ? { model: command.model } : {}), + ...(command.branch !== undefined ? { branch: command.branch } : {}), + ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}), + updatedAt: command.createdAt, + }, + }; + case "message.send": + return { + eventId, + type: "message.sent", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + id: command.messageId, + role: command.role, + text: command.text, + threadId: command.threadId, + createdAt: command.createdAt, + streaming: command.streaming === true, + }, + }; + case "thread.session": + return { + eventId, + type: "thread.session-set", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + session: command.session, + }, + }; + case "git.readModel.upsert": + return { + eventId, + type: "git.read-model-upsert", + aggregateType: "project", + aggregateId: command.projectId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + projectId: command.projectId, + branch: command.branch, + hasWorkingTreeChanges: command.hasWorkingTreeChanges, + aheadCount: command.aheadCount, + behindCount: command.behindCount, + updatedAt: command.createdAt, + }, + }; + case "thread.turnDiff.complete": + return { + eventId, + type: "thread.turn-diff-completed", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + turnId: command.turnId, + completedAt: command.completedAt, + ...(command.status !== undefined ? { status: command.status } : {}), + files: command.files, + ...(command.assistantMessageId !== undefined + ? { assistantMessageId: command.assistantMessageId } + : {}), + ...(command.checkpointTurnCount !== undefined + ? { checkpointTurnCount: command.checkpointTurnCount } + : {}), + }, + }; + case "thread.revert": + return { + eventId, + type: "thread.reverted", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + turnCount: command.turnCount, + messageCount: command.messageCount, + }, + }; + } +} + +const decodeUnknownCommand = Schema.decodeUnknownEither(OrchestrationCommandSchema); + +export const makeOrchestrationEngine = Effect.gen(function* () { + const logger = createLogger("orchestration"); + const eventStore = yield* OrchestrationEventRepository; + const runtime = Runtime.defaultRuntime; + + let readModel = createEmptyReadModel(new Date().toISOString()); + + const commandQueue = yield* Queue.unbounded(); + const readModelPubSub = yield* PubSub.unbounded(); + const eventPubSub = yield* PubSub.unbounded(); + + const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>(); + const domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); + + const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => + Effect.gen(function* () { + const eventBase = mapCommandToEvent(envelope.command); + const savedEvent = yield* eventStore.append(eventBase); + readModel = yield* reduceEvent(readModel, savedEvent); + + const snapshot = readModel; + yield* Effect.all([ + PubSub.publish(eventPubSub, savedEvent), + PubSub.publish(readModelPubSub, snapshot), + ]); + + for (const listener of domainEventListeners) { + listener(savedEvent); + } + for (const listener of readModelListeners) { + listener(snapshot); + } + + yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); + }).pipe( + Effect.catchAllCause((cause) => + Deferred.fail( + envelope.result, + asError(Cause.squash(cause), "Unknown command processing error"), + ).pipe(Effect.asVoid), + ), + ); + + const bootstrapReadModel: Effect.Effect = Effect.gen(function* () { + const existingEvents = yield* eventStore.readAll(); + for (const event of existingEvents) { + readModel = yield* reduceEvent(readModel, event); + } + }); + + yield* bootstrapReadModel; + + const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope))); + const workerFiber: Fiber.RuntimeFiber = Runtime.runFork(runtime)(worker); + logger.info("orchestration engine started", { + sequence: readModel.sequence, + }); + + yield* Effect.addFinalizer(() => Fiber.interrupt(workerFiber).pipe(Effect.orDie)); + + const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () => + Effect.sync((): OrchestrationReadModel => readModel); + + const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) => + eventStore.readFromSequence(fromSequenceExclusive); + + const dispatch: OrchestrationEngineShape["dispatch"] = (command) => + Effect.gen(function* () { + const result = yield* Deferred.make<{ sequence: number }, Error>(); + yield* Queue.offer(commandQueue, { command, result }); + return yield* Deferred.await(result); + }).pipe(Effect.mapError((error) => asError(error, "Queue offer failed"))); + + const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) => + Effect.gen(function* () { + let payload: unknown = command; + if (typeof command === "string") { + try { + payload = JSON.parse(command) as unknown; + } catch { + return yield* Effect.fail( + new Error("Invalid orchestration command: payload is not valid JSON"), + ); + } + } + const decoded = decodeUnknownCommand(payload); + if (Either.isLeft(decoded)) { + return yield* Effect.fail( + new Error(`Invalid orchestration command: ${decoded.left.toString()}`), + ); + } + return yield* dispatch(decoded.right); + }); + + const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) => + Effect.sync(() => { + readModelListeners.add(callback); + return () => { + readModelListeners.delete(callback); + }; + }); + + const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) => + Effect.sync(() => { + domainEventListeners.add(callback); + return () => { + domainEventListeners.delete(callback); + }; + }); + + return { + getSnapshot, + replayEvents, + dispatchUnknown, + dispatch, + subscribeToReadModel, + subscribeToDomainEvents, + } satisfies OrchestrationEngineShape; +}); + +export const OrchestrationEngineLive = Layer.scoped( + OrchestrationEngineService, + makeOrchestrationEngine, +); diff --git a/apps/server/src/orchestration/Service.ts b/apps/server/src/orchestration/Service.ts new file mode 100644 index 0000000000000000000000000000000000000000..9959f897a4cfa0e6de8ee66d0b5405789892c14b --- /dev/null +++ b/apps/server/src/orchestration/Service.ts @@ -0,0 +1,25 @@ +import type { + OrchestrationCommand, + OrchestrationEvent, + OrchestrationReadModel, +} from "@t3tools/contracts"; +import { Context } from "effect"; +import type { Effect } from "effect"; + +export interface OrchestrationEngineShape { + readonly getSnapshot: () => Effect.Effect; + readonly replayEvents: (fromSequenceExclusive: number) => Effect.Effect; + readonly dispatchUnknown: (command: unknown) => Effect.Effect<{ sequence: number }, Error>; + readonly dispatch: (command: OrchestrationCommand) => Effect.Effect<{ sequence: number }, Error>; + readonly subscribeToReadModel: ( + callback: (snapshot: OrchestrationReadModel) => void, + ) => Effect.Effect<() => void>; + readonly subscribeToDomainEvents: ( + callback: (event: OrchestrationEvent) => void, + ) => Effect.Effect<() => void>; +} + +export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")< + OrchestrationEngineService, + OrchestrationEngineShape +>() {} diff --git a/apps/server/src/orchestration/engine.test.ts b/apps/server/src/orchestration/engine.test.ts deleted file mode 100644 index d8c7813cfc1dac21dc83711609525ce36a5dc467..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/engine.test.ts +++ /dev/null @@ -1,388 +0,0 @@ -import fs from "node:fs"; -import os from "node:os"; -import path from "node:path"; - -import type { OrchestrationEvent } from "@t3tools/contracts"; -import { Effect } from "effect"; -import { afterEach, describe, expect, it } from "vitest"; - -import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents"; -import { OrchestrationEngine } from "./engine"; - -import { Layer, ManagedRuntime } from "effect"; - -import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite"; - -import { OrchestrationLive } from "./layers"; -import { OrchestrationEngineService } from "./services"; - -const tempDirs: string[] = []; - -function makeTempDir(prefix: string): string { - const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); - tempDirs.push(dir); - return dir; -} - -afterEach(() => { - for (const dir of tempDirs.splice(0, tempDirs.length)) { - fs.rmSync(dir, { recursive: true, force: true }); - } -}); - -export async function createOrchestrationSystem(stateDir: string) { - const dbPath = path.join(stateDir, "orchestration.sqlite"); - const orchestrationLayer = OrchestrationLive.pipe( - Layer.provide(makeSqlitePersistenceLive(dbPath)), - ); - const runtime = ManagedRuntime.make(orchestrationLayer); - const engine = await runtime.runPromise(OrchestrationEngineService); - return { - engine, - dispose: () => runtime.dispose(), - }; -} - -describe("OrchestrationEngine", () => { - it("replays to the same deterministic snapshot", async () => { - const stateDir = makeTempDir("t3code-orchestration-"); - const createdAt = new Date().toISOString(); - const projectId = "project-1"; - const threadId = "thread-1"; - - const firstSystem = await createOrchestrationSystem(stateDir); - const engineA = firstSystem.engine; - await engineA.dispatch({ - type: "thread.create", - commandId: "cmd-1", - threadId, - projectId, - title: "Thread", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }); - await engineA.dispatch({ - type: "message.send", - commandId: "cmd-3", - threadId, - messageId: "msg-1", - role: "user", - text: "hello", - streaming: false, - createdAt, - }); - const snapshotA = engineA.getSnapshot(); - await firstSystem.dispose(); - - const secondSystem = await createOrchestrationSystem(stateDir); - const engineB = secondSystem.engine; - const snapshotB = engineB.getSnapshot(); - expect(snapshotB).toEqual(snapshotA); - await secondSystem.dispose(); - }); - - it("fans out read-model updates to subscribers", async () => { - const stateDir = makeTempDir("t3code-orchestration-fanout-"); - const system = await createOrchestrationSystem(stateDir); - const engine = system.engine; - const updates: number[] = []; - const unsubscribe = engine.subscribeToReadModel((snapshot) => { - updates.push(snapshot.sequence); - }); - await engine.dispatch({ - type: "thread.create", - commandId: "cmd-thread", - threadId: "thread-2", - projectId: "project-2", - title: "fanout", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: new Date().toISOString(), - }); - unsubscribe(); - expect(updates.length).toBeGreaterThan(0); - await system.dispose(); - }); - - it("replays append-only events from sequence", async () => { - const stateDir = makeTempDir("t3code-orchestration-replay-"); - const system = await createOrchestrationSystem(stateDir); - const engine = system.engine; - await engine.dispatch({ - type: "thread.create", - commandId: "cmd-a", - threadId: "thread-replay", - projectId: "project-replay", - title: "replay", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: new Date().toISOString(), - }); - await engine.dispatch({ - type: "thread.delete", - commandId: "cmd-b", - threadId: "thread-replay", - createdAt: new Date().toISOString(), - }); - const events = await engine.replayEvents(0); - expect(events.length).toBe(2); - expect(events[0]?.type).toBe("thread.created"); - expect(events[1]?.type).toBe("thread.deleted"); - await system.dispose(); - }); - - it("stores completed turn summaries even when no files changed", async () => { - const stateDir = makeTempDir("t3code-orchestration-turn-diff-"); - const firstSystem = await createOrchestrationSystem(stateDir); - const engine = firstSystem.engine; - const completedAt = new Date().toISOString(); - await engine.dispatch({ - type: "thread.create", - commandId: "cmd-thread-turn-diff", - threadId: "thread-turn-diff", - projectId: "project-turn-diff", - title: "Turn diff thread", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: completedAt, - }); - await engine.dispatch({ - type: "thread.turnDiff.complete", - commandId: "cmd-turn-diff-complete", - threadId: "thread-turn-diff", - turnId: "turn-1", - completedAt, - status: "completed", - files: [], - checkpointTurnCount: 1, - createdAt: completedAt, - }); - - const firstThread = engine - .getSnapshot() - .threads.find((thread) => thread.id === "thread-turn-diff"); - expect(firstThread?.turnDiffSummaries).toEqual([ - { - turnId: "turn-1", - completedAt, - status: "completed", - files: [], - checkpointTurnCount: 1, - }, - ]); - await firstSystem.dispose(); - - const secondSystem = await createOrchestrationSystem(stateDir); - const restartedEngine = secondSystem.engine; - const restartedThread = restartedEngine - .getSnapshot() - .threads.find((thread) => thread.id === "thread-turn-diff"); - expect(restartedThread?.turnDiffSummaries).toEqual([ - { - turnId: "turn-1", - completedAt, - status: "completed", - files: [], - checkpointTurnCount: 1, - }, - ]); - await secondSystem.dispose(); - }); - - it("reverts thread messages and turn summaries to a checkpoint", async () => { - const stateDir = makeTempDir("t3code-orchestration-revert-"); - const firstSystem = await createOrchestrationSystem(stateDir); - const engine = firstSystem.engine; - const createdAt = new Date().toISOString(); - await engine.dispatch({ - type: "thread.create", - commandId: "cmd-thread-revert", - threadId: "thread-revert", - projectId: "project-revert", - title: "Revert thread", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }); - await engine.dispatch({ - type: "message.send", - commandId: "cmd-msg-1-user", - threadId: "thread-revert", - messageId: "user-1", - role: "user", - text: "first", - streaming: false, - createdAt, - }); - await engine.dispatch({ - type: "message.send", - commandId: "cmd-msg-1-assistant", - threadId: "thread-revert", - messageId: "assistant:turn-1", - role: "assistant", - text: "first-response", - streaming: false, - createdAt, - }); - await engine.dispatch({ - type: "thread.turnDiff.complete", - commandId: "cmd-turn-1-complete", - threadId: "thread-revert", - turnId: "turn-1", - completedAt: createdAt, - status: "completed", - files: [], - assistantMessageId: "assistant:turn-1", - checkpointTurnCount: 1, - createdAt, - }); - - const createdAtSecond = new Date(Date.now() + 1_000).toISOString(); - await engine.dispatch({ - type: "message.send", - commandId: "cmd-msg-2-user", - threadId: "thread-revert", - messageId: "user-2", - role: "user", - text: "second", - streaming: false, - createdAt: createdAtSecond, - }); - await engine.dispatch({ - type: "message.send", - commandId: "cmd-msg-2-assistant", - threadId: "thread-revert", - messageId: "assistant:turn-2", - role: "assistant", - text: "second-response", - streaming: false, - createdAt: createdAtSecond, - }); - await engine.dispatch({ - type: "thread.turnDiff.complete", - commandId: "cmd-turn-2-complete", - threadId: "thread-revert", - turnId: "turn-2", - completedAt: createdAtSecond, - status: "completed", - files: [], - assistantMessageId: "assistant:turn-2", - checkpointTurnCount: 2, - createdAt: createdAtSecond, - }); - - await engine.dispatch({ - type: "thread.revert", - commandId: "cmd-thread-revert-apply", - threadId: "thread-revert", - turnCount: 1, - messageCount: 2, - createdAt: createdAtSecond, - }); - - const thread = engine.getSnapshot().threads.find((entry) => entry.id === "thread-revert"); - expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]); - expect(thread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]); - expect(thread?.latestTurnId).toBe("turn-1"); - await firstSystem.dispose(); - - const secondSystem = await createOrchestrationSystem(stateDir); - const restarted = secondSystem.engine; - const restartedThread = restarted - .getSnapshot() - .threads.find((entry) => entry.id === "thread-revert"); - expect(restartedThread?.messages.map((message) => message.id)).toEqual([ - "user-1", - "assistant:turn-1", - ]); - expect(restartedThread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]); - await secondSystem.dispose(); - }); - - it("allows stop to be called multiple times", async () => { - const inMemoryStore: OrchestrationEventRepositoryShape = { - append(event) { - return Effect.succeed({ ...event, sequence: 1 }); - }, - readFromSequence() { - return Effect.succeed([]); - }, - readAll() { - return Effect.succeed([]); - }, - }; - const engine = new OrchestrationEngine(inMemoryStore); - await engine.start(); - await engine.stop(); - await expect(engine.stop()).resolves.toBeUndefined(); - }); - - it("keeps processing queued commands after a storage failure", async () => { - const events: OrchestrationEvent[] = []; - let nextSequence = 1; - let shouldFailFirstAppend = true; - - const flakyStore: OrchestrationEventRepositoryShape = { - append(event) { - if (shouldFailFirstAppend) { - shouldFailFirstAppend = false; - return Effect.die(new Error("append failed")); - } - const savedEvent = { - ...event, - sequence: nextSequence, - } satisfies OrchestrationEvent; - nextSequence += 1; - events.push(savedEvent); - return Effect.succeed(savedEvent); - }, - readFromSequence(sequenceExclusive) { - return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive)); - }, - readAll() { - return Effect.succeed(events); - }, - }; - - const engine = new OrchestrationEngine(flakyStore); - const createdAt = new Date().toISOString(); - await engine.start(); - - await expect( - engine.dispatch({ - type: "thread.create", - commandId: "cmd-flaky-1", - threadId: "thread-flaky-fail", - projectId: "project-flaky", - title: "flaky-fail", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ).rejects.toThrow("append failed"); - - const result = await engine.dispatch({ - type: "thread.create", - commandId: "cmd-flaky-2", - threadId: "thread-flaky-ok", - projectId: "project-flaky", - title: "flaky-ok", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }); - - expect(result.sequence).toBe(1); - expect(engine.getSnapshot().sequence).toBe(1); - - await engine.stop(); - }); -}); diff --git a/apps/server/src/orchestration/engine.ts b/apps/server/src/orchestration/engine.ts deleted file mode 100644 index 1f66c21d058af45101aca3024588070e41972dbd..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/engine.ts +++ /dev/null @@ -1,309 +0,0 @@ -import type { - OrchestrationCommand, - OrchestrationEvent, - OrchestrationReadModel, -} from "@t3tools/contracts"; -import { OrchestrationCommandSchema } from "@t3tools/contracts"; -import { - PubSub, - Queue, - Schema, - Stream, - Effect, - Fiber, - Runtime, - Either, - Deferred, - Cause, -} from "effect"; - -import { createLogger } from "../logger"; -import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents"; -import { createEmptyReadModel, reduceEvent } from "./reducer"; - -type CommandEnvelope = { - command: OrchestrationCommand; - result: Deferred.Deferred<{ sequence: number }, Error>; -}; - -function asError(error: unknown, fallbackMessage: string): Error { - return error instanceof Error ? error : new Error(fallbackMessage); -} - -function mapCommandToEvent(command: OrchestrationCommand): Omit { - const eventId = crypto.randomUUID(); - switch (command.type) { - case "thread.create": - return { - eventId, - type: "thread.created", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.threadId, - projectId: command.projectId, - title: command.title, - model: command.model, - branch: command.branch, - worktreePath: command.worktreePath, - createdAt: command.createdAt, - updatedAt: command.createdAt, - }, - }; - case "thread.delete": - return { - eventId, - type: "thread.deleted", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.threadId, - deletedAt: command.createdAt, - }, - }; - case "thread.meta.update": - return { - eventId, - type: "thread.meta-updated", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - ...(command.title !== undefined ? { title: command.title } : {}), - ...(command.model !== undefined ? { model: command.model } : {}), - ...(command.branch !== undefined ? { branch: command.branch } : {}), - ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}), - updatedAt: command.createdAt, - }, - }; - case "message.send": - return { - eventId, - type: "message.sent", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.messageId, - role: command.role, - text: command.text, - threadId: command.threadId, - createdAt: command.createdAt, - streaming: command.streaming === true, - }, - }; - case "thread.session": - return { - eventId, - type: "thread.session-set", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - session: command.session, - }, - }; - case "git.readModel.upsert": - return { - eventId, - type: "git.read-model-upsert", - aggregateType: "project", - aggregateId: command.projectId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - projectId: command.projectId, - branch: command.branch, - hasWorkingTreeChanges: command.hasWorkingTreeChanges, - aheadCount: command.aheadCount, - behindCount: command.behindCount, - updatedAt: command.createdAt, - }, - }; - case "thread.turnDiff.complete": - return { - eventId, - type: "thread.turn-diff-completed", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - turnId: command.turnId, - completedAt: command.completedAt, - ...(command.status !== undefined ? { status: command.status } : {}), - files: command.files, - ...(command.assistantMessageId !== undefined - ? { assistantMessageId: command.assistantMessageId } - : {}), - ...(command.checkpointTurnCount !== undefined - ? { checkpointTurnCount: command.checkpointTurnCount } - : {}), - }, - }; - case "thread.revert": - return { - eventId, - type: "thread.reverted", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - turnCount: command.turnCount, - messageCount: command.messageCount, - }, - }; - } -} - -export class OrchestrationEngine { - private readonly logger = createLogger("orchestration"); - private readonly runtime = Runtime.defaultRuntime; - private readonly eventStore: OrchestrationEventRepositoryShape; - - private readModel: OrchestrationReadModel; - private commandQueue: Queue.Queue; - private readModelPubSub: PubSub.PubSub; - private eventPubSub: PubSub.PubSub; - private workerFiber: Fiber.RuntimeFiber | null = null; - private readonly readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>(); - private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); - - constructor(eventStore: OrchestrationEventRepositoryShape) { - this.eventStore = eventStore; - this.readModel = createEmptyReadModel(new Date().toISOString()); - this.commandQueue = Runtime.runSync(this.runtime)(Queue.unbounded()); - this.readModelPubSub = Runtime.runSync(this.runtime)( - PubSub.unbounded(), - ); - this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded()); - } - - private processEnvelope(envelope: CommandEnvelope): Effect.Effect { - return Effect.gen(this, function* () { - const eventBase = mapCommandToEvent(envelope.command); - const savedEvent = yield* this.eventStore.append(eventBase); - this.readModel = yield* reduceEvent(this.readModel, savedEvent); - - const snapshot = this.readModel; - yield* Effect.all([ - PubSub.publish(this.eventPubSub, savedEvent), - PubSub.publish(this.readModelPubSub, snapshot), - ]); - - for (const listener of this.domainEventListeners) { - listener(savedEvent); - } - for (const listener of this.readModelListeners) { - listener(snapshot); - } - - yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); - }).pipe( - Effect.catchAllCause((cause) => - Deferred.fail( - envelope.result, - asError(Cause.squash(cause), "Unknown command processing error"), - ).pipe(Effect.asVoid), - ), - ); - } - - private bootstrapReadModel(): Effect.Effect { - return Effect.gen(this, function* () { - const existingEvents = yield* this.eventStore.readAll(); - for (const event of existingEvents) { - this.readModel = yield* reduceEvent(this.readModel, event); - } - }); - } - - async start(): Promise { - if (this.workerFiber) { - return; - } - - await Runtime.runPromise(this.runtime)(this.bootstrapReadModel()); - - const worker = Stream.fromQueue(this.commandQueue).pipe( - Stream.runForEach((envelope) => this.processEnvelope(envelope)), - ); - - this.workerFiber = Runtime.runFork(this.runtime)(worker); - this.logger.info("orchestration engine started", { - sequence: this.readModel.sequence, - }); - } - - async stop(): Promise { - if (this.workerFiber) { - await Runtime.runPromise(this.runtime)(Fiber.interrupt(this.workerFiber)); - this.workerFiber = null; - } - } - - getSnapshot(): OrchestrationReadModel { - return this.readModel; - } - - async replayEvents(fromSequenceExclusive: number): Promise { - return Runtime.runPromise(this.runtime)( - this.eventStore.readFromSequence(fromSequenceExclusive), - ); - } - - async dispatchUnknown(command: unknown): Promise<{ sequence: number }> { - const decode = Schema.decodeUnknownEither(OrchestrationCommandSchema); - let payload: unknown = command; - if (typeof command === "string") { - try { - payload = JSON.parse(command) as unknown; - } catch { - throw new Error("Invalid orchestration command: payload is not valid JSON"); - } - } - const decoded = decode(payload); - if (Either.isLeft(decoded)) { - const issues = decoded.left.toString(); - throw new Error(`Invalid orchestration command: ${issues}`); - } - return this.dispatch(decoded.right); - } - - async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> { - const program = Effect.gen(this, function* () { - const result = yield* Deferred.make<{ sequence: number }, Error>(); - yield* Queue.offer(this.commandQueue, { command, result }); - return yield* Deferred.await(result); - }); - return Runtime.runPromise(this.runtime)(program).catch((error) => { - throw asError(error, "Queue offer failed"); - }); - } - - subscribeToReadModel(callback: (snapshot: OrchestrationReadModel) => void): () => void { - this.readModelListeners.add(callback); - return () => { - this.readModelListeners.delete(callback); - }; - } - - subscribeToDomainEvents(callback: (event: OrchestrationEvent) => void): () => void { - this.domainEventListeners.add(callback); - return () => { - this.domainEventListeners.delete(callback); - }; - } -} diff --git a/apps/server/src/orchestration/layers.ts b/apps/server/src/orchestration/layers.ts deleted file mode 100644 index 18670b49ff6f7d0dcdcd5d96c53949f9dbf9a1ba..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/layers.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { Effect, Layer } from "effect"; - -import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents"; -import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents"; -import { OrchestrationEngine } from "./engine"; -import { OrchestrationEngineService } from "./services"; - -export const OrchestrationEngineLive = Layer.scoped( - OrchestrationEngineService, - Effect.acquireRelease( - Effect.gen(function* () { - const eventRepository = yield* OrchestrationEventRepository; - const engine = new OrchestrationEngine(eventRepository); - yield* Effect.promise(() => engine.start()); - return engine; - }), - (engine) => Effect.promise(() => engine.stop()), - ), -); - -export const OrchestrationLive = OrchestrationEngineLive.pipe( - Layer.provide(OrchestrationEventRepositoryLive), -); diff --git a/apps/server/src/orchestration/services.ts b/apps/server/src/orchestration/services.ts deleted file mode 100644 index f6621a54f3233a38b8df9858c6b55d4da2e9e319..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/services.ts +++ /dev/null @@ -1,8 +0,0 @@ -import { Context } from "effect"; - -import { OrchestrationEngine } from "./engine"; - -export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")< - OrchestrationEngineService, - OrchestrationEngine ->() {} diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index d63b8677888da8866aa81583dddd6add3605e98f..42fb00d352692958f6d4f8165cb502fea86e21ee 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -36,13 +36,13 @@ import { import { TerminalManager } from "./terminalManager"; import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings"; import { searchWorkspaceEntries } from "./workspaceEntries"; -import type { OrchestrationEngine } from "./orchestration/engine"; -import { OrchestrationLive } from "./orchestration/layers"; -import { OrchestrationEngineService } from "./orchestration/services"; +import { OrchestrationEngineLive } from "./orchestration/Layer"; +import { OrchestrationEngineService, type OrchestrationEngineShape } from "./orchestration/Service"; import { ProjectRepositoryLive } from "./persistence/Layers/Projects"; import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects"; import assert from "node:assert"; +import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents"; const MIME_TYPES: Record = { ".html": "text/html; charset=utf-8", @@ -136,7 +136,7 @@ export function createServer(options: ServerOptions) { unknown > | null = null; let projectRepository: ProjectRepositoryShape | null = null; - let orchestrationEngine: OrchestrationEngine | null = null; + let orchestrationEngine: OrchestrationEngineShape | null = null; const clients = new Set(); const logger = createLogger("ws"); const logWebSocketEvents = @@ -169,141 +169,157 @@ export function createServer(options: ServerOptions) { return effectRuntime; }; - function attachOrchestrationSubscriptions() { - const orchestrationEngine = getOrchestrationEngine(); - - unsubscribeReadModel = orchestrationEngine.subscribeToReadModel((snapshot) => { - const push: WsPush = { - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.readModel, - data: { - sequence: snapshot.sequence, - snapshot, - } satisfies OrchestrationReadModelPush, - }; - const message = JSON.stringify(push); - let recipients = 0; - for (const client of clients) { - if (client.readyState === client.OPEN) { - client.send(message); - recipients += 1; + async function attachOrchestrationSubscriptions( + runtime: ManagedRuntime.ManagedRuntime, + orchestrationEngine: OrchestrationEngineShape, + ) { + unsubscribeReadModel = await runtime.runPromise( + orchestrationEngine.subscribeToReadModel((snapshot) => { + const push: WsPush = { + type: "push", + channel: ORCHESTRATION_WS_CHANNELS.readModel, + data: { + sequence: snapshot.sequence, + snapshot, + } satisfies OrchestrationReadModelPush, + }; + 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); - }); + logOutgoingPush(push, recipients); + }), + ); - unsubscribeDomainEvents = orchestrationEngine.subscribeToDomainEvents((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; + unsubscribeDomainEvents = await runtime.runPromise( + orchestrationEngine.subscribeToDomainEvents((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); - }); + logOutgoingPush(push, recipients); + }), + ); } // Forward provider events to all connected WebSocket clients providerManager.on("event", (event) => { void (async () => { const liveOrchestrationEngine = orchestrationEngine; - if (!liveOrchestrationEngine) { + const runtime = effectRuntime; + if (!liveOrchestrationEngine || !runtime) { return; } - const snapshot = liveOrchestrationEngine.getSnapshot(); + const snapshot = await runtime.runPromise(liveOrchestrationEngine.getSnapshot()); const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId); if (!thread) return; const now = event.createdAt; if (event.method === "turn/started" || event.method === "turn/completed") { - await liveOrchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - sessionId: event.sessionId, - provider: event.provider, - status: event.method === "turn/started" ? "running" : "ready", + await runtime.runPromise( + liveOrchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), threadId: thread.id, - activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null, - createdAt: thread.session?.createdAt ?? now, - updatedAt: now, - lastError: null, - }, - createdAt: now, - }); + session: { + sessionId: event.sessionId, + provider: event.provider, + status: event.method === "turn/started" ? "running" : "ready", + threadId: thread.id, + activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null, + createdAt: thread.session?.createdAt ?? now, + updatedAt: now, + lastError: null, + }, + createdAt: now, + }), + ); } if (event.textDelta && event.textDelta.length > 0) { const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`; - await liveOrchestrationEngine.dispatch({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: thread.id, - messageId: assistantMessageId, - role: "assistant", - text: event.textDelta, - streaming: true, - createdAt: now, - }); + await runtime.runPromise( + liveOrchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: event.textDelta, + streaming: true, + createdAt: now, + }), + ); } if (event.method === "turn/completed") { const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`; - await liveOrchestrationEngine.dispatch({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: thread.id, - messageId: assistantMessageId, - role: "assistant", - text: "", - streaming: false, - createdAt: now, - }); + await runtime.runPromise( + liveOrchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: "", + streaming: false, + createdAt: now, + }), + ); } if (event.method === "checkpoint/captured") { const payload = asObject(event.payload); const turnId = event.turnId ?? asString(payload?.turnId); if (turnId) { - await liveOrchestrationEngine.dispatch({ - type: "thread.turnDiff.complete", - commandId: crypto.randomUUID(), - threadId: thread.id, - turnId, - completedAt: now, - ...(asString(payload?.status) !== null - ? { status: asString(payload?.status) ?? undefined } - : {}), - files: [], - assistantMessageId: `assistant:${turnId}`, - ...(asNumber(payload?.turnCount) !== null - ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined } - : {}), - createdAt: now, - }); + await runtime.runPromise( + liveOrchestrationEngine.dispatch({ + type: "thread.turnDiff.complete", + commandId: crypto.randomUUID(), + threadId: thread.id, + turnId, + completedAt: now, + ...(asString(payload?.status) !== null + ? { status: asString(payload?.status) ?? undefined } + : {}), + files: [], + assistantMessageId: `assistant:${turnId}`, + ...(asNumber(payload?.turnCount) !== null + ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined } + : {}), + createdAt: now, + }), + ); } } if (event.kind === "error" && event.message) { - await liveOrchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - sessionId: event.sessionId, - provider: event.provider, - status: "error", + await runtime.runPromise( + liveOrchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), threadId: thread.id, - activeTurnId: event.turnId ?? null, - createdAt: thread.session?.createdAt ?? now, - updatedAt: now, - lastError: event.message, - }, - createdAt: now, - }); + 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, + }), + ); } })().catch(() => undefined); }); @@ -410,6 +426,7 @@ export function createServer(options: ServerOptions) { wss.on("connection", (ws) => { clients.add(ws); const orchestrationEngine = getOrchestrationEngine(); + const runtime = getEffectRuntime(); // Send welcome message with project info const segments = cwd.split(/[/\\]/).filter(Boolean); @@ -423,16 +440,24 @@ export function createServer(options: ServerOptions) { logOutgoingPush(welcome, 1); ws.send(JSON.stringify(welcome)); - const snapshotPush: WsPush = { - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.readModel, - data: { - sequence: orchestrationEngine.getSnapshot().sequence, - snapshot: orchestrationEngine.getSnapshot(), - } satisfies OrchestrationReadModelPush, - }; - logOutgoingPush(snapshotPush, 1); - ws.send(JSON.stringify(snapshotPush)); + void runtime + .runPromise(orchestrationEngine.getSnapshot()) + .then((snapshot) => { + if (ws.readyState !== ws.OPEN) { + return; + } + const snapshotPush: WsPush = { + type: "push", + channel: ORCHESTRATION_WS_CHANNELS.readModel, + data: { + sequence: snapshot.sequence, + snapshot, + } satisfies OrchestrationReadModelPush, + }; + logOutgoingPush(snapshotPush, 1); + ws.send(JSON.stringify(snapshotPush)); + }) + .catch(() => undefined); ws.on("message", (raw) => { void handleMessage(ws, raw); @@ -509,32 +534,36 @@ export function createServer(options: ServerOptions) { const params = request.params as { sessionId?: string }; const sessionId = params.sessionId; if (typeof sessionId === "string") { - const snapshot = orchestrationEngine.getSnapshot(); + const snapshot = await effectRuntime.runPromise(orchestrationEngine.getSnapshot()); const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId); if (thread) { const now = new Date().toISOString(); - await orchestrationEngine.dispatch({ - type: "thread.revert", - commandId: crypto.randomUUID(), - threadId: thread.id, - turnCount: result.turnCount, - messageCount: result.messageCount, - createdAt: now, - }); - if (thread.session) { - await orchestrationEngine.dispatch({ - type: "thread.session", + await effectRuntime.runPromise( + orchestrationEngine.dispatch({ + type: "thread.revert", commandId: crypto.randomUUID(), threadId: thread.id, - session: { - ...thread.session, - status: "ready", - activeTurnId: null, - updatedAt: now, - lastError: null, - }, + turnCount: result.turnCount, + messageCount: result.messageCount, createdAt: now, - }); + }), + ); + if (thread.session) { + await effectRuntime.runPromise( + orchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), + threadId: thread.id, + session: { + ...thread.session, + status: "ready", + activeTurnId: null, + updatedAt: now, + lastError: null, + }, + createdAt: now, + }), + ); } } } @@ -605,16 +634,18 @@ export function createServer(options: ServerOptions) { const projects = await effectRuntime.runPromise(projectRepository.list()); const project = projects.find((entry) => entry.cwd === params.cwd); if (project) { - await orchestrationEngine.dispatch({ - type: "git.readModel.upsert", - commandId: crypto.randomUUID(), - projectId: project.id, - branch: status.branch, - hasWorkingTreeChanges: status.hasWorkingTreeChanges, - aheadCount: status.aheadCount, - behindCount: status.behindCount, - createdAt: new Date().toISOString(), - }); + await effectRuntime.runPromise( + orchestrationEngine.dispatch({ + type: "git.readModel.upsert", + commandId: crypto.randomUUID(), + projectId: project.id, + branch: status.branch, + hasWorkingTreeChanges: status.hasWorkingTreeChanges, + aheadCount: status.aheadCount, + behindCount: status.behindCount, + createdAt: new Date().toISOString(), + }), + ); } return status; } @@ -677,19 +708,21 @@ export function createServer(options: ServerOptions) { }; case ORCHESTRATION_WS_METHODS.getSnapshot: - return orchestrationEngine.getSnapshot(); + return effectRuntime.runPromise(orchestrationEngine.getSnapshot()); case ORCHESTRATION_WS_METHODS.dispatchCommand: - return orchestrationEngine.dispatchUnknown(request.params); + return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params)); case ORCHESTRATION_WS_METHODS.replayEvents: - return orchestrationEngine.replayEvents( - Math.max( - 0, - Math.floor( - Number( - (request.params as { fromSequenceExclusive?: number } | undefined) - ?.fromSequenceExclusive ?? 0, + return effectRuntime.runPromise( + orchestrationEngine.replayEvents( + Math.max( + 0, + Math.floor( + Number( + (request.params as { fromSequenceExclusive?: number } | undefined) + ?.fromSequenceExclusive ?? 0, + ), ), ), ), @@ -703,11 +736,14 @@ export function createServer(options: ServerOptions) { async function createEffectRuntime() { const dbPath = path.join(resolvedStateDir, "orchestration.sqlite"); const persistenceLayer = makeSqlitePersistenceLive(dbPath); - const runtime = ManagedRuntime.make( - Layer.mergeAll(OrchestrationLive, ProjectRepositoryLive).pipe( - Layer.provide(persistenceLayer), - ), + const orchestrationLayer = Layer.provide( + OrchestrationEngineLive, + OrchestrationEventRepositoryLive, ); + const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive).pipe( + Layer.provide(persistenceLayer), + ); + const runtime = ManagedRuntime.make(layer); try { const [nextOrchestrationEngine, repository] = await Promise.all([ @@ -717,7 +753,7 @@ export function createServer(options: ServerOptions) { orchestrationEngine = nextOrchestrationEngine; projectRepository = repository; await runtime.runPromise(repository.pruneMissing()); - attachOrchestrationSubscriptions(); + await attachOrchestrationSubscriptions(runtime, nextOrchestrationEngine); } catch (error) { await runtime.dispose().catch(() => undefined); throw error;