feat(peer): link bots across two daemons

crates/bus/src/schema.rs

@@ -7,13 +7,15 @@
 mod base;
 mod decisions;
 mod history;
+mod peers;
 
 use base::MIGRATION_1;
 use decisions::MIGRATION_12;
 use history::{
     MIGRATION_10, MIGRATION_11, MIGRATION_2, MIGRATION_3, MIGRATION_4, MIGRATION_5, MIGRATION_6,
     MIGRATION_7, MIGRATION_8, MIGRATION_9,
 };
+use peers::MIGRATION_14;
 
 pub const MIGRATIONS: &[&str] = &[
     MIGRATION_1,
@@ -29,4 +31,5 @@ pub const MIGRATIONS: &[&str] = &[
     MIGRATION_11,
     MIGRATION_12,
     "ALTER TABLE bot ADD COLUMN runtime TEXT NOT NULL DEFAULT 'claude_code' CHECK(runtime IN ('claude_code', 'codex_cli'));",
+    MIGRATION_14,
 ];

crates/bus/src/schema/peers.rs

@@ -0,0 +1,46 @@
+//! Migration 14: peer daemons and linked bots.
+//!
+//! A linked bot is an ordinary `bot` row with `peer_id` set, so every foreign
+//! key that assumes a local bot (message senders, task ends, deliveries) keeps
+//! holding. The mapping tables carry the other side's ids: a message or task
+//! that crosses the link has an id on each daemon, and threading, dedupe and
+//! closing a mirrored task all need to translate between them.
+
+pub(super) const MIGRATION_14: &str = r#"
+CREATE TABLE peer (
+    id           TEXT PRIMARY KEY,
+    name         TEXT NOT NULL UNIQUE,
+    daemon_id    TEXT UNIQUE,
+    url          TEXT,
+    created_at   TEXT NOT NULL,
+    last_seen_at TEXT,
+    revoked_at   TEXT
+);
+
+ALTER TABLE bot ADD COLUMN peer_id TEXT REFERENCES peer(id);
+ALTER TABLE bot ADD COLUMN remote_bot_id TEXT;
+CREATE UNIQUE INDEX idx_bot_remote ON bot(peer_id, remote_bot_id, project_id)
+    WHERE peer_id IS NOT NULL AND deleted_at IS NULL;
+
+CREATE TABLE peer_link (
+    peer_id TEXT NOT NULL REFERENCES peer(id),
+    bot_id  TEXT NOT NULL REFERENCES bot(id),
+    PRIMARY KEY (peer_id, bot_id)
+);
+
+CREATE TABLE peer_message (
+    peer_id           TEXT NOT NULL REFERENCES peer(id),
+    remote_message_id TEXT NOT NULL,
+    message_id        TEXT NOT NULL,
+    PRIMARY KEY (peer_id, remote_message_id)
+);
+CREATE INDEX idx_peer_message_local ON peer_message(peer_id, message_id);
+
+CREATE TABLE peer_task (
+    peer_id        TEXT NOT NULL REFERENCES peer(id),
+    remote_task_id TEXT NOT NULL,
+    task_id        TEXT NOT NULL,
+    PRIMARY KEY (peer_id, remote_task_id)
+);
+CREATE INDEX idx_peer_task_local ON peer_task(peer_id, task_id);
+"#;

crates/bus/src/types.rs

@@ -7,11 +7,15 @@ mod decision_enums;
 mod decisions;
 mod entities;
 mod enums;
+mod peers;
+mod revisions;
 
 pub use decision_enums::*;
 pub use decisions::*;
 pub use entities::*;
 pub use enums::*;
+pub use peers::*;
+pub use revisions::*;
 
 pub type Id = String;
 

crates/bus/src/types/entities.rs

@@ -51,9 +51,23 @@ pub struct Bot {
     /// leave `list_bots`, addressing, and the population cap.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub deleted_at: Option<DateTime<Utc>>,
+    /// Set on a linked bot: one that runs on this peer daemon and stands in
+    /// here as a message target. It has no workspace and no runtime.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub peer_id: Option<Id>,
+    /// The linked bot's id on its peer.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub remote_bot_id: Option<Id>,
     pub created_at: DateTime<Utc>,
 }
 
+impl Bot {
+    /// True for a bot that runs on a peer daemon rather than here.
+    pub fn is_linked(&self) -> bool {
+        self.peer_id.is_some()
+    }
+}
+
 #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum BotRuntime {
@@ -71,64 +85,6 @@ impl BotRuntime {
     }
 }
 
