Effect persistence layer

README.md

@@ -34,6 +34,15 @@ T3 Code runs as a **Node.js WebSocket server** that wraps `codex app-server` (JS
 - `/apps/desktop`: Electron shell. Spawns a desktop-scoped `t3` backend process and loads the shared web app.
 - `/packages/contracts`: Shared Zod schemas and TypeScript contracts for provider events, WebSocket protocol, and model/session types.
 
+## Persistence architecture
+
+Server persistence now runs on the Effect v4 beta SQL stack with SQLite:
+
+- Effect SQL drivers are selected by runtime (Bun in development, Node in production).
+- Schema bootstrapping is managed through the Effect SQL migrator.
+- Persistence reads/writes are organized into Effect-backed repositories under `apps/server/src/persistence/`.
+- State synchronization (`state.bootstrap`, ordered `state.event`, `state.catchUp`) is still exposed through the existing WebSocket API surface, with typed state-event payloads shared through `@t3tools/contracts`.
+
 ## Codex prerequisites
 
 - Install Codex CLI so `codex` is on your PATH.

apps/server/package.json

@@ -17,7 +17,10 @@
     "test": "vitest run"
   },
   "dependencies": {
+    "@effect/sql-sqlite-bun": "^4.0.0-beta.6",
+    "@effect/sql-sqlite-node": "^4.0.0-beta.6",
     "@pierre/diffs": "^1.1.0-beta.16",
+    "effect": "^4.0.0-beta.6",
     "node-pty": "^1.1.0",
     "open": "^10.1.0",
     "ws": "^8.18.0"

apps/server/src/persistence/config.ts

@@ -0,0 +1,16 @@
+import path from "node:path";
+
+export interface PersistenceConfig {
+  dbPath: string;
+  legacyProjectsJsonPath?: string;
+}
+
+export function resolvePersistenceConfig(input: PersistenceConfig): PersistenceConfig {
+  const resolved: PersistenceConfig = {
+    dbPath: path.resolve(input.dbPath),
+  };
+  if (input.legacyProjectsJsonPath) {
+    resolved.legacyProjectsJsonPath = path.resolve(input.legacyProjectsJsonPath);
+  }
+  return resolved;
+}

apps/server/src/persistence/domain/appSettings.ts

@@ -0,0 +1,25 @@
+import {
+  type AppSettings,
+  type AppSettingsUpdateInput,
+  appSettingsSchema,
+  appSettingsUpdateInputSchema,
+} from "@t3tools/contracts";
+
+export function resolveAppSettings(metadataValue: unknown): AppSettings {
+  const parsed = appSettingsSchema.safeParse(metadataValue);
+  if (parsed.success) {
+    return parsed.data;
+  }
+  return appSettingsSchema.parse({});
+}
+
+export function buildUpdatedAppSettings(
+  current: AppSettings,
+  rawPatch: AppSettingsUpdateInput,
+): AppSettings {
+  const patch = appSettingsUpdateInputSchema.parse(rawPatch);
+  return appSettingsSchema.parse({
+    ...current,
+    ...patch,
+  });
+}

apps/server/src/persistence/domain/messages.ts

@@ -0,0 +1,39 @@
+import {
+  type ProviderSendTurnInput,
+  type StateMessage,
+  stateMessageSchema,
+} from "@t3tools/contracts";
+
+export function messageDocId(threadId: string, messageId: string): string {
+  return `message:${threadId}:${messageId}`;
+}
+
+export function buildUserTurnMessage(input: {
+  turn: ProviderSendTurnInput;
+  threadId: string;
+  messageId: string;
+  createdAt: string;
+}): StateMessage {
+  const text = input.turn.clientMessageText ?? input.turn.input ?? "";
+  const inputAttachments = input.turn.attachments ?? [];
+  const attachments =
+    inputAttachments.length > 0
+      ? inputAttachments.map((attachment, index) => ({
+          type: "image" as const,
+          id: `${input.messageId}:image:${index + 1}`,
+          name: attachment.name,
+          mimeType: attachment.mimeType,
+          sizeBytes: attachment.sizeBytes,
+        }))
+      : undefined;
+  return stateMessageSchema.parse({
+    id: input.messageId,
+    threadId: input.threadId,
+    role: "user",
+    text,
+    ...(attachments ? { attachments } : {}),
+    createdAt: input.createdAt,
+    updatedAt: input.createdAt,
+    streaming: false,
+  });
+}

apps/server/src/persistence/domain/projects.ts

@@ -0,0 +1,24 @@
+import fs from "node:fs";
+import path from "node:path";
+
+export 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;
+}
+
+export function isDirectory(cwd: string): boolean {
+  try {
+    return fs.statSync(cwd).isDirectory();
+  } catch {
+    return false;
+  }
+}
+
+export function inferProjectName(cwd: string): string {
+  const name = path.basename(cwd);
+  return name.length > 0 ? name : "project";
+}

