Harness implementation research: link daemon, agent pill, VM-as-runtime

.gitignore

@@ -80,8 +80,13 @@ studio-demo/
 apps/mesh/test-results/
 apps/mesh/playwright-report/
 
-# Generated docs (root-level only, not apps/docs/)
-/docs
+# Generated docs (root-level only, not apps/docs/). Use /docs/* not /docs
+# so we can un-ignore specific subdirs below — git won't traverse into an
+# ignored directory.
+/docs/*
+# Hand-written specs/plans (superpowers brainstorming + writing-plans output)
+# are first-class engineering artifacts and should be tracked.
+!/docs/superpowers/
 
 # Local dev data directory
 .deco

apps/mesh/migrations/083-thread-run-locally.ts

@@ -0,0 +1,28 @@
+/**
+ * Migration 078: Add run_locally column to threads
+ *
+ * Persists the "Run locally" choice per thread (Phase 10 of the remote
+ * harness dispatch project). When true, every message on this thread is
+ * dispatched to the thread owner's link daemon instead of the
+ * in-cluster sandbox.
+ *
+ * The column is set on first-message creation from the POST body and is
+ * not modified afterwards, so DBOS replays and subsequent messages all
+ * agree on where the run executes. Defaults to false so every existing
+ * thread continues to behave as a normal cluster-side thread.
+ */
+
+import type { Kysely } from "kysely";
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  await db.schema
+    .alterTable("threads")
+    .addColumn("run_locally", "boolean", (col) =>
+      col.notNull().defaultTo(false),
+    )
+    .execute();
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+  await db.schema.alterTable("threads").dropColumn("run_locally").execute();
+}

apps/mesh/migrations/084-drop-host-sandbox-rows.ts

@@ -0,0 +1,29 @@
+/**
+ * Migration 079: Drop sandbox_runner_state rows for the retired `host`
+ * runner.
+ *
+ * The `host` SandboxProvider was the local-dev shortcut that spawned the
+ * sandbox daemon as a child of the mesh process. It has been retired in
+ * favor of the laptop-side `deco link` daemon (auto-spawned by
+ * `bun run dev --local-sandbox-provider`), which exercises the same
+ * remote-cli + remote-user code paths production uses.
+ *
+ * Any `runner_kind = 'host'` rows left in dev databases are orphaned
+ * pointers to daemon PIDs/ports that no longer exist; the new code path
+ * never reads them. Delete them so the table doesn't accumulate dead
+ * state. Sandbox state is ephemeral by design — the runner rehydrates
+ * from a healthy daemon, or provisions a new one on next ensure().
+ */
+
+import { sql, type Kysely } from "kysely";
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  await sql`delete from sandbox_runner_state where runner_kind = 'host'`.execute(
+    db,
+  );
+}
+
+export async function down(_db: Kysely<unknown>): Promise<void> {
+  // Irreversible — the host runner is gone, restoring deleted rows
+  // would point at processes that no longer exist.
+}

apps/mesh/migrations/085-rename-runner-kind.ts

