feat(peers): linked Dispatch instances — pairing, shadow agents, and remote activity

apps/server/src/agents/events.ts

@@ -8,6 +8,47 @@ import type {
   AgentRecord,
 } from "./types.js";
 
+/**
+ * Append one row to the `agent_events` history. Fire-and-forget by design:
+ * losing a history row must never block the live status indicator.
+ *
+ * `at` exists for MIRRORED peer events, which already happened on another
+ * machine — their history row keeps the originating timestamp so the timeline
+ * reads in the order the remote agent actually worked in. It doubles as the
+ * dedupe identity: a peer snapshot replays the whole agent list after every
+ * reconnect, so the insert is skipped when this agent already has a row at that
+ * exact instant. Locally-written events omit it and get `now()`.
+ */
+export function appendAgentEventHistory(
+  pool: Pool,
+  logger: FastifyBaseLogger,
+  id: string,
+  input: AgentLatestEventInput,
+  at?: string
+): void {
+  pool
+    .query(
+      `INSERT INTO agent_events (agent_id, event_type, message, metadata, agent_type, agent_name, project_dir, created_at)
+       SELECT $1, $2, $3, $4::jsonb, type, name, COALESCE(git_context->>'repoRoot', cwd),
+              COALESCE($5::timestamptz, now())
+       FROM agents WHERE id = $1
+         AND ($5::timestamptz IS NULL OR NOT EXISTS (
+           SELECT 1 FROM agent_events
+            WHERE agent_id = $1 AND created_at = $5::timestamptz
+         ))`,
+      [
+        id,
+        input.type,
+        input.message,
+        JSON.stringify(input.metadata ?? {}),
+        at ?? null,
+      ]
+    )
+    .catch((err) =>
+      logger.warn({ err }, "Failed to insert agent event history")
+    );
+}
+
 /**
  * Persist a latest-event update for `id`. Two writes:
  *   1. UPDATE the agent's `latest_event_*` columns synchronously. Throws
@@ -52,19 +93,7 @@ export async function writeLatestEvent(
     throw new AgentError("Agent not found.", 404);
   }
 
-  // Append to event history (fire-and-forget — keeping this off the critical
-  // path means the agent status indicator updates even if the history table
-  // is briefly unavailable).
-  pool
-    .query(
-      `INSERT INTO agent_events (agent_id, event_type, message, metadata, agent_type, agent_name, project_dir)
-       SELECT $1, $2, $3, $4::jsonb, type, name, COALESCE(git_context->>'repoRoot', cwd)
-       FROM agents WHERE id = $1`,
-      [id, input.type, message, JSON.stringify(input.metadata ?? {})]
-    )
-    .catch((err) =>
-      logger.warn({ err }, "Failed to insert agent event history")
-    );
+  appendAgentEventHistory(pool, logger, id, { ...input, message });
 }
 
 /**
@@ -113,16 +142,7 @@ export async function writeLatestEventIfCurrent(
     return false;
   }
 
-  pool
-    .query(
-      `INSERT INTO agent_events (agent_id, event_type, message, metadata, agent_type, agent_name, project_dir)
-       SELECT $1, $2, $3, $4::jsonb, type, name, COALESCE(git_context->>'repoRoot', cwd)
-       FROM agents WHERE id = $1`,
-      [id, input.type, message, JSON.stringify(input.metadata ?? {})]
-    )
-    .catch((err) =>
-      logger.warn({ err }, "Failed to insert agent event history")
-    );
+  appendAgentEventHistory(pool, logger, id, { ...input, message });
 
   return true;
 }

apps/server/src/agents/manager.ts

@@ -587,6 +587,55 @@ export class AgentManager {
     );
   }
 
+  /**
+   * Insert a shadow row for an agent that runs on a linked instance. No tmux
+   * session, no setup script — status and events are mirrored from the peer.
+   */
+  async createShadowAgent(input: {
+    peerId: string;
+    remoteId: string;
+    name: string;
+    type: AgentType;
+    cwd: string;
+    status?: AgentStatus;
+    parentAgentId?: string;
+    launchedByAgentId?: string;
+  }): Promise<AgentRecord> {
+    const id = this.newAgentId();
+    await this.pool.query(
+      `INSERT INTO agents (id, name, type, role, status, cwd, peer_id, remote_id, parent_agent_id, launched_by_agent_id, codex_args, updated_at)
+       VALUES ($1, $2, $3, 'standard', $4, $5, $6, $7, $8, $9, '[]'::jsonb, NOW())`,
+      [
+        id,
+        input.name,
+        input.type,
+        input.status ?? "creating",
+        input.cwd,
+        input.peerId,
+        input.remoteId,
+        input.parentAgentId ?? null,
+        input.launchedByAgentId ?? input.parentAgentId ?? null,
+      ]
+    );
+    return await this.getRequiredAgent(id);
+  }
+
+  /** Mirror a peer-reported status onto a shadow row. */
+  async updateShadowAgent(
+    id: string,
+    update: { status?: AgentStatus; name?: string }
+  ): Promise<AgentRecord | null> {
+    await this.pool.query(
+      `UPDATE agents
+          SET status = COALESCE($2, status),
+              name = COALESCE($3, name),
+              updated_at = NOW()
+        WHERE id = $1 AND peer_id IS NOT NULL`,
+      [id, update.status ?? null, update.name ?? null]
+    );
+    return await this.getAgent(id);
+  }
+
   private async launchInertAgent(opts: {
     id: string;
     type: AgentType;
@@ -982,6 +1031,20 @@ export class AgentManager {
 
   async getTerminalAccess(id: string): Promise<AgentTerminalAccess> {
     const agent = await this.getRequiredAgent(id);
+    if (agent.peerId) {
+      // Shadow row: the pane lives on another instance and is not proxied.
+      // Name it the way every other surface does — "inst_9f2c62a1b0" means
+      // nothing to the person reading this.
+      const peer = await this.pool.query<{ name: string }>(
+        `SELECT name FROM peers WHERE id = $1`,
+        [agent.peerId]
+      );
+      const label = peer.rows[0]?.name ?? agent.peerId;
+      return {
+        mode: "inert",
+        message: `This agent runs on linked instance "${label}" — its terminal is not available here.`,
+      };
+    }
     if (agent.status !== "running" && agent.status !== "creating") {
       throw new AgentError("Agent is not running.", 409);
     }
@@ -1437,6 +1500,8 @@ export class AgentManager {
         template_id AS "templateId",
         auto_review AS "autoReview",
         cli_session_id AS "cliSessionId",
+        peer_id AS "peerId",
+        remote_id AS "remoteId",
         (
           SELECT unified_review.id
           FROM reviews unified_review

apps/server/src/agents/reconciler.ts

@@ -108,7 +108,9 @@ async function reconcileAgentStatuses(
   await diagnostics.maybeMaintenanceLogs();
 
   const result = await pool.query(
-    "SELECT id, tmux_session AS \"tmuxSession\", status, updated_at AS \"updatedAt\" FROM agents WHERE deleted_at IS NULL AND status IN ('running', 'stopping', 'creating', 'archiving')"
+    // Shadow rows (peer_id set) have no local pane by design — their status is
+    // mirrored from the owning peer, so local tmux reconciliation must skip them.
+    "SELECT id, tmux_session AS \"tmuxSession\", status, updated_at AS \"updatedAt\" FROM agents WHERE deleted_at IS NULL AND peer_id IS NULL AND status IN ('running', 'stopping', 'creating', 'archiving')"
   );
 
   const reconciled: AgentRecord[] = [];

apps/server/src/agents/types.ts

@@ -130,6 +130,10 @@ export type AgentRecord = {
   templateId: string | null;
   autoReview: boolean;
   cliSessionId: string | null;
+  /** Set on shadow rows: the linked instance actually running this agent. */
+  peerId: string | null;
+  /** The agent's id on that peer instance. */
+  remoteId: string | null;
   createdAt: string;
   updatedAt: string;
 };

apps/server/src/db/migrations/0042_peers.sql

@@ -0,0 +1,125 @@
+-- Linked Dispatch instances: pairing, per-direction credentials, and the
+-- pair-time launch policy. Generalized from the browser-extension pairing
+-- shape (0035); differences: the code renders on the accepting side,
+-- credentials are issued in BOTH directions, and the caller's tailnet
+-- StableID is pinned alongside the token.
+
+CREATE TABLE IF NOT EXISTS peers (
+  id text PRIMARY KEY,                       -- the peer's instance_id (inst_*)
+  -- Two names, because they answer different questions. `reported_name` is what
+  -- the peer calls itself (its own instance_name, or hostname); `name` is what
+  -- THIS instance calls it, seeded from reported_name and editable here. The
+  -- local label is the one agents type as `location`, so "Cloud" can mean
+  -- different machines on different laptops without renaming anything remote.
+  name text NOT NULL,
+  reported_name text,
+  url text NOT NULL,                         -- base URL we dial (MagicDNS or public)
+  tailnet_stable_id text,                    -- the peer node's durable tailscale ID
+  outbound_token text NOT NULL,              -- bearer WE present to THEM
+  created_at timestamptz NOT NULL DEFAULT now(),
+  last_seen_at timestamptz,
+  revoked_at timestamptz
+);
+
+-- One peer per local label, so `location: "Cloud"` is never ambiguous. Partial
+-- so a revoked peer's label is free for reuse.
+CREATE UNIQUE INDEX IF NOT EXISTS peers_active_name_idx
+  ON peers (lower(name))
+  WHERE revoked_at IS NULL;
+
+-- Bearer tokens THEY present to US, plus the standing pair-time policy.
+--
+-- Capabilities are a SET, not a flag. Pairing grants separable things — run
+-- code here, inject prompts into agents here, do either with the sandbox off,
+-- and see what agents here are doing — and a single boolean cannot describe
+-- them. Each route gates on its own column, so a launch-only CI box or a
+-- message-only observer is a policy row rather than a protocol version.
+CREATE TABLE IF NOT EXISTS peer_credentials (
+  id uuid PRIMARY KEY,
+  peer_id text NOT NULL REFERENCES peers(id) ON DELETE CASCADE,
+  token_hash text NOT NULL UNIQUE,
+  tailnet_stable_id text,                    -- pinned caller identity; NULL only for non-tailnet pairings
+  allow_launch boolean NOT NULL DEFAULT true,
+  allow_message boolean NOT NULL DEFAULT true,
+  -- Off by default even when launching is allowed: fullAccess disables the
+  -- sandbox, and "may launch here" should not silently mean "may launch
+  -- unsandboxed here". Opt in per pairing.
+  allow_full_access boolean NOT NULL DEFAULT false,
+  -- Gates the LATEST-EVENT payload on the peer event feed, not the feed itself.
+  -- Id/name/type/status always cross — without them a shadow row could never
+  -- track its remote agent at all. What this grant adds is the event text an
+  -- agent writes about itself ("Refactoring auth middleware"), which says far
+  -- more about what this machine is doing than a status enum does. On by
+  -- default: a shadow whose progress never moves is the thing it exists to fix.
+  allow_events boolean NOT NULL DEFAULT true,
+  created_at timestamptz NOT NULL DEFAULT now(),
+  last_used_at timestamptz,
+  revoked_at timestamptz
+);
+
+CREATE INDEX IF NOT EXISTS peer_credentials_active_idx
+  ON peer_credentials (token_hash)
+  WHERE revoked_at IS NULL;
+
+-- Short-lived pairing offers displayed on THIS (accepting) instance.
+CREATE TABLE IF NOT EXISTS peer_pairings (
+  id uuid PRIMARY KEY,
+  code_hash text NOT NULL,
+  allow_launch boolean NOT NULL DEFAULT true,
+  allow_message boolean NOT NULL DEFAULT true,
+  allow_full_access boolean NOT NULL DEFAULT false,
+  allow_events boolean NOT NULL DEFAULT true,
+  require_tailnet boolean NOT NULL DEFAULT true,
+  created_at timestamptz NOT NULL DEFAULT now(),
+  expires_at timestamptz NOT NULL,
+  claimed_at timestamptz,
+  peer_id text REFERENCES peers(id) ON DELETE SET NULL
+);
+
+CREATE INDEX IF NOT EXISTS peer_pairings_expires_idx
+  ON peer_pairings (expires_at);
+
+-- Shadow rows: an agent launched on a peer gets a LOCAL agents row (a local
+-- id, no tmux session) so every existing consumer — sidebar, list_agents,
+-- message targeting, SSE — keeps working unchanged. Only the ORIGINATING
+-- instance holds the shadow; on the executing instance it is a plain agent.
+ALTER TABLE agents ADD COLUMN IF NOT EXISTS peer_id text;
+ALTER TABLE agents ADD COLUMN IF NOT EXISTS remote_id text;
+
+-- Deliberately no index on (peer_id, remote_id): building one at startup
+-- migration write-locks a potentially large agents table, and shadow lookups
+-- filter a table that stays small in practice.
+
+-- Messages are CONTENT: losing one loses work, and a blind retry would
+-- double-inject. Hence a durable sender-side outbox with backoff, and a
+-- receiver-side receipt per idempotency key. Status/events are STATE and get
+-- neither — the next event supersedes a dropped one.
+CREATE TABLE IF NOT EXISTS peer_outbox (
+  id uuid PRIMARY KEY,
+  peer_id text NOT NULL REFERENCES peers(id) ON DELETE CASCADE,
+  path text NOT NULL,
+  body jsonb NOT NULL,
+  idempotency_key uuid NOT NULL,
+  attempts int NOT NULL DEFAULT 0,
+  next_attempt_at timestamptz NOT NULL DEFAULT now(),
+  last_error text,
+  created_at timestamptz NOT NULL DEFAULT now(),
+  delivered_at timestamptz,
+  -- Retry is not forever. A peer that is merely gone — a machine nobody
+  -- explicitly unlinked — would otherwise accumulate rows that are re-attempted
+  -- until the end of time, and each attempt costs a 30s connect timeout. Past
+  -- the cap the row is dead-lettered: kept for inspection, never sent again.
+  dead_lettered_at timestamptz
+);
+
+CREATE INDEX IF NOT EXISTS peer_outbox_due_idx
+  ON peer_outbox (peer_id, next_attempt_at)
+  WHERE delivered_at IS NULL AND dead_lettered_at IS NULL;
+
+CREATE TABLE IF NOT EXISTS peer_message_receipts (
+  peer_id text NOT NULL,
+  idempotency_key uuid NOT NULL,
+  delivered boolean NOT NULL DEFAULT false,
+  received_at timestamptz NOT NULL DEFAULT now(),
+  PRIMARY KEY (peer_id, idempotency_key)
+);