apps/server/src/persistence/domain/providerProjection.ts

@@ -0,0 +1,42 @@
+import type { ProviderEvent } from "@t3tools/contracts";
+
+export function asObject(value: unknown): Record<string, unknown> | undefined {
+  if (!value || typeof value !== "object") {
+    return undefined;
+  }
+  return value as Record<string, unknown>;
+}
+
+export function asString(value: unknown): string | undefined {
+  return typeof value === "string" && value.length > 0 ? value : undefined;
+}
+
+export function normalizeProviderItemType(value: string | undefined): string | undefined {
+  if (!value) return undefined;
+  const normalized = value.trim();
+  if (normalized.length === 0) return undefined;
+  return normalized.replace(/[_\-\s]+/g, "").toLowerCase();
+}
+
+export function parseThreadIdFromEventPayload(payload: unknown): string | null {
+  const record = asObject(payload);
+  const threadId = asString(record?.threadId) ?? asString(record?.thread_id);
+  if (threadId) return threadId;
+  const thread = asObject(record?.thread);
+  return asString(thread?.id) ?? null;
+}
+
+export function parseTurnIdFromEvent(event: ProviderEvent): string | null {
+  if (event.turnId) return event.turnId;
+  const payload = asObject(event.payload);
+  const turn = asObject(payload?.turn);
+  return asString(turn?.id) ?? null;
+}
+
+export function parseAssistantItemId(event: ProviderEvent): string | null {
+  const payload = asObject(event.payload);
+  const item = asObject(payload?.item);
+  const itemType = asString(item?.type);
+  if (itemType !== "agentMessage") return null;
+  return asString(item?.id) ?? event.itemId ?? null;
+}

apps/server/src/persistence/domain/stateSync.ts

@@ -0,0 +1,41 @@
+import {
+  type StateBootstrapResult,
+  type StateBootstrapThread,
+  type StateCatchUpResult,
+  type StateEvent,
+  type StateListMessagesResult,
+  type StateMessage,
+  type StateProject,
+  stateBootstrapResultSchema,
+  stateCatchUpResultSchema,
+  stateListMessagesResultSchema,
+} from "@t3tools/contracts";
+
+export function buildStateBootstrapResult(input: {
+  projects: StateProject[];
+  threads: StateBootstrapThread[];
+  lastStateSeq: number;
+}): StateBootstrapResult {
+  return stateBootstrapResultSchema.parse(input);
+}
+
+export function buildStateCatchUpResult(input: {
+  events: StateEvent[];
+  lastStateSeq: number;
+}): StateCatchUpResult {
+  return stateCatchUpResultSchema.parse(input);
+}
+
+export function buildStateListMessagesResult(input: {
+  messages: StateMessage[];
+  total: number;
+  offset: number;
+  pageSize: number;
+}): StateListMessagesResult {
+  const nextOffset = input.offset + input.pageSize;
+  return stateListMessagesResultSchema.parse({
+    messages: input.messages,
+    total: input.total,
+    nextOffset: nextOffset < input.total ? nextOffset : null,
+  });
+}

apps/server/src/persistence/domain/threads.ts