-/// One field-level change to a bot's identity: the audit trail that makes
-/// unsupervised self-management reversible.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct BotRevision {
-    pub id: Id,
-    pub bot_id: Id,
-    /// `user` or `bot:<id>`.
-    pub changed_by: String,
-    pub field: RevisionField,
-    pub old_value: String,
-    pub new_value: String,
-    pub created_at: DateTime<Utc>,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum RevisionField {
-    Name,
-    Avatar,
-    Description,
-    Instructions,
-    /// Lifecycle markers rather than field edits; `new_value` carries the
-    /// reason, so a bot's whole life is one ordered list.
-    Created,
-    Deleted,
-}
-
-impl RevisionField {
-    pub fn as_str(self) -> &'static str {
-        match self {
-            RevisionField::Name => "name",
-            RevisionField::Avatar => "avatar",
-            RevisionField::Description => "description",
-            RevisionField::Instructions => "instructions",
-            RevisionField::Created => "created",
-            RevisionField::Deleted => "deleted",
-        }
-    }
-
-    pub fn parse(s: &str) -> Option<Self> {
-        match s {
-            "name" => Some(RevisionField::Name),
-            "avatar" => Some(RevisionField::Avatar),
-            "description" => Some(RevisionField::Description),
-            "instructions" => Some(RevisionField::Instructions),
-            "created" => Some(RevisionField::Created),
-            "deleted" => Some(RevisionField::Deleted),
-            _ => None,
-        }
-    }
-
-    /// Whether reverting this entry means writing `old_value` back. Lifecycle
-    /// markers are history, not state, so they are not revertible.
-    pub fn is_revertible(self) -> bool {
-        !matches!(self, RevisionField::Created | RevisionField::Deleted)
-    }
-}
-
 /// A bot's DM thread. Every bot has exactly one, created with it.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Conversation {

crates/bus/src/types/peers.rs

@@ -0,0 +1,44 @@
+//! Peer daemons: another Gravity daemon of the same owner whose bots this
+//! one can exchange messages with. See `docs/peer-bots.md`.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use super::Id;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Peer {
+    pub id: Id,
+    /// The owner's name for the other machine, e.g. `win-pc`.
+    pub name: String,
+    /// The other daemon's stable id, learned at the first handshake. A token
+    /// is bound to the first daemon that presents it.
+    pub daemon_id: Option<String>,
+    /// Where to dial, on the side that dials. `None` on the side that listens.
+    pub url: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub last_seen_at: Option<DateTime<Utc>>,
+    pub revoked_at: Option<DateTime<Utc>>,
+}
+
+/// Largest file sent with a forwarded result.
+pub const MAX_PEER_ARTIFACT_BYTES: u64 = 16 * 1024 * 1024;
+
+/// Largest total of files sent with one forwarded result.
+pub const MAX_PEER_ARTIFACTS_TOTAL_BYTES: u64 = 48 * 1024 * 1024;
+
+/// A bot as a peer describes it: enough to stand it in as a linked bot.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RemoteBot {
+    pub id: Id,
+    pub name: String,
+    #[serde(default)]
+    pub description: String,
+    #[serde(default)]
+    pub avatar: String,
+    #[serde(default)]
+    pub runtime: super::BotRuntime,
+    /// The project it works in on its own daemon. Informational only.
+    #[serde(default)]
+    pub project: String,
+}

crates/bus/src/types/revisions.rs

@@ -0,0 +1,64 @@
+//! The bot identity audit trail.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use super::Id;
+
+/// One field-level change to a bot's identity: the audit trail that makes
+/// unsupervised self-management reversible.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BotRevision {
+    pub id: Id,
+    pub bot_id: Id,
+    /// `user` or `bot:<id>`.
+    pub changed_by: String,
+    pub field: RevisionField,
+    pub old_value: String,
+    pub new_value: String,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RevisionField {
+    Name,
+    Avatar,
+    Description,
+    Instructions,
+    /// Lifecycle markers rather than field edits; `new_value` carries the
+    /// reason, so a bot's whole life is one ordered list.
+    Created,
+    Deleted,
+}
+
+impl RevisionField {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            RevisionField::Name => "name",
+            RevisionField::Avatar => "avatar",
+            RevisionField::Description => "description",
+            RevisionField::Instructions => "instructions",
+            RevisionField::Created => "created",
+            RevisionField::Deleted => "deleted",
+        }
+    }
+
+    pub fn parse(s: &str) -> Option<Self> {
+        match s {
+            "name" => Some(RevisionField::Name),
+            "avatar" => Some(RevisionField::Avatar),
+            "description" => Some(RevisionField::Description),
+            "instructions" => Some(RevisionField::Instructions),
+            "created" => Some(RevisionField::Created),
+            "deleted" => Some(RevisionField::Deleted),
+            _ => None,
+        }
+    }
+
+    /// Whether reverting this entry means writing `old_value` back. Lifecycle
+    /// markers are history, not state, so they are not revertible.
+    pub fn is_revertible(self) -> bool {
+        !matches!(self, RevisionField::Created | RevisionField::Deleted)
+    }
+}

