From 7a5ea7a5d8d2a779cafb6413306eed554977825c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 22 Feb 2026 01:00:48 -0800 Subject: [PATCH] more effect: ProjectRegistry and shared runtime --- apps/server/src/index.ts | 3 - apps/server/src/orchestration/engine.test.ts | 106 ++-- apps/server/src/orchestration/engine.ts | 59 +- apps/server/src/orchestration/layers.test.ts | 47 -- apps/server/src/orchestration/layers.ts | 13 +- apps/server/src/orchestration/reducer.test.ts | 97 ++++ apps/server/src/orchestration/reducer.ts | 526 +++++++++--------- apps/server/src/orchestration/runtime.ts | 22 - apps/server/src/orchestration/services.ts | 9 - .../src/orchestration/uiContractInventory.ts | 10 +- .../Layers/OrchestrationEvents.test.ts | 10 +- .../persistence/Layers/OrchestrationEvents.ts | 37 +- .../src/persistence/Layers/Projects.test.ts | 83 +++ .../server/src/persistence/Layers/Projects.ts | 339 +++++++++++ apps/server/src/persistence/Layers/Sqlite.ts | 41 ++ apps/server/src/persistence/Migrations.ts | 5 +- .../Migrations/001_OrchestrationEvents.ts | 2 +- .../persistence/Migrations/002_Projects.ts | 22 + .../src/persistence/Services/Projects.ts | 25 + apps/server/src/projectRegistry.test.ts | 113 ---- apps/server/src/projectRegistry.ts | 235 -------- apps/server/src/wsServer.test.ts | 53 +- apps/server/src/wsServer.ts | 287 +++++----- apps/web/src/components/Sidebar.tsx | 67 +-- apps/web/src/routes/__root.tsx | 48 +- apps/web/src/store.ts | 49 +- packages/contracts/src/orchestration.ts | 31 -- 27 files changed, 1204 insertions(+), 1135 deletions(-) delete mode 100644 apps/server/src/orchestration/layers.test.ts create mode 100644 apps/server/src/orchestration/reducer.test.ts delete mode 100644 apps/server/src/orchestration/runtime.ts create mode 100644 apps/server/src/persistence/Layers/Projects.test.ts create mode 100644 apps/server/src/persistence/Layers/Projects.ts create mode 100644 apps/server/src/persistence/Layers/Sqlite.ts create mode 100644 apps/server/src/persistence/Migrations/002_Projects.ts create mode 100644 apps/server/src/persistence/Services/Projects.ts delete mode 100644 apps/server/src/projectRegistry.test.ts delete mode 100644 apps/server/src/projectRegistry.ts diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 159dbbc2a88d5576acec002c83c46b23d9de7bc2..00886cb8d054eaf67739814678efe5c64e8f2d34 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -6,7 +6,6 @@ import { fileURLToPath } from "node:url"; import { fixPath } from "./fixPath"; import { createLogger } from "./logger"; -import { ProjectRegistry } from "./projectRegistry"; import { createServer } from "./wsServer"; fixPath(); @@ -102,7 +101,6 @@ async function main() { const port = requestedPort ?? (mode === "desktop" ? DEFAULT_PORT : await findAvailablePort(DEFAULT_PORT)); const stateDir = resolveStateDir(process.env.T3CODE_STATE_DIR); - const projectRegistry = new ProjectRegistry(stateDir); const devUrl = process.env.VITE_DEV_SERVER_URL; const noBrowser = parseBooleanEnv(process.env.T3CODE_NO_BROWSER) ?? mode === "desktop"; const authToken = process.env.T3CODE_AUTH_TOKEN; @@ -121,7 +119,6 @@ async function main() { stateDir, staticDir, devUrl, - projectRegistry, authToken, }); await server.start(); diff --git a/apps/server/src/orchestration/engine.test.ts b/apps/server/src/orchestration/engine.test.ts index 666dff3d5bbeb52214e6243243eb4788d930aaf7..d8c7813cfc1dac21dc83711609525ce36a5dc467 100644 --- a/apps/server/src/orchestration/engine.test.ts +++ b/apps/server/src/orchestration/engine.test.ts @@ -8,7 +8,13 @@ import { afterEach, describe, expect, it } from "vitest"; import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents"; import { OrchestrationEngine } from "./engine"; -import { createOrchestrationSystem } from "./runtime"; + +import { Layer, ManagedRuntime } from "effect"; + +import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite"; + +import { OrchestrationLive } from "./layers"; +import { OrchestrationEngineService } from "./services"; const tempDirs: string[] = []; @@ -24,6 +30,19 @@ afterEach(() => { } }); +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-"); @@ -34,17 +53,8 @@ describe("OrchestrationEngine", () => { const firstSystem = await createOrchestrationSystem(stateDir); const engineA = firstSystem.engine; await engineA.dispatch({ - type: "project.create", + type: "thread.create", commandId: "cmd-1", - projectId, - name: "demo", - cwd: "/tmp/demo", - model: "gpt-5-codex", - createdAt, - }); - await engineA.dispatch({ - type: "thread.create", - commandId: "cmd-2", threadId, projectId, title: "Thread", @@ -82,12 +92,14 @@ describe("OrchestrationEngine", () => { updates.push(snapshot.sequence); }); await engine.dispatch({ - type: "project.create", - commandId: "cmd-project", + type: "thread.create", + commandId: "cmd-thread", + threadId: "thread-2", projectId: "project-2", - name: "fanout", - cwd: "/tmp/fanout", + title: "fanout", model: "gpt-5-codex", + branch: null, + worktreePath: null, createdAt: new Date().toISOString(), }); unsubscribe(); @@ -100,24 +112,26 @@ describe("OrchestrationEngine", () => { const system = await createOrchestrationSystem(stateDir); const engine = system.engine; await engine.dispatch({ - type: "project.create", + type: "thread.create", commandId: "cmd-a", + threadId: "thread-replay", projectId: "project-replay", - name: "replay", - cwd: "/tmp/replay", + title: "replay", model: "gpt-5-codex", + branch: null, + worktreePath: null, createdAt: new Date().toISOString(), }); await engine.dispatch({ - type: "project.delete", + type: "thread.delete", commandId: "cmd-b", - projectId: "project-replay", + threadId: "thread-replay", createdAt: new Date().toISOString(), }); const events = await engine.replayEvents(0); expect(events.length).toBe(2); - expect(events[0]?.type).toBe("project.created"); - expect(events[1]?.type).toBe("project.deleted"); + expect(events[0]?.type).toBe("thread.created"); + expect(events[1]?.type).toBe("thread.deleted"); await system.dispose(); }); @@ -126,15 +140,6 @@ describe("OrchestrationEngine", () => { const firstSystem = await createOrchestrationSystem(stateDir); const engine = firstSystem.engine; const completedAt = new Date().toISOString(); - await engine.dispatch({ - type: "project.create", - commandId: "cmd-project-turn-diff", - projectId: "project-turn-diff", - name: "turn-diff", - cwd: "/tmp/turn-diff", - model: "gpt-5-codex", - createdAt: completedAt, - }); await engine.dispatch({ type: "thread.create", commandId: "cmd-thread-turn-diff", @@ -158,7 +163,9 @@ describe("OrchestrationEngine", () => { createdAt: completedAt, }); - const firstThread = engine.getSnapshot().threads.find((thread) => thread.id === "thread-turn-diff"); + const firstThread = engine + .getSnapshot() + .threads.find((thread) => thread.id === "thread-turn-diff"); expect(firstThread?.turnDiffSummaries).toEqual([ { turnId: "turn-1", @@ -192,15 +199,6 @@ describe("OrchestrationEngine", () => { const firstSystem = await createOrchestrationSystem(stateDir); const engine = firstSystem.engine; const createdAt = new Date().toISOString(); - await engine.dispatch({ - type: "project.create", - commandId: "cmd-project-revert", - projectId: "project-revert", - name: "revert", - cwd: "/tmp/revert", - model: "gpt-5-codex", - createdAt, - }); await engine.dispatch({ type: "thread.create", commandId: "cmd-thread-revert", @@ -334,9 +332,7 @@ describe("OrchestrationEngine", () => { append(event) { if (shouldFailFirstAppend) { shouldFailFirstAppend = false; - return Effect.sync(() => { - throw new Error("append failed"); - }); + return Effect.die(new Error("append failed")); } const savedEvent = { ...event, @@ -360,28 +356,32 @@ describe("OrchestrationEngine", () => { await expect( engine.dispatch({ - type: "project.create", + type: "thread.create", commandId: "cmd-flaky-1", - projectId: "project-flaky-fail", - name: "flaky-fail", - cwd: "/tmp/flaky-fail", + 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: "project.create", + type: "thread.create", commandId: "cmd-flaky-2", - projectId: "project-flaky-ok", - name: "flaky-ok", - cwd: "/tmp/flaky-ok", + 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().projects.map((project) => project.id)).toEqual(["project-flaky-ok"]); + 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 index 4069645dd90f416e67954030d3ddd081a1b62b4d..fda67ce71a38e14ea86c90e5b3d53f79ce584d63 100644 --- a/apps/server/src/orchestration/engine.ts +++ b/apps/server/src/orchestration/engine.ts @@ -34,36 +34,6 @@ function asError(error: unknown, fallbackMessage: string): Error { function mapCommandToEvent(command: OrchestrationCommand): Omit { const eventId = crypto.randomUUID(); switch (command.type) { - case "project.create": - return { - eventId, - type: "project.created", - aggregateType: "project", - aggregateId: command.projectId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.projectId, - name: command.name, - cwd: command.cwd, - model: command.model, - createdAt: command.createdAt, - updatedAt: command.createdAt, - }, - }; - case "project.delete": - return { - eventId, - type: "project.deleted", - aggregateType: "project", - aggregateId: command.projectId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.projectId, - deletedAt: command.createdAt, - }, - }; case "thread.create": return { eventId, @@ -226,10 +196,7 @@ export class OrchestrationEngine { return Effect.gen(this, function* () { const eventBase = mapCommandToEvent(envelope.command); const savedEvent = yield* this.eventStore.append(eventBase); - - yield* Effect.sync(() => { - this.readModel = reduceEvent(this.readModel, savedEvent); - }); + this.readModel = yield* reduceEvent(this.readModel, savedEvent); const snapshot = this.readModel; yield* Effect.all([ @@ -237,14 +204,12 @@ export class OrchestrationEngine { PubSub.publish(this.readModelPubSub, snapshot), ]); - yield* Effect.sync(() => { - for (const listener of this.domainEventListeners) { - listener(savedEvent); - } - for (const listener of this.readModelListeners) { - listener(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( @@ -257,14 +222,12 @@ export class OrchestrationEngine { ); } - private bootstrapReadModel(): Effect.Effect { + private bootstrapReadModel(): Effect.Effect { return Effect.gen(this, function* () { const existingEvents = yield* this.eventStore.readAll(); - yield* Effect.sync(() => { - for (const event of existingEvents) { - this.readModel = reduceEvent(this.readModel, event); - } - }); + for (const event of existingEvents) { + this.readModel = yield* reduceEvent(this.readModel, event); + } }); } diff --git a/apps/server/src/orchestration/layers.test.ts b/apps/server/src/orchestration/layers.test.ts deleted file mode 100644 index 440af03bbdb0d97dd09141dfc8ca622ca0c5a11c..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/layers.test.ts +++ /dev/null @@ -1,47 +0,0 @@ -import fs from "node:fs"; -import os from "node:os"; -import path from "node:path"; - -import { afterEach, describe, expect, it } from "vitest"; - -import { createOrchestrationSystem } from "./runtime"; - -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 }); - } -}); - -describe("orchestration layers", () => { - it("creates an engine through Context.Tag wiring", async () => { - const stateDir = makeTempDir("t3code-orchestration-layer-"); - const createdAt = new Date().toISOString(); - const firstSystem = await createOrchestrationSystem(stateDir); - const { engine } = firstSystem; - - await engine.dispatch({ - type: "project.create", - commandId: "cmd-layer-1", - projectId: "project-layer-1", - name: "layer", - cwd: "/tmp/layer", - model: "gpt-5-codex", - createdAt, - }); - const firstSnapshot = engine.getSnapshot(); - await firstSystem.dispose(); - - const secondSystem = await createOrchestrationSystem(stateDir); - const restarted = secondSystem.engine; - expect(restarted.getSnapshot()).toEqual(firstSnapshot); - await secondSystem.dispose(); - }); -}); diff --git a/apps/server/src/orchestration/layers.ts b/apps/server/src/orchestration/layers.ts index 63009c9c9b7bfc24fe6f5ce32debb7545b7a9664..18670b49ff6f7d0dcdcd5d96c53949f9dbf9a1ba 100644 --- a/apps/server/src/orchestration/layers.ts +++ b/apps/server/src/orchestration/layers.ts @@ -1,18 +1,9 @@ -import path from "node:path"; - import { Effect, Layer } from "effect"; import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents"; -import { makeSqliteOrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents"; +import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents"; import { OrchestrationEngine } from "./engine"; -import { OrchestrationConfig, OrchestrationEngineService } from "./services"; - -export const OrchestrationEventRepositoryLive = Layer.unwrapEffect( - Effect.map(OrchestrationConfig, ({ stateDir }) => { - const dbPath = path.join(stateDir, "orchestration.sqlite"); - return makeSqliteOrchestrationEventRepositoryLive(dbPath); - }), -); +import { OrchestrationEngineService } from "./services"; export const OrchestrationEngineLive = Layer.scoped( OrchestrationEngineService, diff --git a/apps/server/src/orchestration/reducer.test.ts b/apps/server/src/orchestration/reducer.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..6ea2a6c81352297792950e60623fb6a3377dc78c --- /dev/null +++ b/apps/server/src/orchestration/reducer.test.ts @@ -0,0 +1,97 @@ +import type { OrchestrationEvent } from "@t3tools/contracts"; +import { Effect } from "effect"; +import { describe, expect, it } from "vitest"; + +import { createEmptyReadModel, reduceEvent } from "./reducer"; + +describe("orchestration reducer", () => { + it("applies thread.created events using runtime schema decoding", async () => { + const now = new Date().toISOString(); + const model = createEmptyReadModel(now); + const event: OrchestrationEvent = { + sequence: 1, + eventId: "event-thread-created", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: now, + commandId: "cmd-1", + payload: { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }; + + const next = await Effect.runPromise(reduceEvent(model, event)); + expect(next.sequence).toBe(1); + expect(next.threads).toEqual([ + { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + turnDiffSummaries: [], + error: null, + }, + ]); + }); + + it("fails when event payload cannot be decoded by runtime schema", async () => { + const now = new Date().toISOString(); + const model = createEmptyReadModel(now); + const event: OrchestrationEvent = { + sequence: 1, + eventId: "event-message-invalid", + type: "message.sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: now, + commandId: "cmd-1", + payload: { + id: "message-1", + threadId: "thread-1", + text: "hello", + createdAt: now, + streaming: false, + }, + }; + + await expect(Effect.runPromise(reduceEvent(model, event))).rejects.toBeDefined(); + }); + + it("keeps reducer forward-compatible for unknown event types", async () => { + const now = new Date().toISOString(); + const model = createEmptyReadModel(now); + const event: OrchestrationEvent = { + sequence: 7, + eventId: "event-unknown", + type: "custom.event.ignored", + aggregateType: "unknown", + aggregateId: "aggregate-1", + occurredAt: "2026-01-01T00:00:00.000Z", + commandId: null, + payload: { test: true }, + }; + + const next = await Effect.runPromise(reduceEvent(model, event)); + expect(next.sequence).toBe(7); + expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z"); + expect(next.threads).toEqual([]); + }); +}); diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/reducer.ts index 14e2c237fb86c6ca1eb497aa10ef7e63c41c5335..1b05441b1f5e7ce6aa89e37081a8a0bf94f1961d 100644 --- a/apps/server/src/orchestration/reducer.ts +++ b/apps/server/src/orchestration/reducer.ts @@ -2,11 +2,20 @@ import type { OrchestrationEvent, OrchestrationGitReadModel, OrchestrationMessage, - OrchestrationProject, OrchestrationReadModel, OrchestrationSession, OrchestrationThread, } from "@t3tools/contracts"; +import { + OrchestrationGitReadModelSchema, + OrchestrationMessageSchema, + OrchestrationReadModelSchema, + OrchestrationSessionSchema, + OrchestrationThreadSchema, + OrchestrationTurnDiffFileSchema, + OrchestrationTurnDiffSummarySchema, +} from "@t3tools/contracts"; +import { Effect, Schema } from "effect"; type ThreadPatch = Partial>; @@ -18,38 +27,84 @@ function updateThread( return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread)); } -function asObject(value: unknown): Record | null { - if (!value || typeof value !== "object") return null; - return value as Record; -} +const decodeReadModelSync = Schema.decodeUnknownSync(OrchestrationReadModelSchema); +const decodeMessage = Schema.decodeUnknown(OrchestrationMessageSchema); +const decodeSession = Schema.decodeUnknown(OrchestrationSessionSchema); +const decodeThread = Schema.decodeUnknown(OrchestrationThreadSchema); +const decodeGitReadModel = Schema.decodeUnknown(OrchestrationGitReadModelSchema); +const decodeTurnDiffSummary = Schema.decodeUnknown(OrchestrationTurnDiffSummarySchema); -function asString(value: unknown): string | null { - return typeof value === "string" ? value : null; -} +const ThreadCreatedPayloadSchema = Schema.Struct({ + id: Schema.String, + projectId: Schema.String, + title: Schema.String, + model: Schema.String, + branch: Schema.NullOr(Schema.String), + worktreePath: Schema.NullOr(Schema.String), + createdAt: Schema.String, + updatedAt: Schema.String, +}); +const decodeThreadCreatedPayload = Schema.decodeUnknown(ThreadCreatedPayloadSchema); -function asNumber(value: unknown): number | null { - return typeof value === "number" && Number.isFinite(value) ? value : null; -} +const ThreadMetaUpdatedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + title: Schema.optional(Schema.String), + model: Schema.optional(Schema.String), + branch: Schema.optional(Schema.NullOr(Schema.String)), + worktreePath: Schema.optional(Schema.NullOr(Schema.String)), + updatedAt: Schema.String, +}); +const decodeThreadMetaUpdatedPayload = Schema.decodeUnknown(ThreadMetaUpdatedPayloadSchema); -function asArray(value: unknown): unknown[] { - return Array.isArray(value) ? value : []; -} +const MessageSentPayloadSchema = Schema.Struct({ + id: Schema.String, + role: Schema.Literal("user", "assistant"), + text: Schema.String, + threadId: Schema.String, + createdAt: Schema.String, + streaming: Schema.Boolean, +}); +const decodeMessageSentPayload = Schema.decodeUnknown(MessageSentPayloadSchema); + +const ThreadSessionSetPayloadSchema = Schema.Struct({ + threadId: Schema.String, + session: OrchestrationSessionSchema, +}); +const decodeThreadSessionSetPayload = Schema.decodeUnknown(ThreadSessionSetPayloadSchema); + +const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + turnId: Schema.String, + completedAt: Schema.String, + status: Schema.optional(Schema.String), + files: Schema.Array(OrchestrationTurnDiffFileSchema), + assistantMessageId: Schema.optional(Schema.String), + checkpointTurnCount: Schema.optional(Schema.Number), +}); +const decodeThreadTurnDiffCompletedPayload = Schema.decodeUnknown( + ThreadTurnDiffCompletedPayloadSchema, +); + +const ThreadRevertedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + turnCount: Schema.Number, + messageCount: Schema.Number, +}); +const decodeThreadRevertedPayload = Schema.decodeUnknown(ThreadRevertedPayloadSchema); export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { - return { + return decodeReadModelSync({ sequence: 0, - projects: [], threads: [], gitByProjectId: {}, updatedAt: nowIso, - }; + }); } export function reduceEvent( model: OrchestrationReadModel, event: OrchestrationEvent, -): OrchestrationReadModel { - const payload = asObject(event.payload); +): Effect.Effect { const nextBase: OrchestrationReadModel = { ...model, sequence: event.sequence, @@ -57,266 +112,215 @@ export function reduceEvent( }; switch (event.type) { - case "project.created": { - if (!payload) return nextBase; - const project: OrchestrationProject = { - id: asString(payload.id) ?? event.aggregateId, - name: asString(payload.name) ?? "project", - cwd: asString(payload.cwd) ?? "", - model: asString(payload.model) ?? "gpt-5-codex", - createdAt: asString(payload.createdAt) ?? event.occurredAt, - updatedAt: asString(payload.updatedAt) ?? event.occurredAt, - }; - const existing = nextBase.projects.find((entry) => entry.id === project.id); - return { - ...nextBase, - projects: existing - ? nextBase.projects.map((entry) => (entry.id === project.id ? project : entry)) - : [...nextBase.projects, project], - }; - } - case "project.deleted": - return { - ...nextBase, - projects: nextBase.projects.filter((project) => project.id !== event.aggregateId), - threads: nextBase.threads.filter((thread) => thread.projectId !== event.aggregateId), - }; - case "thread.created": { - if (!payload) return nextBase; - const thread: OrchestrationThread = { - id: asString(payload.id) ?? event.aggregateId, - projectId: asString(payload.projectId) ?? "", - title: asString(payload.title) ?? "New thread", - model: asString(payload.model) ?? "gpt-5-codex", - branch: asString(payload.branch), - worktreePath: asString(payload.worktreePath), - createdAt: asString(payload.createdAt) ?? event.occurredAt, - updatedAt: asString(payload.updatedAt) ?? event.occurredAt, - latestTurnId: null, - latestTurnStartedAt: null, - latestTurnCompletedAt: null, - latestTurnDurationMs: null, - messages: [], - session: null, - turnDiffSummaries: [], - error: null, - }; - const existing = nextBase.threads.find((entry) => entry.id === thread.id); - return { - ...nextBase, - threads: existing - ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry)) - : [...nextBase.threads, thread], - }; - } + case "thread.created": + return Effect.gen(function* () { + const payload = yield* decodeThreadCreatedPayload(event.payload); + const thread: OrchestrationThread = yield* decodeThread({ + id: payload.id, + projectId: payload.projectId, + title: payload.title, + model: payload.model, + branch: payload.branch, + worktreePath: payload.worktreePath, + createdAt: payload.createdAt, + updatedAt: payload.updatedAt, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + turnDiffSummaries: [], + error: null, + }); + const existing = nextBase.threads.find((entry) => entry.id === thread.id); + return { + ...nextBase, + threads: existing + ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry)) + : [...nextBase.threads, thread], + }; + }); case "thread.deleted": - return { + return Effect.succeed({ ...nextBase, threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId), - }; - case "thread.meta-updated": { - if (!payload) return nextBase; - const threadId = asString(payload.threadId) ?? event.aggregateId; - const thread = nextBase.threads.find((entry) => entry.id === threadId); - if (!thread) return nextBase; - return { - ...nextBase, - threads: updateThread(nextBase.threads, threadId, { - ...(asString(payload.title) !== null ? { title: asString(payload.title) ?? thread.title } : {}), - ...(asString(payload.model) !== null ? { model: asString(payload.model) ?? thread.model } : {}), - ...(payload.branch !== undefined ? { branch: asString(payload.branch) } : {}), - ...(payload.worktreePath !== undefined ? { worktreePath: asString(payload.worktreePath) } : {}), - updatedAt: asString(payload.updatedAt) ?? event.occurredAt, + }); + case "thread.meta-updated": + return decodeThreadMetaUpdatedPayload(event.payload).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + ...(payload.title !== undefined ? { title: payload.title } : {}), + ...(payload.model !== undefined ? { model: payload.model } : {}), + ...(payload.branch !== undefined ? { branch: payload.branch } : {}), + ...(payload.worktreePath !== undefined ? { worktreePath: payload.worktreePath } : {}), + updatedAt: payload.updatedAt, + }), + }; }), - }; - } - case "message.sent": { - if (!payload) return nextBase; - const threadId = asString(payload.threadId) ?? event.aggregateId; - const message: OrchestrationMessage = { - id: asString(payload.id) ?? crypto.randomUUID(), - role: (asString(payload.role) as OrchestrationMessage["role"] | null) ?? "user", - text: asString(payload.text) ?? "", - createdAt: asString(payload.createdAt) ?? event.occurredAt, - streaming: payload.streaming === true, - }; - const targetThread = nextBase.threads.find((thread) => thread.id === threadId); - if (!targetThread) return nextBase; - const existingMessage = targetThread.messages.find((entry) => entry.id === message.id); - const nextMessages = existingMessage - ? targetThread.messages.map((entry) => - entry.id === message.id - ? { - ...entry, - text: - message.streaming + ); + case "message.sent": + return Effect.gen(function* () { + const payload = yield* decodeMessageSentPayload(event.payload); + const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId); + if (!targetThread) { + return nextBase; + } + + const message: OrchestrationMessage = yield* decodeMessage({ + id: payload.id, + role: payload.role, + text: payload.text, + createdAt: payload.createdAt, + streaming: payload.streaming, + }); + const existingMessage = targetThread.messages.find((entry) => entry.id === message.id); + const nextMessages = existingMessage + ? targetThread.messages.map((entry) => + entry.id === message.id + ? { + ...entry, + text: message.streaming ? `${entry.text}${message.text}` : message.text.length > 0 ? message.text : entry.text, - streaming: message.streaming, - createdAt: message.createdAt, - } - : entry, - ) - : [...targetThread.messages, message]; - return { - ...nextBase, - threads: updateThread(nextBase.threads, threadId, { - messages: nextMessages, - updatedAt: event.occurredAt, - }), - }; - } - case "thread.session-set": { - if (!payload) return nextBase; - const threadId = asString(payload.threadId) ?? event.aggregateId; - const thread = nextBase.threads.find((entry) => entry.id === threadId); - if (!thread) return nextBase; - const sessionPayload = asObject(payload.session); - if (!sessionPayload) return nextBase; - const session: OrchestrationSession = { - sessionId: asString(sessionPayload.sessionId) ?? crypto.randomUUID(), - status: - (asString(sessionPayload.status) as OrchestrationSession["status"] | null) ?? "connecting", - provider: - (asString(sessionPayload.provider) as OrchestrationSession["provider"] | null) ?? "codex", - threadId, - activeTurnId: asString(sessionPayload.activeTurnId), - createdAt: asString(sessionPayload.createdAt) ?? event.occurredAt, - updatedAt: asString(sessionPayload.updatedAt) ?? event.occurredAt, - lastError: asString(sessionPayload.lastError), - }; - return { - ...nextBase, - threads: updateThread(nextBase.threads, threadId, { - session, - updatedAt: event.occurredAt, - error: session.lastError, - }), - }; - } - case "thread.turn-diff-completed": { - if (!payload) return nextBase; - const threadId = asString(payload.threadId) ?? event.aggregateId; - const thread = nextBase.threads.find((entry) => entry.id === threadId); - const turnId = asString(payload.turnId); - if (!thread || !turnId) return nextBase; + streaming: message.streaming, + createdAt: message.createdAt, + } + : entry, + ) + : [...targetThread.messages, message]; - const files = asArray(payload.files).flatMap((filePayload) => { - const file = asObject(filePayload); - const filePath = asString(file?.path); - if (!filePath) { - return []; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: nextMessages, + updatedAt: event.occurredAt, + }), + }; + }); + case "thread.session-set": + return Effect.gen(function* () { + const payload = yield* decodeThreadSessionSetPayload(event.payload); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; } - const nextFile: { - path: string; - kind?: string; - additions?: number; - deletions?: number; - } = { path: filePath }; - const kind = asString(file?.kind); - if (kind !== null) { - nextFile.kind = kind; - } - const additions = asNumber(file?.additions); - if (additions !== null) { - nextFile.additions = additions; - } - const deletions = asNumber(file?.deletions); - if (deletions !== null) { - nextFile.deletions = deletions; - } - return [nextFile]; + const session: OrchestrationSession = yield* decodeSession({ + ...payload.session, + threadId: payload.threadId, + }); + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + session, + updatedAt: event.occurredAt, + error: session.lastError, + }), + }; }); + case "thread.turn-diff-completed": + return Effect.gen(function* () { + const payload = yield* decodeThreadTurnDiffCompletedPayload(event.payload); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } - const nextSummary = { - turnId, - completedAt: asString(payload.completedAt) ?? event.occurredAt, - ...(asString(payload.status) !== null ? { status: asString(payload.status) ?? undefined } : {}), - files, - ...(asString(payload.assistantMessageId) !== null - ? { assistantMessageId: asString(payload.assistantMessageId) ?? undefined } - : {}), - ...(asNumber(payload.checkpointTurnCount) !== null - ? { checkpointTurnCount: asNumber(payload.checkpointTurnCount) ?? undefined } - : {}), - }; - const turnDiffSummaries = [...thread.turnDiffSummaries.filter((summary) => summary.turnId !== turnId), nextSummary] - .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); + const nextSummary = yield* decodeTurnDiffSummary({ + turnId: payload.turnId, + completedAt: payload.completedAt, + ...(payload.status !== undefined ? { status: payload.status } : {}), + files: payload.files, + ...(payload.assistantMessageId !== undefined + ? { assistantMessageId: payload.assistantMessageId } + : {}), + ...(payload.checkpointTurnCount !== undefined + ? { checkpointTurnCount: payload.checkpointTurnCount } + : {}), + }); + const turnDiffSummaries = [ + ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId), + nextSummary, + ].toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); - return { - ...nextBase, - threads: updateThread(nextBase.threads, threadId, { - latestTurnId: turnId, - latestTurnCompletedAt: nextSummary.completedAt, - turnDiffSummaries, - updatedAt: event.occurredAt, - }), - }; - } - case "thread.reverted": { - if (!payload) return nextBase; - const threadId = asString(payload.threadId) ?? event.aggregateId; - const thread = nextBase.threads.find((entry) => entry.id === threadId); - if (!thread) return nextBase; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + latestTurnId: payload.turnId, + latestTurnCompletedAt: nextSummary.completedAt, + turnDiffSummaries, + updatedAt: event.occurredAt, + }), + }; + }); + case "thread.reverted": + return decodeThreadRevertedPayload(event.payload).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } - const targetTurnCount = Math.max(0, Math.floor(asNumber(payload.turnCount) ?? 0)); - const targetMessageCount = Math.max(0, Math.floor(asNumber(payload.messageCount) ?? 0)); - const sortedSummaries = [...thread.turnDiffSummaries].toSorted((left, right) => - left.completedAt.localeCompare(right.completedAt), - ); - const inferredTurnCountByTurnId = new Map(); - for (let index = 0; index < sortedSummaries.length; index += 1) { - const summary = sortedSummaries[index]; - if (!summary) continue; - inferredTurnCountByTurnId.set(summary.turnId, index + 1); - } + const targetTurnCount = Math.max(0, Math.floor(payload.turnCount)); + const targetMessageCount = Math.max(0, Math.floor(payload.messageCount)); + const sortedSummaries = [...thread.turnDiffSummaries].toSorted((left, right) => + left.completedAt.localeCompare(right.completedAt), + ); + const inferredTurnCountByTurnId = new Map(); + for (let index = 0; index < sortedSummaries.length; index += 1) { + const summary = sortedSummaries[index]; + if (!summary) continue; + inferredTurnCountByTurnId.set(summary.turnId, index + 1); + } - const turnDiffSummaries = thread.turnDiffSummaries - .filter((summary) => { - const checkpointTurnCount = - summary.checkpointTurnCount ?? inferredTurnCountByTurnId.get(summary.turnId) ?? 0; - return checkpointTurnCount <= targetTurnCount; - }) - .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); - const latestSummary = - turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null; + const turnDiffSummaries = thread.turnDiffSummaries + .filter((summary) => { + const checkpointTurnCount = + summary.checkpointTurnCount ?? inferredTurnCountByTurnId.get(summary.turnId) ?? 0; + return checkpointTurnCount <= targetTurnCount; + }) + .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); + const latestSummary = + turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null; - return { - ...nextBase, - threads: updateThread(nextBase.threads, threadId, { - messages: thread.messages.slice(0, Math.min(targetMessageCount, thread.messages.length)), - turnDiffSummaries, - latestTurnId: latestSummary?.turnId ?? null, - latestTurnStartedAt: null, - latestTurnCompletedAt: latestSummary?.completedAt ?? null, - latestTurnDurationMs: null, - updatedAt: event.occurredAt, + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: thread.messages.slice( + 0, + Math.min(targetMessageCount, thread.messages.length), + ), + turnDiffSummaries, + latestTurnId: latestSummary?.turnId ?? null, + latestTurnStartedAt: null, + latestTurnCompletedAt: latestSummary?.completedAt ?? null, + latestTurnDurationMs: null, + updatedAt: event.occurredAt, + }), + }; }), - }; - } - case "git.read-model-upsert": { - if (!payload) return nextBase; - const projectId = asString(payload.projectId) ?? event.aggregateId; - const gitReadModel: OrchestrationGitReadModel = { - projectId, - branch: asString(payload.branch), - hasWorkingTreeChanges: payload.hasWorkingTreeChanges === true, - aheadCount: asNumber(payload.aheadCount) ?? 0, - behindCount: asNumber(payload.behindCount) ?? 0, - updatedAt: asString(payload.updatedAt) ?? event.occurredAt, - }; - return { - ...nextBase, - gitByProjectId: { - ...nextBase.gitByProjectId, - [projectId]: gitReadModel, - }, - }; - } + ); + case "git.read-model-upsert": + return decodeGitReadModel(event.payload).pipe( + Effect.map((gitReadModel: OrchestrationGitReadModel) => ({ + ...nextBase, + gitByProjectId: { + ...nextBase.gitByProjectId, + [gitReadModel.projectId]: gitReadModel, + }, + })), + ); default: - return nextBase; + return Effect.succeed(nextBase); } } diff --git a/apps/server/src/orchestration/runtime.ts b/apps/server/src/orchestration/runtime.ts deleted file mode 100644 index 67851be2f5432b2083636d55b4c610b1c7cea941..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/runtime.ts +++ /dev/null @@ -1,22 +0,0 @@ -import { Layer, ManagedRuntime } from "effect"; - -import type { OrchestrationEngine } from "./engine"; -import { OrchestrationLive } from "./layers"; -import { OrchestrationConfig, OrchestrationEngineService } from "./services"; - -export interface OrchestrationSystem { - readonly engine: OrchestrationEngine; - readonly dispose: () => Promise; -} - -export async function createOrchestrationSystem(stateDir: string): Promise { - const orchestrationLayer = OrchestrationLive.pipe( - Layer.provide(Layer.succeed(OrchestrationConfig, { stateDir })), - ); - const runtime = ManagedRuntime.make(orchestrationLayer); - const engine = await runtime.runPromise(OrchestrationEngineService); - return { - engine, - dispose: () => runtime.dispose(), - }; -} diff --git a/apps/server/src/orchestration/services.ts b/apps/server/src/orchestration/services.ts index 148a83ef8bdb000fa7f720c7b1b01b580a8fa573..f6621a54f3233a38b8df9858c6b55d4da2e9e319 100644 --- a/apps/server/src/orchestration/services.ts +++ b/apps/server/src/orchestration/services.ts @@ -2,15 +2,6 @@ import { Context } from "effect"; import { OrchestrationEngine } from "./engine"; -export interface OrchestrationConfigShape { - readonly stateDir: string; -} - -export class OrchestrationConfig extends Context.Tag("orchestration/Config")< - OrchestrationConfig, - OrchestrationConfigShape ->() {} - export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")< OrchestrationEngineService, OrchestrationEngine diff --git a/apps/server/src/orchestration/uiContractInventory.ts b/apps/server/src/orchestration/uiContractInventory.ts index d138b5871245e14b26ab06e99d202ae6eb156ca9..eb0dbed858ced9141e97265f68f81ffd7bbf6e5d 100644 --- a/apps/server/src/orchestration/uiContractInventory.ts +++ b/apps/server/src/orchestration/uiContractInventory.ts @@ -7,11 +7,6 @@ export interface UiEntityContract { // Explicit UI inventory used to keep server read models aligned // with what the current frontend renders. export const UI_ENTITY_CONTRACTS: UiEntityContract[] = [ - { - entity: "Project", - requiredFields: ["id", "name", "cwd", "model"], - consumedBy: ["apps/web/src/components/Sidebar.tsx"], - }, { entity: "Thread", requiredFields: [ @@ -61,6 +56,9 @@ export const UI_ENTITY_CONTRACTS: UiEntityContract[] = [ { entity: "GitReadModel", requiredFields: ["projectId", "branch", "hasWorkingTreeChanges", "aheadCount", "behindCount"], - consumedBy: ["apps/web/src/components/GitActionsControl.tsx", "apps/web/src/components/BranchToolbar.tsx"], + consumedBy: [ + "apps/web/src/components/GitActionsControl.tsx", + "apps/web/src/components/BranchToolbar.tsx", + ], }, ]; diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts index 57f75fb55d610bbe2486f22d6fe1450fa0cfa00e..b9768172b286a07ae63b6b153f8a4751081de42b 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts @@ -35,12 +35,12 @@ describe("OrchestrationEventRepository", () => { const saved = await Runtime.runPromise(runtime)( first.append({ eventId: "event-1", - type: "project.created", - aggregateType: "project", - aggregateId: "project-1", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-1", - payload: { id: "project-1", name: "demo" }, + payload: { id: "thread-1", projectId: "project-1", title: "demo" }, }), ); expect(saved.sequence).toBe(1); @@ -92,7 +92,7 @@ describe("OrchestrationEventRepository", () => { `, ) .all() as Array>; - expect(migrationRows).toHaveLength(1); + expect(migrationRows.length).toBeGreaterThanOrEqual(1); expect(Object.values(migrationRows[0] ?? {}).length).toBeGreaterThan(0); db.close(); }); diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.ts index 56bdf08bdb8a5799e684a5406c2b9627dcb66107..29f256582697364e481d742ef0c62218c751e30d 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.ts @@ -1,15 +1,10 @@ -import fs from "node:fs"; -import path from "node:path"; - import type { OrchestrationEvent } from "@t3tools/contracts"; import { OrchestrationEventSchema } from "@t3tools/contracts"; -import type { SqlClient as SqlClientService } from "@effect/sql/SqlClient"; import * as SqlClient from "@effect/sql/SqlClient"; import * as SqlSchema from "@effect/sql/SqlSchema"; -import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"; -import { Context, Effect, Layer, Schema } from "effect"; +import { Effect, Layer, Schema } from "effect"; -import { runMigrations } from "../Migrations"; +import { makeSqlitePersistenceLive } from "./Sqlite"; import { OrchestrationEventRepository, type OrchestrationEventRepositoryShape, @@ -45,11 +40,6 @@ const ReadFromSequenceRequestSchema = Schema.Struct({ limit: Schema.Number, }); -class OrchestrationSql extends Context.Tag("persistence/OrchestrationSql")< - OrchestrationSql, - SqlClientService ->() {} - function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { return decodeEvent({ sequence: row.sequence, @@ -64,7 +54,7 @@ function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { } const makeRepository = Effect.gen(function* () { - const sql = yield* OrchestrationSql; + const sql = yield* SqlClient.SqlClient; const appendEventRow = SqlSchema.single({ Request: AppendEventRequestSchema, @@ -155,24 +145,5 @@ export const OrchestrationEventRepositoryLive = Layer.effect( ); export function makeSqliteOrchestrationEventRepositoryLive(dbPath: string) { - const SqliteClientLive = Effect.gen(function* () { - yield* Effect.try({ - try: () => fs.mkdirSync(path.dirname(dbPath), { recursive: true }), - catch: (error) => Effect.die(error), - }) - return SqliteClient.layer({ filename: dbPath }); - }).pipe(Layer.unwrapEffect) - - const OrchestrationSqlLive = Layer.scoped( - OrchestrationSql, - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - yield* sql`PRAGMA journal_mode = WAL;`; - yield* sql`PRAGMA foreign_keys = ON;`; - yield* runMigrations; - return sql; - }), - ).pipe(Layer.provide(SqliteClientLive)); - - return OrchestrationEventRepositoryLive.pipe(Layer.provide(OrchestrationSqlLive)); + return OrchestrationEventRepositoryLive.pipe(Layer.provide(makeSqlitePersistenceLive(dbPath))); } diff --git a/apps/server/src/persistence/Layers/Projects.test.ts b/apps/server/src/persistence/Layers/Projects.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..a53ca6b221063ee12eda225dd02165b8f44facde --- /dev/null +++ b/apps/server/src/persistence/Layers/Projects.test.ts @@ -0,0 +1,83 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { Layer, ManagedRuntime } from "effect"; +import { afterEach, describe, expect, it } from "vitest"; + +import { ProjectRepository } from "../Services/Projects"; +import { ProjectRepositoryLive } from "./Projects"; +import { makeSqlitePersistenceLive } from "./Sqlite"; + +function makeProjectRepositoryTest(dbPath: string) { + return ProjectRepositoryLive.pipe(Layer.provide(makeSqlitePersistenceLive(dbPath))); +} + +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 }); + } +}); + +describe("ProjectRepository", () => { + it("persists projects and deduplicates adds by cwd", async () => { + const stateDir = makeTempDir("t3code-project-repo-state-"); + const projectDir = makeTempDir("t3code-project-repo-project-"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + + const firstRuntime = ManagedRuntime.make(makeProjectRepositoryTest(dbPath)); + const first = await firstRuntime.runPromise(ProjectRepository); + + const created = await firstRuntime.runPromise(first.add({ cwd: projectDir })); + expect(created.created).toBe(true); + + const duplicate = await firstRuntime.runPromise(first.add({ cwd: projectDir })); + expect(duplicate.created).toBe(false); + expect(duplicate.project.id).toBe(created.project.id); + + await firstRuntime.dispose(); + + const secondRuntime = ManagedRuntime.make(makeProjectRepositoryTest(dbPath)); + const second = await secondRuntime.runPromise(ProjectRepository); + const listed = await secondRuntime.runPromise(second.list()); + + expect(listed).toHaveLength(1); + expect(listed[0]?.cwd).toBe(projectDir); + + await secondRuntime.dispose(); + }); + + it("prunes missing project paths", async () => { + const stateDir = makeTempDir("t3code-project-repo-prune-state-"); + const existing = makeTempDir("t3code-project-repo-prune-existing-"); + const missing = makeTempDir("t3code-project-repo-prune-missing-"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + + const runtime = ManagedRuntime.make(makeProjectRepositoryTest(dbPath)); + const repository = await runtime.runPromise(ProjectRepository); + + const existingProject = await runtime.runPromise(repository.add({ cwd: existing })); + const missingProject = await runtime.runPromise(repository.add({ cwd: missing })); + + expect(existingProject.created).toBe(true); + expect(missingProject.created).toBe(true); + + fs.rmSync(missing, { recursive: true, force: true }); + + await runtime.runPromise(repository.pruneMissing()); + const listed = await runtime.runPromise(repository.list()); + + expect(listed).toHaveLength(1); + expect(listed[0]?.cwd).toBe(existing); + + await runtime.dispose(); + }); +}); diff --git a/apps/server/src/persistence/Layers/Projects.ts b/apps/server/src/persistence/Layers/Projects.ts new file mode 100644 index 0000000000000000000000000000000000000000..dc50541bc0cb1adc8b9d1f2df87d2aa15a1a1a96 --- /dev/null +++ b/apps/server/src/persistence/Layers/Projects.ts @@ -0,0 +1,339 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs"; +import path from "node:path"; + +import type { + ProjectAddInput, + ProjectAddResult, + ProjectListResult, + ProjectRecord, + ProjectScript, + ProjectUpdateScriptsResult, +} from "@t3tools/contracts"; +import { + normalizeProjectScripts, + projectAddInputSchema, + projectRecordSchema, + projectRemoveInputSchema, + projectScriptsSchema, + projectUpdateScriptsInputSchema, +} from "@t3tools/contracts"; +import * as SqlClient from "@effect/sql/SqlClient"; +import * as SqlSchema from "@effect/sql/SqlSchema"; +import { Effect, Layer, Option, Schema } from "effect"; + +import { ProjectRepository, type ProjectRepositoryShape } from "../Services/Projects"; + +const ProjectRowSchema = Schema.Struct({ + id: Schema.String, + cwd: Schema.String, + name: Schema.String, + scriptsJson: Schema.String, + createdAt: Schema.String, + updatedAt: Schema.String, +}); + +type ProjectRow = Schema.Schema.Type; + +const FindByIdRequestSchema = Schema.Struct({ + id: Schema.String, +}); + +const FindByCwdRequestSchema = Schema.Struct({ + cwd: Schema.String, +}); + +const InsertProjectRequestSchema = Schema.Struct({ + id: Schema.String, + cwd: Schema.String, + name: Schema.String, + scriptsJson: Schema.String, + createdAt: Schema.String, + updatedAt: Schema.String, +}); + +const UpdateScriptsRequestSchema = Schema.Struct({ + id: Schema.String, + scriptsJson: Schema.String, + updatedAt: Schema.String, +}); + +const DeleteProjectRequestSchema = Schema.Struct({ + id: Schema.String, +}); + +function cloneScripts(scripts: readonly ProjectScript[]): ProjectScript[] { + const cloned: ProjectScript[] = []; + for (const script of scripts) { + cloned.push({ + id: script.id, + name: script.name, + command: script.command, + icon: script.icon, + runOnWorktreeCreate: script.runOnWorktreeCreate, + }); + } + return cloned; +} + +function normalizeCwd(rawCwd: string): string { + const resolved = path.resolve(rawCwd.trim()); + const normalized = path.normalize(resolved); + if (process.platform === "win32") { + return normalized.toLowerCase(); + } + return normalized; +} + +function isDirectory(cwd: string): boolean { + try { + return fs.statSync(cwd).isDirectory(); + } catch { + return false; + } +} + +function inferProjectName(cwd: string): string { + const name = path.basename(cwd); + return name.length > 0 ? name : "project"; +} + +function parseScriptsJson(raw: string): ProjectScript[] { + try { + const parsed = JSON.parse(raw) as unknown; + const scripts = projectScriptsSchema.parse(parsed); + return normalizeProjectScripts(scripts); + } catch { + return []; + } +} + +function rowToProjectRecord(row: ProjectRow): ProjectRecord { + return projectRecordSchema.parse({ + id: row.id, + cwd: row.cwd, + name: row.name, + scripts: parseScriptsJson(row.scriptsJson), + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }); +} + +const makeRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const listProjectRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectRowSchema, + execute: () => + sql` + SELECT + id, + cwd, + name, + scripts_json AS "scriptsJson", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM projects + ORDER BY created_at DESC + `, + }); + + const findProjectById = SqlSchema.findOne({ + Request: FindByIdRequestSchema, + Result: ProjectRowSchema, + execute: ({ id }) => + sql` + SELECT + id, + cwd, + name, + scripts_json AS "scriptsJson", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM projects + WHERE id = ${id} + `, + }); + + const findProjectByCwd = SqlSchema.findOne({ + Request: FindByCwdRequestSchema, + Result: ProjectRowSchema, + execute: ({ cwd }) => + sql` + SELECT + id, + cwd, + name, + scripts_json AS "scriptsJson", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM projects + WHERE cwd = ${cwd} + `, + }); + + const insertProject = SqlSchema.single({ + Request: InsertProjectRequestSchema, + Result: ProjectRowSchema, + execute: (request) => + sql` + INSERT INTO projects ( + id, + cwd, + name, + scripts_json, + created_at, + updated_at + ) + VALUES ( + ${request.id}, + ${request.cwd}, + ${request.name}, + ${request.scriptsJson}, + ${request.createdAt}, + ${request.updatedAt} + ) + RETURNING + id, + cwd, + name, + scripts_json AS "scriptsJson", + created_at AS "createdAt", + updated_at AS "updatedAt" + `, + }); + + const updateProjectScripts = SqlSchema.single({ + Request: UpdateScriptsRequestSchema, + Result: ProjectRowSchema, + execute: (request) => + sql` + UPDATE projects + SET + scripts_json = ${request.scriptsJson}, + updated_at = ${request.updatedAt} + WHERE id = ${request.id} + RETURNING + id, + cwd, + name, + scripts_json AS "scriptsJson", + created_at AS "createdAt", + updated_at AS "updatedAt" + `, + }); + + const deleteProject = SqlSchema.void({ + Request: DeleteProjectRequestSchema, + execute: ({ id }) => sql`DELETE FROM projects WHERE id = ${id}`, + }); + + const list: ProjectRepositoryShape["list"] = () => + listProjectRows(undefined).pipe( + Effect.map((rows): ProjectListResult => rows.map(rowToProjectRecord)), + Effect.map((projects) => + projects.map((project) => ({ ...project, scripts: cloneScripts(project.scripts) })), + ), + Effect.orDie, + ); + + const add: ProjectRepositoryShape["add"] = (rawInput) => + Effect.gen(function* () { + const input = projectAddInputSchema.parse(rawInput as ProjectAddInput); + const normalizedCwd = normalizeCwd(input.cwd); + if (!isDirectory(normalizedCwd)) { + return yield* Effect.fail(new Error(`Project path does not exist: ${normalizedCwd}`)); + } + + const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe(Effect.orDie); + if (Option.isSome(existing)) { + const project = rowToProjectRecord(existing.value); + const result: ProjectAddResult = { + project: { + ...project, + scripts: cloneScripts(project.scripts), + }, + created: false, + }; + return result; + } + + const now = new Date().toISOString(); + const row = yield* insertProject({ + id: randomUUID(), + cwd: normalizedCwd, + name: inferProjectName(normalizedCwd), + scriptsJson: "[]", + createdAt: now, + updatedAt: now, + }).pipe(Effect.orDie); + + const created = rowToProjectRecord(row); + const result: ProjectAddResult = { + project: { + ...created, + scripts: cloneScripts(created.scripts), + }, + created: true, + }; + return result; + }); + + const remove: ProjectRepositoryShape["remove"] = (rawInput) => + Effect.gen(function* () { + const input = projectRemoveInputSchema.parse(rawInput); + yield* deleteProject({ id: input.id }).pipe(Effect.orDie); + }); + + const updateScripts: ProjectRepositoryShape["updateScripts"] = (rawInput) => + Effect.gen(function* () { + const input = projectUpdateScriptsInputSchema.parse(rawInput); + const existing = yield* findProjectById({ id: input.id }).pipe(Effect.orDie); + if (Option.isNone(existing)) { + return yield* Effect.fail(new Error(`Project not found: ${input.id}`)); + } + + const nextScripts = normalizeProjectScripts(projectScriptsSchema.parse(input.scripts)); + const nextUpdatedAt = new Date().toISOString(); + const updatedRow = yield* updateProjectScripts({ + id: input.id, + scriptsJson: JSON.stringify(nextScripts), + updatedAt: nextUpdatedAt, + }).pipe(Effect.orDie); + + const updated = rowToProjectRecord(updatedRow); + const result: ProjectUpdateScriptsResult = { + project: { + ...updated, + scripts: cloneScripts(updated.scripts), + }, + }; + return result; + }); + + const pruneMissing: ProjectRepositoryShape["pruneMissing"] = () => + listProjectRows(undefined).pipe( + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => { + const normalizedCwd = normalizeCwd(row.cwd); + if (isDirectory(normalizedCwd)) { + return Effect.void; + } + return deleteProject({ id: row.id }).pipe(Effect.orDie); + }), + ), + Effect.asVoid, + Effect.orDie, + ); + + return { + list, + add, + remove, + updateScripts, + pruneMissing, + } satisfies ProjectRepositoryShape; +}); + +export const ProjectRepositoryLive = Layer.effect(ProjectRepository, makeRepository); diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts new file mode 100644 index 0000000000000000000000000000000000000000..735a4252215400d5dd99d6738855189abd46c96b --- /dev/null +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -0,0 +1,41 @@ +import fs from "node:fs"; +import path from "node:path"; + +import * as SqlClient from "@effect/sql/SqlClient"; +import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"; +import { Context, Effect, Layer } from "effect"; + +import { runMigrations } from "../Migrations"; + +export interface PersistenceConfigShape { + readonly dbPath: string; +} + +export class PersistenceConfig extends Context.Tag("persistence/Config")< + PersistenceConfig, + PersistenceConfigShape +>() {} + +export function makeSqlitePersistenceLive(dbPath: string) { + const SqliteClientLive = Layer.unwrapEffect( + Effect.sync(() => { + fs.mkdirSync(path.dirname(dbPath), { recursive: true }); + return SqliteClient.layer({ filename: dbPath }); + }), + ); + + return Layer.scoped( + SqlClient.SqlClient, + Effect.gen(function* () { + const sql = yield* SqliteClient.SqliteClient; + yield* sql`PRAGMA journal_mode = WAL;`; + yield* sql`PRAGMA foreign_keys = ON;`; + yield* runMigrations; + return sql; + }), + ).pipe(Layer.provide(SqliteClientLive)); +} + +export const SqlitePersistenceLive = Layer.unwrapEffect( + Effect.map(PersistenceConfig, ({ dbPath }) => makeSqlitePersistenceLive(dbPath)), +); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 129a6f3c6adc9385161ee1852f5702acb386d496..628fecf88a560ea4d68bf1cd596cbe6f7994652b 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -13,7 +13,7 @@ import * as Layer from "effect/Layer"; // Import all migrations statically import Migration0001 from "./Migrations/001_OrchestrationEvents.ts"; - +import Migration0002 from "./Migrations/002_Projects.ts"; /** * Migration loader with all migrations defined inline. @@ -27,6 +27,7 @@ import Migration0001 from "./Migrations/001_OrchestrationEvents.ts"; */ const loader = Migrator.fromRecord({ "1_OrchestrationEvents": Migration0001, + "2_Projects": Migration0002, }); /** @@ -64,4 +65,4 @@ export const runMigrations = run({ loader }); * ) * ``` */ -export const MigrationsLive = Layer.effectDiscard(runMigrations); \ No newline at end of file +export const MigrationsLive = Layer.effectDiscard(runMigrations); diff --git a/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts b/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts index f2c3b9f9a2c2c8e4f5f17586419a4575bb4c5a81..dac64e9e282f0f9bca7ad2b0a4d158af158d3dd2 100644 --- a/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts +++ b/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts @@ -21,4 +21,4 @@ export default Effect.gen(function* () { CREATE INDEX IF NOT EXISTS idx_orch_events_aggregate ON orchestration_events(aggregate_type, aggregate_id, sequence) `; -}); \ No newline at end of file +}); diff --git a/apps/server/src/persistence/Migrations/002_Projects.ts b/apps/server/src/persistence/Migrations/002_Projects.ts new file mode 100644 index 0000000000000000000000000000000000000000..7ec9785f699fb3e9bc87fb6eac3e5af5cdbddb04 --- /dev/null +++ b/apps/server/src/persistence/Migrations/002_Projects.ts @@ -0,0 +1,22 @@ +import { SqlClient } from "@effect/sql"; +import * as Effect from "effect/Effect"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS projects ( + id TEXT PRIMARY KEY, + cwd TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + scripts_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projects_created_at + ON projects(created_at DESC) + `; +}); diff --git a/apps/server/src/persistence/Services/Projects.ts b/apps/server/src/persistence/Services/Projects.ts new file mode 100644 index 0000000000000000000000000000000000000000..24288b3a673da920821736f6ece217c4255e7ecc --- /dev/null +++ b/apps/server/src/persistence/Services/Projects.ts @@ -0,0 +1,25 @@ +import type { + ProjectAddInput, + ProjectAddResult, + ProjectListResult, + ProjectRemoveInput, + ProjectUpdateScriptsInput, + ProjectUpdateScriptsResult, +} from "@t3tools/contracts"; +import { Context } from "effect"; +import type { Effect } from "effect"; + +export interface ProjectRepositoryShape { + readonly list: () => Effect.Effect; + readonly add: (input: ProjectAddInput) => Effect.Effect; + readonly remove: (input: ProjectRemoveInput) => Effect.Effect; + readonly updateScripts: ( + input: ProjectUpdateScriptsInput, + ) => Effect.Effect; + readonly pruneMissing: () => Effect.Effect; +} + +export class ProjectRepository extends Context.Tag("persistence/ProjectRepository")< + ProjectRepository, + ProjectRepositoryShape +>() {} diff --git a/apps/server/src/projectRegistry.test.ts b/apps/server/src/projectRegistry.test.ts deleted file mode 100644 index 199e69d000a666efb610a39f68e9c12974345b48..0000000000000000000000000000000000000000 --- a/apps/server/src/projectRegistry.test.ts +++ /dev/null @@ -1,113 +0,0 @@ -import fs from "node:fs"; -import os from "node:os"; -import path from "node:path"; - -import { afterEach, describe, expect, it } from "vitest"; - -import { ProjectRegistry } from "./projectRegistry"; - -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 }); - } -}); - -describe("ProjectRegistry scripts", () => { - it("stores and reloads project scripts", () => { - const stateDir = makeTempDir("t3code-project-registry-state-"); - const projectDir = makeTempDir("t3code-project-registry-project-"); - const registry = new ProjectRegistry(stateDir); - - const created = registry.add({ cwd: projectDir }).project; - const updated = registry.updateScripts({ - id: created.id, - scripts: [ - { - id: "run", - name: "Run", - command: "bun run dev", - icon: "play", - runOnWorktreeCreate: false, - }, - { - id: "setup", - name: "Setup", - command: "bun install", - icon: "configure", - runOnWorktreeCreate: true, - }, - ], - }); - - expect(updated.project.scripts).toHaveLength(2); - expect(updated.project.scripts[1]?.id).toBe("setup"); - expect(updated.project.scripts[1]?.runOnWorktreeCreate).toBe(true); - - const reloaded = new ProjectRegistry(stateDir); - const listed = reloaded.list(); - expect(listed).toHaveLength(1); - expect(listed[0]?.scripts).toHaveLength(2); - expect(listed[0]?.scripts[0]?.id).toBe("run"); - expect(listed[0]?.scripts[1]?.id).toBe("setup"); - }); - - it("deduplicates script ids and allows only one setup script", () => { - const stateDir = makeTempDir("t3code-project-registry-state-"); - const projectDir = makeTempDir("t3code-project-registry-project-"); - const registry = new ProjectRegistry(stateDir); - const created = registry.add({ cwd: projectDir }).project; - - const updated = registry.updateScripts({ - id: created.id, - scripts: [ - { - id: "setup", - name: "Setup", - command: "echo setup", - icon: "configure", - runOnWorktreeCreate: true, - }, - { - id: "setup", - name: "Setup Duplicate", - command: "echo duplicate", - icon: "configure", - runOnWorktreeCreate: true, - }, - { - id: "another", - name: "Another Setup", - command: "echo another", - icon: "play", - runOnWorktreeCreate: true, - }, - ], - }); - - expect(updated.project.scripts).toHaveLength(2); - expect(updated.project.scripts[0]?.id).toBe("setup"); - expect(updated.project.scripts[0]?.runOnWorktreeCreate).toBe(true); - expect(updated.project.scripts[1]?.id).toBe("another"); - expect(updated.project.scripts[1]?.runOnWorktreeCreate).toBe(false); - }); - - it("throws when updating scripts for an unknown project id", () => { - const stateDir = makeTempDir("t3code-project-registry-state-"); - const registry = new ProjectRegistry(stateDir); - - expect(() => - registry.updateScripts({ - id: "missing-project", - scripts: [], - }), - ).toThrowError("Project not found"); - }); -}); diff --git a/apps/server/src/projectRegistry.ts b/apps/server/src/projectRegistry.ts deleted file mode 100644 index c849192d53ec4064425467c67416dc94f0c9902f..0000000000000000000000000000000000000000 --- a/apps/server/src/projectRegistry.ts +++ /dev/null @@ -1,235 +0,0 @@ -import { randomUUID } from "node:crypto"; -import fs from "node:fs"; -import path from "node:path"; - -import { - type ProjectAddInput, - type ProjectAddResult, - type ProjectListResult, - type ProjectRecord, - type ProjectRemoveInput, - type ProjectScript, - type ProjectUpdateScriptsInput, - type ProjectUpdateScriptsResult, - normalizeProjectScripts, - projectAddInputSchema, - projectRecordSchema, - projectRemoveInputSchema, - projectScriptsSchema, - projectUpdateScriptsInputSchema, -} from "@t3tools/contracts"; - -function cloneScripts(scripts: readonly ProjectScript[]): ProjectScript[] { - const cloned: ProjectScript[] = []; - for (const script of scripts) { - cloned.push({ - id: script.id, - name: script.name, - command: script.command, - icon: script.icon, - runOnWorktreeCreate: script.runOnWorktreeCreate, - }); - } - return cloned; -} - -function normalizeCwd(rawCwd: string): string { - const resolved = path.resolve(rawCwd.trim()); - const normalized = path.normalize(resolved); - if (process.platform === "win32") { - return normalized.toLowerCase(); - } - return normalized; -} - -function isDirectory(cwd: string): boolean { - try { - return fs.statSync(cwd).isDirectory(); - } catch { - return false; - } -} - -function inferProjectName(cwd: string): string { - const name = path.basename(cwd); - return name.length > 0 ? name : "project"; -} - -export class ProjectRegistry { - private readonly stateDir: string; - private readonly filePath: string; - private projects: ProjectRecord[]; - - constructor(stateDir: string) { - this.stateDir = stateDir; - this.filePath = path.join(this.stateDir, "projects.json"); - fs.mkdirSync(this.stateDir, { recursive: true }); - this.projects = this.loadFromDisk(); - } - - list(): ProjectListResult { - return this.projects.map((project) => ({ - ...project, - scripts: cloneScripts(project.scripts), - })); - } - - add(raw: ProjectAddInput): ProjectAddResult { - const input = projectAddInputSchema.parse(raw); - const normalizedCwd = normalizeCwd(input.cwd); - if (!isDirectory(normalizedCwd)) { - throw new Error(`Project path does not exist: ${normalizedCwd}`); - } - - const existing = this.projects.find((project) => normalizeCwd(project.cwd) === normalizedCwd); - if (existing) { - return { - project: { - ...existing, - scripts: cloneScripts(existing.scripts), - }, - created: false, - }; - } - - const now = new Date().toISOString(); - const project: ProjectRecord = { - id: randomUUID(), - cwd: normalizedCwd, - name: inferProjectName(normalizedCwd), - scripts: [], - createdAt: now, - updatedAt: now, - }; - - this.projects = [project, ...this.projects]; - this.persist(); - return { - project: { - ...project, - scripts: cloneScripts(project.scripts), - }, - created: true, - }; - } - - remove(raw: ProjectRemoveInput): void { - const input = projectRemoveInputSchema.parse(raw); - const next = this.projects.filter((project) => project.id !== input.id); - if (next.length === this.projects.length) { - return; - } - - this.projects = next; - this.persist(); - } - - updateScripts(raw: ProjectUpdateScriptsInput): ProjectUpdateScriptsResult { - const input = projectUpdateScriptsInputSchema.parse(raw); - const index = this.projects.findIndex((project) => project.id === input.id); - if (index < 0) { - throw new Error(`Project not found: ${input.id}`); - } - - const nextScripts = normalizeProjectScripts(projectScriptsSchema.parse(input.scripts)); - const nextUpdatedAt = new Date().toISOString(); - const nextProject: ProjectRecord = { - ...this.projects[index]!, - scripts: nextScripts, - updatedAt: nextUpdatedAt, - }; - - const nextProjects = [...this.projects]; - nextProjects[index] = nextProject; - this.projects = nextProjects; - this.persist(); - - return { - project: { ...nextProject, scripts: cloneScripts(nextProject.scripts) }, - }; - } - - private loadFromDisk(): ProjectRecord[] { - const loaded = this.readPersistedProjects(); - const deduped = new Map(); - const normalizedProjects: ProjectRecord[] = []; - - for (const project of loaded) { - const normalizedCwd = normalizeCwd(project.cwd); - if (!isDirectory(normalizedCwd)) { - continue; - } - if (deduped.has(normalizedCwd)) { - continue; - } - - const normalizedProject: ProjectRecord = { - ...project, - cwd: normalizedCwd, - name: - project.name.trim().length > 0 ? project.name.trim() : inferProjectName(normalizedCwd), - scripts: normalizeProjectScripts(project.scripts), - }; - deduped.set(normalizedCwd, normalizedProject); - normalizedProjects.push(normalizedProject); - } - - const changed = - normalizedProjects.length !== loaded.length || - normalizedProjects.some((project, index) => { - const source = loaded[index]; - if (!source) return true; - return ( - source.id !== project.id || - source.cwd !== project.cwd || - source.name !== project.name || - JSON.stringify(source.scripts) !== JSON.stringify(project.scripts) || - source.createdAt !== project.createdAt || - source.updatedAt !== project.updatedAt - ); - }); - - if (changed) { - this.projects = normalizedProjects; - this.persist(); - } - - return normalizedProjects; - } - - private readPersistedProjects(): ProjectRecord[] { - try { - const raw = fs.readFileSync(this.filePath, "utf8"); - const parsed = JSON.parse(raw) as { projects?: unknown }; - if (!parsed || !Array.isArray(parsed.projects)) { - return []; - } - - const projects: ProjectRecord[] = []; - for (const candidate of parsed.projects) { - const row = projectRecordSchema.safeParse(candidate); - if (!row.success) { - continue; - } - projects.push(row.data); - } - return projects; - } catch { - return []; - } - } - - private persist(): void { - const payload = JSON.stringify( - { - version: 1, - projects: this.projects, - }, - null, - 2, - ); - const tempFile = `${this.filePath}.tmp-${process.pid}-${Date.now()}`; - fs.writeFileSync(tempFile, payload); - fs.renameSync(tempFile, this.filePath); - } -} diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index a9b9cc110241bb7586a1555ba643a22c8d5f0260..c977418f05d69acd3bef5c52c031892196c4e0c3 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -3,6 +3,7 @@ import os from "node:os"; import path from "node:path"; import { EventEmitter } from "node:events"; +import { Layer, ManagedRuntime } from "effect"; import { describe, expect, it, afterEach, vi } from "vitest"; import { createServer } from "./wsServer"; import WebSocket from "ws"; @@ -16,8 +17,8 @@ import { type WsPush, type WsResponse, } from "@t3tools/contracts"; -import { ProjectRegistry } from "./projectRegistry"; import { compileResolvedKeybindingRule, DEFAULT_KEYBINDINGS } from "./keybindings"; +import { ProjectRepository } from "./persistence/Services/Projects"; import type { TerminalClearInput, TerminalCloseInput, @@ -28,6 +29,8 @@ import type { TerminalWriteInput, } from "@t3tools/contracts"; import type { TerminalManager } from "./terminalManager"; +import { ProjectRepositoryLive } from "./persistence/Layers/Projects"; +import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; interface PendingMessages { queue: unknown[]; @@ -249,9 +252,9 @@ describe("WebSocket Server", () => { return createServer({ port: 0, cwd: options.cwd ?? "/test/project", + stateDir, ...(options.devUrl ? { devUrl: options.devUrl } : {}), ...(options.authToken ? { authToken: options.authToken } : {}), - projectRegistry: new ProjectRegistry(stateDir), ...(options.gitManager ? { gitManager: options.gitManager as never } : {}), ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}), }); @@ -888,7 +891,11 @@ describe("WebSocket Server", () => { it("supports projects.searchEntries", async () => { const workspace = makeTempDir("t3code-ws-workspace-entries-"); fs.mkdirSync(path.join(workspace, "src", "components"), { recursive: true }); - fs.writeFileSync(path.join(workspace, "src", "components", "Composer.tsx"), "export {};", "utf8"); + fs.writeFileSync( + path.join(workspace, "src", "components", "Composer.tsx"), + "export {};", + "utf8", + ); fs.writeFileSync(path.join(workspace, "README.md"), "# test", "utf8"); fs.mkdirSync(path.join(workspace, ".git"), { recursive: true }); fs.writeFileSync(path.join(workspace, ".git", "HEAD"), "ref: refs/heads/main\n", "utf8"); @@ -1025,36 +1032,18 @@ describe("WebSocket Server", () => { it("prunes missing projects on startup", async () => { const stateDir = makeTempDir("t3code-ws-prune-state-"); const existing = makeTempDir("t3code-ws-existing-project-"); - const missing = path.join(stateDir, "definitely-missing"); - const now = new Date().toISOString(); - const projectsFile = path.join(stateDir, "projects.json"); + const missing = makeTempDir("t3code-ws-missing-project-"); - fs.writeFileSync( - projectsFile, - JSON.stringify( - { - version: 1, - projects: [ - { - id: "project-existing", - cwd: existing, - name: "existing", - createdAt: now, - updatedAt: now, - }, - { - id: "project-missing", - cwd: missing, - name: "missing", - createdAt: now, - updatedAt: now, - }, - ], - }, - null, - 2, + const projectRuntime = ManagedRuntime.make( + ProjectRepositoryLive.pipe( + Layer.provide(makeSqlitePersistenceLive(path.join(stateDir, "orchestration.sqlite"))), ), ); + const projectRepository = await projectRuntime.runPromise(ProjectRepository); + await projectRuntime.runPromise(projectRepository.add({ cwd: existing })); + await projectRuntime.runPromise(projectRepository.add({ cwd: missing })); + await projectRuntime.dispose(); + fs.rmSync(missing, { recursive: true, force: true }); server = createTestServer({ stateDir, cwd: "/test" }); await server.start(); @@ -1067,9 +1056,9 @@ describe("WebSocket Server", () => { const response = await sendRequest(ws, WS_METHODS.projectsList); expect(response.error).toBeUndefined(); - const listed = response.result as Array<{ id: string }>; + const listed = response.result as Array<{ cwd: string }>; expect(listed).toHaveLength(1); - expect(listed[0]?.id).toBe("project-existing"); + expect(listed[0]?.cwd).toBe(existing); }); it("rejects websocket connections without a valid auth token", async () => { diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index dd8f3fd6eb1a02e19212168465f01a8f833207d9..d63b8677888da8866aa81583dddd6add3605e98f 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -18,10 +18,10 @@ import { type WsResponse, wsRequestSchema, } from "@t3tools/contracts"; +import { Layer, ManagedRuntime } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; -import { ProjectRegistry } from "./projectRegistry"; import { ProviderManager } from "./providerManager"; import { GitManager } from "./gitManager"; import { @@ -36,7 +36,13 @@ import { import { TerminalManager } from "./terminalManager"; import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings"; import { searchWorkspaceEntries } from "./workspaceEntries"; -import { createOrchestrationSystem, type OrchestrationSystem } from "./orchestration/runtime"; +import type { OrchestrationEngine } from "./orchestration/engine"; +import { OrchestrationLive } from "./orchestration/layers"; +import { OrchestrationEngineService } from "./orchestration/services"; +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"; const MIME_TYPES: Record = { ".html": "text/html; charset=utf-8", @@ -61,12 +67,31 @@ export interface ServerOptions { staticDir?: string | undefined; devUrl?: string | undefined; logWebSocketEvents?: boolean | undefined; - projectRegistry?: ProjectRegistry | undefined; gitManager?: GitManager | undefined; terminalManager?: TerminalManager | undefined; authToken?: string | undefined; } +const isServerNotRunningError = (error: unknown): boolean => { + if (!(error instanceof Error)) return false; + const maybeCode = (error as NodeJS.ErrnoException).code; + return ( + maybeCode === "ERR_SERVER_NOT_RUNNING" || error.message.toLowerCase().includes("not running") + ); +}; + +function rejectUpgrade(socket: Duplex, statusCode: number, message: string): void { + socket.write( + `HTTP/1.1 ${statusCode} ${statusCode === 401 ? "Unauthorized" : "Bad Request"}\r\n` + + "Connection: close\r\n" + + "Content-Type: text/plain\r\n" + + `Content-Length: ${Buffer.byteLength(message)}\r\n` + + "\r\n" + + message, + ); + socket.destroy(); +} + function parseBooleanEnv(value: string | undefined): boolean | undefined { if (value === undefined) return undefined; const normalized = value.trim().toLowerCase(); @@ -99,18 +124,19 @@ export function createServer(options: ServerOptions) { staticDir, devUrl, logWebSocketEvents: explicitLogWsEvents, - projectRegistry: providedRegistry, - gitManager: providedGitManager, - terminalManager: providedTerminalManager, + gitManager = new GitManager(), + terminalManager = new TerminalManager(), authToken, } = options; const providerManager = new ProviderManager(); - const terminalManager = providedTerminalManager ?? new TerminalManager(); - const projectRegistry = - providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata")); - const gitManager = providedGitManager ?? new GitManager(); + const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata"); - let orchestrationSystem: OrchestrationSystem | null = null; + let effectRuntime: ManagedRuntime.ManagedRuntime< + ProjectRepository | OrchestrationEngineService, + unknown + > | null = null; + let projectRepository: ProjectRepositoryShape | null = null; + let orchestrationEngine: OrchestrationEngine | null = null; const clients = new Set(); const logger = createLogger("ws"); const logWebSocketEvents = @@ -128,15 +154,23 @@ export function createServer(options: ServerOptions) { }); } - function requireOrchestrationEngine() { - if (!orchestrationSystem) { - throw new Error("Orchestration engine is not started"); - } - return orchestrationSystem.engine; - } + const getOrchestrationEngine = () => { + assert(orchestrationEngine, "Orchestration engine is not started"); + return orchestrationEngine; + }; + + const getProjectRepository = () => { + assert(projectRepository, "Project repository is not started"); + return projectRepository; + }; + + const getEffectRuntime = () => { + assert(effectRuntime, "Effect runtime is not started"); + return effectRuntime; + }; function attachOrchestrationSubscriptions() { - const orchestrationEngine = requireOrchestrationEngine(); + const orchestrationEngine = getOrchestrationEngine(); unsubscribeReadModel = orchestrationEngine.subscribeToReadModel((snapshot) => { const push: WsPush = { @@ -179,17 +213,16 @@ export function createServer(options: ServerOptions) { // Forward provider events to all connected WebSocket clients providerManager.on("event", (event) => { void (async () => { - const liveOrchestrationSystem = orchestrationSystem; - if (!liveOrchestrationSystem) { + const liveOrchestrationEngine = orchestrationEngine; + if (!liveOrchestrationEngine) { return; } - const orchestrationEngine = liveOrchestrationSystem.engine; - const snapshot = orchestrationEngine.getSnapshot(); + const snapshot = 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 orchestrationEngine.dispatch({ + await liveOrchestrationEngine.dispatch({ type: "thread.session", commandId: crypto.randomUUID(), threadId: thread.id, @@ -208,7 +241,7 @@ export function createServer(options: ServerOptions) { } if (event.textDelta && event.textDelta.length > 0) { const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`; - await orchestrationEngine.dispatch({ + await liveOrchestrationEngine.dispatch({ type: "message.send", commandId: crypto.randomUUID(), threadId: thread.id, @@ -221,7 +254,7 @@ export function createServer(options: ServerOptions) { } if (event.method === "turn/completed") { const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`; - await orchestrationEngine.dispatch({ + await liveOrchestrationEngine.dispatch({ type: "message.send", commandId: crypto.randomUUID(), threadId: thread.id, @@ -236,13 +269,15 @@ export function createServer(options: ServerOptions) { const payload = asObject(event.payload); const turnId = event.turnId ?? asString(payload?.turnId); if (turnId) { - await orchestrationEngine.dispatch({ + 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 } : {}), + ...(asString(payload?.status) !== null + ? { status: asString(payload?.status) ?? undefined } + : {}), files: [], assistantMessageId: `assistant:${turnId}`, ...(asNumber(payload?.turnCount) !== null @@ -253,7 +288,7 @@ export function createServer(options: ServerOptions) { } } if (event.kind === "error" && event.message) { - await orchestrationEngine.dispatch({ + await liveOrchestrationEngine.dispatch({ type: "thread.session", commandId: crypto.randomUUID(), threadId: thread.id, @@ -350,18 +385,6 @@ export function createServer(options: ServerOptions) { // WebSocket server — upgrades from the HTTP server const wss = new WebSocketServer({ noServer: true }); - function rejectUpgrade(socket: Duplex, statusCode: number, message: string): void { - socket.write( - `HTTP/1.1 ${statusCode} ${statusCode === 401 ? "Unauthorized" : "Bad Request"}\r\n` + - "Connection: close\r\n" + - "Content-Type: text/plain\r\n" + - `Content-Length: ${Buffer.byteLength(message)}\r\n` + - "\r\n" + - message, - ); - socket.destroy(); - } - httpServer.on("upgrade", (request, socket, head) => { if (authToken) { let providedToken: string | null = null; @@ -386,7 +409,7 @@ export function createServer(options: ServerOptions) { wss.on("connection", (ws) => { clients.add(ws); - const orchestrationEngine = requireOrchestrationEngine(); + const orchestrationEngine = getOrchestrationEngine(); // Send welcome message with project info const segments = cwd.split(/[/\\]/).filter(Boolean); @@ -453,7 +476,9 @@ export function createServer(options: ServerOptions) { } async function routeRequest(request: WsRequest): Promise { - const orchestrationEngine = requireOrchestrationEngine(); + const orchestrationEngine = getOrchestrationEngine(); + const projectRepository = getProjectRepository(); + const effectRuntime = getEffectRuntime(); switch (request.method) { case WS_METHODS.providersStartSession: @@ -479,79 +504,56 @@ export function createServer(options: ServerOptions) { case WS_METHODS.providersGetCheckpointDiff: return providerManager.getCheckpointDiff(request.params as never); - case WS_METHODS.providersRevertToCheckpoint: - { - const result = await providerManager.revertToCheckpoint(request.params as never); - const params = request.params as { sessionId?: string }; - const sessionId = params.sessionId; - if (typeof sessionId === "string") { - const snapshot = orchestrationEngine.getSnapshot(); - const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId); - if (thread) { - const now = new Date().toISOString(); + case WS_METHODS.providersRevertToCheckpoint: { + const result = await providerManager.revertToCheckpoint(request.params as never); + const params = request.params as { sessionId?: string }; + const sessionId = params.sessionId; + if (typeof sessionId === "string") { + const snapshot = 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.revert", + type: "thread.session", commandId: crypto.randomUUID(), threadId: thread.id, - turnCount: result.turnCount, - messageCount: result.messageCount, + session: { + ...thread.session, + status: "ready", + activeTurnId: null, + updatedAt: now, + lastError: null, + }, createdAt: now, }); - if (thread.session) { - await orchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - ...thread.session, - status: "ready", - activeTurnId: null, - updatedAt: now, - lastError: null, - }, - createdAt: now, - }); - } } } - return result; } + return result; + } case WS_METHODS.projectsList: - return projectRegistry.list(); + return effectRuntime.runPromise(projectRepository.list()); case WS_METHODS.projectsAdd: - { - const result = await projectRegistry.add(request.params as never); - await orchestrationEngine.dispatch({ - type: "project.create", - commandId: crypto.randomUUID(), - projectId: result.project.id, - name: result.project.name, - cwd: result.project.cwd, - model: "gpt-5-codex", - createdAt: new Date().toISOString(), - }); - return result; - } + return effectRuntime.runPromise(projectRepository.add(request.params as never)); case WS_METHODS.projectsRemove: - { - const params = request.params as { id: string }; - projectRegistry.remove(request.params as never); - await orchestrationEngine.dispatch({ - type: "project.delete", - commandId: crypto.randomUUID(), - projectId: params.id, - createdAt: new Date().toISOString(), - }); - return undefined; - } + return effectRuntime.runPromise(projectRepository.remove(request.params as never)); case WS_METHODS.projectsSearchEntries: return searchWorkspaceEntries(request.params as never); case WS_METHODS.projectsUpdateScripts: - return projectRegistry.updateScripts(request.params as never); + return effectRuntime.runPromise(projectRepository.updateScripts(request.params as never)); case WS_METHODS.shellOpenInEditor: { const params = request.params as { @@ -597,25 +599,25 @@ export function createServer(options: ServerOptions) { return undefined; } - case WS_METHODS.gitStatus: - { - const params = request.params as { cwd: string }; - const status = await gitManager.status(request.params as never); - const project = projectRegistry.list().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(), - }); - } - return status; + case WS_METHODS.gitStatus: { + const params = request.params as { cwd: string }; + const status = await gitManager.status(request.params as never); + 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(), + }); } + return status; + } case WS_METHODS.gitPull: return pullGitBranch(request.params as never); @@ -698,37 +700,57 @@ export function createServer(options: ServerOptions) { } } - async function ensureOrchestrationSystemStarted() { - if (orchestrationSystem) { - return; + 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), + ), + ); + + try { + const [nextOrchestrationEngine, repository] = await Promise.all([ + runtime.runPromise(OrchestrationEngineService), + runtime.runPromise(ProjectRepository), + ]); + orchestrationEngine = nextOrchestrationEngine; + projectRepository = repository; + await runtime.runPromise(repository.pruneMissing()); + attachOrchestrationSubscriptions(); + } catch (error) { + await runtime.dispose().catch(() => undefined); + throw error; } - orchestrationSystem = await createOrchestrationSystem(resolvedStateDir); - attachOrchestrationSubscriptions(); + return runtime; } - async function disposeOrchestrationSystem() { + async function disposeEffectRuntime() { unsubscribeReadModel(); unsubscribeDomainEvents(); unsubscribeReadModel = noop; unsubscribeDomainEvents = noop; - if (!orchestrationSystem) { + const runtime = effectRuntime; + effectRuntime = null; + orchestrationEngine = null; + projectRepository = null; + + if (!runtime) { return; } - const system = orchestrationSystem; - orchestrationSystem = null; - await system.dispose(); + await runtime.dispose(); } async function start() { - await ensureOrchestrationSystemStarted(); + effectRuntime = await createEffectRuntime(); return new Promise((resolve, reject) => { const onError = (error: Error) => { httpServer.off("error", onError); - void disposeOrchestrationSystem().finally(() => reject(error)); + void disposeEffectRuntime().finally(() => reject(error)); }; httpServer.once("error", onError); const onListening = () => { @@ -744,7 +766,7 @@ export function createServer(options: ServerOptions) { } async function stop(): Promise { - await disposeOrchestrationSystem(); + await disposeEffectRuntime(); terminalManager.off("event", onTerminalEvent); providerManager.stopAll(); providerManager.dispose(); @@ -755,15 +777,6 @@ export function createServer(options: ServerOptions) { } clients.clear(); - const isServerNotRunningError = (error: unknown): boolean => { - if (!(error instanceof Error)) return false; - const maybeCode = (error as NodeJS.ErrnoException).code; - return ( - maybeCode === "ERR_SERVER_NOT_RUNNING" || - error.message.toLowerCase().includes("not running") - ); - }; - const closeWebSocketServer = new Promise((resolve, reject) => { wss.close((error) => { if (error && !isServerNotRunningError(error)) { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 0d2ad398f2cddce94d91d416e3bd20027fbac782..7371b19bc411b60333feab25f40951adda40361a 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -28,13 +28,6 @@ function formatRelativeTime(iso: string): string { return `${Math.floor(hours / 24)}d ago`; } -function inferProjectName(cwd: string): string { - const normalized = cwd.replace(/\\/g, "/").replace(/\/+$/, ""); - const parts = normalized.split("/"); - const last = parts[parts.length - 1]; - return last && last.length > 0 ? last : "project"; -} - interface ThreadStatusPill { label: "Working" | "Connecting" | "Completed" | "Awaiting response"; colorClass: string; @@ -195,23 +188,16 @@ export default function Sidebar() { return; } - const projectId = crypto.randomUUID(); - const now = new Date().toISOString(); - const name = inferProjectName(cwd); - if (isElectron) { - // Keep project registry in sync for desktop integrations. - await api.projects.add({ cwd }); + const result = await api.projects.add({ cwd }); + const projects = await api.projects.list(); + dispatch({ type: "SYNC_PROJECTS", projects }); + + const hasThread = state.threads.some((thread) => thread.projectId === result.project.id); + if (hasThread) { + focusMostRecentThreadForProject(result.project.id); + return; } - await api.orchestration.dispatchCommand({ - type: "project.create", - commandId: crypto.randomUUID(), - projectId, - name, - cwd, - model: DEFAULT_MODEL, - createdAt: now, - }); - handleNewThread(projectId); + handleNewThread(result.project.id); } finally { setIsAddingProject(false); setNewCwd(""); @@ -225,6 +211,7 @@ export default function Sidebar() { handleNewThread, isAddingProject, state.projects, + state.threads, ], ); @@ -381,28 +368,20 @@ export default function Sidebar() { ); if (!confirmed) return; - if (isElectron) { - try { - await api.projects.remove({ id: projectId }); - } catch (error) { - const message = - error instanceof Error ? error.message : "Unknown error deleting project."; - console.error("Failed to remove project", { projectId, error }); - toastManager.add({ - type: "error", - title: `Failed to delete "${project.name}"`, - description: message, - }); - return; - } + try { + await api.projects.remove({ id: projectId }); + const projects = await api.projects.list(); + dispatch({ type: "SYNC_PROJECTS", projects }); + } catch (error) { + const message = + error instanceof Error ? error.message : "Unknown error deleting project."; + console.error("Failed to remove project", { projectId, error }); + toastManager.add({ + type: "error", + title: `Failed to delete "${project.name}"`, + description: message, + }); } - - await api.orchestration.dispatchCommand({ - type: "project.delete", - commandId: crypto.randomUUID(), - projectId, - createdAt: new Date().toISOString(), - }); }, [api, dispatch, state.projects, state.threads], ); diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index 1ef0eabe4bcbf6440ab2bed923c068ca20d40c42..8a6ee0752bc72e1099a2a202b38a60d0396a5f7f 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -126,9 +126,9 @@ function EventRouter() { useEffect(() => { if (!api) return; - void api.orchestration - .getSnapshot() - .then((snapshot) => { + void Promise.all([api.projects.list(), api.orchestration.getSnapshot()]) + .then(([projects, snapshot]) => { + dispatch({ type: "SYNC_PROJECTS", projects }); dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot }); }) .catch(() => undefined); @@ -150,8 +150,8 @@ function AutoProjectBootstrap() { useEffect(() => { if (!api) return; - // Browser mode bootstraps from server welcome. - // Electron bootstraps from persisted projects via DesktopProjectBootstrap. + // Browser mode auto-adds the current cwd project via server welcome. + // Desktop skips this because projects are already loaded from the repository. if (isElectron) return; return onServerWelcome((payload) => { @@ -166,40 +166,38 @@ function AutoProjectBootstrap() { } bootstrappedRef.current = true; - const projectId = crypto.randomUUID(); - const threadId = crypto.randomUUID(); const now = new Date().toISOString(); - void api.orchestration - .dispatchCommand({ - type: "project.create", - commandId: crypto.randomUUID(), - projectId, - name: payload.projectName, - cwd: payload.cwd, - model: DEFAULT_MODEL, - createdAt: now, - }) - .then(() => - api.orchestration.dispatchCommand({ + void api.projects + .add({ cwd: payload.cwd }) + .then(async (result) => { + const projects = await api.projects.list(); + dispatch({ type: "SYNC_PROJECTS", projects }); + + const hasThread = state.threads.some((thread) => thread.projectId === result.project.id); + if (hasThread) { + return; + } + + return api.orchestration.dispatchCommand({ type: "thread.create", commandId: crypto.randomUUID(), - threadId, - projectId, + threadId: crypto.randomUUID(), + projectId: result.project.id, title: "New thread", model: DEFAULT_MODEL, branch: null, worktreePath: null, createdAt: now, - }), - ) + }); + }) .catch(() => undefined); }); - }, [api, state.projects, dispatch]); + }, [api, state.projects, state.threads, dispatch]); return null; } function DesktopProjectBootstrap() { - // Desktop now hydrates from orchestration read-model snapshots. + // Desktop hydration runs through EventRouter project + orchestration sync. return null; } diff --git a/apps/web/src/store.ts b/apps/web/src/store.ts index 1df450ac06a24187c753e7becde0b397b063146f..46de08875d1f5abc13af10fe2a9875b0c436f484 100644 --- a/apps/web/src/store.ts +++ b/apps/web/src/store.ts @@ -10,9 +10,10 @@ import { import { type OrchestrationReadModel, + type ProjectListResult, normalizeProjectScripts, } from "@t3tools/contracts"; -import { resolveModelSlug } from "./model-logic"; +import { DEFAULT_MODEL, resolveModelSlug } from "./model-logic"; import { DEFAULT_THREAD_TERMINAL_HEIGHT, DEFAULT_THREAD_TERMINAL_ID, @@ -29,6 +30,7 @@ import { type Action = | { type: "SYNC_SERVER_READ_MODEL"; readModel: OrchestrationReadModel } + | { type: "SYNC_PROJECTS"; projects: ProjectListResult } | { type: "SET_PROJECT_SCRIPTS"; projectId: string; scripts: ProjectScript[] } | { type: "SET_THREADS_HYDRATED"; hydrated: boolean } | { type: "TOGGLE_PROJECT"; projectId: string } @@ -136,6 +138,29 @@ function updateThread( return threads.map((t) => (t.id === threadId ? updater(t) : t)); } +function mapProjects( + incoming: ProjectListResult, + previous: Project[], +): Project[] { + return incoming.map((project) => { + const existing = + previous.find((entry) => entry.id === project.id) ?? + previous.find((entry) => entry.cwd === project.cwd); + return { + id: project.id, + name: project.name, + cwd: project.cwd, + model: existing?.model ?? DEFAULT_MODEL, + expanded: + existing?.expanded ?? + (persistedExpandedProjectCwds.size > 0 + ? persistedExpandedProjectCwds.has(project.cwd) + : true), + scripts: normalizeProjectScripts(project.scripts), + }; + }); +} + function normalizeTerminalIds(terminalIds: string[]): string[] { const ids = terminalIds.map((id) => id.trim()).filter((id) => id.length > 0); @@ -330,21 +355,6 @@ export function reducer(state: AppState, action: Action): AppState { switch (action.type) { case "SYNC_SERVER_READ_MODEL": { const existingThreadById = new Map(state.threads.map((thread) => [thread.id, thread] as const)); - const projects = action.readModel.projects.map((project) => { - const existing = state.projects.find((entry) => entry.id === project.id); - return { - id: project.id, - name: project.name, - cwd: project.cwd, - model: resolveModelSlug(project.model), - expanded: - existing?.expanded ?? - (persistedExpandedProjectCwds.size > 0 - ? persistedExpandedProjectCwds.has(project.cwd) - : true), - scripts: existing?.scripts ?? [], - }; - }); const threads = action.readModel.threads.map((thread) => { const existing = existingThreadById.get(thread.id); return normalizeThreadTerminals({ @@ -392,12 +402,17 @@ export function reducer(state: AppState, action: Action): AppState { }); return { ...state, - projects, threads, threadsHydrated: true, }; } + case "SYNC_PROJECTS": + return { + ...state, + projects: mapProjects(action.projects, state.projects), + }; + case "SET_PROJECT_SCRIPTS": return { ...state, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index e9f53c19db9476b5f149348c3ef1e6efa6a8db79..22f1c6411f185bd5c1a97183eca375c3bea9ec1f 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -80,17 +80,6 @@ export const OrchestrationThreadSchema = Schema.Struct({ export type OrchestrationThread = Schema.Schema.Type; -export const OrchestrationProjectSchema = Schema.Struct({ - id: IdSchema, - name: Schema.String, - cwd: Schema.String, - model: Schema.String, - createdAt: IsoDateTimeSchema, - updatedAt: IsoDateTimeSchema, -}); - -export type OrchestrationProject = Schema.Schema.Type; - export const OrchestrationGitReadModelSchema = Schema.Struct({ projectId: IdSchema, branch: Schema.NullOr(Schema.String), @@ -104,7 +93,6 @@ export type OrchestrationGitReadModel = Schema.Schema.Type; -export const CreateProjectCommandSchema = Schema.Struct({ - type: Schema.Literal("project.create"), - commandId: IdSchema, - projectId: IdSchema, - name: Schema.String, - cwd: Schema.String, - model: Schema.String, - createdAt: IsoDateTimeSchema, -}); - -export const DeleteProjectCommandSchema = Schema.Struct({ - type: Schema.Literal("project.delete"), - commandId: IdSchema, - projectId: IdSchema, - createdAt: IsoDateTimeSchema, -}); - export const CreateThreadCommandSchema = Schema.Struct({ type: Schema.Literal("thread.create"), commandId: IdSchema, @@ -212,8 +183,6 @@ export const RevertThreadCommandSchema = Schema.Struct({ }); export const OrchestrationCommandSchema = Schema.Union( - CreateProjectCommandSchema, - DeleteProjectCommandSchema, CreateThreadCommandSchema, DeleteThreadCommandSchema, UpdateThreadMetaCommandSchema, -- 2.54.0