@@ -0,0 +1,117 @@
+import type { StateThread } from "@t3tools/contracts";
+
+const MAX_TERMINAL_COUNT = 4;
+const DEFAULT_TERMINAL_ID = "default";
+
+export function normalizeTerminalIds(ids: readonly string[]): string[] {
+  const normalized = [
+    ...new Set(ids.map((id) => id.trim()).filter((id) => id.length > 0)),
+  ].slice(0, MAX_TERMINAL_COUNT);
+  if (normalized.length > 0) {
+    return normalized;
+  }
+  return [DEFAULT_TERMINAL_ID];
+}
+
+function normalizeRunningTerminalIds(
+  runningTerminalIds: readonly string[],
+  terminalIds: readonly string[],
+): string[] {
+  if (runningTerminalIds.length === 0) {
+    return [];
+  }
+
+  const validTerminalIds = new Set(terminalIds);
+  return [...new Set(runningTerminalIds)]
+    .map((id) => id.trim())
+    .filter((id) => id.length > 0 && validTerminalIds.has(id))
+    .slice(0, MAX_TERMINAL_COUNT);
+}
+
+export function fallbackGroupId(terminalId: string): string {
+  return `group-${terminalId}`;
+}
+
+function assignUniqueGroupId(groupId: string, usedGroupIds: Set<string>): string {
+  if (!usedGroupIds.has(groupId)) {
+    usedGroupIds.add(groupId);
+    return groupId;
+  }
+
+  let suffix = 2;
+  while (usedGroupIds.has(`${groupId}-${suffix}`)) {
+    suffix += 1;
+  }
+  const uniqueGroupId = `${groupId}-${suffix}`;
+  usedGroupIds.add(uniqueGroupId);
+  return uniqueGroupId;
+}
+
+function normalizeTerminalGroups(
+  groups: StateThread["terminalGroups"],
+  terminalIds: readonly string[],
+): StateThread["terminalGroups"] {
+  const validTerminalIds = new Set(terminalIds);
+  const assignedTerminalIds = new Set<string>();
+  const usedGroupIds = new Set<string>();
+  const normalizedGroups: StateThread["terminalGroups"] = [];
+
+  for (const group of groups) {
+    const groupTerminalIds = [
+      ...new Set(group.terminalIds.map((id) => id.trim()).filter((id) => id.length > 0)),
+    ].filter((terminalId) => {
+      if (!validTerminalIds.has(terminalId)) return false;
+      if (assignedTerminalIds.has(terminalId)) return false;
+      return true;
+    });
+    if (groupTerminalIds.length === 0) continue;
+    for (const terminalId of groupTerminalIds) {
+      assignedTerminalIds.add(terminalId);
+    }
+    const baseGroupId =
+      group.id.trim().length > 0
+        ? group.id.trim()
+        : fallbackGroupId(groupTerminalIds[0] ?? DEFAULT_TERMINAL_ID);
+    normalizedGroups.push({
+      id: assignUniqueGroupId(baseGroupId, usedGroupIds),
+      terminalIds: groupTerminalIds,
+    });
+  }
+
+  for (const terminalId of terminalIds) {
+    if (assignedTerminalIds.has(terminalId)) continue;
+    normalizedGroups.push({
+      id: assignUniqueGroupId(fallbackGroupId(terminalId), usedGroupIds),
+      terminalIds: [terminalId],
+    });
+  }
+
+  if (normalizedGroups.length > 0) {
+    return normalizedGroups;
+  }
+
+  return [{ id: fallbackGroupId(DEFAULT_TERMINAL_ID), terminalIds: [DEFAULT_TERMINAL_ID] }];
+}
+
+export function normalizeThread(thread: StateThread): StateThread {
+  const terminalIds = normalizeTerminalIds(thread.terminalIds);
+  const runningTerminalIds = normalizeRunningTerminalIds(thread.runningTerminalIds, terminalIds);
+  const activeTerminalId = terminalIds.includes(thread.activeTerminalId)
+    ? thread.activeTerminalId
+    : (terminalIds[0] ?? DEFAULT_TERMINAL_ID);
+  const terminalGroups = normalizeTerminalGroups(thread.terminalGroups, terminalIds);
+  const activeGroupId =
+    terminalGroups.find((group) => group.id === thread.activeTerminalGroupId)?.id ??
+    terminalGroups.find((group) => group.terminalIds.includes(activeTerminalId))?.id ??
+    terminalGroups[0]?.id ??
+    fallbackGroupId(activeTerminalId);
+
+  return {
+    ...thread,
+    terminalIds,
+    runningTerminalIds,
+    activeTerminalId,
+    terminalGroups,
+    activeTerminalGroupId: activeGroupId,
+  };
+}

apps/server/src/persistence/domain/turnSummaries.test.ts

@@ -0,0 +1,46 @@
+import { describe, expect, test } from "vitest";
+
+import { mergeTurnSummaryFiles, summarizeUnifiedDiff } from "./turnSummaries";
+
+describe("turnSummaries domain helpers", () => {
+  test("summarizeUnifiedDiff returns per-file diff stats", () => {
+    const diff = [
+      "diff --git a/src/example.ts b/src/example.ts",
+      "index 1111111..2222222 100644",
+      "--- a/src/example.ts",
+      "+++ b/src/example.ts",
+      "@@ -1,2 +1,3 @@",
+      " line1",
+      "-line2",
+      "+line2-updated",
+      "+line3",
+      "",
+    ].join("\n");
+
+    expect(summarizeUnifiedDiff(diff)).toEqual([
+      {
+        path: "src/example.ts",
+        kind: "change",
+        additions: 2,
+        deletions: 1,
+      },
+    ]);
+  });
+
+  test("mergeTurnSummaryFiles merges by path while preserving prior fields", () => {
+    const existing = [
+      { path: "a.ts", kind: "modified" as const, additions: 1, deletions: 2 },
+      { path: "b.ts", kind: "deleted" as const, additions: 0, deletions: 4 },
+    ];
+    const incoming = [
+      { path: "a.ts", additions: 3, deletions: 5 },
+      { path: "c.ts", kind: "added" as const, additions: 7, deletions: 0 },
+    ];
+
+    expect(mergeTurnSummaryFiles(existing, incoming)).toEqual([
+      { path: "a.ts", kind: "modified", additions: 3, deletions: 5 },
+      { path: "b.ts", kind: "deleted", additions: 0, deletions: 4 },
+      { path: "c.ts", kind: "added", additions: 7, deletions: 0 },
+    ]);
+  });
+});