@@ -0,0 +1,121 @@
+/**
+ * Migration 080: Rename sandbox_runner_state.runner_kind →
+ * sandbox_provider_kind, rewrite vmMap JSON keys in virtualmcps, and drop
+ * threads.run_locally.
+ *
+ * (1) ALTER TABLE sandbox_runner_state RENAME COLUMN runner_kind TO
+ *     sandbox_provider_kind — aligns the column name with the
+ *     SandboxProviderKind type rename done in Task 1.1.
+ *
+ * (2) Walk every virtualmcps.metadata.vmMap[user][branch] entry and rename
+ *     the JSON key `runnerKind` → `sandboxProviderKind`. The UPDATE is
+ *     idempotent: it only rewrites entries where `runnerKind` exists and
+ *     `sandboxProviderKind` does not, so re-running is safe.
+ *
+ * (3) DROP COLUMN threads.run_locally — the VM now owns runner-choice; the
+ *     dispatch layer reads from VmMapEntry.sandboxProviderKind instead.
+ */
+
+import { sql, type Kysely } from "kysely";
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  await sql`
+    ALTER TABLE sandbox_runner_state
+    RENAME COLUMN runner_kind TO sandbox_provider_kind
+  `.execute(db);
+
+  // Rewrite vmMap JSON keys. Wrapped in a DO block so it is a no-op in test
+  // environments where the virtualmcps table was never created (PGlite
+  // migration tests run in isolation and may not have the full schema).
+  await sql`
+    DO $$ BEGIN
+      UPDATE virtualmcps v
+      SET metadata = jsonb_set(
+        v.metadata,
+        '{vmMap}',
+        COALESCE(
+          (
+            SELECT jsonb_object_agg(
+              user_key,
+              COALESCE(
+                (
+                  SELECT jsonb_object_agg(
+                    branch_key,
+                    CASE
+                      WHEN entry ? 'runnerKind' AND NOT entry ? 'sandboxProviderKind'
+                        THEN jsonb_set(entry, '{sandboxProviderKind}', entry->'runnerKind') - 'runnerKind'
+                      ELSE entry
+                    END
+                  )
+                  FROM jsonb_each(user_map) AS branches(branch_key, entry)
+                ),
+                '{}'::jsonb
+              )
+            )
+            FROM jsonb_each(v.metadata->'vmMap') AS users(user_key, user_map)
+          ),
+          '{}'::jsonb
+        )
+      )
+      WHERE v.metadata ? 'vmMap'
+        AND jsonb_typeof(v.metadata->'vmMap') = 'object';
+    EXCEPTION WHEN undefined_table THEN
+      NULL;
+    END; $$
+  `.execute(db);
+
+  await db.schema.alterTable("threads").dropColumn("run_locally").execute();
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+  // Re-add threads.run_locally before the other reversals so that if
+  // anything downstream reads the column it still exists.
+  await db.schema
+    .alterTable("threads")
+    .addColumn("run_locally", "boolean", (col) =>
+      col.notNull().defaultTo(false),
+    )
+    .execute();
+
+  await sql`
+    ALTER TABLE sandbox_runner_state
+    RENAME COLUMN sandbox_provider_kind TO runner_kind
+  `.execute(db);
+
+  await sql`
+    DO $$ BEGIN
+      UPDATE virtualmcps v
+      SET metadata = jsonb_set(
+        v.metadata,
+        '{vmMap}',
+        COALESCE(
+          (
+            SELECT jsonb_object_agg(
+              user_key,
+              COALESCE(
+                (
+                  SELECT jsonb_object_agg(
+                    branch_key,
+                    CASE
+                      WHEN entry ? 'sandboxProviderKind' AND NOT entry ? 'runnerKind'
+                        THEN jsonb_set(entry, '{runnerKind}', entry->'sandboxProviderKind') - 'sandboxProviderKind'
+                      ELSE entry
+                    END
+                  )
+                  FROM jsonb_each(user_map) AS branches(branch_key, entry)
+                ),
+                '{}'::jsonb
+              )
+            )
+            FROM jsonb_each(v.metadata->'vmMap') AS users(user_key, user_map)
+          ),
+          '{}'::jsonb
+        )
+      )
+      WHERE v.metadata ? 'vmMap'
+        AND jsonb_typeof(v.metadata->'vmMap') = 'object';
+    EXCEPTION WHEN undefined_table THEN
+      NULL;
+    END; $$
+  `.execute(db);
+}

apps/mesh/migrations/086-thread-pins-and-vm-map-rekey.ts

