feat(FN-3841): add worktree database hydration to executor

Implements worktree database hydration for the executor (`worktree-db-hydrate.ts`), enabling the task executor to restore its database state from a worktree on resume, with integration in `executor.ts` and comprehensive test coverage across the new module and worktree scenarios. Documentation and a

Fusion-Task-Id: FN-3841

packages/engine/src/worktree-db-hydrate.ts

@@ -0,0 +1,172 @@
+import { existsSync } from "node:fs";
+import { join } from "node:path";
+import { DatabaseSync, TaskStore } from "@fusion/core";
+
+export interface HydrateWorktreeDbParams {
+  rootDir: string;
+  worktreePath: string;
+  taskId: string;
+  store: Pick<TaskStore, "getTask">;
+  logger: { warn: (message: string) => void };
+}
+
+export interface HydrateWorktreeDbResult {
+  tasksCopied: number;
+  documentsCopied: number;
+  degraded: boolean;
+  reason?: string;
+}
+
+const MAX_DEPTH = 5;
+const MAX_IDS = 50;
+
+function getDbPath(projectDir: string): string {
+  return join(projectDir, ".fusion", "fusion.db");
+}
+
+function getColumns(db: DatabaseSync, table: "tasks" | "task_documents"): string[] {
+  const rows = db.prepare(`PRAGMA table_info('${table}')`).all() as Array<{ name?: string }>;
+  return rows.map((row) => row.name).filter((name): name is string => typeof name === "string" && name.length > 0);
+}
+
+function intersectColumns(src: string[], dst: string[]) {
+  const dstSet = new Set(dst);
+  const shared = src.filter((column) => dstSet.has(column));
+  const dropped = src.filter((column) => !dstSet.has(column));
+  return { shared, dropped };
+}
+
+async function resolveDependencyIds(taskId: string, store: Pick<TaskStore, "getTask">): Promise<string[]> {
+  const visited = new Set<string>();
+  const queue: Array<{ id: string; depth: number }> = [{ id: taskId, depth: 0 }];
+
+  while (queue.length > 0 && visited.size < MAX_IDS) {
+    const current = queue.shift();
+    if (!current || visited.has(current.id)) continue;
+    visited.add(current.id);
+    if (current.depth >= MAX_DEPTH) continue;
+
+    const task = await store.getTask(current.id);
+    const deps = Array.isArray(task?.dependencies) ? task.dependencies : [];
+    for (const depId of deps) {
+      if (!visited.has(depId) && queue.length + visited.size < MAX_IDS) {
+        queue.push({ id: depId, depth: current.depth + 1 });
+      }
+    }
+  }
+
+  return Array.from(visited);
+}
+
+function ensureWorktreeSchema(worktreePath: string): void {
+  const schemaStore = new TaskStore(worktreePath);
+  schemaStore.close();
+}
+
+export async function hydrateWorktreeDb({
+  rootDir,
+  worktreePath,
+  taskId,
+  store,
+  logger,
+}: HydrateWorktreeDbParams): Promise<HydrateWorktreeDbResult> {
+  if (rootDir === worktreePath) {
+    return { tasksCopied: 0, documentsCopied: 0, degraded: false, reason: "root_worktree" };
+  }
+
+  let srcDb: DatabaseSync | undefined;
+  let dstDb: DatabaseSync | undefined;
+
+  try {
+    const ids = await resolveDependencyIds(taskId, store);
+    if (ids.length === 0) {
+      return { tasksCopied: 0, documentsCopied: 0, degraded: false, reason: "no_ids" };
+    }
+
+    const srcDbPath = getDbPath(rootDir);
+    const dstDbPath = getDbPath(worktreePath);
+
+    if (!existsSync(srcDbPath)) {
+      return { tasksCopied: 0, documentsCopied: 0, degraded: true, reason: "source_db_missing" };
+    }
+
+    if (!existsSync(dstDbPath)) {
+      ensureWorktreeSchema(worktreePath);
+    }
+
+    srcDb = new DatabaseSync(srcDbPath);
+    dstDb = new DatabaseSync(dstDbPath);
+
+    dstDb.exec("PRAGMA journal_mode = WAL");
+
+    const srcTaskCols = getColumns(srcDb, "tasks");
+    const dstTaskCols = getColumns(dstDb, "tasks");
+    const srcDocCols = getColumns(srcDb, "task_documents");
+    const dstDocCols = getColumns(dstDb, "task_documents");
+
+    const { shared: taskColumns, dropped: droppedTaskColumns } = intersectColumns(srcTaskCols, dstTaskCols);
+    const { shared: docColumns, dropped: droppedDocColumns } = intersectColumns(srcDocCols, dstDocCols);
+
+    if (taskColumns.length === 0 || docColumns.length === 0) {
+      throw new Error("schema intersection empty");
+    }
+
+    const dropped = [...droppedTaskColumns.map((c) => `tasks.${c}`), ...droppedDocColumns.map((c) => `task_documents.${c}`)];
+    if (dropped.length > 0) {
+      logger.warn(`Worktree DB hydration dropped columns for ${taskId}: ${dropped.join(", ")}`);
+    }
+
+    const placeholders = ids.map(() => "?").join(", ");
+    const taskColumnList = taskColumns.join(", ");
+    const docColumnList = docColumns.join(", ");
+    const taskValuePlaceholders = taskColumns.map(() => "?").join(", ");
+    const docValuePlaceholders = docColumns.map(() => "?").join(", ");
+
+    const taskRows = srcDb
+      .prepare(`SELECT ${taskColumnList} FROM tasks WHERE id IN (${placeholders})`)
+      .all(...ids) as Array<Record<string, unknown>>;
+
+    const documentRows = srcDb
+      .prepare(`SELECT ${docColumnList} FROM task_documents WHERE taskId IN (${placeholders})`)
+      .all(...ids) as Array<Record<string, unknown>>;
+
+    const insertTask = dstDb.prepare(
+      `INSERT OR REPLACE INTO tasks (${taskColumnList}) VALUES (${taskValuePlaceholders})`,
+    );
+    const insertDocument = dstDb.prepare(
+      `INSERT OR REPLACE INTO task_documents (${docColumnList}) VALUES (${docValuePlaceholders})`,
+    );
+
+    dstDb.exec("BEGIN IMMEDIATE");
+    try {
+      for (const row of taskRows) {
+        insertTask.run(...taskColumns.map((column) => row[column]));
+      }
+      for (const row of documentRows) {
+        insertDocument.run(...docColumns.map((column) => row[column]));
+      }
+      dstDb.exec("COMMIT");
+    } catch (error) {
+      dstDb.exec("ROLLBACK");
+      throw error;
+    }
+
+    return {
+      tasksCopied: taskRows.length,
+      documentsCopied: documentRows.length,
+      degraded: false,
+    };
+  } catch (error) {
+    const reason = error instanceof Error ? error.message : String(error);
+    logger.warn(`Worktree DB hydration failed for ${taskId}: ${reason} (${worktreePath})`);
+    return {
+      tasksCopied: 0,
+      documentsCopied: 0,
+      degraded: true,
+      reason,
+    };
+  } finally {
+    srcDb?.close();
+    dstDb?.close();
+  }
+}