crates/gravityd/Cargo.toml

@@ -29,6 +29,7 @@ hex = { workspace = true }
 futures = { workspace = true }
 bytes = { workspace = true }
 tokio-tungstenite = "0.24"
+base64 = "0.22"
 
 [dev-dependencies]
 tempfile = "3"

crates/gravityd/src/app.rs

@@ -39,6 +39,9 @@ pub struct AppState {
     /// Runtime override for `cfg.auto_compact_window`, shared with the
     /// supervisor and persisted in the `meta` table.
     pub auto_compact: AutoCompactOverride,
+    /// Live links to peer daemons, shared by the delivery worker (forwarding
+    /// to linked bots) and the control plane (pairing and linking).
+    pub peers: crate::peer::PeerHub,
     pub started_at: Instant,
     /// Wall-clock start, kept alongside the monotonic `started_at` purely so
     /// [`AppState::stale_build`] can compare it against a file mtime.
@@ -75,6 +78,7 @@ impl AppState {
             secrets,
             supervisor,
             auto_compact,
+            peers: crate::peer::PeerHub::default(),
             started_at: Instant::now(),
             started_wall: SystemTime::now(),
         }))

crates/gravityd/src/botmgmt.rs

@@ -185,6 +185,10 @@ pub fn apply_identity_edit(
 
 /// Rewrite the daemon-owned files for a bot after its identity changed.
 pub fn reprovision(app: &Arc<AppState>, bot: &Bot) -> anyhow::Result<()> {
+    // A linked bot's files live on its peer; there is nothing to write here.
+    if bot.is_linked() {
+        return Ok(());
+    }
     let project = app
         .db
         .get_project(&bot.project_id)?
@@ -199,7 +203,12 @@ pub fn reprovision(app: &Arc<AppState>, bot: &Bot) -> anyhow::Result<()> {
 /// `CLAUDE.md` pick up the new layout. The write is content-hash guarded, so
 /// this is a no-op once each bot is current.
 pub fn regenerate_all_system_md(app: &Arc<AppState>) -> anyhow::Result<()> {
-    for bot in app.db.list_bots(None)? {
+    for bot in app
+        .db
+        .list_bots(None)?
+        .into_iter()
+        .filter(|b| !b.is_linked())
+    {
         if let Err(e) = reprovision(app, &bot) {
             tracing::warn!(bot_id = %bot.id, error = %e, "system.md regeneration failed");
         }

crates/gravityd/src/db.rs

@@ -26,6 +26,7 @@ mod decisions_list;
 mod decisions_tests;
 mod deliveries;
 mod devices;
+mod peers;
 mod projects;
 mod revisions;
 mod routines;

crates/gravityd/src/db/bot_runtime.rs

@@ -57,6 +57,8 @@ impl Db {
             dir_name: dir_name.to_string(),
             created_by_bot_id: created_by_bot_id.map(|s| s.to_string()),
             deleted_at: None,
+            peer_id: None,
+            remote_bot_id: None,
             created_at: now(),
         };
         let conn = self.lock();

crates/gravityd/src/db/bots.rs

@@ -18,7 +18,7 @@ const TOMBSTONE_SEP: char = '#';
 impl Db {
     // ---- bots ----
 
-    fn bot_from_row(r: &Row<'_>) -> rusqlite::Result<Bot> {
+    pub(super) fn bot_from_row(r: &Row<'_>) -> rusqlite::Result<Bot> {
         Ok(Bot {
             id: r.get(0)?,
             project_id: r.get(1)?,
@@ -35,15 +35,19 @@ impl Db {
                 "codex_cli" => BotRuntime::CodexCli,
                 _ => BotRuntime::ClaudeCode,
             },
+            peer_id: r.get(12)?,
+            remote_bot_id: r.get(13)?,
             // Runtime fields are overlaid by the supervisor.
             state: BotState::Stopped,
             state_reason: String::new(),
             unread_count: 0,
         })
     }
 
-    const BOT_COLS: &'static str = "id, project_id, name, description, avatar, instructions, \
-         workspace_path, created_at, dir_name, created_by_bot_id, deleted_at, runtime";
+    pub(super) const BOT_COLS: &'static str =
+        "id, project_id, name, description, avatar, instructions, \
+         workspace_path, created_at, dir_name, created_by_bot_id, deleted_at, runtime, peer_id, \
+         remote_bot_id";
 
     /// Restricts a query to bots that still exist for addressing purposes.
     const LIVE: &'static str = "deleted_at IS NULL";