feat(plugins): allow .sql in packages + make the 8 migrators concurrency-safe

middleware/assets/boilerplate/agent-integration/scripts/build-zip.mjs

@@ -68,6 +68,10 @@ const INCLUDE = [
   'manifest.yaml',
   'package.json',
   'dist',
+  // Plugin-owned SQL. The host allows .sql in a package (zipExtractor.ts); without
+  // this entry the directory is silently dropped here and the install succeeds with
+  // no schema at all — a green install and a missing table.
+  'migrations',
   'skills',
   'assets',
   'README.md',

middleware/assets/boilerplate/agent-pure-llm/scripts/build-zip.mjs

@@ -68,6 +68,10 @@ const INCLUDE = [
   'manifest.yaml',
   'package.json',
   'dist',
+  // Plugin-owned SQL. The host allows .sql in a package (zipExtractor.ts); without
+  // this entry the directory is silently dropped here and the install succeeds with
+  // no schema at all — a green install and a missing table.
+  'migrations',
   'skills',
   'assets',
   'README.md',

middleware/packages/harness-knowledge-graph-neon/src/migrator.ts

@@ -1,40 +1,106 @@
 import { readdir, readFile } from 'node:fs/promises';
 import { dirname, join } from 'node:path';
 import { fileURLToPath } from 'node:url';
-import type { Pool } from 'pg';
+import type { Pool, PoolClient } from 'pg';
 
 const MIGRATIONS_DIR = join(
   dirname(fileURLToPath(import.meta.url)),
   'migrations',
 );
 
+const LEDGER_DDL = `
+      CREATE TABLE IF NOT EXISTS _graph_migrations (
+        id          TEXT PRIMARY KEY,
+        applied_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+      );
+    `;
+
+/**
+ * Advisory-lock coordinates. The namespace is shared by every SQL migrator;
+ * 4400 is `LOCK_NS_REGISTRY` (embedding registry) and 4401 the stale-vector
+ * clear, so 4410 keeps migrations clear of both. The second key is
+ * `hashtext(<ledger table>)`, so each subsystem serialises against its own
+ * replicas only and never against a different subsystem's migrations.
+ */
+const LOCK_NS_MIGRATIONS = 4_410;
+const LOCK_KEY = '_graph_migrations';
+
+/**
+ * How long a replica waits for the migration lock before giving up.
+ *
+ * This is the tightest of the eight budgets and therefore sets the value for
+ * all of them: `activate()` in `plugin.ts` calls `waitForPostgres` (6s budget)
+ * and then this migrator, inside a `ToolPluginRuntime` timeout of 10s. 6s + 2s
+ * leaves 2s of headroom for the migrations themselves, and the blocking
+ * `pg_advisory_lock` is unusable here for exactly that reason. Exported so the
+ * budget can be asserted in a test instead of trusted in a comment.
+ */
+export const GRAPH_MIGRATION_LOCK_WAIT_MS = 2_000;
+const LOCK_POLL_MS = 100;
+
+/**
+ * Apply pending knowledge-graph SQL migrations against the graph pool.
+ *
+ * Concurrency: read-ledger → filter → apply is not safe on its own. Two
+ * replicas booting together both see the same pending list and both execute
+ * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the
+ * loser's boot fails, and the kernel treats the knowledge graph as required,
+ * so that is a crash-loop). The apply loop therefore runs under a
+ * session-scoped advisory lock, taken with `pg_try_advisory_lock` and only
+ * after the ledger says there is work to do — the steady-state boot takes no
+ * lock at all and pays nothing.
+ */
 export async function runGraphMigrations(
   pool: Pool,
   log: (msg: string) => void = () => undefined,
 ): Promise<void> {
   const client = await pool.connect();
+  // Tracks whether THIS session provably holds the advisory lock. It is the
+  // only input to `client.release()` below: a connection that cannot prove it
+  // released the lock is destroyed rather than pooled, because ending the
+  // session is the only other way a session-scoped lock goes away.
+  let lockHeld = false;
   try {
-    await client.query(`
-      CREATE TABLE IF NOT EXISTS _graph_migrations (
-        id          TEXT PRIMARY KEY,
-        applied_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+    await ensureLedger(client);
+
+    // Ledger first, lock second. The overwhelmingly common boot has nothing
+    // pending, and that boot must not pay for — or queue behind — a lock.
+    let pending = await pendingMigrations(client);
+    if (pending.length === 0) return;
+
+    const deadline = Date.now() + GRAPH_MIGRATION_LOCK_WAIT_MS;
+    for (;;) {
+      lockHeld = await tryAcquireMigrationLock(client);
+      if (lockHeld) break;
+      const remaining = deadline - Date.now();
+      if (remaining <= 0) break;
+      await sleep(Math.min(LOCK_POLL_MS, remaining));
+    }
+
+    if (!lockHeld) {
+      // Never a silent skip: re-read the ledger. If the holder finished while
+      // we waited, this replica's schema IS current and the boot continues.
+      pending = await pendingMigrations(client);
+      if (pending.length === 0) {
+        log('[graph] migrations applied by another replica while waiting');
+        return;
+      }
+      // Otherwise the work is genuinely still owed. Failing loudly is the only
+      // honest option; the message says "timed out" deliberately, so
+      // `bootstrap.retryErroredPlugins` classifies it as transient and
+      // re-attempts activation instead of latching the plugin `errored`.
+      throw new Error(
+        `[graph] timed out after ${String(GRAPH_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` +
+          `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`,
       );
-    `);
-
-    const applied = new Set(
-      (
-        await client.query<{ id: string }>(
-          'SELECT id FROM _graph_migrations',
-        )
-      ).rows.map((r) => r.id),
-    );
+    }
 
-    const files = (await readdir(MIGRATIONS_DIR))
-      .filter((f) => f.endsWith('.sql'))
-      .sort();
+    // Re-read UNDER the lock. The pre-lock read is a fast path, not a
+    // decision: the replica we queued behind may have applied part or all of
+    // that list before it released.
+    pending = await pendingMigrations(client);
 
-    for (const file of files) {
-      if (applied.has(file)) continue;
+    for (const file of pending) {
       const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8');
       log(`[graph] applying migration ${file}`);
       await client.query('BEGIN');
@@ -50,7 +116,102 @@ export async function runGraphMigrations(
         throw err;
       }
     }
+
+    // Unlock on the success path only, and inside `try` — never in `finally`.
+    // In `finally` it would run on a possibly half-open connection whose pool
+    // sets no `statement_timeout`, so it could hang the release indefinitely,
+    // and an unlock that throws there would replace the original migration
+    // error. On the failure path `lockHeld` stays true and the connection is
+    // destroyed instead, which releases the lock with the session.
+    if (await releaseMigrationLock(client)) lockHeld = false;
   } finally {
-    client.release();
+    client.release(lockHeld);
+  }
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+/**
+ * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE
+ * TABLE` of the same name: the existence check and the catalog insert are
+ * separate steps, so two replicas booting together can both pass the check and
+ * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on
+ * a system catalog index). The table exists either way, so one retry settles
+ * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs
+ * outside any transaction, so the failed statement leaves nothing to roll back.
+ */
+async function ensureLedger(client: PoolClient): Promise<void> {
+  try {
+    await client.query(LEDGER_DDL);
+  } catch (err) {
+    if (!isDuplicateObjectError(err)) throw err;
+    await client.query(LEDGER_DDL);
+  }
+}
+
+function isDuplicateObjectError(err: unknown): boolean {
+  const code = (err as { code?: unknown } | null)?.code;
+  return code === '42P07' || code === '23505';
+}
+
+/**
+ * The ledger read, expressed as the list of files still owed. Called twice on
+ * the locking path — once before the lock and once after acquiring it.
+ */
+async function pendingMigrations(client: PoolClient): Promise<string[]> {
+  const applied = new Set(
+    (
+      await client.query<{ id: string }>(
+        'SELECT id FROM _graph_migrations',
+      )
+    ).rows.map((r) => r.id),
+  );
+
+  const files = (await readdir(MIGRATIONS_DIR))
+    .filter((f) => f.endsWith('.sql'))
+    .sort();
+
+  return files.filter((f) => !applied.has(f));
+}
+
+/**
+ * Take the migration lock without ever blocking the backend, and REPORT
+ * whether it was taken. The boolean is the whole point: a caller that cannot
+ * distinguish "acquired" from "someone else holds it" cannot release anything
+ * either.
+ */
+async function tryAcquireMigrationLock(client: PoolClient): Promise<boolean> {
+  const result = await client.query<{ locked: boolean }>(
+    'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked',
+    [LOCK_NS_MIGRATIONS, LOCK_KEY],
+  );
+  // A fake/limited driver that does not model advisory locks returns no row;
+  // treat that as acquired so unit tests still exercise the migrations. Mirrors
+  // `tryAcquireRegistryLock` in `vectorColumnMigration.ts`.
+  const row = result.rows[0];
+  return row === undefined || row.locked !== false;
+}
+
+/**
+ * Release the session lock, and REPORT whether it actually went. `false` is
+ * what makes the caller destroy the connection instead of pooling it — the
+ * only other way a session-scoped lock is released. Swallowing the answer
+ * hands a connection that may still hold the lock back to the pool, where it
+ * blocks every later replica's migration for the connection's lifetime.
+ */
+async function releaseMigrationLock(client: PoolClient): Promise<boolean> {
+  try {
+    const result = await client.query<{ unlocked: boolean }>(
+      'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked',
+      [LOCK_NS_MIGRATIONS, LOCK_KEY],
+    );
+    // "No row" mirrors the acquire side: a driver that does not model advisory
+    // locks never took one, so nothing is leaked by pooling the connection.
+    const row = result.rows[0];
+    return row === undefined || row.unlocked !== false;
+  } catch {
+    return false;
   }
 }