fix(tasktree): serialize schema migrations

internal/tasktree/dispatcher.ts

@@ -111,7 +111,14 @@ export function createTaskTreeDispatcher(options: TaskTreeDispatcherOptions) {
       else throw codedError("unsupported_operation", `unsupported operation: ${String(operation)}`);
       return { ok: true, operation, data };
     } catch (error: any) {
-      return { ok: false, operation: command.operation, error: { code: error?.code ?? "tasktree_error", message: error?.message ?? String(error) } };
+      const migrationDetail = error?.tasktreeMigrationId
+        ? { migrationId: String(error.tasktreeMigrationId), sqlState: error?.code ? String(error.code) : undefined }
+        : undefined;
+      return {
+        ok: false,
+        operation: command.operation,
+        error: { code: error?.code ?? "tasktree_error", message: error?.message ?? String(error), ...(migrationDetail ? { detail: migrationDetail } : {}) },
+      };
     }
   };
 }

internal/tasktree/store.ts

@@ -11,58 +11,100 @@ import type { MdtodoImportPlan } from "./mdtodo-import.ts";
 
 const { Pool } = pg;
 
-const schema = [
+type SchemaMigration = { id: string; statements: string[] };
-  `CREATE TABLE IF NOT EXISTS tasktree_schema_migrations (migration_id TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
+
-  `CREATE TABLE IF NOT EXISTS tasktree_groups (
+const migrationLedgerStatement = `CREATE TABLE IF NOT EXISTS tasktree_schema_migrations (
-    id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '',
+  migration_id TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`;
-    created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
+const schemaMigrations: SchemaMigration[] = [
-  `CREATE TABLE IF NOT EXISTS tasktree_tasks (
+  {
-    id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE,
+    id: "tasktree-20260716-v1",
-    parent_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, kind TEXT NOT NULL CHECK (kind IN ('task','subtask','subsubtask')),
+    statements: [
-    title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending'
+      `CREATE TABLE IF NOT EXISTS tasktree_groups (
-      CHECK (status IN ('pending','in_progress','completed','blocked')),
+        id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '',
-    start_at TIMESTAMPTZ, due_at TIMESTAMPTZ, sort_order INTEGER NOT NULL DEFAULT 0,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+      `CREATE TABLE IF NOT EXISTS tasktree_tasks (
-    CHECK (due_at IS NULL OR start_at IS NULL OR due_at >= start_at))`,
+        id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE,
-  `CREATE TABLE IF NOT EXISTS tasktree_milestones (
+        parent_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, kind TEXT NOT NULL CHECK (kind IN ('task','subtask')),
-    id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE,
+        title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending'
-    task_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, title TEXT NOT NULL,
+          CHECK (status IN ('pending','in_progress','completed','blocked')),
-    occurs_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
+        start_at TIMESTAMPTZ, due_at TIMESTAMPTZ, sort_order INTEGER NOT NULL DEFAULT 0,
-  `CREATE TABLE IF NOT EXISTS tasktree_execution_reports (
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-    id TEXT PRIMARY KEY, task_id TEXT NOT NULL REFERENCES tasktree_tasks(id) ON DELETE CASCADE,
+        CHECK (due_at IS NULL OR start_at IS NULL OR due_at >= start_at))`,
-    title TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'succeeded',
+      `CREATE TABLE IF NOT EXISTS tasktree_milestones (
-    created_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
+        id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE,
-  `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_group_order ON tasktree_tasks(group_id, sort_order, created_at)`,
+        task_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, title TEXT NOT NULL,
-  `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_parent ON tasktree_tasks(parent_id)`,
+        occurs_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
-  `CREATE INDEX IF NOT EXISTS idx_tasktree_reports_task ON tasktree_execution_reports(task_id, created_at DESC)`
+      `CREATE TABLE IF NOT EXISTS tasktree_execution_reports (
-  ,`CREATE TABLE IF NOT EXISTS tasktree_backup_state (
+        id TEXT PRIMARY KEY, task_id TEXT NOT NULL REFERENCES tasktree_tasks(id) ON DELETE CASCADE,
+        title TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'succeeded',
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
+      `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_group_order ON tasktree_tasks(group_id, sort_order, created_at)`,
+      `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_parent ON tasktree_tasks(parent_id)`,
+      `CREATE INDEX IF NOT EXISTS idx_tasktree_reports_task ON tasktree_execution_reports(task_id, created_at DESC)`,
+    ],
+  },
+  { id: "tasktree-20260717-v2-subsubtask", statements: [
+    `ALTER TABLE tasktree_tasks DROP CONSTRAINT IF EXISTS tasktree_tasks_kind_check`,
+    `ALTER TABLE tasktree_tasks ADD CONSTRAINT tasktree_tasks_kind_check CHECK (kind IN ('task','subtask','subsubtask'))`,
+  ] },
+  { id: "tasktree-20260718-v3-github-backup", statements: [
+    `CREATE TABLE IF NOT EXISTS tasktree_backup_state (
     target TEXT PRIMARY KEY, pending BOOLEAN NOT NULL DEFAULT false,
     requested_at TIMESTAMPTZ, request_reason TEXT NOT NULL DEFAULT '',
     last_attempt_at TIMESTAMPTZ, last_success_at TIMESTAMPTZ,
     content_sha256 TEXT NOT NULL DEFAULT '', git_commit TEXT NOT NULL DEFAULT '',
     status TEXT NOT NULL DEFAULT 'never' CHECK (status IN ('never','pending','succeeded','failed')),
     error_code TEXT NOT NULL DEFAULT '', error_message TEXT NOT NULL DEFAULT '',
-    updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
-  ,`ALTER TABLE tasktree_tasks DROP CONSTRAINT IF EXISTS tasktree_tasks_kind_check`
+  ] },
-  ,`ALTER TABLE tasktree_tasks ADD CONSTRAINT tasktree_tasks_kind_check CHECK (kind IN ('task','subtask','subsubtask'))`
 ];
+const schemaMigrationIds = schemaMigrations.map((migration) => migration.id);
+const schemaAdvisoryLockKeys = [0x48574c42, 0x54545245] as const;
 
 export class TaskTreeStore {
   private readonly pool: pg.Pool;
   private ready = false;
+  private schemaPromise: Promise<void> | null = null;
 
   constructor(databaseUrl: string) {
     if (!databaseUrl) throw new Error("TASKTREE_DATABASE_URL or DATABASE_URL is required");
-    this.pool = new Pool({ connectionString: databaseUrl, max: 4, connectionTimeoutMillis: 5000, ssl: { rejectUnauthorized: false } });
+    this.pool = new Pool({ connectionString: databaseUrl, max: 4, connectionTimeoutMillis: 5000, ssl: taskTreePostgresSsl(databaseUrl) });
   }
 
   async ensureSchema() {
     if (this.ready) return;
+    this.schemaPromise ??= this.migrateSchema();
+    try {
+      await this.schemaPromise;
+    } catch (error) {
+      this.schemaPromise = null;
+      throw error;
+    }
+  }
+
+  private async migrateSchema() {
+    if (await schemaIsCurrent(this.pool)) {
+      this.ready = true;
+      return;
+    }
     const client = await this.pool.connect();
     try {
       await client.query("BEGIN");
-      for (const statement of schema) await client.query(statement);
+      await client.query("SELECT pg_advisory_xact_lock($1, $2)", [...schemaAdvisoryLockKeys]);
-      await client.query("INSERT INTO tasktree_schema_migrations (migration_id) VALUES ($1),($2),($3) ON CONFLICT DO NOTHING", ["tasktree-20260716-v1", "tasktree-20260717-v2-subsubtask", "tasktree-20260718-v3-github-backup"]);
+      const ledger = await client.query("SELECT to_regclass('tasktree_schema_migrations') AS relation");
+      if (ledger.rows[0]?.relation == null) await client.query(migrationLedgerStatement);
+      const appliedResult = await client.query("SELECT migration_id FROM tasktree_schema_migrations");
+      const applied = new Set(appliedResult.rows.map((row) => String(row.migration_id)));
+      for (const migration of schemaMigrations) {
+        if (applied.has(migration.id)) continue;
+        try {
+          for (const statement of migration.statements) await client.query(statement);
+          await client.query("INSERT INTO tasktree_schema_migrations (migration_id) VALUES ($1)", [migration.id]);
+        } catch (error) {
+          if (error && typeof error === "object") Object.assign(error, { tasktreeMigrationId: migration.id });
+          throw error;
+        }
+      }
       await client.query("COMMIT");
       this.ready = true;
     } catch (error) {
@@ -581,6 +623,21 @@ export function projectTimelineTasks(tasks: TaskItem[], reports: ExecutionReport
   });
 }
 
+async function schemaIsCurrent(pool: pg.Pool): Promise<boolean> {
+  const relation = await pool.query("SELECT to_regclass('tasktree_schema_migrations') AS relation");
+  if (relation.rows[0]?.relation == null) return false;
+  try {
+    const applied = await pool.query(
+      "SELECT COUNT(*)::int AS count FROM tasktree_schema_migrations WHERE migration_id = ANY($1::text[])",
+      [schemaMigrationIds],
+    );
+    return Number(applied.rows[0]?.count ?? 0) === schemaMigrationIds.length;
+  } catch (error: any) {
+    if (error?.code === "42P01") return false;
+    throw error;
+  }
+}
+
 function groupRow(row: any): TaskGroup { return { id: row.id, name: row.name, description: row.description, createdAt: iso(row.created_at), updatedAt: iso(row.updated_at) }; }
 function overviewRow(row: any): TaskGroupOverview { return { group: groupRow(row), taskCount: Number(row.task_count), subtaskCount: Number(row.subtask_count), subsubtaskCount: Number(row.subsubtask_count), reportCount: Number(row.report_count), startAt: nullableIso(row.start_at), dueAt: nullableIso(row.due_at) }; }
 function taskRow(row: any): TaskItem { return { id: row.id, groupId: row.group_id, parentId: row.parent_id, kind: row.kind, title: row.title, description: row.description, status: row.status, startAt: nullableIso(row.start_at), dueAt: nullableIso(row.due_at), sortOrder: row.sort_order, createdAt: iso(row.created_at), updatedAt: iso(row.updated_at) }; }
@@ -605,6 +662,13 @@ function backupStateRow(row: any): TaskTreeBackupState {
 function iso(value: unknown): string { return new Date(value as any).toISOString(); }
 function nullableIso(value: unknown): string | null { return value == null ? null : iso(value); }
 function domainError(code: string, message: string) { return Object.assign(new Error(message), { code }); }
+function taskTreePostgresSsl(databaseUrl: string) {
+  try {
+    return new URL(databaseUrl).searchParams.get("sslmode") === "disable" ? false : { rejectUnauthorized: false };
+  } catch {
+    return { rejectUnauthorized: false };
+  }
+}
 function ensureTimeRange(startAt: string | null, dueAt: string | null) {
   if (startAt !== null && dueAt !== null && new Date(startAt).valueOf() > new Date(dueAt).valueOf()) {
     throw domainError("invalid_time_range", "startAt must be before or equal to dueAt");

package.json

@@ -18,6 +18,7 @@
     "harnessrl:native:l1-smoke": "node scripts/harnessrl-native-l1-smoke.mjs",
     "tasktree:backup:l1-smoke": "bun scripts/tasktree-backup-native-l1-smoke.mjs",
     "tasktree:backup:restore:l1-smoke": "bun scripts/tasktree-backup-restore-l1-smoke.mjs",
+    "tasktree:schema:concurrency-smoke": "bun scripts/tasktree-schema-concurrency-smoke.mjs",
     "workbench:api:dev": "bun --watch cmd/hwlab-workbench-api/main.ts",
     "workbench:worker:dev": "bun --watch cmd/hwlab-workbench-worker/main.ts",
     "workbench:web:dev": "cd web/hwlab-cloud-web && WORKBENCH_VITE_USE_POLLING=1 bun run dev",

scripts/tasktree-schema-concurrency-smoke.mjs

@@ -0,0 +1,151 @@
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { fileURLToPath } from "node:url";
+
+import pg from "pg";
+
+const { Pool } = pg;
+const clients = positiveInteger(process.env.TASKTREE_MIGRATION_SMOKE_CLIENTS, 8);
+const rounds = positiveInteger(process.env.TASKTREE_MIGRATION_SMOKE_ROUNDS, 3);
+const adminUrl = process.env.TASKTREE_MIGRATION_SMOKE_DATABASE_URL
+  || "postgresql:///postgres?host=/var/run/postgresql&user=root";
+const databaseName = `tasktree_smoke_${randomUUID().replaceAll("-", "")}`;
+const cliPath = fileURLToPath(new URL("../tools/hwlab-cli/bin/hwlab-cli.ts", import.meta.url));
+const scopedUrl = new URL(adminUrl);
+scopedUrl.pathname = `/${databaseName}`;
+scopedUrl.searchParams.set("sslmode", "disable");
+scopedUrl.searchParams.set("options", "-clock_timeout=1500ms");
+const databaseUrl = scopedUrl.toString();
+const admin = new Pool({ connectionString: adminUrl, max: 2 });
+const scoped = new Pool({ connectionString: databaseUrl, max: 4 });
+
+try {
+  await admin.query(`CREATE DATABASE ${databaseName}`);
+
+  const fresh = await Promise.all(Array.from({ length: clients }, () => runCli(["group", "list"])));
+  fresh.forEach((result) => assertEnvelope(result, "group.list"));
+  await assertMigrations();
+
+  const createdGroup = await runCli(["group", "create", "--name", `Concurrency smoke ${Date.now()}`]);
+  assertEnvelope(createdGroup, "group.create");
+  const groupId = createdGroup.data.id;
+  const createdTask = await runCli(["task", "create", "--group", groupId, "--title", "Read envelope seed"]);
+  assertEnvelope(createdTask, "task.create");
+  const taskId = createdTask.data.id;
+  const report = await runCli(["report", "write", "--task", taskId, "--title", "Smoke", "--body", "passed"]);
+  assertEnvelope(report, "report.write");
+  assert.equal(report.data.mutation, true);
+
+  await scoped.query(
+    "DELETE FROM tasktree_schema_migrations WHERE migration_id = ANY($1::text[])",
+    [["tasktree-20260717-v2-subsubtask", "tasktree-20260718-v3-github-backup"]],
+  );
+  await scoped.query("ALTER TABLE tasktree_tasks DROP CONSTRAINT tasktree_tasks_kind_check");
+  await scoped.query(
+    "INSERT INTO tasktree_tasks (id,group_id,kind,title) VALUES ($1,$2,$3,$4)",
+    ["tt_invalid_migration_smoke", groupId, "invalid", "Migration rollback sentinel"],
+  );
+  const failedMigration = await runCli(["group", "list"], true);
+  assert.equal(failedMigration.ok, false);
+  assert.equal(failedMigration.error.code, "23514");
+  assert.deepEqual(failedMigration.error.detail, {
+    migrationId: "tasktree-20260717-v2-subsubtask",
+    sqlState: "23514",
+  });
+  const afterFailure = await scoped.query("SELECT migration_id FROM tasktree_schema_migrations ORDER BY migration_id");
+  assert.deepEqual(afterFailure.rows.map((row) => row.migration_id), ["tasktree-20260716-v1"]);
+
+  await scoped.query("DELETE FROM tasktree_tasks WHERE id=$1", ["tt_invalid_migration_smoke"]);
+  assertEnvelope(await runCli(["group", "list"]), "group.list");
+  await assertMigrations();
+
+  const blocker = await scoped.connect();
+  try {
+    await blocker.query("BEGIN");
+    await blocker.query("SELECT id FROM tasktree_tasks LIMIT 1");
+    const commands = [
+      ["group", "list"],
+      ["task", "list", "--group", groupId],
+      ["report", "list", "--task", taskId],
+    ];
+    for (let round = 0; round < rounds; round += 1) {
+      const results = await Promise.all(Array.from({ length: clients }, (_, index) => runCli(commands[index % commands.length])));
+      results.forEach((result, index) => assertEnvelope(result, ["group.list", "task.list", "report.list"][index % commands.length]));
+    }
+    await blocker.query("COMMIT");
+  } catch (error) {
+    await blocker.query("ROLLBACK").catch(() => {});
+    throw error;
+  } finally {
+    blocker.release();
+  }
+
+  console.log(JSON.stringify({
+    ok: true,
+    operation: "tasktree.schema-concurrency-smoke",
+    data: {
+      clients,
+      rounds,
+      freshCliEnvelopes: fresh.length,
+      existingCliEnvelopes: clients * rounds,
+      migrationRollbackVerified: true,
+      existingReadsWithTableLockVerified: true,
+      migrationIds: await migrationIds(),
+    },
+  }, null, 2));
+} finally {
+  await scoped.end().catch(() => {});
+  await admin.query(`DROP DATABASE IF EXISTS ${databaseName} WITH (FORCE)`).catch(() => {});
+  await admin.end().catch(() => {});
+}
+
+async function runCli(args, allowFailure = false) {
+  const child = spawn(process.execPath, [cliPath, "tasktree", ...args], {
+    cwd: fileURLToPath(new URL("..", import.meta.url)),
+    env: { ...process.env, TASKTREE_DATABASE_URL: databaseUrl, DATABASE_URL: "" },
+    stdio: ["ignore", "pipe", "pipe"],
+  });
+  let stdout = "";
+  let stderr = "";
+  child.stdout.setEncoding("utf8").on("data", (chunk) => { stdout += chunk; });
+  child.stderr.setEncoding("utf8").on("data", (chunk) => { stderr += chunk; });
+  const exitCode = await new Promise((resolve, reject) => {
+    child.once("error", reject);
+    child.once("close", resolve);
+  });
+  assert.equal(stderr.trim(), "", `CLI stderr must stay empty: ${stderr}`);
+  assert.ok(stdout.trim(), "CLI must return one JSON envelope on stdout");
+  const result = JSON.parse(stdout);
+  if (!allowFailure) assert.equal(exitCode, 0, JSON.stringify(result));
+  return result;
+}
+
+function assertEnvelope(result, operation) {
+  assert.equal(result.ok, true, JSON.stringify(result));
+  assert.equal(result.operation, operation);
+  assert.ok(Object.hasOwn(result, "data"), "successful CLI envelope must contain data");
+}
+
+async function assertMigrations() {
+  assert.deepEqual(await migrationIds(), [
+    "tasktree-20260716-v1",
+    "tasktree-20260717-v2-subsubtask",
+    "tasktree-20260718-v3-github-backup",
+  ]);
+  const constraint = await scoped.query(
+    "SELECT COUNT(*)::int AS count FROM pg_constraint WHERE conrelid='tasktree_tasks'::regclass AND conname='tasktree_tasks_kind_check'",
+  );
+  assert.equal(constraint.rows[0].count, 1);
+}
+
+async function migrationIds() {
+  const result = await scoped.query("SELECT migration_id FROM tasktree_schema_migrations ORDER BY migration_id");
+  return result.rows.map((row) => row.migration_id);
+}
+
+function positiveInteger(value, fallback) {
+  const parsed = Number(value ?? fallback);
+  assert.ok(Number.isInteger(parsed) && parsed > 0, "smoke concurrency parameters must be positive integers");
+  return parsed;
+}