From e7acf559e154bea817f3342145f91af32d3bb720 Mon Sep 17 00:00:00 2001 From: olafura Date: Thu, 20 Aug 2026 17:20:30 +0200 Subject: [PATCH 1/2] feat(server): list threads in a T3 state directory Add `vp run thread:list`, which prints the live threads of an existing T3 state database with their workspace roots and titles, or the full records as JSON. The source may be a workspace containing `.t3`, the T3 base directory, or a direct state directory, and `--state dev` selects a main-checkout dev database. The database is opened read-only. This is the first step of thread transfer; export and import build on the same state-directory resolution. Co-Authored-By: Claude Fable 5 --- apps/server/scripts/list-threads.test.ts | 202 +++++++++++++++++++++++ apps/server/scripts/list-threads.ts | 65 ++++++++ apps/server/scripts/thread-transfer.ts | 172 +++++++++++++++++++ docs/internals/scripts.md | 4 + package.json | 1 + 5 files changed, 444 insertions(+) create mode 100644 apps/server/scripts/list-threads.test.ts create mode 100644 apps/server/scripts/list-threads.ts create mode 100644 apps/server/scripts/thread-transfer.ts diff --git a/apps/server/scripts/list-threads.test.ts b/apps/server/scripts/list-threads.test.ts new file mode 100644 index 000000000000..ced3120c611c --- /dev/null +++ b/apps/server/scripts/list-threads.test.ts @@ -0,0 +1,202 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as NodeSqliteClient from "../src/persistence/NodeSqliteClient.ts"; +import { listThreads, ThreadTransferError } from "./thread-transfer.ts"; + +interface FixtureInput { + readonly stateDir: string; + readonly projects: ReadonlyArray<{ + readonly projectId: string; + readonly workspaceRoot: string; + readonly deletedAt?: string; + }>; + readonly threads: ReadonlyArray<{ + readonly threadId: string; + readonly projectId: string; + readonly title: string; + readonly updatedAt: string; + readonly deletedAt?: string; + }>; +} + +/** Seeds the projection tables `thread:list` reads, mirroring the server's schema. */ +const createFixtureDatabase = Effect.fn("createThreadListFixtureDatabase")(function* ( + input: FixtureInput, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const databasePath = path.join(input.stateDir, "state.sqlite"); + yield* fs.makeDirectory(input.stateDir, { recursive: true }); + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE projection_projects ( + project_id TEXT PRIMARY KEY, + title TEXT NOT NULL, + workspace_root TEXT NOT NULL, + updated_at TEXT NOT NULL, + deleted_at TEXT + )`; + yield* sql`CREATE TABLE projection_threads ( + thread_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + title TEXT NOT NULL, + updated_at TEXT NOT NULL, + deleted_at TEXT + )`; + for (const project of input.projects) { + yield* sql`INSERT INTO projection_projects ( + project_id, title, workspace_root, updated_at, deleted_at + ) VALUES ( + ${project.projectId}, ${`Project ${project.projectId}`}, ${project.workspaceRoot}, + '2026-08-20T12:00:00.000Z', ${project.deletedAt ?? null} + )`; + } + for (const thread of input.threads) { + yield* sql`INSERT INTO projection_threads ( + thread_id, project_id, title, updated_at, deleted_at + ) VALUES ( + ${thread.threadId}, ${thread.projectId}, ${thread.title}, ${thread.updatedAt}, + ${thread.deletedAt ?? null} + )`; + } + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: databasePath }))); + return databasePath; +}); + +it.layer(NodeServices.layer)("thread list", (it) => { + it.effect("lists live projects and threads, newest thread first", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-list-" }); + const workspace = path.join(root, "workspace"); + const otherWorkspace = path.join(root, "other"); + yield* createFixtureDatabase({ + stateDir: path.join(workspace, ".t3", "userdata"), + projects: [ + { projectId: "project-b", workspaceRoot: otherWorkspace }, + { projectId: "project-a", workspaceRoot: workspace }, + { + projectId: "project-gone", + workspaceRoot: path.join(root, "gone"), + deletedAt: "2026-08-21T13:00:00.000Z", + }, + ], + threads: [ + { + threadId: "thread-old", + projectId: "project-a", + title: "Older thread", + updatedAt: "2026-08-19T12:00:00.000Z", + }, + { + threadId: "thread-new", + projectId: "project-b", + title: "Newer\tthread", + updatedAt: "2026-08-20T12:00:00.000Z", + }, + { + threadId: "thread-gone", + projectId: "project-a", + title: "Deleted thread", + updatedAt: "2026-08-21T12:00:00.000Z", + deletedAt: "2026-08-21T13:00:00.000Z", + }, + ], + }); + + const listing = yield* listThreads({ source: workspace }); + assert.deepStrictEqual(listing, { + projects: [ + { + id: "project-b", + title: "Project project-b", + workspaceRoot: otherWorkspace, + updatedAt: "2026-08-20T12:00:00.000Z", + }, + { + id: "project-a", + title: "Project project-a", + workspaceRoot: workspace, + updatedAt: "2026-08-20T12:00:00.000Z", + }, + ], + threads: [ + { + id: "thread-new", + title: "Newer\tthread", + projectId: "project-b", + projectTitle: "Project project-b", + workspaceRoot: otherWorkspace, + updatedAt: "2026-08-20T12:00:00.000Z", + }, + { + id: "thread-old", + title: "Older thread", + projectId: "project-a", + projectTitle: "Project project-a", + workspaceRoot: workspace, + updatedAt: "2026-08-19T12:00:00.000Z", + }, + ], + }); + }), + ); + + it.effect("resolves the base directory, a direct state directory, and --state dev", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-list-state-" }); + const baseDir = path.join(root, ".t3"); + const thread = { title: "Thread", updatedAt: "2026-08-20T12:00:00.000Z" }; + yield* createFixtureDatabase({ + stateDir: path.join(baseDir, "userdata"), + projects: [{ projectId: "project-userdata", workspaceRoot: root }], + threads: [{ threadId: "thread-userdata", projectId: "project-userdata", ...thread }], + }); + yield* createFixtureDatabase({ + stateDir: path.join(baseDir, "dev"), + projects: [{ projectId: "project-dev", workspaceRoot: root }], + threads: [{ threadId: "thread-dev", projectId: "project-dev", ...thread }], + }); + + const fromBase = yield* listThreads({ source: baseDir }); + assert.deepStrictEqual( + fromBase.threads.map((entry) => entry.id), + ["thread-userdata"], + ); + const fromStateDir = yield* listThreads({ source: path.join(baseDir, "userdata") }); + assert.deepStrictEqual(fromStateDir, fromBase); + const fromDev = yield* listThreads({ source: baseDir, state: "dev" }); + assert.deepStrictEqual( + fromDev.projects.map((entry) => entry.id), + ["project-dev"], + ); + assert.deepStrictEqual( + fromDev.threads.map((entry) => entry.id), + ["thread-dev"], + ); + }), + ); + + it.effect("reports every location it probed when no database exists", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-list-missing-" }); + const error = yield* listThreads({ source: root }).pipe(Effect.flip); + assert.instanceOf(error, ThreadTransferError); + assert.equal(error.operation, "resolve directory"); + assert.equal( + error.detail, + `No T3 userdata database found at '${path.join(root, "state.sqlite")}', '${path.join(root, "userdata", "state.sqlite")}', or '${path.join(root, ".t3", "userdata", "state.sqlite")}'.`, + ); + }), + ); +}); diff --git a/apps/server/scripts/list-threads.ts b/apps/server/scripts/list-threads.ts new file mode 100644 index 000000000000..f7b07c1a5239 --- /dev/null +++ b/apps/server/scripts/list-threads.ts @@ -0,0 +1,65 @@ +#!/usr/bin/env node + +import * as NodeRuntime from "@effect/platform-node/NodeRuntime"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { fromJsonStringPretty } from "@t3tools/shared/schemaJson"; +import * as Console from "effect/Console"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import { Command, Flag } from "effect/unstable/cli"; + +import { listThreads, ThreadListing, ThreadTransferState } from "./thread-transfer.ts"; + +const oneLine = (value: string) => value.replaceAll(/[\r\n\t]+/g, " "); +const encodeListing = Schema.encodeEffect(fromJsonStringPretty(ThreadListing)); + +export const listThreadsCommand = Command.make( + "list-threads", + { + source: Flag.string("source").pipe( + Flag.withDescription("Workspace root, T3 base directory, or direct state directory."), + ), + state: Flag.choice("state", ThreadTransferState.literals).pipe( + Flag.withDefault("userdata"), + Flag.withDescription("State directory below the T3 base directory; defaults to userdata."), + ), + json: Flag.boolean("json").pipe( + Flag.withDefault(false), + Flag.withDescription("Print the complete thread list as JSON."), + ), + }, + ({ source, state, json }) => + Effect.gen(function* () { + const listing = yield* listThreads({ source, state }); + if (json) { + yield* Console.log(yield* encodeListing(listing)); + return; + } + if (listing.projects.length === 0) { + yield* Console.log("No projects found."); + } else { + yield* Console.log("PROJECT_ID\tWORKSPACE\tTITLE"); + for (const project of listing.projects) { + yield* Console.log( + `${project.id}\t${oneLine(project.workspaceRoot)}\t${oneLine(project.title)}`, + ); + } + } + yield* Console.log(""); + if (listing.threads.length === 0) { + yield* Console.log("No threads found."); + return; + } + yield* Console.log("THREAD_ID\tPROJECT_ID\tTITLE"); + for (const thread of listing.threads) { + yield* Console.log(`${thread.id}\t${thread.projectId}\t${oneLine(thread.title)}`); + } + }), +).pipe(Command.withDescription("List the projects and threads stored in a T3 state directory.")); + +if (import.meta.main) { + Command.run(listThreadsCommand, { version: "0.0.0" }).pipe( + Effect.provide(NodeServices.layer), + NodeRuntime.runMain, + ); +} diff --git a/apps/server/scripts/thread-transfer.ts b/apps/server/scripts/thread-transfer.ts new file mode 100644 index 000000000000..45c85f5e1694 --- /dev/null +++ b/apps/server/scripts/thread-transfer.ts @@ -0,0 +1,172 @@ +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import { isSqlError } from "effect/unstable/sql/SqlError"; + +import * as NodeSqliteClient from "../src/persistence/NodeSqliteClient.ts"; + +export const ThreadTransferState = Schema.Literals(["userdata", "dev"]); +export type ThreadTransferState = typeof ThreadTransferState.Type; + +export class ThreadTransferError extends Schema.TaggedErrorClass()( + "ThreadTransferError", + { + operation: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + return `${this.operation}: ${this.detail}`; + } +} + +export interface ListThreadsInput { + /** Workspace root, T3 base directory, or direct state directory. */ + readonly source: string; + readonly state?: ThreadTransferState | undefined; +} + +export const ListedProject = Schema.Struct({ + id: Schema.String, + title: Schema.String, + workspaceRoot: Schema.String, + updatedAt: Schema.NullOr(Schema.String), +}); +export type ListedProject = typeof ListedProject.Type; + +export const ListedThread = Schema.Struct({ + id: Schema.String, + title: Schema.String, + projectId: Schema.String, + projectTitle: Schema.String, + workspaceRoot: Schema.String, + updatedAt: Schema.NullOr(Schema.String), +}); +export type ListedThread = typeof ListedThread.Type; + +/** The live projects of a state directory and the live threads across all its projects. */ +export const ThreadListing = Schema.Struct({ + projects: Schema.Array(ListedProject), + threads: Schema.Array(ListedThread), +}); +export type ThreadListing = typeof ThreadListing.Type; + +interface StateLocation { + readonly stateDir: string; + readonly databasePath: string; + readonly workspaceRoot: string | null; +} + +interface ProjectRow { + readonly projectId: string; + readonly title: string; + readonly workspaceRoot: string; + readonly updatedAt: string | null; + readonly deletedAt: string | null; +} + +interface ListedThreadRow { + readonly threadId: string; + readonly projectId: string; + readonly title: string; + readonly updatedAt: string | null; +} + +const transferError = (operation: string, detail: string, cause?: unknown): ThreadTransferError => + new ThreadTransferError({ operation, detail, ...(cause === undefined ? {} : { cause }) }); + +/** Runs `effect` against the state database, reporting SQL failures as `operation` errors. */ +const withThreadDatabase = + (location: StateLocation, operation: string, access: "read" | "update") => + (effect: Effect.Effect) => + effect.pipe( + Effect.provideService(SqlClient.SafeIntegers, access === "read"), + Effect.provide( + NodeSqliteClient.layer({ filename: location.databasePath, readonly: access === "read" }), + ), + Effect.mapError((cause) => + isSqlError(cause) + ? transferError(operation, `Could not ${access} '${location.databasePath}'.`, cause) + : cause, + ), + ); + +/** Accepts a state directory, a T3 base directory, or a workspace root holding `.t3/`. */ +const resolveStateLocation = Effect.fn("resolveThreadTransferStateLocation")(function* ( + directory: string, + state: ThreadTransferState = "userdata", +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = path.resolve(directory); + const candidates = [ + { stateDir: root, workspaceRoot: null }, + { stateDir: path.join(root, state), workspaceRoot: null }, + { stateDir: path.join(root, ".t3", state), workspaceRoot: root }, + ].map( + (candidate): StateLocation => ({ + ...candidate, + databasePath: path.join(candidate.stateDir, "state.sqlite"), + }), + ); + for (const candidate of candidates) { + const exists = yield* fs.exists(candidate.databasePath).pipe(Effect.orElseSucceed(() => false)); + if (exists) return candidate; + } + const [direct, base, nested] = candidates.map((candidate) => `'${candidate.databasePath}'`); + return yield* transferError( + "resolve directory", + `No T3 ${state} database found at ${direct}, ${base}, or ${nested}.`, + ); +}); + +export const listThreads = Effect.fn("listThreads")(function* (input: ListThreadsInput) { + const location = yield* resolveStateLocation(input.source, input.state); + return yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const projects = yield* sql` + SELECT + project_id AS "projectId", + title, + workspace_root AS "workspaceRoot", + updated_at AS "updatedAt", + deleted_at AS "deletedAt" + FROM projection_projects`; + const projectsById = new Map(projects.map((project) => [project.projectId, project])); + const threads = yield* sql` + SELECT thread_id AS "threadId", project_id AS "projectId", title, updated_at AS "updatedAt" + FROM projection_threads + WHERE deleted_at IS NULL`; + const listing: ThreadListing = { + projects: projects + .filter((project) => project.deletedAt === null) + .map((project) => ({ + id: project.projectId, + title: project.title, + workspaceRoot: project.workspaceRoot, + updatedAt: project.updatedAt, + })) + .sort((left, right) => left.workspaceRoot.localeCompare(right.workspaceRoot)), + threads: threads + .map((thread): ListedThread => { + const project = projectsById.get(thread.projectId); + return { + id: thread.threadId, + title: thread.title, + projectId: thread.projectId, + projectTitle: project?.title ?? thread.projectId, + workspaceRoot: project?.workspaceRoot ?? "", + updatedAt: thread.updatedAt, + }; + }) + .sort((left, right) => { + const updated = (right.updatedAt ?? "").localeCompare(left.updatedAt ?? ""); + return updated !== 0 ? updated : left.id.localeCompare(right.id); + }), + }; + return listing; + }).pipe(withThreadDatabase(location, "list threads", "read")); +}); diff --git a/docs/internals/scripts.md b/docs/internals/scripts.md index b6cb014932e1..baf4b5fa6b78 100644 --- a/docs/internals/scripts.md +++ b/docs/internals/scripts.md @@ -63,6 +63,10 @@ authenticated. - `vp run lint:mobile`: Mobile native static analysis (`scripts/mobile-native-static-check.ts`). - `node apps/server/scripts/t3-sqlite-state.ts --base-dir ...`: Inspects or seeds an isolated T3 SQLite database; writes create a private backup first. +- `vp run thread:list --source `: Lists the projects (id, workspace root, title) and + live threads (id, project id, title) of an existing T3 state database. The source can be a workspace containing `.t3`, the T3 base + directory, or a direct state directory containing `state.sqlite`. State defaults to `userdata`; + pass `--state dev` for a main-checkout dev database. Pass `--json` for structured output. ## Desktop artifacts diff --git a/package.json b/package.json index 3fc66d0dd021..454ca8af9d54 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ "dev:marketing": "vp run --filter @t3tools/marketing dev", "dev:desktop": "node scripts/dev-runner.ts dev:desktop", "migrate-dev-db": "node apps/server/scripts/migrate-dev-db.ts", + "thread:list": "node apps/server/scripts/list-threads.ts", "start": "vp run --filter t3 start", "start:desktop": "vp run --filter @t3tools/desktop start", "start:marketing": "vp run --filter @t3tools/marketing preview", From 94240e5d5083f00bfbf7818ee8afee3cce223b93 Mon Sep 17 00:00:00 2001 From: olafura Date: Thu, 20 Aug 2026 17:21:05 +0200 Subject: [PATCH 2/2] feat(server): export and import threads between T3 state directories Add `vp run thread:export` and `vp run thread:import`. Export writes one thread's orchestration events and image attachments (terminal logs only with `--include-terminal-logs`, since they may hold credentials) into a self-contained JSON archive with per-file checksums. Import validates the archive (every event belongs to the thread, decodes against the orchestration contract, and carries unique ids and stream versions), refuses a destination that already holds the thread, backs the destination database up with VACUUM INTO, remaps the thread onto the target project, clears worktree paths that do not exist on the destination, and writes only the events: the destination server replays them above the projectors' recorded sequence on its next start and rebuilds the read model itself. Copying projection rows too would make that replay append onto already-complete rows. The live ~/.t3/userdata database is refused unless `--dangerous-allow-t3-directory` is passed, which is how a thread moves from a dev checkout back into the real install once its server is stopped. The source and destination accept the same directory forms and `--state dev` selection as `thread:list`. `ensureDevDbNotInUse` is the renamed dev-db guard from migrate-dev-db.ts, reused to refuse importing into a running server's database. Co-Authored-By: Claude Fable 5 --- apps/server/scripts/export-thread.ts | 44 ++ apps/server/scripts/import-thread.ts | 61 ++ apps/server/scripts/list-threads.ts | 11 +- apps/server/scripts/migrate-dev-db.ts | 8 +- apps/server/scripts/thread-transfer.test.ts | 581 ++++++++++++++++ apps/server/scripts/thread-transfer.ts | 697 +++++++++++++++++++- docs/internals/scripts.md | 27 +- package.json | 2 + 8 files changed, 1415 insertions(+), 16 deletions(-) create mode 100644 apps/server/scripts/export-thread.ts create mode 100644 apps/server/scripts/import-thread.ts create mode 100644 apps/server/scripts/thread-transfer.test.ts diff --git a/apps/server/scripts/export-thread.ts b/apps/server/scripts/export-thread.ts new file mode 100644 index 000000000000..efb9fd236890 --- /dev/null +++ b/apps/server/scripts/export-thread.ts @@ -0,0 +1,44 @@ +#!/usr/bin/env node + +import * as NodeRuntime from "@effect/platform-node/NodeRuntime"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Console from "effect/Console"; +import * as Effect from "effect/Effect"; +import { Command, Flag } from "effect/unstable/cli"; + +import { exportThread, threadTransferFlags } from "./thread-transfer.ts"; + +export const exportThreadCommand = Command.make( + "export-thread", + { + source: threadTransferFlags.directory("source"), + state: threadTransferFlags.state, + threadId: Flag.string("thread-id").pipe(Flag.withDescription("Thread to export.")), + output: Flag.string("output").pipe(Flag.withDescription("Archive JSON file to create.")), + includeTerminalLogs: Flag.boolean("include-terminal-logs").pipe( + Flag.withDefault(false), + Flag.withDescription("Include persisted terminal history, which may contain secrets."), + ), + }, + ({ source, state, threadId, output, includeTerminalLogs }) => + Effect.gen(function* () { + const result = yield* exportThread({ + source, + state, + threadId, + output, + includeTerminalLogs, + }); + yield* Console.log(`Exported '${result.title}' (${result.threadId}) to ${result.output}`); + yield* Console.log( + ` ${result.eventCount} events, ${result.attachmentCount} attachments, ${result.terminalLogCount} terminal logs`, + ); + }), +).pipe(Command.withDescription("Export one T3 thread and its supporting files.")); + +if (import.meta.main) { + Command.run(exportThreadCommand, { version: "0.0.0" }).pipe( + Effect.provide(NodeServices.layer), + NodeRuntime.runMain, + ); +} diff --git a/apps/server/scripts/import-thread.ts b/apps/server/scripts/import-thread.ts new file mode 100644 index 000000000000..013e8fa78485 --- /dev/null +++ b/apps/server/scripts/import-thread.ts @@ -0,0 +1,61 @@ +#!/usr/bin/env node + +import * as NodeRuntime from "@effect/platform-node/NodeRuntime"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Console from "effect/Console"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import { Command, Flag } from "effect/unstable/cli"; + +import { importThread, threadTransferFlags } from "./thread-transfer.ts"; + +export const importThreadCommand = Command.make( + "import-thread", + { + archive: Flag.string("archive").pipe(Flag.withDescription("Thread archive JSON to import.")), + destination: threadTransferFlags.directory("destination"), + state: threadTransferFlags.state, + targetProjectId: Flag.string("target-project-id").pipe( + Flag.optional, + Flag.withDescription("Project id when it cannot be inferred from the destination path."), + ), + dangerousAllowT3Directory: Flag.boolean("dangerous-allow-t3-directory").pipe( + Flag.withDefault(false), + Flag.withDescription( + "Allow importing into the live ~/.t3/userdata database. Stop the T3 server that uses it first.", + ), + ), + }, + ({ archive, destination, state, targetProjectId, dangerousAllowT3Directory }) => + Effect.gen(function* () { + const result = yield* importThread({ + archive, + destination, + state, + targetProjectId: Option.getOrUndefined(targetProjectId), + dangerousAllowT3Directory, + }); + yield* Console.log( + `Imported '${result.title}' (${result.threadId}) into ${result.targetProjectTitle}`, + ); + yield* Console.log( + ` ${result.eventCount} events, ${result.attachmentCount} attachments, ${result.terminalLogCount} terminal logs`, + ); + yield* Console.log(` Database backup: ${result.backup}`); + if (result.droppedWorktreePaths.length > 0) { + yield* Console.log( + ` Cleared worktree paths missing here (the thread will use the project workspace): ${result.droppedWorktreePaths.join(", ")}`, + ); + } + yield* Console.log( + "Start the destination T3 server; it rebuilds the thread's read model from the imported events.", + ); + }), +).pipe(Command.withDescription("Import one T3 thread into an isolated project database.")); + +if (import.meta.main) { + Command.run(importThreadCommand, { version: "0.0.0" }).pipe( + Effect.provide(NodeServices.layer), + NodeRuntime.runMain, + ); +} diff --git a/apps/server/scripts/list-threads.ts b/apps/server/scripts/list-threads.ts index f7b07c1a5239..920d71337966 100644 --- a/apps/server/scripts/list-threads.ts +++ b/apps/server/scripts/list-threads.ts @@ -8,7 +8,7 @@ import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import { Command, Flag } from "effect/unstable/cli"; -import { listThreads, ThreadListing, ThreadTransferState } from "./thread-transfer.ts"; +import { listThreads, ThreadListing, threadTransferFlags } from "./thread-transfer.ts"; const oneLine = (value: string) => value.replaceAll(/[\r\n\t]+/g, " "); const encodeListing = Schema.encodeEffect(fromJsonStringPretty(ThreadListing)); @@ -16,13 +16,8 @@ const encodeListing = Schema.encodeEffect(fromJsonStringPretty(ThreadListing)); export const listThreadsCommand = Command.make( "list-threads", { - source: Flag.string("source").pipe( - Flag.withDescription("Workspace root, T3 base directory, or direct state directory."), - ), - state: Flag.choice("state", ThreadTransferState.literals).pipe( - Flag.withDefault("userdata"), - Flag.withDescription("State directory below the T3 base directory; defaults to userdata."), - ), + source: threadTransferFlags.directory("source"), + state: threadTransferFlags.state, json: Flag.boolean("json").pipe( Flag.withDefault(false), Flag.withDescription("Print the complete thread list as JSON."), diff --git a/apps/server/scripts/migrate-dev-db.ts b/apps/server/scripts/migrate-dev-db.ts index 0958f2149f45..53b1d56e0c29 100644 --- a/apps/server/scripts/migrate-dev-db.ts +++ b/apps/server/scripts/migrate-dev-db.ts @@ -188,7 +188,9 @@ const isProcessAlive = (pid: number): boolean => { * wal_checkpoint(TRUNCATE) reports busy while another connection holds the * WAL. A leftover -shm alone is not a signal — read-only connections cannot * clean it up on close. */ -const ensureNotInUse = Effect.fn("ensureDevDbNotInUse")(function* (databasePath: string) { +export const ensureDevDbNotInUse = Effect.fn("ensureDevDbNotInUse")(function* ( + databasePath: string, +) { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; @@ -403,7 +405,7 @@ export const runMigrateDevDb = Effect.fn("runMigrateDevDb")(function* ( } yield* fs.makeDirectory(stateDir, { recursive: true }); - yield* ensureNotInUse(databasePath); + yield* ensureDevDbNotInUse(databasePath); const wrapPhase = (phase: MigrateDevDbPhaseError["phase"], phaseDatabasePath: string) => @@ -466,7 +468,7 @@ export const runMigrateDevDb = Effect.fn("runMigrateDevDb")(function* ( yield* Console.log(`Compacting into ${databasePath}...`); // Re-check right before the swap: a dev server started while the // snapshot was migrating and pruning must not lose its database. - yield* ensureNotInUse(databasePath); + yield* ensureDevDbNotInUse(databasePath); yield* removeDatabaseFiles(databasePath); yield* Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/apps/server/scripts/thread-transfer.test.ts b/apps/server/scripts/thread-transfer.test.ts new file mode 100644 index 000000000000..5cddb11a7a73 --- /dev/null +++ b/apps/server/scripts/thread-transfer.test.ts @@ -0,0 +1,581 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Encoding from "effect/Encoding"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as NodeSqliteClient from "../src/persistence/NodeSqliteClient.ts"; +import { + exportThread, + importThread, + listThreads, + ThreadArchive, + ThreadTransferError, +} from "./thread-transfer.ts"; + +const withDatabase = + (databasePath: string, readonly = false) => + (effect: Effect.Effect) => + effect.pipe(Effect.provide(NodeSqliteClient.layer({ filename: databasePath, readonly }))); + +const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); +const decodeArchive = Schema.decodeUnknownSync(Schema.fromJsonString(ThreadArchive)); +const encodeArchive = Schema.encodeSync(Schema.fromJsonString(ThreadArchive)); + +const failImport = (...args: Parameters) => + importThread(...args).pipe( + Effect.flip, + Effect.map((error) => { + assert.instanceOf(error, ThreadTransferError); + return error; + }), + ); + +const failExport = (...args: Parameters) => + exportThread(...args).pipe( + Effect.flip, + Effect.map((error) => { + assert.instanceOf(error, ThreadTransferError); + return error; + }), + ); + +const decodeProjectPayload = Schema.decodeUnknownSync( + Schema.fromJsonString( + Schema.Struct({ + projectId: Schema.String, + worktreePath: Schema.optional(Schema.NullOr(Schema.String)), + }), + ), +); + +const TERMINAL_LOG_CONTENTS = "terminal output\n"; + +/** Writes one terminal history file owned by `threadId` below `stateDir` and returns its name. */ +const writeTerminalLog = Effect.fn("writeThreadTransferTerminalLog")(function* ( + stateDir: string, + threadId: string, + contents = TERMINAL_LOG_CONTENTS, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = path.join(stateDir, "logs", "terminals"); + const fileName = `terminal_${Encoding.encodeBase64Url(threadId)}.log`; + yield* fs.makeDirectory(directory, { recursive: true }); + yield* fs.writeFileString(path.join(directory, fileName), contents); + return fileName; +}); + +interface FixtureInput { + readonly workspace: string; + readonly projectId: string; + readonly threadId?: string; + readonly state?: "userdata" | "dev"; + readonly worktreePath?: string; +} + +/** + * Seeds the subset of the server schema the transfer scripts touch. The thread + * event is a contract-valid `thread.created` because the importer decodes + * archives against `@t3tools/contracts`. + */ +const createFixtureDatabase = Effect.fn("createThreadTransferFixtureDatabase")(function* ( + input: FixtureInput, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const stateDir = path.join(input.workspace, ".t3", input.state ?? "userdata"); + const databasePath = path.join(stateDir, "state.sqlite"); + yield* fs.makeDirectory(stateDir, { recursive: true }); + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE projection_projects ( + project_id TEXT PRIMARY KEY, + title TEXT NOT NULL, + workspace_root TEXT NOT NULL, + updated_at TEXT NOT NULL, + deleted_at TEXT + )`; + yield* sql`CREATE TABLE projection_threads ( + thread_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + title TEXT NOT NULL, + worktree_path TEXT, + updated_at TEXT NOT NULL, + deleted_at TEXT + )`; + yield* sql`CREATE TABLE projection_thread_messages ( + message_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + text TEXT NOT NULL + )`; + yield* sql`CREATE TABLE orchestration_events ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + aggregate_kind TEXT NOT NULL, + stream_id TEXT NOT NULL, + stream_version INTEGER NOT NULL, + event_type TEXT NOT NULL, + occurred_at TEXT NOT NULL, + command_id TEXT, + causation_event_id TEXT, + correlation_id TEXT, + actor_kind TEXT NOT NULL, + payload_json TEXT NOT NULL, + metadata_json TEXT NOT NULL, + UNIQUE (aggregate_kind, stream_id, stream_version) + )`; + yield* sql`INSERT INTO projection_projects ( + project_id, title, workspace_root, updated_at, deleted_at + ) VALUES ( + ${input.projectId}, + ${`Project ${input.projectId}`}, + ${input.workspace}, + '2026-08-20T12:00:00.000Z', + NULL + )`; + if (input.threadId === undefined) return; + + yield* sql`INSERT INTO projection_threads ( + thread_id, project_id, title, worktree_path, updated_at, deleted_at + ) VALUES ( + ${input.threadId}, ${input.projectId}, 'Image rendering thread', + ${input.worktreePath ?? null}, '2026-08-20T12:00:00.000Z', NULL + )`; + yield* sql`INSERT INTO projection_thread_messages ( + message_id, thread_id, text + ) VALUES ('message-v1', ${input.threadId}, 'Render this image')`; + yield* sql`INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, + command_id, causation_event_id, correlation_id, actor_kind, payload_json, metadata_json + ) VALUES ( + 'event-v1', 'thread', ${input.threadId}, 0, 'thread.created', + '2026-08-20T12:00:00.000Z', NULL, NULL, NULL, 'client', + ${encodeUnknownJson({ + threadId: input.threadId, + projectId: input.projectId, + title: "Image rendering thread", + modelSelection: { provider: "codex", model: "gpt-5-codex" }, + branch: null, + worktreePath: input.worktreePath ?? null, + createdAt: "2026-08-20T12:00:00.000Z", + updatedAt: "2026-08-20T12:00:00.000Z", + })}, '{}' + )`; + }).pipe(withDatabase(databasePath)); + return databasePath; +}); + +/** Seeds a v1 source state with one thread and one attachment, returning what the tests need. */ +const createAttachmentSource = Effect.fn("createThreadTransferAttachmentSource")(function* ( + source: string, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + yield* createFixtureDatabase({ + workspace: source, + projectId: "project-source", + threadId: "thread-v1", + }); + const attachmentName = "thread-v1-00000000-0000-4000-8000-000000000001.png"; + const attachmentsDir = path.join(source, ".t3", "userdata", "attachments"); + yield* fs.makeDirectory(attachmentsDir, { recursive: true }); + yield* fs.writeFile( + path.join(attachmentsDir, attachmentName), + Uint8Array.from([137, 80, 78, 71]), + ); + return { attachmentName }; +}); + +const readThreadEventIds = (databasePath: string, threadId: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly eventId: string }>` + SELECT event_id AS "eventId" FROM orchestration_events + WHERE stream_id = ${threadId} ORDER BY sequence`; + return rows.map((row) => row.eventId); + }).pipe(withDatabase(databasePath, true)); + +it.layer(NodeServices.layer)("thread transfer", (it) => { + it.effect("moves a v1 thread, its projection rows, and its image into another project", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-v1-" }); + const source = path.join(root, "source"); + const destination = path.join(root, "destination"); + const archivePath = path.join(root, "thread.json"); + const { attachmentName } = yield* createAttachmentSource(source); + const destinationDatabase = yield* createFixtureDatabase({ + workspace: destination, + projectId: "project-target", + }); + const sourceStateDir = path.join(source, ".t3", "userdata"); + const terminalLogName = yield* writeTerminalLog(sourceStateDir, "thread-v1"); + yield* writeTerminalLog(sourceStateDir, "unrelated", "unrelated output\n"); + + const exported = yield* exportThread({ + source, + threadId: "thread-v1", + output: archivePath, + includeTerminalLogs: true, + }); + assert.equal(exported.attachmentCount, 1); + assert.equal(exported.terminalLogCount, 1); + + const imported = yield* importThread( + { archive: archivePath, destination }, + { sharedHome: path.join(root, "shared-home") }, + ); + assert.equal(imported.targetProjectId, "project-target"); + assert.deepStrictEqual( + yield* readThreadEventIds(imported.backup, "thread-v1"), + [], + "backup holds the pre-import database", + ); + assert.isTrue( + yield* fs.exists(path.join(destination, ".t3", "userdata", "attachments", attachmentName)), + ); + assert.equal(imported.terminalLogCount, 1); + assert.equal( + yield* fs.readFileString( + path.join(destination, ".t3", "userdata", "logs", "terminals", terminalLogName), + ), + TERMINAL_LOG_CONTENTS, + ); + + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const threads = yield* sql`SELECT thread_id FROM projection_threads`; + const events = yield* sql<{ readonly payload: string }>` + SELECT payload_json AS payload FROM orchestration_events WHERE stream_id = 'thread-v1'`; + assert.deepStrictEqual(threads, [], "read model is left for the server to rebuild"); + assert.equal(decodeProjectPayload(events[0]!.payload).projectId, "project-target"); + }).pipe(withDatabase(destinationDatabase, true)); + }), + ); + + it.effect("keeps worktree paths that exist on the destination and clears the rest", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-worktree-" }); + const sharedHome = path.join(root, "shared-home"); + const presentWorktree = path.join(root, "present-worktree"); + yield* fs.makeDirectory(presentWorktree, { recursive: true }); + const missingWorktree = path.join(root, "missing-worktree"); + + const readImportedWorktree = (destinationDatabase: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const events = yield* sql<{ readonly payload: string }>` + SELECT payload_json AS payload FROM orchestration_events WHERE stream_id = 'thread-v1'`; + return decodeProjectPayload(events[0]!.payload).worktreePath; + }).pipe(withDatabase(destinationDatabase, true)); + + for (const [name, worktreePath, expected] of [ + ["present", presentWorktree, presentWorktree], + ["missing", missingWorktree, null], + ] as const) { + const source = path.join(root, name, "source"); + const destination = path.join(root, name, "destination"); + const archivePath = path.join(root, name, "thread.json"); + yield* createFixtureDatabase({ + workspace: source, + projectId: "project-source", + threadId: "thread-v1", + worktreePath, + }); + const destinationDatabase = yield* createFixtureDatabase({ + workspace: destination, + projectId: "project-target", + }); + yield* exportThread({ source, threadId: "thread-v1", output: archivePath }); + const imported = yield* importThread({ archive: archivePath, destination }, { sharedHome }); + assert.deepStrictEqual( + imported.droppedWorktreePaths, + expected === null ? [worktreePath] : [], + name, + ); + assert.equal(yield* readImportedWorktree(destinationDatabase), expected, name); + } + }), + ); + + it.effect("falls back to the thread.created payload when the projection row is gone", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-no-row-" }); + const source = path.join(root, "source"); + const archivePath = path.join(root, "thread.json"); + const sourceDatabase = yield* createFixtureDatabase({ + workspace: source, + projectId: "project-source", + threadId: "thread-v1", + }); + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`DELETE FROM projection_threads WHERE thread_id = 'thread-v1'`; + }).pipe(withDatabase(sourceDatabase)); + + const exported = yield* exportThread({ source, threadId: "thread-v1", output: archivePath }); + assert.equal(exported.title, "thread-v1"); + const archive = decodeArchive(yield* fs.readFileString(archivePath)); + assert.equal(archive.thread.sourceProjectId, "project-source"); + }), + ); + + it.effect("exports every terminal history the server attributes to the thread", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-terminals-" }); + const source = path.join(root, "source"); + const archivePath = path.join(root, "thread.json"); + yield* createFixtureDatabase({ + workspace: source, + projectId: "project-source", + threadId: "thread-v1", + }); + const threadPart = `terminal_${Encoding.encodeBase64Url("thread-v1")}`; + const owned = [ + `${threadPart}.log`, + `${threadPart}_${Encoding.encodeBase64Url("terminal-1")}.log`, + "thread-v1.log", + ]; + const foreign = [ + `terminal_${Encoding.encodeBase64Url("thread-v10")}.log`, + `terminal_${Encoding.encodeBase64Url("thread-v10")}_${Encoding.encodeBase64Url("t")}.log`, + "thread-v10.log", + ]; + const logsDir = path.join(source, ".t3", "userdata", "logs", "terminals"); + yield* fs.makeDirectory(logsDir, { recursive: true }); + for (const name of [...owned, ...foreign]) { + yield* fs.writeFileString(path.join(logsDir, name), `${name}\n`); + } + + const exported = yield* exportThread({ + source, + threadId: "thread-v1", + output: archivePath, + includeTerminalLogs: true, + }); + assert.equal(exported.terminalLogCount, owned.length); + const archive = decodeArchive(yield* fs.readFileString(archivePath)); + assert.deepStrictEqual( + archive.terminalLogs.map((log) => log.fileName).sort(), + [...owned].sort(), + ); + }), + ); + + it.effect("refuses imports that would collide or touch shared state", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-refusals-" }); + const source = path.join(root, "source"); + const archivePath = path.join(root, "thread.json"); + const sharedHome = path.join(root, "shared-home"); + yield* createAttachmentSource(source); + yield* exportThread({ source, threadId: "thread-v1", output: archivePath }); + + const missing = yield* failImport( + { archive: archivePath, destination: path.join(root, "nowhere") }, + { sharedHome }, + ); + assert.match(missing.detail, /^No T3 userdata database found at /); + + const collision = yield* failImport( + { archive: archivePath, destination: source }, + { sharedHome }, + ); + assert.equal(collision.detail, "Thread 'thread-v1' already exists in the destination."); + + yield* createFixtureDatabase({ + workspace: sharedHome, + projectId: "project-shared", + }); + const shared = yield* failImport( + { archive: archivePath, destination: path.join(sharedHome, ".t3") }, + { sharedHome: path.join(sharedHome, ".t3") }, + ); + assert.equal( + shared.detail, + "Refusing to mutate the shared ~/.t3/userdata database. Choose an isolated destination or pass --dangerous-allow-t3-directory.", + ); + const allowed = yield* importThread( + { + archive: archivePath, + destination: path.join(sharedHome, ".t3"), + dangerousAllowT3Directory: true, + }, + { sharedHome: path.join(sharedHome, ".t3") }, + ); + assert.equal(allowed.targetProjectId, "project-shared"); + }), + ); + + it.effect("rejects archives whose events stray from the thread or its contract", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-events-" }); + const source = path.join(root, "source"); + const destination = path.join(root, "destination"); + const archivePath = path.join(root, "thread.json"); + const sharedHome = path.join(root, "shared-home"); + yield* createAttachmentSource(source); + const destinationDatabase = yield* createFixtureDatabase({ + workspace: destination, + projectId: "project-target", + }); + yield* exportThread({ source, threadId: "thread-v1", output: archivePath }); + const archive = decodeArchive(yield* fs.readFileString(archivePath)); + const [created] = archive.events; + + const foreignPath = path.join(root, "foreign.json"); + yield* fs.writeFileString( + foreignPath, + encodeArchive({ + ...archive, + events: [created!, { ...created!, eventId: "event-other", streamId: "thread-other" }], + }), + ); + const foreign = yield* failImport({ archive: foreignPath, destination }, { sharedHome }); + assert.equal(foreign.operation, "read archive"); + assert.equal(foreign.detail, "Event 'event-other' does not belong to thread 'thread-v1'."); + + const skewedPath = path.join(root, "skewed.json"); + yield* fs.writeFileString( + skewedPath, + encodeArchive({ + ...archive, + events: [{ ...created!, payloadJson: encodeUnknownJson({ threadId: "thread-v1" }) }], + }), + ); + const skewed = yield* failImport({ archive: skewedPath, destination }, { sharedHome }); + assert.equal( + skewed.detail, + "Event 'event-v1' (thread.created) does not match this checkout's orchestration contract. Run the import from the destination's checkout.", + ); + assert.deepStrictEqual(yield* readThreadEventIds(destinationDatabase, "thread-v1"), []); + assert.isFalse(yield* fs.exists(path.join(destination, ".t3", "userdata", "attachments"))); + }), + ); + + it.effect("rejects a corrupt archive file before writing anything to the destination", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-checksum-" }); + const source = path.join(root, "source"); + const destination = path.join(root, "destination"); + const archivePath = path.join(root, "thread.json"); + yield* createAttachmentSource(source); + const destinationDatabase = yield* createFixtureDatabase({ + workspace: destination, + projectId: "project-target", + }); + const terminalLogName = yield* writeTerminalLog( + path.join(source, ".t3", "userdata"), + "thread-v1", + ); + yield* exportThread({ + source, + threadId: "thread-v1", + output: archivePath, + includeTerminalLogs: true, + }); + const archive = decodeArchive(yield* fs.readFileString(archivePath)); + const tampered = { + ...archive, + terminalLogs: archive.terminalLogs.map((log) => ({ ...log, sha256: "0".repeat(64) })), + }; + yield* fs.writeFileString(archivePath, encodeArchive(tampered)); + + const error = yield* failImport( + { archive: archivePath, destination }, + { sharedHome: path.join(root, "shared-home") }, + ); + assert.equal(error.detail, `Terminal log '${terminalLogName}' failed its checksum.`); + assert.isFalse(yield* fs.exists(path.join(destination, ".t3", "userdata", "attachments"))); + assert.isFalse( + yield* fs.exists(path.join(destination, ".t3", "userdata", "logs", "terminals")), + ); + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const events = yield* sql`SELECT event_id FROM orchestration_events`; + assert.deepStrictEqual(events, []); + }).pipe(withDatabase(destinationDatabase, true)); + }), + ); + + it.effect("rolls back every event and removes the files it wrote when one insert fails", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "thread-transfer-rollback-" }); + const source = path.join(root, "source"); + const destination = path.join(root, "destination"); + const archivePath = path.join(root, "thread.json"); + const { attachmentName } = yield* createAttachmentSource(source); + const destinationDatabase = yield* createFixtureDatabase({ + workspace: destination, + projectId: "project-target", + }); + // The second archived event collides with an event id the destination + // already holds, so the first insert succeeds and the second fails. + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, + command_id, causation_event_id, correlation_id, actor_kind, payload_json, metadata_json + ) VALUES ( + 'event-archived', 'thread', 'thread-other', 0, 'thread.created', + '2026-08-20T12:00:00.000Z', NULL, NULL, NULL, 'client', '{}', '{}' + )`; + }).pipe(withDatabase(destinationDatabase)); + yield* exportThread({ source, threadId: "thread-v1", output: archivePath }); + const archive = decodeArchive(yield* fs.readFileString(archivePath)); + const [created] = archive.events; + yield* fs.writeFileString( + archivePath, + encodeArchive({ + ...archive, + events: [ + created!, + { + ...created!, + eventId: "event-archived", + streamVersion: 1, + eventType: "thread.archived", + payloadJson: encodeUnknownJson({ + threadId: "thread-v1", + archivedAt: "2026-08-20T13:00:00.000Z", + updatedAt: "2026-08-20T13:00:00.000Z", + }), + }, + ], + }), + ); + + const error = yield* failImport( + { archive: archivePath, destination }, + { sharedHome: path.join(root, "shared-home") }, + ); + assert.equal(error.operation, "import thread"); + assert.isFalse( + yield* fs.exists(path.join(destination, ".t3", "userdata", "attachments", attachmentName)), + ); + assert.deepStrictEqual(yield* readThreadEventIds(destinationDatabase, "thread-v1"), []); + assert.deepStrictEqual(yield* readThreadEventIds(destinationDatabase, "thread-other"), [ + "event-archived", + ]); + }), + ); +}); diff --git a/apps/server/scripts/thread-transfer.ts b/apps/server/scripts/thread-transfer.ts index 45c85f5e1694..9f819b3d7360 100644 --- a/apps/server/scripts/thread-transfer.ts +++ b/apps/server/scripts/thread-transfer.ts @@ -1,15 +1,92 @@ +// @effect-diagnostics nodeBuiltinImport:off - node modules provide hashing and the shared-home guard. +import * as NodeCrypto from "node:crypto"; +import * as NodeOS from "node:os"; + +import { OrchestrationEvent } from "@t3tools/contracts"; +import { fromJsonStringPretty } from "@t3tools/shared/schemaJson"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Encoding from "effect/Encoding"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import { Flag } from "effect/unstable/cli"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { isSqlError } from "effect/unstable/sql/SqlError"; +import { + parseAttachmentIdFromRelativePath, + parseThreadSegmentFromAttachmentId, + toSafeThreadAttachmentSegment, +} from "../src/attachmentStore.ts"; import * as NodeSqliteClient from "../src/persistence/NodeSqliteClient.ts"; +import { ensureDevDbNotInUse } from "./migrate-dev-db.ts"; +const ThreadArchiveEvent = Schema.Struct({ + eventId: Schema.String, + aggregateKind: Schema.String, + streamId: Schema.String, + streamVersion: Schema.Number, + eventType: Schema.String, + occurredAt: Schema.String, + commandId: Schema.NullOr(Schema.String), + causationEventId: Schema.NullOr(Schema.String), + correlationId: Schema.NullOr(Schema.String), + actorKind: Schema.String, + payloadJson: Schema.String, + metadataJson: Schema.String, +}); +type ThreadArchiveEvent = typeof ThreadArchiveEvent.Type; +const ThreadArchiveFile = Schema.Struct({ + fileName: Schema.String, + sha256: Schema.String, + dataBase64: Schema.String, +}); + +/** + * One thread's canonical events plus the files that travel with it. Every + * event belongs to the archived thread. + */ +export const ThreadArchive = Schema.Struct({ + format: Schema.Literal("t3-thread-export"), + version: Schema.Literal(1), + exportedAt: Schema.String, + thread: Schema.Struct({ + id: Schema.String, + title: Schema.String, + sourceProjectId: Schema.String, + sourceWorkspaceRoot: Schema.String, + }), + events: Schema.Array(ThreadArchiveEvent), + attachments: Schema.Array(ThreadArchiveFile), + terminalLogs: Schema.Array(ThreadArchiveFile).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + ), +}); +export type ThreadArchive = typeof ThreadArchive.Type; export const ThreadTransferState = Schema.Literals(["userdata", "dev"]); export type ThreadTransferState = typeof ThreadTransferState.Type; +/** CLI flags shared by the list, export, and import commands. */ +export const threadTransferFlags = { + directory: (name: "source" | "destination") => + Flag.string(name).pipe( + Flag.withDescription("Workspace root, T3 base directory, or direct state directory."), + ), + state: Flag.choice("state", ThreadTransferState.literals).pipe( + Flag.withDefault("userdata"), + Flag.withDescription("State directory below the T3 base directory; defaults to userdata."), + ), +}; + +const decodeThreadArchive = Schema.decodeEffect(Schema.fromJsonString(ThreadArchive)); +const encodeThreadArchive = Schema.encodeEffect(fromJsonStringPretty(ThreadArchive)); +const decodeUnknownJson = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown)); +const encodeUnknownJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); +const decodeOrchestrationEvent = Schema.decodeUnknownEffect(OrchestrationEvent); +const hasProjectId = Schema.is(Schema.Struct({ projectId: Schema.String })); +const hasWorktreePath = Schema.is(Schema.Struct({ worktreePath: Schema.String })); + export class ThreadTransferError extends Schema.TaggedErrorClass()( "ThreadTransferError", { @@ -23,6 +100,25 @@ export class ThreadTransferError extends Schema.TaggedErrorClass>; + +interface RawEventRow extends RawSqliteRow { + readonly eventId: string; + readonly aggregateKind: string; + readonly streamId: string; + readonly streamVersion: number; + readonly eventType: string; + readonly occurredAt: string; + readonly commandId: string | null; + readonly causationEventId: string | null; + readonly correlationId: string | null; + readonly actorKind: string; + readonly payloadJson: string; + readonly metadataJson: string; +} + +interface ProjectRow extends RawSqliteRow { readonly projectId: string; readonly title: string; readonly workspaceRoot: string; @@ -68,7 +186,7 @@ interface ProjectRow { readonly deletedAt: string | null; } -interface ListedThreadRow { +interface ListedThreadRow extends RawSqliteRow { readonly threadId: string; readonly projectId: string; readonly title: string; @@ -123,6 +241,131 @@ const resolveStateLocation = Effect.fn("resolveThreadTransferStateLocation")(fun ); }); +const tableExists = Effect.fn("threadTransferTableExists")(function* (table: string) { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM sqlite_master + WHERE type = 'table' AND name = ${table}`; + return Number(rows[0]?.count ?? 0) > 0; +}); + +const readProjectIdFromPayload = Effect.fn("readThreadTransferProjectIdFromPayload")(function* ( + payloadJson: string, +) { + const payload = yield* decodeUnknownJson(payloadJson).pipe( + Effect.mapError((cause) => transferError("read thread", "Invalid event payload JSON.", cause)), + ); + return hasProjectId(payload) ? payload.projectId : null; +}); + +interface ImportRewrite { + readonly sourceProjectId: string; + readonly targetProjectId: string; + /** Worktree paths cleared because they do not exist on the destination. */ + readonly droppedWorktreePaths: Set; +} + +/** + * Rewrites the destination-specific fields of an event payload: the project + * id (so the thread lands in the target project) and any worktree path that + * does not exist on the destination. A worktree belongs to the source + * checkout, so a missing path is cleared and the thread falls back to the + * project workspace root; existing paths are kept so same-machine moves stay + * put. Other payloads pass through untouched. + */ +const rewriteEventPayload = Effect.fn("rewriteThreadTransferEventPayload")(function* ( + payloadJson: string, + rewrite: ImportRewrite, +) { + const fs = yield* FileSystem.FileSystem; + const payload = yield* decodeUnknownJson(payloadJson).pipe( + Effect.mapError((cause) => + transferError("import thread", "Invalid event payload JSON.", cause), + ), + ); + let next: Record | null = null; + if (hasProjectId(payload) && payload.projectId === rewrite.sourceProjectId) { + next = { ...payload, projectId: rewrite.targetProjectId }; + } + if (hasWorktreePath(payload)) { + const exists = yield* fs.exists(payload.worktreePath).pipe(Effect.orElseSucceed(() => false)); + if (!exists) { + rewrite.droppedWorktreePaths.add(payload.worktreePath); + next = { ...(next ?? payload), worktreePath: null }; + } + } + if (next === null) return payloadJson; + return yield* encodeUnknownJson(next).pipe( + Effect.mapError((cause) => + transferError("import thread", "Could not encode event JSON.", cause), + ), + ); +}); + +function isAttachmentForThread(fileName: string, threadId: string): boolean { + const segment = toSafeThreadAttachmentSegment(threadId); + if (segment === null) return false; + const attachmentId = parseAttachmentIdFromRelativePath(fileName); + return attachmentId !== null && parseThreadSegmentFromAttachmentId(attachmentId) === segment; +} + +/** Mirrors the ownership rule of `deleteAllHistoryForThread` in `src/terminal/Manager.ts`. */ +function isTerminalLogForThread(fileName: string, threadId: string): boolean { + const safeThreadId = `terminal_${Encoding.encodeBase64Url(threadId)}`; + const legacyThreadId = threadId.replace(/[^a-zA-Z0-9._-]/g, "_"); + return ( + fileName === `${safeThreadId}.log` || + fileName === `${legacyThreadId}.log` || + fileName.startsWith(`${safeThreadId}_`) + ); +} + +/** A per-thread file family that lives below the state directory and travels with the archive. */ +interface ThreadFileKind { + readonly label: "Attachment" | "Terminal log"; + readonly directory: ReadonlyArray; + readonly belongsToThread: (fileName: string, threadId: string) => boolean; +} + +const ATTACHMENTS: ThreadFileKind = { + label: "Attachment", + directory: ["attachments"], + belongsToThread: isAttachmentForThread, +}; + +const TERMINAL_LOGS: ThreadFileKind = { + label: "Terminal log", + directory: ["logs", "terminals"], + belongsToThread: isTerminalLogForThread, +}; + +const sha256Hex = (data: Uint8Array) => NodeCrypto.createHash("sha256").update(data).digest("hex"); + +const loadThreadFiles = Effect.fn("loadThreadTransferFiles")(function* ( + location: StateLocation, + kind: ThreadFileKind, + threadId: string, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = path.join(location.stateDir, ...kind.directory); + if (!(yield* fs.exists(directory))) return []; + const names = (yield* fs.readDirectory(directory)) + .filter((name) => kind.belongsToThread(name, threadId)) + .sort(); + return yield* Effect.forEach(names, (fileName) => + Effect.gen(function* () { + const data = yield* fs.readFile(path.join(directory, fileName)); + return { + fileName, + sha256: sha256Hex(data), + dataBase64: Buffer.from(data).toString("base64"), + }; + }), + ); +}); + export const listThreads = Effect.fn("listThreads")(function* (input: ListThreadsInput) { const location = yield* resolveStateLocation(input.source, input.state); return yield* Effect.gen(function* () { @@ -170,3 +413,453 @@ export const listThreads = Effect.fn("listThreads")(function* (input: ListThread return listing; }).pipe(withThreadDatabase(location, "list threads", "read")); }); + +/** Every archived event must belong to the archived thread and be unique in its stream. */ +const validateArchiveEvents = (archive: ThreadArchive): ThreadTransferError | null => { + if (archive.events.length === 0) { + return transferError("read archive", `Thread '${archive.thread.id}' has no events.`); + } + const eventIds = new Set(); + const streamVersions = new Set(); + for (const event of archive.events) { + if (event.aggregateKind !== "thread" || event.streamId !== archive.thread.id) { + return transferError( + "read archive", + `Event '${event.eventId}' does not belong to thread '${archive.thread.id}'.`, + ); + } + if (eventIds.has(event.eventId) || streamVersions.has(event.streamVersion)) { + return transferError( + "read archive", + `Event '${event.eventId}' repeats an event id or stream version.`, + ); + } + eventIds.add(event.eventId); + streamVersions.add(event.streamVersion); + } + return null; +}; + +const loadArchive = Effect.fn("loadThreadTransferArchive")(function* (filePath: string) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const resolved = path.resolve(filePath); + const contents = yield* fs + .readFileString(resolved) + .pipe( + Effect.mapError((cause) => + transferError("read archive", `Could not read '${resolved}'.`, cause), + ), + ); + const archive = yield* decodeThreadArchive(contents).pipe( + Effect.mapError((cause) => + transferError("read archive", `'${resolved}' is not a T3 thread archive.`, cause), + ), + ); + const invalid = validateArchiveEvents(archive); + return invalid === null ? archive : yield* invalid; +}); + +/** Reads one thread's events and files from the source into an archive. */ +const buildThreadArchive = Effect.fn("buildThreadTransferArchive")(function* ( + location: StateLocation, + input: ExportThreadInput, +) { + const sql = yield* SqlClient.SqlClient; + if (!(yield* tableExists("orchestration_events"))) { + return yield* transferError("export thread", "The source has no orchestration event log."); + } + const rawEvents = yield* sql` + SELECT + event_id AS "eventId", + aggregate_kind AS "aggregateKind", + stream_id AS "streamId", + stream_version AS "streamVersion", + event_type AS "eventType", + occurred_at AS "occurredAt", + command_id AS "commandId", + causation_event_id AS "causationEventId", + correlation_id AS "correlationId", + actor_kind AS "actorKind", + payload_json AS "payloadJson", + metadata_json AS "metadataJson" + FROM orchestration_events + WHERE aggregate_kind = 'thread' AND stream_id = ${input.threadId} + ORDER BY sequence`; + if (rawEvents.length === 0) { + return yield* transferError( + "export thread", + `Thread '${input.threadId}' has no canonical events.`, + ); + } + const threadRows = yield* sql<{ readonly projectId: string; readonly title: string }>` + SELECT project_id AS "projectId", title + FROM projection_threads + WHERE thread_id = ${input.threadId}`; + const createdEvent = rawEvents.find((event) => event.eventType === "thread.created"); + const eventProjectId = + createdEvent === undefined ? null : yield* readProjectIdFromPayload(createdEvent.payloadJson); + const sourceProjectId = threadRows[0]?.projectId ?? eventProjectId; + if (sourceProjectId === null) { + return yield* transferError( + "export thread", + `Could not resolve the project for thread '${input.threadId}'.`, + ); + } + const projectRows = yield* sql` + SELECT project_id AS "projectId", title, workspace_root AS "workspaceRoot" + FROM projection_projects + WHERE project_id = ${sourceProjectId}`; + const project = projectRows[0]; + if (project === undefined) { + return yield* transferError( + "export thread", + `Project '${sourceProjectId}' is missing from the source database.`, + ); + } + const exportedAt = DateTime.formatIso(yield* DateTime.now); + return { + format: "t3-thread-export", + version: 1, + exportedAt, + thread: { + id: input.threadId, + title: threadRows[0]?.title ?? input.threadId, + sourceProjectId, + sourceWorkspaceRoot: project.workspaceRoot, + }, + events: rawEvents.map((event) => ({ + eventId: event.eventId, + aggregateKind: event.aggregateKind, + streamId: event.streamId, + streamVersion: Number(event.streamVersion), + eventType: event.eventType, + occurredAt: event.occurredAt, + commandId: event.commandId, + causationEventId: event.causationEventId, + correlationId: event.correlationId, + actorKind: event.actorKind, + payloadJson: event.payloadJson, + metadataJson: event.metadataJson, + })), + attachments: yield* loadThreadFiles(location, ATTACHMENTS, input.threadId), + terminalLogs: + input.includeTerminalLogs === true + ? yield* loadThreadFiles(location, TERMINAL_LOGS, input.threadId) + : [], + } satisfies ThreadArchive; +}); + +export const exportThread = Effect.fn("exportThread")(function* (input: ExportThreadInput) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const location = yield* resolveStateLocation(input.source, input.state); + const output = path.resolve(input.output); + if (yield* fs.exists(output)) { + return yield* transferError("export thread", `Output '${output}' already exists.`); + } + + const archive = yield* buildThreadArchive(location, input).pipe( + withThreadDatabase(location, "export thread", "read"), + ); + + const encoded = yield* encodeThreadArchive(archive).pipe( + Effect.mapError((cause) => transferError("export thread", "Could not encode archive.", cause)), + ); + yield* fs.makeDirectory(path.dirname(output), { recursive: true }); + yield* fs + .writeFileString(output, encoded) + .pipe( + Effect.mapError((cause) => + transferError("export thread", `Could not write '${output}'.`, cause), + ), + ); + yield* fs.chmod(output, 0o600); + return { + output, + threadId: archive.thread.id, + title: archive.thread.title, + eventCount: archive.events.length, + attachmentCount: archive.attachments.length, + terminalLogCount: archive.terminalLogs.length, + } as const; +}); + +const resolveTargetProject = Effect.fn("resolveThreadTransferTargetProject")(function* ( + workspaceRoot: string | null, + explicitProjectId: string | undefined, +) { + const sql = yield* SqlClient.SqlClient; + const path = yield* Path.Path; + const projects = yield* sql` + SELECT project_id AS "projectId", title, workspace_root AS "workspaceRoot" + FROM projection_projects + WHERE deleted_at IS NULL + ORDER BY updated_at DESC`; + if (explicitProjectId !== undefined) { + const project = projects.find((candidate) => candidate.projectId === explicitProjectId); + if (project !== undefined) return project; + return yield* transferError( + "import thread", + `Target project '${explicitProjectId}' does not exist in the destination.`, + ); + } + if (workspaceRoot !== null) { + const normalizedRoot = path.resolve(workspaceRoot); + const project = projects.find( + (candidate) => path.resolve(candidate.workspaceRoot) === normalizedRoot, + ); + if (project !== undefined) return project; + } + if (projects.length === 1) return projects[0]!; + return yield* transferError( + "import thread", + "Could not infer the target project. Pass --target-project-id.", + ); +}); + +/** Validates one archive file family and returns the files that still need writing. */ +const stageArchiveFiles = Effect.fn("stageThreadTransferArchiveFiles")(function* ( + location: StateLocation, + kind: ThreadFileKind, + files: ThreadArchive["attachments"], + threadId: string, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const destinationDir = path.join(location.stateDir, ...kind.directory); + const pending: Array<{ readonly path: string; readonly data: Uint8Array }> = []; + for (const file of files) { + if (path.basename(file.fileName) !== file.fileName) { + return yield* transferError( + "import thread", + `${kind.label} name '${file.fileName}' is not safe.`, + ); + } + if (!kind.belongsToThread(file.fileName, threadId)) { + return yield* transferError( + "import thread", + `${kind.label} '${file.fileName}' does not belong to this thread.`, + ); + } + const data = Uint8Array.from(Buffer.from(file.dataBase64, "base64")); + const hash = sha256Hex(data); + if (hash !== file.sha256) { + return yield* transferError( + "import thread", + `${kind.label} '${file.fileName}' failed its checksum.`, + ); + } + const destination = path.join(destinationDir, file.fileName); + if (yield* fs.exists(destination)) { + const existing = yield* fs.readFile(destination); + const existingHash = sha256Hex(existing); + if (existingHash !== hash) { + return yield* transferError( + "import thread", + `${kind.label} '${file.fileName}' already exists with different contents.`, + ); + } + continue; + } + pending.push({ path: destination, data }); + } + return pending; +}); + +/** The archive's events with their payloads rewritten for the destination. */ +const prepareEvents = Effect.fn("prepareThreadTransferEvents")(function* ( + archive: ThreadArchive, + rewrite: ImportRewrite, +) { + return yield* Effect.forEach(archive.events, (event) => + rewriteEventPayload(event.payloadJson, rewrite).pipe( + Effect.map((payloadJson): ThreadArchiveEvent => ({ ...event, payloadJson })), + ), + ); +}); + +/** + * The destination server decodes every event against its orchestration + * contract at startup and refuses to start on one it cannot read, so an + * archive from a differently-versioned source is rejected here, before any + * write. + */ +const ensureEventsDecode = Effect.fn("ensureThreadTransferEventsDecode")(function* ( + events: ReadonlyArray, +) { + for (const event of events) { + yield* Effect.all([ + decodeUnknownJson(event.payloadJson), + decodeUnknownJson(event.metadataJson), + ]).pipe( + Effect.flatMap(([payload, metadata]) => + decodeOrchestrationEvent({ + sequence: 0, + eventId: event.eventId, + aggregateKind: event.aggregateKind, + aggregateId: event.streamId, + type: event.eventType, + occurredAt: event.occurredAt, + commandId: event.commandId, + causationEventId: event.causationEventId, + correlationId: event.correlationId, + payload, + metadata, + }), + ), + Effect.mapError((cause) => + transferError( + "import thread", + `Event '${event.eventId}' (${event.eventType}) does not match this checkout's orchestration contract. Run the import from the destination's checkout.`, + cause, + ), + ), + ); + } +}); + +const insertEvents = Effect.fn("insertThreadTransferEvents")(function* ( + events: ReadonlyArray, +) { + const sql = yield* SqlClient.SqlClient; + const columns = [ + "event_id", + "aggregate_kind", + "stream_id", + "stream_version", + "event_type", + "occurred_at", + "command_id", + "causation_event_id", + "correlation_id", + "actor_kind", + "payload_json", + "metadata_json", + ]; + const insertSql = `INSERT INTO orchestration_events (${columns.join(", ")}) VALUES (${columns.map(() => "?").join(", ")})`; + for (const event of events) { + const params: ReadonlyArray = [ + event.eventId, + event.aggregateKind, + event.streamId, + event.streamVersion, + event.eventType, + event.occurredAt, + event.commandId, + event.causationEventId, + event.correlationId, + event.actorKind, + event.payloadJson, + event.metadataJson, + ]; + yield* sql.unsafe(insertSql, params).unprepared; + } +}); + +export const importThread = Effect.fn("importThread")(function* ( + input: ImportThreadInput, + options: ImportThreadOptions = {}, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const location = yield* resolveStateLocation(input.destination, input.state); + const sharedHome = path.resolve(options.sharedHome ?? path.join(NodeOS.homedir(), ".t3")); + const sharedDatabase = path.join(sharedHome, "userdata", "state.sqlite"); + const [canonicalDatabase, canonicalSharedDatabase] = yield* Effect.all([ + fs.realPath(location.databasePath).pipe(Effect.orElseSucceed(() => location.databasePath)), + fs.realPath(sharedDatabase).pipe(Effect.orElseSucceed(() => sharedDatabase)), + ]); + if (canonicalDatabase === canonicalSharedDatabase && input.dangerousAllowT3Directory !== true) { + return yield* transferError( + "import thread", + "Refusing to mutate the shared ~/.t3/userdata database. Choose an isolated destination or pass --dangerous-allow-t3-directory.", + ); + } + yield* ensureDevDbNotInUse(location.databasePath).pipe( + Effect.mapError((cause) => transferError("import thread", cause.message, cause)), + ); + const archive = yield* loadArchive(input.archive); + + return yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.unsafe("PRAGMA busy_timeout = 5000").unprepared; + yield* sql.unsafe("PRAGMA foreign_keys = ON").unprepared; + const targetProject = yield* resolveTargetProject( + location.workspaceRoot, + input.targetProjectId, + ); + const existingEvents = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM orchestration_events + WHERE aggregate_kind = 'thread' AND stream_id = ${archive.thread.id}`; + const existingThreads = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM projection_threads + WHERE thread_id = ${archive.thread.id}`; + if (Number(existingEvents[0]?.count ?? 0) > 0 || Number(existingThreads[0]?.count ?? 0) > 0) { + return yield* transferError( + "import thread", + `Thread '${archive.thread.id}' already exists in the destination.`, + ); + } + + const pendingFiles = [ + ...(yield* stageArchiveFiles(location, ATTACHMENTS, archive.attachments, archive.thread.id)), + ...(yield* stageArchiveFiles( + location, + TERMINAL_LOGS, + archive.terminalLogs, + archive.thread.id, + )), + ]; + const rewrite: ImportRewrite = { + sourceProjectId: archive.thread.sourceProjectId, + targetProjectId: targetProject.projectId, + droppedWorktreePaths: new Set(), + }; + const events = yield* prepareEvents(archive, rewrite); + yield* ensureEventsDecode(events); + + const timestamp = DateTime.formatIso(yield* DateTime.now).replaceAll(":", "-"); + const backupPath = `${location.databasePath}.backup-thread-import-${timestamp}`; + yield* sql`VACUUM INTO ${backupPath}`; + yield* fs.chmod(backupPath, 0o600); + + const writtenFiles: Array = []; + yield* Effect.gen(function* () { + for (const file of pendingFiles) { + yield* fs.makeDirectory(path.dirname(file.path), { recursive: true }); + yield* fs.writeFile(file.path, file.data); + writtenFiles.push(file.path); + yield* fs.chmod(file.path, 0o600); + } + // Only events are written. The destination server derives the thread's + // read model from them on its next start, when the projectors replay + // every event above their recorded sequence. Copying projection rows as + // well would make that replay append onto already-complete rows. + yield* sql.withTransaction(insertEvents(events)); + }).pipe( + Effect.onError(() => + Effect.forEach( + writtenFiles, + (filePath) => fs.remove(filePath).pipe(Effect.orElseSucceed(() => undefined)), + { discard: true }, + ), + ), + ); + + return { + database: location.databasePath, + backup: backupPath, + threadId: archive.thread.id, + title: archive.thread.title, + targetProjectId: targetProject.projectId, + targetProjectTitle: targetProject.title, + eventCount: events.length, + attachmentCount: archive.attachments.length, + terminalLogCount: archive.terminalLogs.length, + droppedWorktreePaths: [...rewrite.droppedWorktreePaths], + } as const; + }).pipe(withThreadDatabase(location, "import thread", "update")); +}); diff --git a/docs/internals/scripts.md b/docs/internals/scripts.md index baf4b5fa6b78..67b8096776ad 100644 --- a/docs/internals/scripts.md +++ b/docs/internals/scripts.md @@ -64,9 +64,30 @@ authenticated. - `node apps/server/scripts/t3-sqlite-state.ts --base-dir ...`: Inspects or seeds an isolated T3 SQLite database; writes create a private backup first. - `vp run thread:list --source `: Lists the projects (id, workspace root, title) and - live threads (id, project id, title) of an existing T3 state database. The source can be a workspace containing `.t3`, the T3 base - directory, or a direct state directory containing `state.sqlite`. State defaults to `userdata`; - pass `--state dev` for a main-checkout dev database. Pass `--json` for structured output. + live threads (id, project id, title) of an existing T3 state database, so you can pick a + `--thread-id` to export and a `--target-project-id` to import into. It accepts the same directory + forms and `--state dev` selection as the transfer commands. Pass `--json` for structured output. +- `vp run thread:export --source --thread-id --output `: Exports + one thread, including its image attachments. The source can be a workspace containing `.t3`, the + T3 base directory, or a direct state directory containing `state.sqlite`. + State defaults to `userdata`; pass `--state dev` for a main-checkout dev database. Pass + `--include-terminal-logs` to also export persisted terminal history. Terminal logs may contain + credentials or other sensitive output, so they are excluded by default. +- `vp run thread:import --archive --destination `: Imports the thread + into an isolated T3 project directory, remaps it to the destination project, and backs up the + destination database first. Only the thread's events and files are written; the destination + server rebuilds the read model from them on its next start. Events are checked against the + running checkout's orchestration contract before anything is written, so run the import from the + destination's checkout. The destination accepts the same + directory forms and `--state dev` selection as export. Stop the destination server before + importing. The live `~/.t3/userdata` database is refused unless you pass + `--dangerous-allow-t3-directory`, which is how a thread moves from a dev checkout back into the + real install. Pass `--target-project-id ` when the destination contains more than one project + and its workspace path does not identify the target. Worktree paths that do not exist on the + destination are cleared, so the thread falls back to the project workspace. Not transferred: the + provider's own session (the imported thread starts a fresh provider session on its next turn) + and checkpoint git refs (revert and per-turn diffs need the source repository, so they are + unavailable for the imported thread). ## Desktop artifacts diff --git a/package.json b/package.json index 454ca8af9d54..a94f7d19cac1 100644 --- a/package.json +++ b/package.json @@ -12,6 +12,8 @@ "dev:desktop": "node scripts/dev-runner.ts dev:desktop", "migrate-dev-db": "node apps/server/scripts/migrate-dev-db.ts", "thread:list": "node apps/server/scripts/list-threads.ts", + "thread:export": "node apps/server/scripts/export-thread.ts", + "thread:import": "node apps/server/scripts/import-thread.ts", "start": "vp run --filter t3 start", "start:desktop": "vp run --filter @t3tools/desktop start", "start:marketing": "vp run --filter @t3tools/marketing preview",