@@ -0,0 +1,121 @@
+/**
+ * Migration 081: per-thread runner + harness pinning, and vmMap re-key.
+ *
+ * Adds:
+ *   - `threads.sandbox_provider_kind` (nullable, backfilled on first message)
+ *   - `threads.harness_id` (nullable, backfilled on first message)
+ *
+ * Re-keys `virtualmcps.metadata.vmMap` from 2-level to 3-level:
+ *   vmMap[user][branch] = entry
+ *   →
+ *   vmMap[user][branch][entry.sandboxProviderKind ?? 'docker'] = entry
+ * Idempotent: only re-keys entries that still have `vmId` at the second level
+ * (which marks them as the 2-level legacy shape). The sandboxProviderKind
+ * field stays on the entry for one release as a tolerant-reader fallback.
+ */
+
+import { sql, type Kysely } from "kysely";
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  // 1. Thread columns (nullable; populated on first message by the
+  //    POST /messages handler).
+  await db.schema
+    .alterTable("threads")
+    .addColumn("sandbox_provider_kind", "text")
+    .execute();
+  await db.schema
+    .alterTable("threads")
+    .addColumn("harness_id", "text")
+    .execute();
+
+  // 2. vmMap re-key. Wrapped in a DO block so it no-ops if `virtualmcps`
+  //    doesn't exist (PGlite test bootstrap order).
+  await sql`
+    DO $$ BEGIN
+      UPDATE virtualmcps v
+      SET metadata = jsonb_set(
+        v.metadata,
+        '{vmMap}',
+        (
+          SELECT jsonb_object_agg(
+            user_key,
+            (
+              SELECT jsonb_object_agg(
+                branch_key,
+                CASE
+                  -- Already 3-level (entry is itself a map without vmId): pass through.
+                  WHEN NOT (entry ? 'vmId') THEN entry
+                  -- Legacy 2-level: wrap under sandboxProviderKind.
+                  ELSE jsonb_build_object(
+                    COALESCE(entry->>'sandboxProviderKind', 'docker'),
+                    entry
+                  )
+                END
+              )
+              FROM jsonb_each(user_map) AS branches(branch_key, entry)
+            )
+          )
+          FROM jsonb_each(v.metadata->'vmMap') AS users(user_key, user_map)
+        )
+      )
+      WHERE v.metadata ? 'vmMap'
+        AND EXISTS (
+          SELECT 1
+          FROM jsonb_each(v.metadata->'vmMap') AS users(user_key, user_map)
+          JOIN jsonb_each(user_map) AS branches(branch_key, entry) ON true
+          WHERE entry ? 'vmId'
+        );
+    EXCEPTION WHEN undefined_table THEN
+      -- PGlite migration order: virtualmcps not yet created. Safe no-op.
+      NULL;
+    END; $$
+  `.execute(db);
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+  // Drop the columns.
+  await db.schema.alterTable("threads").dropColumn("harness_id").execute();
+  await db.schema
+    .alterTable("threads")
+    .dropColumn("sandbox_provider_kind")
+    .execute();
+
+  // Reverse the vmMap re-key: collapse each (user, branch, kind) back to
+  // (user, branch). When multiple kinds exist for the same branch, the first
+  // one (in JSON iteration order) wins; the others are dropped. Acceptable
+  // because down() is a manual recovery path, not a production rollback.
+  await sql`
+    DO $$ BEGIN
+      UPDATE virtualmcps v
+      SET metadata = jsonb_set(
+        v.metadata,
+        '{vmMap}',
+        (
+          SELECT jsonb_object_agg(
+            user_key,
+            (
+              SELECT jsonb_object_agg(
+                branch_key,
+                CASE
+                  -- Already 2-level (entry has vmId): pass through.
+                  WHEN entry ? 'vmId' THEN entry
+                  -- 3-level: pick the first kind's entry.
+                  ELSE (
+                    SELECT inner_entry
+                    FROM jsonb_each(entry) AS kinds(kind_key, inner_entry)
+                    LIMIT 1
+                  )
+                END
+              )
+              FROM jsonb_each(user_map) AS branches(branch_key, entry)
+            )
+          )
+          FROM jsonb_each(v.metadata->'vmMap') AS users(user_key, user_map)
+        )
+      )
+      WHERE v.metadata ? 'vmMap';
+    EXCEPTION WHEN undefined_table THEN
+      NULL;
+    END; $$
+  `.execute(db);
+}