From 1f79522ecb1e2bc896a261c9580f7c8d866de155 Mon Sep 17 00:00:00 2001 From: Manuel Rinaldi Date: Thu, 1 Oct 2026 17:55:01 +0300 Subject: [PATCH 1/2] docs: design peer bots across two daemons Co-Authored-By: Claude Opus 5.5 --- docs/peer-bots.md | 227 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 227 insertions(+) create mode 100644 docs/peer-bots.md diff --git a/docs/peer-bots.md b/docs/peer-bots.md new file mode 100644 index 0000000..b40de06 --- /dev/null +++ b/docs/peer-bots.md @@ -0,0 +1,227 @@ +# Peer bots: one team across two machines + +Status: design, not yet implemented. + +A team can span two Gravity daemons. The motivating case is an app built for +macOS and Windows: a lead and a Mac developer run on the Mac, a Windows +developer runs on a Windows PC, and the lead hands the Windows side its work +without the user switching the desktop app between daemons to relay messages +or fetch files. + +Each bot still runs, with its own terminal and runtime, on the machine that +owns it. What crosses machines is the bus: messages, tasks, results, and the +artifacts those results attach. + +## Assumptions + +- Both daemons belong to one owner and reach each other over Tailscale. Like + remote desktop clients today, the link is plain `ws://` on the tailnet; the + tailnet is the transport security, and a peer token is the authentication. +- Two machines is the target. Nothing below limits a daemon to one peer, but + multi-peer routing (A relaying for B to C) is out of scope: messages only + ever travel one link. +- The desktop UI is due for a rework, so this version adds the minimum UI to + pair, link, and tell a linked bot apart, and leaves the rest for the rework. + +## Concepts + +**Peer.** Another daemon this one is paired with. Stored in a `peer` table: +local id, display name (`win-pc`), the remote daemon's stable id, the URL to +dial if this side dials, and `last_seen_at`. Each daemon gets a stable +`daemon_id` in `meta` the first time it starts. + +**Linked bot.** A `bot` row whose `peer_id` is set. It stands in for a bot +that runs on the peer, with the peer's id for it in `remote_bot_id`. It has no +workspace and no runtime: the supervisor never starts it. Everything else +treats it as a bot in the project. It appears in `list_bots`, is addressed by +name, owns a DM conversation, and is the `from`/`to` of task rows. Because it is +a real row, every foreign key and guardrail that assumes a local bot keeps +working unchanged. + +Linking is symmetric. When the Mac lead first messages the Windows developer, +the Windows daemon creates a linked bot for the lead in the Windows +developer's project. The Windows developer's replies and results are then +ordinary sends to a local bot, which the Windows delivery worker forwards +back. + +## Pairing + +1. On the Windows daemon (which must bind its Tailscale address, as for any + remote client), the owner creates an invite. It returns a one-time code + with the daemon's tailnet URL and a fresh peer token. +2. On the Mac, the owner adds the peer with that code. The Mac daemon stores + the URL and token (`secrets/peer-.token`), dials, and both sides + record each other's `daemon_id` and name. +3. Revoking a peer on either side deletes its token, and that side's linked + bots stop accepting deliveries (they fail with "peer revoked"). + +The dialing side holds one long-lived WebSocket to `/peer` on the other, and +traffic flows both ways over it. Only the listening daemon needs to be +reachable. The dialer reconnects with backoff, and `last_seen_at` drives the +online/offline badge. + +A peer token authenticates only the `/peer` route. It carries no control-plane +grants: a peer cannot list projects, attach terminals, rule on decisions, or +reach a bot that has not been linked to it. + +## Linking a bot + +The owner picks a project on the Mac, a peer, and one of the peer's bots (the +peer answers a `list_bots` frame with id, name, description, and runtime). The +Mac creates the linked bot under the remote bot's name. If the name is already +taken in the project, the link is refused until one of the bots is renamed, +since names are what bots address each other by. + +The peer records the link: the remote bot accepts frames from this peer only +once it is linked. That is the exposure boundary. Linking a bot exposes that +one bot to the peer, and nothing else in its project. + +Unlinking archives the linked bot locally, like deleting any bot: its open +tasks are cancelled and its history is kept. + +## Message flow + +Sending is unchanged up to the delivery worker. `send_message`, `send_user_message`, +`complete_task`, and `cancel_task` insert messages, open or close task rows, and +enqueue a delivery to the recipient bot. The single new branch is in +`DeliveryWorker::attempt`: when the recipient is a linked bot, the worker +forwards the message over the peer link instead of calling +`Supervisor::deliver`. + +- Peer offline: `NotReady`, so the delivery waits without spending attempts, + exactly like a bot that is still starting. +- Peer rejected the frame (bot unlinked, task unknown, peer revoked): + `Failed`, with the peer's reason as the delivery error. +- Peer acknowledged: `delivered`. The ack means the peer has durably stored + the message and enqueued its own local delivery. It does not mean the remote + bot has read it, which matches what `delivered` means locally. + +A forwarded frame carries: + +```json +{ + "type": "message", + "id": "", + "to_bot_id": "", + "from": { "kind": "bot", "bot_id": "", "name": "lead" }, + "kind": "task | reply | note | done | chat", + "body": "…", + "ref": "", + "task": { "id": "", "deadline_at": "…", "hop_count": 2 }, + "closes_task": { "id": "", "state": "done | cancelled" }, + "artifacts": [{ "name": "build.log", "bytes": "" }] +} +``` + +The receiver handles it in one transaction: + +1. **Dedupe.** `(peer_id, id)` in `peer_message` means it was already stored, + so the receiver acks again and does nothing else. +2. **Resolve the recipient.** `to_bot_id` must be a live, local, non-linked + bot linked to this peer. +3. **Resolve the sender.** `kind: bot` uses the linked bot for + `(peer, from.bot_id)`, creating it in the recipient's project on first + contact. `kind: user` is the owner speaking from the other machine; it + arrives as `chat` with the user's authority, which is the reason peer + tokens are only issued between the owner's own daemons. +4. **Insert and enqueue** with `messaging::send_dm`, mapping `ref` through + `peer_message` so threading survives the hop. +5. **Mirror the task.** A `task` frame opens a local task from the linked + sender to the recipient. It keeps the sender's `deadline_at`, and its + `hop_count` is the sender's, so the hop limit holds across the whole chain. + `peer_task` maps the two task ids. A `closes_task` frame flips the mapped + local task (`try_close_task`) before inserting the `done` or cancel note, + exactly as the local tool would. + +Replies count against the reply budget on both sides, because each side holds +its own copy of the task. Loops are refused locally by the existing chain +check: the Windows developer cannot delegate back to the lead's linked bot +while working the lead's task, because that linked bot is already in the +chain. Deadlines expire independently on both sides from the same timestamp. + +### Artifacts + +`complete_task` already lists artifact paths in the `done` body. When the +recipient is a linked bot, the forwarder also reads each listed file and +sends its bytes in the frame. Only regular files under the sending project's +artifacts directory or the sending bot's workspace are sent. The cap is +16 MiB per file and 48 MiB per message. A file over the cap stays listed by its +remote path, with a note saying it was not transferred. + +The receiver writes the files to +`/peers///` and rewrites the listing in +the stored body to those local paths. The lead reads the result exactly as it +would read a local one. + +## What bots are told + +- `list_bots` adds `"machine": ""` and `"online": bool` for linked + bots. +- The envelope header names the machine: `from windows-dev @ win-pc`. The + sender is still daemon-authenticated, but the authority is a peer bot's, + never the user's, unless the frame came from the owner (`kind: user`). +- The system prompt's bus section says a linked bot runs on another machine + and that paths it mentions outside a transferred artifact are not readable + here. + +## Desktop app (minimum for this version) + +- Settings → Peers: create an invite, add a peer from a code, see each peer's + status, and revoke a peer. +- Project → Link a bot from a peer. +- A linked bot shows its machine and online state in the bot list. Selecting + it shows its conversation and the composer instead of a terminal. Sending to + it from the composer is a forwarded `chat`. + +Watching the Windows bot's terminal from the Mac is left to the UI rework. +It needs the app to hold connections to two daemons at once. + +## Schema + +One append-only migration: + +```sql +CREATE TABLE peer ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + daemon_id TEXT UNIQUE, -- learned at first handshake + url TEXT, -- set on the dialing side + 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 ( -- local bots exposed to a peer + 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 ( -- dedupe and ref mapping + 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 TABLE peer_task ( -- the two halves of a mirrored 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) +); +``` + +## Delivery plan + +1. **Peers and forwarding (daemon).** The schema, `daemon_id`, pairing and + revocation as control-plane requests (plus `gravityd peer …` commands that + drive the local daemon, so this is usable before the UI lands), the `/peer` + route and dialer, linking, and forwarding of every message kind with task + mirroring. Tested with two in-process daemons on the double runtime. +2. **Artifacts.** File transfer on `done`, with path rewriting and the caps. +3. **Desktop app.** The Peers settings, link flow, and linked-bot view. +4. **Live check.** Mac and Windows over Tailscale with real runtimes, then a + `bus-live-test` mode for two daemons. From e052d4cccb9375c6c3b5ad4a098487441e7bed67 Mon Sep 17 00:00:00 2001 From: Manuel Rinaldi Date: Thu, 1 Oct 2026 18:11:18 +0300 Subject: [PATCH 2/2] feat(peer): link bots across two daemons Pair two of the owner's daemons over Tailscale and stand a bot from one in as a linked bot in a project on the other. A linked bot is an ordinary bot row with peer_id set, so addressing, task budgets and history work unchanged; the delivery worker forwards deliveries for it over the /peer link instead of writing to a local session. The receiving daemon stores each forwarded message as if the sender were local: a task is mirrored with the sender's deadline and hop count, a result or cancel closes the mirrored task, and the sender gets a linked bot of its own on first contact so replies route back. Files listed in a result travel with it (from the bot's own directory or the project artifacts only, size-capped) and land in the recipient project's artifacts directory. Pairing is a one-time invite holding a link token bound to the first daemon that presents it. Peers can only list bots, link one, and deliver to bots linked to them; revoking drops the link. `gravityd peer ...` drives pairing and linking until the desktop UI lands. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/bus/src/schema.rs | 3 + crates/bus/src/schema/peers.rs | 46 +++ crates/bus/src/types.rs | 4 + crates/bus/src/types/entities.rs | 72 +---- crates/bus/src/types/peers.rs | 44 +++ crates/bus/src/types/revisions.rs | 64 ++++ crates/gravityd/Cargo.toml | 1 + crates/gravityd/src/app.rs | 4 + crates/gravityd/src/botmgmt.rs | 11 +- crates/gravityd/src/db.rs | 1 + crates/gravityd/src/db/bot_runtime.rs | 2 + crates/gravityd/src/db/bots.rs | 10 +- crates/gravityd/src/db/peers.rs | 292 ++++++++++++++++++ crates/gravityd/src/delivery.rs | 90 ++++-- crates/gravityd/src/lib.rs | 1 + crates/gravityd/src/main.rs | 4 + crates/gravityd/src/mcp/tools.rs | 14 +- crates/gravityd/src/paths/prompt.rs | 6 + crates/gravityd/src/peer.rs | 19 ++ crates/gravityd/src/peer/artifacts.rs | 183 +++++++++++ crates/gravityd/src/peer/cli.rs | 197 ++++++++++++ crates/gravityd/src/peer/forward.rs | 188 ++++++++++++ crates/gravityd/src/peer/frames.rs | 74 +++++ crates/gravityd/src/peer/hub.rs | 182 +++++++++++ crates/gravityd/src/peer/inbound.rs | 76 +++++ crates/gravityd/src/peer/receive.rs | 202 ++++++++++++ crates/gravityd/src/peer/socket.rs | 237 ++++++++++++++ crates/gravityd/src/secrets.rs | 56 ++++ crates/gravityd/src/server.rs | 4 + crates/gravityd/src/supervisor/lifecycle.rs | 5 +- crates/gravityd/src/ws.rs | 1 + crates/gravityd/src/ws/dispatch.rs | 8 + crates/gravityd/src/ws/peers.rs | 197 ++++++++++++ crates/gravityd/src/ws/views.rs | 24 +- crates/gravityd/tests/peer_bots.rs | 322 ++++++++++++++++++++ docs/architecture.md | 4 +- docs/peer-bots.md | 29 +- docs/protocol.md | 6 + third-party/inventory.json | 4 +- 40 files changed, 2594 insertions(+), 94 deletions(-) create mode 100644 crates/bus/src/schema/peers.rs create mode 100644 crates/bus/src/types/peers.rs create mode 100644 crates/bus/src/types/revisions.rs create mode 100644 crates/gravityd/src/db/peers.rs create mode 100644 crates/gravityd/src/peer.rs create mode 100644 crates/gravityd/src/peer/artifacts.rs create mode 100644 crates/gravityd/src/peer/cli.rs create mode 100644 crates/gravityd/src/peer/forward.rs create mode 100644 crates/gravityd/src/peer/frames.rs create mode 100644 crates/gravityd/src/peer/hub.rs create mode 100644 crates/gravityd/src/peer/inbound.rs create mode 100644 crates/gravityd/src/peer/receive.rs create mode 100644 crates/gravityd/src/peer/socket.rs create mode 100644 crates/gravityd/src/ws/peers.rs create mode 100644 crates/gravityd/tests/peer_bots.rs diff --git a/Cargo.lock b/Cargo.lock index 734722b..600e137 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -520,6 +520,7 @@ version = "0.13.0" dependencies = [ "anyhow", "axum", + "base64", "bus", "bytes", "chrono", diff --git a/crates/bus/src/schema.rs b/crates/bus/src/schema.rs index 10283ed..89faee7 100644 --- a/crates/bus/src/schema.rs +++ b/crates/bus/src/schema.rs @@ -7,6 +7,7 @@ mod base; mod decisions; mod history; +mod peers; use base::MIGRATION_1; use decisions::MIGRATION_12; @@ -14,6 +15,7 @@ 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, ]; diff --git a/crates/bus/src/schema/peers.rs b/crates/bus/src/schema/peers.rs new file mode 100644 index 0000000..511f916 --- /dev/null +++ b/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); +"#; diff --git a/crates/bus/src/types.rs b/crates/bus/src/types.rs index 3fc3b5b..e558bfc 100644 --- a/crates/bus/src/types.rs +++ b/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; diff --git a/crates/bus/src/types/entities.rs b/crates/bus/src/types/entities.rs index a86827a..5fc2d14 100644 --- a/crates/bus/src/types/entities.rs +++ b/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>, + /// 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, + /// The linked bot's id on its peer. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub remote_bot_id: Option, pub created_at: DateTime, } +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:`. - pub changed_by: String, - pub field: RevisionField, - pub old_value: String, - pub new_value: String, - pub created_at: DateTime, -} - -#[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 { - 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 { diff --git a/crates/bus/src/types/peers.rs b/crates/bus/src/types/peers.rs new file mode 100644 index 0000000..dc79b28 --- /dev/null +++ b/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, + /// Where to dial, on the side that dials. `None` on the side that listens. + pub url: Option, + pub created_at: DateTime, + pub last_seen_at: Option>, + pub revoked_at: Option>, +} + +/// 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, +} diff --git a/crates/bus/src/types/revisions.rs b/crates/bus/src/types/revisions.rs new file mode 100644 index 0000000..55040be --- /dev/null +++ b/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:`. + pub changed_by: String, + pub field: RevisionField, + pub old_value: String, + pub new_value: String, + pub created_at: DateTime, +} + +#[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 { + 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) + } +} diff --git a/crates/gravityd/Cargo.toml b/crates/gravityd/Cargo.toml index 4ffca78..4f9e980 100644 --- a/crates/gravityd/Cargo.toml +++ b/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" diff --git a/crates/gravityd/src/app.rs b/crates/gravityd/src/app.rs index be3448a..5713890 100644 --- a/crates/gravityd/src/app.rs +++ b/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(), })) diff --git a/crates/gravityd/src/botmgmt.rs b/crates/gravityd/src/botmgmt.rs index ca2fb7c..69a7659 100644 --- a/crates/gravityd/src/botmgmt.rs +++ b/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, 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, 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) -> 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"); } diff --git a/crates/gravityd/src/db.rs b/crates/gravityd/src/db.rs index 4bdafd5..227adc2 100644 --- a/crates/gravityd/src/db.rs +++ b/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; diff --git a/crates/gravityd/src/db/bot_runtime.rs b/crates/gravityd/src/db/bot_runtime.rs index aa1698b..248cdcc 100644 --- a/crates/gravityd/src/db/bot_runtime.rs +++ b/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(); diff --git a/crates/gravityd/src/db/bots.rs b/crates/gravityd/src/db/bots.rs index 0241812..2782fa2 100644 --- a/crates/gravityd/src/db/bots.rs +++ b/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 { + pub(super) fn bot_from_row(r: &Row<'_>) -> rusqlite::Result { Ok(Bot { id: r.get(0)?, project_id: r.get(1)?, @@ -35,6 +35,8 @@ 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(), @@ -42,8 +44,10 @@ impl Db { }) } - 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"; diff --git a/crates/gravityd/src/db/peers.rs b/crates/gravityd/src/db/peers.rs new file mode 100644 index 0000000..aa7a391 --- /dev/null +++ b/crates/gravityd/src/db/peers.rs @@ -0,0 +1,292 @@ +//! Peer daemons, linked bots, and the id mappings for messages and tasks that +//! cross a peer link. See `docs/peer-bots.md`. + +use bus::*; +use rusqlite::{params, OptionalExtension, Row}; + +use super::{parse_ts, ts, Db}; + +/// `meta` key holding this daemon's stable id, which peers bind a token to. +const DAEMON_ID_KEY: &str = "daemon_id"; + +impl Db { + // ---- this daemon ---- + + /// This daemon's stable id, created the first time it is asked for. + pub fn daemon_id(&self) -> anyhow::Result { + if let Some(id) = self.get_meta(DAEMON_ID_KEY)? { + return Ok(id); + } + let id = new_id(); + self.set_meta(DAEMON_ID_KEY, &id)?; + Ok(id) + } + + // ---- peers ---- + + fn peer_from_row(r: &Row<'_>) -> rusqlite::Result { + Ok(Peer { + id: r.get(0)?, + name: r.get(1)?, + daemon_id: r.get(2)?, + url: r.get(3)?, + created_at: parse_ts(&r.get::<_, String>(4)?), + last_seen_at: r.get::<_, Option>(5)?.map(|s| parse_ts(&s)), + revoked_at: r.get::<_, Option>(6)?.map(|s| parse_ts(&s)), + }) + } + + const PEER_COLS: &'static str = + "id, name, daemon_id, url, created_at, last_seen_at, revoked_at"; + + pub fn create_peer(&self, name: &str, url: Option<&str>) -> anyhow::Result { + let peer = Peer { + id: new_id(), + name: name.to_string(), + daemon_id: None, + url: url.map(str::to_string), + created_at: now(), + last_seen_at: None, + revoked_at: None, + }; + self.lock().execute( + "INSERT INTO peer(id, name, url, created_at) VALUES (?1, ?2, ?3, ?4)", + params![peer.id, peer.name, peer.url, ts(peer.created_at)], + )?; + Ok(peer) + } + + pub fn list_peers(&self) -> anyhow::Result> { + let conn = self.lock(); + let mut stmt = conn.prepare(&format!( + "SELECT {} FROM peer ORDER BY name", + Self::PEER_COLS + ))?; + let rows = stmt.query_map([], Self::peer_from_row)?; + Ok(rows.collect::>()?) + } + + pub fn get_peer(&self, peer_id: &str) -> anyhow::Result> { + let conn = self.lock(); + Ok(conn + .query_row( + &format!("SELECT {} FROM peer WHERE id = ?1", Self::PEER_COLS), + params![peer_id], + Self::peer_from_row, + ) + .optional()?) + } + + pub fn get_peer_by_name(&self, name: &str) -> anyhow::Result> { + let conn = self.lock(); + Ok(conn + .query_row( + &format!( + "SELECT {} FROM peer WHERE name = ?1 COLLATE NOCASE", + Self::PEER_COLS + ), + params![name], + Self::peer_from_row, + ) + .optional()?) + } + + /// Binds the peer to the daemon that presented its token. The first + /// daemon wins; any other daemon presenting the same token is refused. + pub fn bind_peer_daemon(&self, peer_id: &str, daemon_id: &str) -> anyhow::Result { + let conn = self.lock(); + conn.execute( + "UPDATE peer SET daemon_id = ?2 WHERE id = ?1 AND daemon_id IS NULL", + params![peer_id, daemon_id], + )?; + let bound: Option = conn.query_row( + "SELECT daemon_id FROM peer WHERE id = ?1", + params![peer_id], + |r| r.get(0), + )?; + Ok(bound.as_deref() == Some(daemon_id)) + } + + pub fn touch_peer(&self, peer_id: &str) -> anyhow::Result<()> { + self.lock().execute( + "UPDATE peer SET last_seen_at = ?2 WHERE id = ?1", + params![peer_id, ts(now())], + )?; + Ok(()) + } + + pub fn revoke_peer(&self, peer_id: &str) -> anyhow::Result { + let changed = self.lock().execute( + "UPDATE peer SET revoked_at = ?2 WHERE id = ?1 AND revoked_at IS NULL", + params![peer_id, ts(now())], + )?; + Ok(changed > 0) + } + + // ---- links ---- + + /// Exposes a local bot to a peer: the peer may now deliver to it. + pub fn expose_bot_to_peer(&self, peer_id: &str, bot_id: &str) -> anyhow::Result<()> { + self.lock().execute( + "INSERT OR IGNORE INTO peer_link(peer_id, bot_id) VALUES (?1, ?2)", + params![peer_id, bot_id], + )?; + Ok(()) + } + + pub fn is_exposed_to_peer(&self, peer_id: &str, bot_id: &str) -> anyhow::Result { + let conn = self.lock(); + let found: Option = conn + .query_row( + "SELECT 1 FROM peer_link WHERE peer_id = ?1 AND bot_id = ?2", + params![peer_id, bot_id], + |r| r.get(0), + ) + .optional()?; + Ok(found.is_some()) + } + + /// The live linked bot standing in for `remote_bot_id` in a project. + pub fn linked_bot( + &self, + peer_id: &str, + remote_bot_id: &str, + project_id: &str, + ) -> anyhow::Result> { + let conn = self.lock(); + let sql = format!( + "SELECT {} FROM bot WHERE peer_id = ?1 AND remote_bot_id = ?2 AND project_id = ?3 \ + AND deleted_at IS NULL", + Self::BOT_COLS + ); + Ok(conn + .query_row( + &sql, + params![peer_id, remote_bot_id, project_id], + Self::bot_from_row, + ) + .optional()?) + } + + /// A bot row standing in for a bot that runs on `peer_id`. It gets a DM + /// conversation like any bot, an empty workspace path, and a directory + /// name no local bot can have, so nothing ever provisions files for it. + pub fn create_linked_bot( + &self, + project_id: &str, + remote: &RemoteBot, + peer_id: &str, + ) -> anyhow::Result { + let dir_name = format!(".peer-{}", &new_id()[..8]); + let mut bot = self.create_bot_with_runtime( + project_id, + &remote.name, + &remote.description, + "", + &remote.avatar, + "", + &dir_name, + None, + remote.runtime, + )?; + self.lock().execute( + "UPDATE bot SET peer_id = ?2, remote_bot_id = ?3 WHERE id = ?1", + params![bot.id, peer_id, remote.id], + )?; + bot.peer_id = Some(peer_id.to_string()); + bot.remote_bot_id = Some(remote.id.clone()); + Ok(bot) + } + + // ---- id mappings ---- + + pub fn map_peer_message( + &self, + peer_id: &str, + remote_message_id: &str, + message_id: &str, + ) -> anyhow::Result<()> { + self.lock().execute( + "INSERT OR IGNORE INTO peer_message(peer_id, remote_message_id, message_id) + VALUES (?1, ?2, ?3)", + params![peer_id, remote_message_id, message_id], + )?; + Ok(()) + } + + /// The local copy of a message the peer knows as `remote_message_id`. + pub fn local_message_for( + &self, + peer_id: &str, + remote_message_id: &str, + ) -> anyhow::Result> { + self.peer_lookup( + "SELECT message_id FROM peer_message WHERE peer_id = ?1 AND remote_message_id = ?2", + peer_id, + remote_message_id, + ) + } + + /// The peer's id for a local message, when the message crossed the link. + pub fn remote_message_for( + &self, + peer_id: &str, + message_id: &str, + ) -> anyhow::Result> { + self.peer_lookup( + "SELECT remote_message_id FROM peer_message WHERE peer_id = ?1 AND message_id = ?2", + peer_id, + message_id, + ) + } + + pub fn map_peer_task( + &self, + peer_id: &str, + remote_task_id: &str, + task_id: &str, + ) -> anyhow::Result<()> { + self.lock().execute( + "INSERT OR IGNORE INTO peer_task(peer_id, remote_task_id, task_id) + VALUES (?1, ?2, ?3)", + params![peer_id, remote_task_id, task_id], + )?; + Ok(()) + } + + /// The peer's id for the other half of a mirrored task. + pub fn remote_task_for(&self, peer_id: &str, task_id: &str) -> anyhow::Result> { + self.peer_lookup( + "SELECT remote_task_id FROM peer_task WHERE peer_id = ?1 AND task_id = ?2", + peer_id, + task_id, + ) + } + + fn peer_lookup(&self, sql: &str, peer_id: &str, key: &str) -> anyhow::Result> { + let conn = self.lock(); + Ok(conn + .query_row(sql, params![peer_id, key], |r| r.get(0)) + .optional()?) + } +} + +impl Db { + /// Every task a message opened, whatever its state. A result or a cancel + /// crossing the link has to find the task it closes after the close. + pub fn tasks_with_origin(&self, origin_message_id: &str) -> anyhow::Result> { + let ids: Vec = { + let conn = self.lock(); + let mut stmt = conn.prepare("SELECT id FROM task WHERE origin_message_id = ?1")?; + let rows = stmt.query_map(params![origin_message_id], |r| r.get(0))?; + rows.collect::>()? + }; + let mut tasks = Vec::with_capacity(ids.len()); + for id in ids { + if let Some(task) = self.get_task(&id)? { + tasks.push(task); + } + } + Ok(tasks) + } +} diff --git a/crates/gravityd/src/delivery.rs b/crates/gravityd/src/delivery.rs index 90384d4..08491e1 100644 --- a/crates/gravityd/src/delivery.rs +++ b/crates/gravityd/src/delivery.rs @@ -3,10 +3,13 @@ //! consumers deduplicate by delivery id / idempotency key. A bot that is not //! ready (stopped, socket unknown) defers without consuming retry attempts. +use std::sync::Arc; + use bus::envelope::{render_message, render_routine}; -use bus::DeliveryState; +use bus::{DeliveryState, Message}; use chrono::Duration as ChronoDuration; +use crate::app::AppState; use crate::config::DeliveryConfig; use crate::db::Db; use crate::events::{Events, Push}; @@ -18,6 +21,8 @@ use crate::supervisor::Supervisor; const MAX_NOT_READY_BACKOFF_SECONDS: i64 = 60; pub struct DeliveryWorker { + /// For forwarding to linked bots, which runs through the peer link. + pub app: Arc, pub db: Db, pub supervisor: Supervisor, pub events: Events, @@ -31,22 +36,22 @@ impl DeliveryWorker { tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tick.tick().await; - if let Err(e) = self.step() { + if let Err(e) = self.step().await { tracing::warn!(error = %e, "delivery worker step failed"); } } } - pub fn step(&self) -> anyhow::Result<()> { + pub async fn step(&self) -> anyhow::Result<()> { self.db.recover_expired_leases()?; let due = self.db.lease_due_deliveries(self.cfg.lease_seconds, 20)?; for delivery in due { - self.attempt(&delivery.id)?; + self.attempt(&delivery.id).await?; } Ok(()) } - fn attempt(&self, delivery_id: &str) -> anyhow::Result<()> { + async fn attempt(&self, delivery_id: &str) -> anyhow::Result<()> { let Some(delivery) = self.db.get_delivery(delivery_id)? else { return Ok(()); }; @@ -55,27 +60,26 @@ impl DeliveryWorker { .mark_delivery_retry(delivery_id, "message missing", chrono::Utc::now(), 0)?; return Ok(()); }; - // A decision notice was already rendered by the registry, which owns - // its header and its phase. Wrapping it in `[msg #N …]` would bury the - // one thing that makes it authority: that it comes from the owner. - let text = if msg.decision_id.is_some() { - msg.body.clone() - } else if msg.sender.kind == bus::SenderKind::Routine { - let run_id = self.db.run_id_for_delivery(delivery_id)?; - render_routine(&msg.sender.name, msg.num, run_id.as_deref(), &msg.body) - } else { - let ref_num = match &msg.ref_message_id { - Some(rid) => self.db.get_message(rid)?.map(|m| m.num), - None => None, - }; - let task_id = self - .db - .open_task_for_message(&msg.id, &delivery.bot_id)? - .map(|t| t.id); - render_message(&msg, ref_num, task_id.as_deref()) + let outcome = match self.db.get_bot(&delivery.bot_id)? { + // A bot that runs on a peer: the peer stores and delivers it. + Some(target) if target.is_linked() => { + match crate::peer::forward(&self.app, &target, &msg).await { + Ok(()) => Ok(()), + Err(crate::peer::ForwardError::Offline) => Err( + crate::supervisor::DeliverError::NotReady("peer offline".to_string()), + ), + Err(crate::peer::ForwardError::Failed(e)) => { + Err(crate::supervisor::DeliverError::Failed(e)) + } + } + } + _ => { + let text = self.render(delivery_id, &delivery.bot_id, &msg)?; + self.supervisor.deliver(&delivery.bot_id, &text) + } }; - match self.supervisor.deliver(&delivery.bot_id, &text) { + match outcome { Ok(()) => { self.db.mark_delivered(delivery_id)?; tracing::info!(delivery_id, bot_id = %delivery.bot_id, message_id = %msg.id, "delivered"); @@ -120,4 +124,42 @@ impl DeliveryWorker { } Ok(()) } + + /// The envelope a local session reads for `msg`. + fn render(&self, delivery_id: &str, bot_id: &str, msg: &Message) -> anyhow::Result { + // A decision notice was already rendered by the registry, which owns + // its header and its phase. Wrapping it in `[msg #N …]` would bury the + // one thing that makes it authority: that it comes from the owner. + Ok(if msg.decision_id.is_some() { + msg.body.clone() + } else if msg.sender.kind == bus::SenderKind::Routine { + let run_id = self.db.run_id_for_delivery(delivery_id)?; + render_routine(&msg.sender.name, msg.num, run_id.as_deref(), &msg.body) + } else { + let ref_num = match &msg.ref_message_id { + Some(rid) => self.db.get_message(rid)?.map(|m| m.num), + None => None, + }; + let task_id = self + .db + .open_task_for_message(&msg.id, bot_id)? + .map(|t| t.id); + render_message(&self.sender_view(msg)?, ref_num, task_id.as_deref()) + }) + } + + /// A linked sender is named with its machine, so a bot can tell a peer + /// on the other daemon from a colleague here. + fn sender_view(&self, msg: &Message) -> anyhow::Result { + let mut view = msg.clone(); + let Some(bot_id) = &msg.sender.bot_id else { + return Ok(view); + }; + if let Some(peer_id) = self.db.get_bot(bot_id)?.and_then(|b| b.peer_id) { + if let Some(peer) = self.db.get_peer(&peer_id)? { + view.sender.name = format!("{} @ {}", msg.sender.name, peer.name); + } + } + Ok(view) + } } diff --git a/crates/gravityd/src/lib.rs b/crates/gravityd/src/lib.rs index 1087461..d6b2268 100644 --- a/crates/gravityd/src/lib.rs +++ b/crates/gravityd/src/lib.rs @@ -18,6 +18,7 @@ pub mod messaging; pub mod model; pub mod overrides; pub mod paths; +pub mod peer; mod permissions; pub mod projectmgmt; pub mod routine_validation; diff --git a/crates/gravityd/src/main.rs b/crates/gravityd/src/main.rs index ece2371..9f9c03d 100644 --- a/crates/gravityd/src/main.rs +++ b/crates/gravityd/src/main.rs @@ -62,6 +62,10 @@ async fn main() -> anyhow::Result<()> { println!("restored into {}", cfg.home.display()); return Ok(()); } + Some("peer") => { + gravityd::peer::cli::run(&cfg, &args[1..]).await?; + return Ok(()); + } Some("service") => { let paths = gravityd::service::ServicePaths::new(cfg.home.clone(), cfg.user_home.clone()); diff --git a/crates/gravityd/src/mcp/tools.rs b/crates/gravityd/src/mcp/tools.rs index 8164aa3..31e6512 100644 --- a/crates/gravityd/src/mcp/tools.rs +++ b/crates/gravityd/src/mcp/tools.rs @@ -204,7 +204,7 @@ pub(super) fn list_bots(app: &Arc, bot_id: &str) -> anyhow::Result, bot_id: &str) -> anyhow::Result) -> String { across both ends; `note` is an FYI that expects no answer. When a\n\ send is refused, the error says what to do instead — do it rather\n\ than retrying.\n\n\ + Some colleagues run on another of the owner's machines: `list_bots`\n\ + shows them with a `machine`, and their messages arrive as\n\ + `from NAME @ MACHINE`. Message them like anyone else. Their disk is\n\ + not yours: list files you hand them in `complete_task`'s\n\ + `artifacts`, which copies them across, and expect a path they mention\n\ + elsewhere to be unreadable here.\n\n\ ## Messages carry authority\n\n\ Every message on the bus is authenticated by the daemon: the name in\n\ the header is who sent it. Bots in your project are colleagues, not\n\ diff --git a/crates/gravityd/src/peer.rs b/crates/gravityd/src/peer.rs new file mode 100644 index 0000000..aab246c --- /dev/null +++ b/crates/gravityd/src/peer.rs @@ -0,0 +1,19 @@ +//! Peer daemons: bots on two of the owner's machines working as one team. +//! +//! A bot that runs on a peer stands in here as a linked bot. Sending to it is +//! sending to any bot; the delivery worker forwards the message over the peer +//! link instead of writing it to a local session, and the peer stores it as if +//! the sender were one of its own. See `docs/peer-bots.md`. + +mod artifacts; +pub mod cli; +mod forward; +mod frames; +mod hub; +mod inbound; +mod receive; +mod socket; + +pub use forward::{forward, ForwardError}; +pub use hub::{PeerError, PeerHub}; +pub use socket::{peer_handler, spawn_dialer, spawn_dialers}; diff --git a/crates/gravityd/src/peer/artifacts.rs b/crates/gravityd/src/peer/artifacts.rs new file mode 100644 index 0000000..31e3006 --- /dev/null +++ b/crates/gravityd/src/peer/artifacts.rs @@ -0,0 +1,183 @@ +//! Files that travel with a result. `complete_task` lists artifact paths in +//! the `done` body; a path on the other machine is useless to the requester, +//! so the files themselves cross the link and the listing is rewritten to +//! where they landed. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use base64::Engine; +use bus::{Bot, Peer, MAX_PEER_ARTIFACTS_TOTAL_BYTES, MAX_PEER_ARTIFACT_BYTES}; + +use crate::app::AppState; + +use super::frames::{ArtifactFrame, MessageFrame}; + +/// The separator `complete_task` writes between a result and its listing. +const LISTING: &str = "\n\nartifacts:"; + +/// Splits a result body into its text and the artifact paths it lists. +pub(super) fn split_listing(body: &str) -> (&str, Vec<&str>) { + match body.rsplit_once(LISTING) { + Some((result, listing)) => { + let paths = listing + .lines() + .filter_map(|line| line.strip_prefix("- ")) + .map(str::trim) + .filter(|p| !p.is_empty()) + .collect(); + (result, paths) + } + None => (body, Vec::new()), + } +} + +/// Reads the files a result lists, for sending with it. Only files inside the +/// sending bot's own directory or its project's artifacts directory are read, +/// so a bot cannot exfiltrate anything else by listing it. +pub(super) fn collect(app: &Arc, sender: &Bot, body: &str) -> Vec { + let (_, paths) = split_listing(body); + let roots = allowed_roots(app, sender); + let workspace = PathBuf::from(&sender.workspace_path); + let mut total = 0u64; + paths + .into_iter() + .map(|listed| { + let skipped = |reason: &str| ArtifactFrame { + path: listed.to_string(), + bytes: None, + skipped: Some(reason.to_string()), + }; + let path = workspace.join(listed); + let Ok(path) = path.canonicalize() else { + return skipped("file not found"); + }; + if !roots.iter().any(|root| path.starts_with(root)) { + return skipped("outside the bot's workspace and the project artifacts"); + } + let size = match path.metadata() { + Ok(meta) if meta.is_file() => meta.len(), + _ => return skipped("not a regular file"), + }; + if size > MAX_PEER_ARTIFACT_BYTES { + return skipped("larger than the 16 MiB transfer limit"); + } + if total + size > MAX_PEER_ARTIFACTS_TOTAL_BYTES { + return skipped("over the 48 MiB limit for one result"); + } + match std::fs::read(&path) { + Ok(bytes) => { + total += size; + ArtifactFrame { + path: listed.to_string(), + bytes: Some(base64::engine::general_purpose::STANDARD.encode(bytes)), + skipped: None, + } + } + Err(_) => skipped("could not be read"), + } + }) + .collect() +} + +fn allowed_roots(app: &Arc, sender: &Bot) -> Vec { + let mut roots = Vec::new(); + if let Some(bot_root) = Path::new(&sender.workspace_path).parent() { + roots.extend(bot_root.canonicalize().ok()); + } + if let Ok(Some(project)) = app.db.get_project(&sender.project_id) { + let artifacts = crate::paths::artifacts_dir(&app.cfg, &project.dir_name); + roots.extend(artifacts.canonicalize().ok()); + } + roots +} + +/// Writes a forwarded result's files into the recipient project's artifacts +/// directory and returns the body with its listing pointing at them. +pub(super) fn store( + app: &Arc, + peer: &Peer, + to: &Bot, + frame: &MessageFrame, +) -> anyhow::Result { + let project = app + .db + .get_project(&to.project_id)? + .ok_or_else(|| anyhow::anyhow!("project missing"))?; + let short: String = frame.id.chars().take(8).collect(); + let dir = crate::paths::artifacts_dir(&app.cfg, &project.dir_name) + .join("peers") + .join(bus::names::dir_name(&peer.name)) + .join(short); + let (result, _) = split_listing(&frame.body); + let mut listing = String::new(); + for artifact in &frame.artifacts { + let line = match (&artifact.bytes, &artifact.skipped) { + (Some(encoded), _) => { + let bytes = base64::engine::general_purpose::STANDARD.decode(encoded)?; + std::fs::create_dir_all(&dir)?; + let target = unique_path(&dir, &file_name(&artifact.path)); + std::fs::write(&target, bytes)?; + target.display().to_string() + } + (None, reason) => format!( + "{} (on {}; not transferred: {})", + artifact.path, + peer.name, + reason.as_deref().unwrap_or("unknown reason") + ), + }; + listing.push_str(&format!("\n- {line}")); + } + Ok(format!("{result}{LISTING}{listing}")) +} + +/// The last component of a path written on either platform. +fn file_name(path: &str) -> String { + let base = path.rsplit(['/', '\\']).next().unwrap_or(path).trim(); + if base.is_empty() || base == "." || base == ".." { + "artifact".to_string() + } else { + base.to_string() + } +} + +/// `dir/name`, or `dir/name-2` and so on when two artifacts share a name. +fn unique_path(dir: &Path, name: &str) -> PathBuf { + let first = dir.join(name); + if !first.exists() { + return first; + } + let (stem, ext) = match name.rsplit_once('.') { + Some((stem, ext)) if !stem.is_empty() => (stem, format!(".{ext}")), + _ => (name, String::new()), + }; + (2..) + .map(|n| dir.join(format!("{stem}-{n}{ext}"))) + .find(|candidate| !candidate.exists()) + .unwrap_or(first) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn splits_the_listing_complete_task_writes() { + let (result, paths) = split_listing("built it\n\nartifacts:\n- out/app.exe\n- log.txt"); + assert_eq!(result, "built it"); + assert_eq!(paths, vec!["out/app.exe", "log.txt"]); + } + + #[test] + fn a_result_without_a_listing_has_no_paths() { + assert_eq!(split_listing("done"), ("done", Vec::new())); + } + + #[test] + fn takes_the_file_name_from_windows_and_unix_paths() { + assert_eq!(file_name(r"C:\work\out\app.exe"), "app.exe"); + assert_eq!(file_name("/tmp/build/log.txt"), "log.txt"); + assert_eq!(file_name("../.."), "artifact"); + } +} diff --git a/crates/gravityd/src/peer/cli.rs b/crates/gravityd/src/peer/cli.rs new file mode 100644 index 0000000..2437d65 --- /dev/null +++ b/crates/gravityd/src/peer/cli.rs @@ -0,0 +1,197 @@ +//! `gravityd peer …`: pairing and linking from a terminal. The commands talk +//! to the running daemon over its control plane with the owner token, so they +//! work while it serves, and they are exactly what the app will send. + +use futures::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio_tungstenite::tungstenite::Message; + +use crate::config::Config; + +const USAGE: &str = "usage: + gravityd peer invite [--url ws://host:port/peer] + gravityd peer add + gravityd peer list + gravityd peer bots + gravityd peer link --project + gravityd peer revoke "; + +pub async fn run(cfg: &Config, args: &[String]) -> anyhow::Result<()> { + let arg = |i: usize| args.get(i).map(String::as_str); + let flag = |name: &str| { + args.iter() + .position(|a| a == name) + .and_then(|i| args.get(i + 1)) + .map(String::as_str) + }; + let mut daemon = Daemon::connect(cfg).await?; + match (arg(0), arg(1), arg(2)) { + (Some("invite"), Some(name), _) => { + let mut req = json!({ "type": "create_peer_invite", "name": name }); + if let Some(url) = flag("--url") { + req["url"] = json!(url); + } + let reply = daemon.request(req).await?; + println!("Invite for {name} (shown once):\n"); + println!("{}", reply["invite"].as_str().unwrap_or_default()); + println!("\nOn {name}, run: gravityd peer add "); + } + (Some("add"), Some(name), Some(invite)) => { + daemon + .request(json!({ "type": "add_peer", "name": name, "invite": invite })) + .await?; + println!("Added {name}. This daemon now keeps a link open to it."); + } + (Some("list"), _, _) => { + for peer in daemon.peers().await? { + let state = if peer["revoked_at"].is_string() { + "revoked" + } else if peer["online"].as_bool() == Some(true) { + "online" + } else { + "offline" + }; + let dials = peer["url"].as_str().map(|u| format!(" dials {u}")); + println!( + "{} {state}{}", + peer["name"].as_str().unwrap_or_default(), + dials.unwrap_or_default() + ); + } + } + (Some("bots"), Some(peer), _) => { + let peer = daemon.peer(peer).await?; + for bot in daemon.peer_bots(&peer).await? { + println!( + "{} ({}) {}", + bot["name"].as_str().unwrap_or_default(), + bot["project"].as_str().unwrap_or_default(), + bot["description"].as_str().unwrap_or_default() + ); + } + } + (Some("link"), Some(peer), Some(bot)) => { + let project_name = flag("--project") + .ok_or_else(|| anyhow::anyhow!("link needs --project "))?; + let peer = daemon.peer(peer).await?; + let remote = daemon + .peer_bots(&peer) + .await? + .into_iter() + .find(|b| { + b["name"] + .as_str() + .is_some_and(|n| n.eq_ignore_ascii_case(bot)) + }) + .ok_or_else(|| anyhow::anyhow!("no bot named '{bot}' on that peer"))?; + let projects = daemon.request(json!({ "type": "list_projects" })).await?; + let project = projects["projects"] + .as_array() + .into_iter() + .flatten() + .find(|p| { + p["name"] + .as_str() + .is_some_and(|n| n.eq_ignore_ascii_case(project_name)) + }) + .ok_or_else(|| anyhow::anyhow!("no project named '{project_name}'"))?; + daemon + .request(json!({ + "type": "link_peer_bot", + "peer_id": peer, + "remote_bot_id": remote["id"], + "project_id": project["id"], + })) + .await?; + println!("Linked {bot} into {project_name}. Bots there can message it by name."); + } + (Some("revoke"), Some(peer), _) => { + let peer = daemon.peer(peer).await?; + daemon + .request(json!({ "type": "revoke_peer", "peer_id": peer })) + .await?; + println!("Revoked. Its token no longer works on this daemon."); + } + _ => anyhow::bail!("{USAGE}"), + } + Ok(()) +} + +type Socket = + tokio_tungstenite::WebSocketStream>; + +struct Daemon { + socket: Socket, + next: u64, +} + +impl Daemon { + async fn connect(cfg: &Config) -> anyhow::Result { + let port = crate::home::runtime_port(&cfg.home).unwrap_or(cfg.port); + let token = std::fs::read_to_string(cfg.secrets_dir().join("client.token")) + .map_err(|e| anyhow::anyhow!("cannot read the owner token: {e}"))?; + let url = format!("ws://127.0.0.1:{port}/ws"); + let (socket, _) = tokio_tungstenite::connect_async(url.as_str()) + .await + .map_err(|e| anyhow::anyhow!("gravityd is not answering on {url}: {e}"))?; + let mut daemon = Self { socket, next: 1 }; + daemon + .request(json!({ + "type": "hello", + "protocol_version": crate::app::PROTOCOL_VERSION, + "token": token.trim(), + "client": "gravityd-cli" + })) + .await?; + Ok(daemon) + } + + /// Sends a request and returns its reply, skipping pushes in between. + async fn request(&mut self, mut req: Value) -> anyhow::Result { + let req_id = self.next; + self.next += 1; + req["req_id"] = json!(req_id); + self.socket.send(Message::Text(req.to_string())).await?; + while let Some(frame) = self.socket.next().await { + let Message::Text(text) = frame? else { + continue; + }; + let reply: Value = serde_json::from_str(&text)?; + if reply["req_id"].as_u64() != Some(req_id) { + continue; + } + if reply["type"] == "error" { + anyhow::bail!("{}", reply["message"].as_str().unwrap_or("request failed")); + } + return Ok(reply); + } + anyhow::bail!("gravityd closed the connection") + } + + async fn peers(&mut self) -> anyhow::Result> { + let reply = self.request(json!({ "type": "list_peers" })).await?; + Ok(reply["peers"].as_array().cloned().unwrap_or_default()) + } + + /// The id of the live peer with this name. + async fn peer(&mut self, name: &str) -> anyhow::Result { + self.peers() + .await? + .into_iter() + .find(|p| { + p["name"] + .as_str() + .is_some_and(|n| n.eq_ignore_ascii_case(name)) + && !p["revoked_at"].is_string() + }) + .and_then(|p| p["id"].as_str().map(str::to_string)) + .ok_or_else(|| anyhow::anyhow!("no peer named '{name}'")) + } + + async fn peer_bots(&mut self, peer_id: &str) -> anyhow::Result> { + let reply = self + .request(json!({ "type": "list_peer_bots", "peer_id": peer_id })) + .await?; + Ok(reply["bots"].as_array().cloned().unwrap_or_default()) + } +} diff --git a/crates/gravityd/src/peer/forward.rs b/crates/gravityd/src/peer/forward.rs new file mode 100644 index 0000000..7011918 --- /dev/null +++ b/crates/gravityd/src/peer/forward.rs @@ -0,0 +1,188 @@ +//! Forwarding a delivery whose recipient is a linked bot: the delivery +//! worker's branch for bots that run on a peer. + +use std::sync::Arc; + +use bus::{Bot, Message, MessageKind, Peer, SenderKind, TaskState}; +use serde_json::json; + +use crate::app::AppState; + +use super::frames::{ClosesFrame, FromFrame, MessageFrame, Received, TaskFrame}; +use super::hub::PeerError; + +pub enum ForwardError { + /// The peer is not connected. Nothing was sent; try again later. + Offline, + Failed(anyhow::Error), +} + +/// Hands `msg` to the peer `target` runs on. `Ok` once the peer has stored it, +/// or once it is clear the message is not the peer's to see. +pub async fn forward(app: &Arc, target: &Bot, msg: &Message) -> Result<(), ForwardError> { + let fail = |e: anyhow::Error| ForwardError::Failed(e); + let peer_id = target + .peer_id + .clone() + .ok_or_else(|| fail(anyhow::anyhow!("not a linked bot")))?; + let peer = app + .db + .get_peer(&peer_id) + .map_err(fail)? + .filter(|p| p.revoked_at.is_none()) + .ok_or_else(|| fail(anyhow::anyhow!("peer revoked")))?; + if !app.peers.is_online(&peer.id) { + return Err(ForwardError::Offline); + } + let frame = { + let app = app.clone(); + let (peer, target, msg) = (peer.clone(), target.clone(), msg.clone()); + tokio::task::spawn_blocking(move || build(&app, &peer, &target, &msg)) + .await + .map_err(|e| fail(e.into()))? + .map_err(fail)? + }; + let Some(frame) = frame else { + return Ok(()); + }; + let task_id = frame.task.as_ref().map(|t| t.id.clone()); + let request = json!({ "type": "message", "message": frame }); + let result = match app.peers.request(&peer.id, request).await { + Ok(result) => result, + Err(PeerError::Offline) => return Err(ForwardError::Offline), + Err(PeerError::Rejected(reason)) => { + return Err(fail(anyhow::anyhow!("{} refused it: {reason}", peer.name))) + } + }; + let received: Received = serde_json::from_value(result).map_err(|e| fail(e.into()))?; + app.db + .map_peer_message(&peer.id, &received.message_id, &msg.id) + .map_err(fail)?; + if let (Some(local), Some(remote)) = (task_id, received.task_id) { + app.db + .map_peer_task(&peer.id, &remote, &local) + .map_err(fail)?; + } + Ok(()) +} + +/// The frame for `msg`, or `None` when it stays on this daemon. +fn build( + app: &Arc, + peer: &Peer, + target: &Bot, + msg: &Message, +) -> anyhow::Result> { + let remote_bot_id = target + .remote_bot_id + .clone() + .ok_or_else(|| anyhow::anyhow!("linked bot has no remote id"))?; + let mut sender_bot = None; + let from = match msg.sender.kind { + // The daemon's own notices (expiries, renames, introductions) describe + // this daemon's view. The peer produces its own for its side. + SenderKind::User if msg.sender.name == "system" => return Ok(None), + SenderKind::User => FromFrame::User, + SenderKind::Bot => { + let id = msg.sender.bot_id.as_deref().unwrap_or_default(); + let bot = app + .db + .get_bot(id)? + .ok_or_else(|| anyhow::anyhow!("sender {id} missing"))?; + anyhow::ensure!( + !bot.is_linked(), + "{} runs on another peer; messages are not relayed between peers", + bot.name + ); + // The recipient's answers come back addressed to the sender, so + // the sender has to accept deliveries from this peer. + app.db.expose_bot_to_peer(&peer.id, &bot.id)?; + let view = super::inbound::remote_view(&bot, String::new()); + sender_bot = Some(bot); + FromFrame::Bot { bot: view } + } + SenderKind::Routine => anyhow::bail!("routine prompts are never forwarded"), + }; + + let ref_message_id = match &msg.ref_message_id { + Some(r) => Some( + app.db + .remote_message_for(&peer.id, r)? + .unwrap_or_else(|| r.clone()), + ), + None => None, + }; + + let task = match msg.kind { + MessageKind::Task => { + let task = app + .db + .open_task_for_message(&msg.id, &target.id)? + .ok_or_else(|| anyhow::anyhow!("task for message {} is no longer open", msg.num))?; + Some(TaskFrame { + id: task.id, + deadline_at: task.deadline_at, + hop_count: task.hop_count, + }) + } + _ => None, + }; + + let closes_task = closes(app, peer, target, msg)?; + let artifacts = match (&sender_bot, msg.kind) { + (Some(bot), MessageKind::Done) => super::artifacts::collect(app, bot, &msg.body), + _ => Vec::new(), + }; + + Ok(Some(MessageFrame { + id: msg.id.clone(), + to_bot_id: remote_bot_id, + from, + kind: msg.kind, + body: msg.body.clone(), + ref_message_id, + task, + closes_task, + artifacts, + })) +} + +/// The peer's half of the task this message closes: a result for a task the +/// linked bot delegated here, or the cancellation of one delegated to it. +fn closes( + app: &Arc, + peer: &Peer, + target: &Bot, + msg: &Message, +) -> anyhow::Result> { + let Some(origin) = &msg.ref_message_id else { + return Ok(None); + }; + let tasks = app.db.tasks_with_origin(origin)?; + let found = match msg.kind { + MessageKind::Done => tasks + .iter() + .find(|t| t.from_bot_id.as_deref() == Some(target.id.as_str())) + .map(|t| (t, TaskState::Done)), + MessageKind::Note => tasks + .iter() + .find(|t| t.to_bot_id == target.id && t.state == TaskState::Cancelled) + .map(|t| (t, TaskState::Cancelled)), + _ => None, + }; + let Some((task, state)) = found else { + anyhow::ensure!( + msg.kind != MessageKind::Done, + "result for a task {} did not delegate", + target.name + ); + return Ok(None); + }; + match app.db.remote_task_for(&peer.id, &task.id)? { + Some(remote) => Ok(Some(ClosesFrame { id: remote, state })), + // A task cancelled before it ever reached the peer has nothing there + // to close; the note still tells the bot to stop. + None if state == TaskState::Cancelled => Ok(None), + None => anyhow::bail!("task {} has no counterpart on {}", task.id, peer.name), + } +} diff --git a/crates/gravityd/src/peer/frames.rs b/crates/gravityd/src/peer/frames.rs new file mode 100644 index 0000000..2843a79 --- /dev/null +++ b/crates/gravityd/src/peer/frames.rs @@ -0,0 +1,74 @@ +//! Frames exchanged on a peer link. Ids in a frame are the sender's own, +//! except `to_bot_id` and `closes_task.id`, which the sender has already +//! translated into the receiver's ids. + +use bus::{MessageKind, RemoteBot, TaskState}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// One message crossing the link, with whatever it opens or closes. +#[derive(Debug, Serialize, Deserialize)] +pub struct MessageFrame { + /// The message's id on the sending daemon. The receiver dedupes on it. + pub id: String, + pub to_bot_id: String, + pub from: FromFrame, + pub kind: MessageKind, + pub body: String, + /// The message this answers, in whichever id the receiver knows it by. + #[serde(rename = "ref", default, skip_serializing_if = "Option::is_none")] + pub ref_message_id: Option, + /// Set when the message opens a task: the receiver mirrors it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub task: Option, + /// Set when the message is the result or the cancellation of a task the + /// receiver holds. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub closes_task: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub artifacts: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum FromFrame { + /// The owner, writing from the other machine's app. + User, + Bot { + bot: RemoteBot, + }, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct TaskFrame { + pub id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub deadline_at: Option>, + pub hop_count: i64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ClosesFrame { + pub id: String, + pub state: TaskState, +} + +/// A file attached to a result. `bytes` is base64; it is absent when the file +/// could not be sent, and `skipped` says why. +#[derive(Debug, Serialize, Deserialize)] +pub struct ArtifactFrame { + /// The path as the sender's result listed it. + pub path: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub skipped: Option, +} + +/// What the receiver answers a message frame with. +#[derive(Debug, Serialize, Deserialize)] +pub struct Received { + pub message_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub task_id: Option, +} diff --git a/crates/gravityd/src/peer/hub.rs b/crates/gravityd/src/peer/hub.rs new file mode 100644 index 0000000..bd43f28 --- /dev/null +++ b/crates/gravityd/src/peer/hub.rs @@ -0,0 +1,182 @@ +//! Live peer links. Each paired daemon has at most one connection, whichever +//! side dialed it; requests go out over it and wait for the matching response. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serde_json::{json, Value}; +use tokio::sync::{mpsc, oneshot, Notify}; + +use crate::app::AppState; + +/// How long a request waits for the peer's response. A forwarded message with +/// artifacts is the slowest frame there is, and a tailnet hop is fast. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(60); + +/// Why a request to a peer did not produce a result. +#[derive(Debug)] +pub enum PeerError { + /// No live link. Nothing was sent, so retrying later is safe. + Offline, + /// The peer answered with a refusal, or never answered. + Rejected(String), +} + +impl std::fmt::Display for PeerError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + PeerError::Offline => write!(f, "peer is offline"), + PeerError::Rejected(reason) => write!(f, "{reason}"), + } + } +} + +type Pending = Arc>>>; + +struct Link { + out: mpsc::UnboundedSender, + pending: Pending, + /// Distinguishes this link from a newer one for the same peer, so the old + /// one closing does not unregister its replacement. + generation: u64, + closed: Arc, +} + +#[derive(Clone, Default)] +pub struct PeerHub { + links: Arc>>, + counter: Arc, +} + +impl PeerHub { + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.links.lock().unwrap_or_else(|e| e.into_inner()) + } + + pub fn is_online(&self, peer_id: &str) -> bool { + self.lock().contains_key(peer_id) + } + + /// Drops the peer's link, if any. Used when the peer is revoked. + pub fn disconnect(&self, peer_id: &str) { + if let Some(link) = self.lock().remove(peer_id) { + link.closed.notify_one(); + } + } + + /// Sends a request and waits for the peer's result. + pub async fn request(&self, peer_id: &str, mut frame: Value) -> Result { + let (out, pending) = { + let links = self.lock(); + let link = links.get(peer_id).ok_or(PeerError::Offline)?; + (link.out.clone(), link.pending.clone()) + }; + let req_id = self.counter.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = oneshot::channel(); + lock_pending(&pending).insert(req_id, tx); + frame["req_id"] = json!(req_id); + if out.send(frame).is_err() { + lock_pending(&pending).remove(&req_id); + return Err(PeerError::Offline); + } + let response = match tokio::time::timeout(REQUEST_TIMEOUT, rx).await { + Ok(Ok(response)) => response, + Ok(Err(_)) => return Err(PeerError::Rejected("peer link closed".to_string())), + Err(_) => { + lock_pending(&pending).remove(&req_id); + return Err(PeerError::Rejected("peer did not answer".to_string())); + } + }; + if response.get("ok").and_then(Value::as_bool) == Some(true) { + Ok(response.get("result").cloned().unwrap_or(Value::Null)) + } else { + let reason = response + .get("error") + .and_then(Value::as_str) + .unwrap_or("peer refused the request"); + Err(PeerError::Rejected(reason.to_string())) + } + } + + /// Runs an authenticated link until either side closes it. Frames from + /// the peer arrive on `inbound`; frames for it go out on `out`. + pub(super) async fn serve( + &self, + app: Arc, + peer_id: String, + mut inbound: mpsc::UnboundedReceiver, + out: mpsc::UnboundedSender, + ) { + let generation = self.counter.fetch_add(1, Ordering::Relaxed); + let pending: Pending = Arc::default(); + let closed = Arc::new(Notify::new()); + let replaced = self.lock().insert( + peer_id.clone(), + Link { + out: out.clone(), + pending: pending.clone(), + generation, + closed: closed.clone(), + }, + ); + if let Some(old) = replaced { + old.closed.notify_one(); + } + tracing::info!(peer_id, "peer link up"); + let _ = app.db.touch_peer(&peer_id); + + loop { + let frame = tokio::select! { + frame = inbound.recv() => frame, + _ = closed.notified() => None, + }; + let Some(frame) = frame else { + break; + }; + if frame.get("type").and_then(Value::as_str) == Some("response") { + let id = frame.get("req_id").and_then(Value::as_u64); + if let Some(tx) = id.and_then(|id| lock_pending(&pending).remove(&id)) { + let _ = tx.send(frame); + } + continue; + } + // Handled off the read loop, so one slow request (a result with + // artifacts to write) does not hold up the responses behind it. + let app = app.clone(); + let peer_id = peer_id.clone(); + let out = out.clone(); + tokio::task::spawn_blocking(move || { + let req_id = frame.get("req_id").cloned().unwrap_or(Value::Null); + let response = match super::inbound::handle(&app, &peer_id, &frame) { + Ok(result) => json!({ + "type": "response", "req_id": req_id, "ok": true, "result": result + }), + Err(e) => json!({ + "type": "response", "req_id": req_id, "ok": false, + "error": format!("{e:#}") + }), + }; + let _ = out.send(response); + }); + } + + let mut links = self.lock(); + if links + .get(&peer_id) + .is_some_and(|link| link.generation == generation) + { + links.remove(&peer_id); + } + drop(links); + let _ = app.db.touch_peer(&peer_id); + tracing::info!(peer_id, "peer link down"); + } +} + +fn lock_pending( + pending: &Pending, +) -> std::sync::MutexGuard<'_, HashMap>> { + pending.lock().unwrap_or_else(|e| e.into_inner()) +} diff --git a/crates/gravityd/src/peer/inbound.rs b/crates/gravityd/src/peer/inbound.rs new file mode 100644 index 0000000..6901847 --- /dev/null +++ b/crates/gravityd/src/peer/inbound.rs @@ -0,0 +1,76 @@ +//! Requests a peer makes of this daemon. A peer can see which bots exist, +//! link one, and deliver messages to the bots linked to it. Nothing else. + +use std::sync::Arc; + +use bus::{Peer, RemoteBot}; +use serde_json::{json, Value}; + +use crate::app::AppState; + +pub(super) fn handle(app: &Arc, peer_id: &str, frame: &Value) -> anyhow::Result { + let peer = app + .db + .get_peer(peer_id)? + .filter(|p| p.revoked_at.is_none()) + .ok_or_else(|| anyhow::anyhow!("peer revoked"))?; + match frame.get("type").and_then(Value::as_str).unwrap_or("") { + "ping" => Ok(json!({})), + "list_bots" => list_bots(app), + "link" => link(app, &peer, frame), + "message" => { + let message = serde_json::from_value(frame["message"].clone())?; + let received = super::receive::receive(app, &peer, message)?; + Ok(serde_json::to_value(received)?) + } + other => anyhow::bail!("unknown peer request '{other}'"), + } +} + +/// The bots a peer could link: every live bot that runs here. +fn list_bots(app: &Arc) -> anyhow::Result { + let projects: std::collections::HashMap = app + .db + .list_projects()? + .into_iter() + .map(|p| (p.id.clone(), crate::db::Db::display_project_name(&p))) + .collect(); + let bots: Vec = app + .db + .list_bots(None)? + .into_iter() + .filter(|bot| !bot.is_linked()) + .map(|bot| { + let project = projects.get(&bot.project_id).cloned().unwrap_or_default(); + remote_view(&bot, project) + }) + .collect(); + Ok(json!({ "bots": bots })) +} + +/// Exposes one local bot to the peer, which is about to stand it in. +fn link(app: &Arc, peer: &Peer, frame: &Value) -> anyhow::Result { + let bot_id = frame + .get("bot_id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("'bot_id' is required"))?; + let bot = app + .db + .get_live_bot(bot_id)? + .filter(|b| !b.is_linked()) + .ok_or_else(|| anyhow::anyhow!("no bot with id {bot_id} runs on this daemon"))?; + app.db.expose_bot_to_peer(&peer.id, &bot.id)?; + tracing::info!(peer = %peer.name, bot = %bot.name, "bot linked to peer"); + Ok(json!({ "bot": remote_view(&bot, String::new()) })) +} + +pub(super) fn remote_view(bot: &bus::Bot, project: String) -> RemoteBot { + RemoteBot { + id: bot.id.clone(), + name: bot.name.clone(), + description: bot.description.clone(), + avatar: bot.avatar.clone(), + runtime: bot.runtime, + project, + } +} diff --git a/crates/gravityd/src/peer/receive.rs b/crates/gravityd/src/peer/receive.rs new file mode 100644 index 0000000..80c4dac --- /dev/null +++ b/crates/gravityd/src/peer/receive.rs @@ -0,0 +1,202 @@ +//! Storing a message a peer forwarded: the receiving half of the delivery +//! worker's forward. Each step mirrors what the local tool would have done had +//! the sender been here, so the recipient cannot tell the difference beyond +//! the machine named in the envelope. + +use std::sync::Arc; + +use bus::{MessageKind, Peer, RemoteBot, Sender, SenderKind, TaskState}; + +use crate::app::AppState; +use crate::events::Push; +use crate::messaging::{self, Dm}; + +use super::frames::{FromFrame, MessageFrame, Received}; + +pub(super) fn receive( + app: &Arc, + peer: &Peer, + frame: MessageFrame, +) -> anyhow::Result { + // At-least-once on the wire: an ack lost to a dropped link means the same + // frame again. The first copy's result is the answer. + if let Some(local) = app.db.local_message_for(&peer.id, &frame.id)? { + let task_id = app + .db + .tasks_with_origin(&local)? + .into_iter() + .next() + .map(|t| t.id); + return Ok(Received { + message_id: local, + task_id, + }); + } + + let to = app + .db + .get_live_bot(&frame.to_bot_id)? + .filter(|b| !b.is_linked()) + .ok_or_else(|| anyhow::anyhow!("no bot with id {} runs here", frame.to_bot_id))?; + anyhow::ensure!( + app.db.is_exposed_to_peer(&peer.id, &to.id)?, + "{} is not linked to this daemon", + to.name + ); + + let (sender, linked) = match &frame.from { + FromFrame::User => (messaging::user_sender(), None), + FromFrame::Bot { bot } => { + anyhow::ensure!( + frame.kind != MessageKind::Chat, + "only the owner sends chat messages" + ); + let linked = linked_sender(app, peer, bot, &to.project_id)?; + let sender = Sender { + kind: SenderKind::Bot, + bot_id: Some(linked.id.clone()), + name: linked.name.clone(), + }; + (sender, Some(linked)) + } + }; + let ref_id = match &frame.ref_message_id { + Some(r) => match app.db.local_message_for(&peer.id, r)? { + Some(local) => Some(local), + None => app.db.get_message(r)?.map(|m| m.id), + }, + None => None, + }; + let body = if frame.artifacts.is_empty() { + frame.body.clone() + } else { + super::artifacts::store(app, peer, &to, &frame)? + }; + + let msg = match &frame.closes_task { + Some(closes) => { + let linked = linked + .as_ref() + .ok_or_else(|| anyhow::anyhow!("only a bot closes a task"))?; + let task = app + .db + .get_task(&closes.id)? + .ok_or_else(|| anyhow::anyhow!("no task {} here", closes.id))?; + match closes.state { + TaskState::Done => { + anyhow::ensure!( + task.to_bot_id == linked.id + && task.from_bot_id.as_deref() == Some(to.id.as_str()), + "task {} is not one {} was given by {}", + task.id, + linked.name, + to.name + ); + // Already closed here (expired while the result was in + // flight) still stores the result: the work was done. + app.db.try_close_task(&task.id, TaskState::Done)?; + let origin = app + .db + .get_message(&task.origin_message_id)? + .ok_or_else(|| anyhow::anyhow!("origin message missing"))?; + let msg = app.db.insert_message( + &origin.conversation_id, + &sender, + MessageKind::Done, + &body, + Some(&origin.id), + None, + )?; + app.events.push(Push::MessageNew { + message: msg.clone(), + }); + let key = format!("{}:{}", msg.id, to.id); + let delivery = app.db.enqueue_delivery(&msg.id, &to.id, &key)?; + app.events.push(Push::DeliveryUpdate { delivery }); + msg + } + TaskState::Cancelled => { + anyhow::ensure!( + task.from_bot_id.as_deref() == Some(linked.id.as_str()) + && task.to_bot_id == to.id, + "task {} is not one {} gave {}", + task.id, + linked.name, + to.name + ); + app.db.try_close_task(&task.id, TaskState::Cancelled)?; + let mut dm = Dm::new(&to.id, &sender, frame.kind, &body); + dm.ref_message_id = ref_id.as_deref(); + messaging::send_dm(&app.db, &app.events, dm)? + } + other => anyhow::bail!("a task cannot be closed as {}", other.as_str()), + } + } + None => { + anyhow::ensure!( + frame.kind != MessageKind::Done, + "a result must name the task it closes" + ); + let mut dm = Dm::new(&to.id, &sender, frame.kind, &body); + dm.ref_message_id = ref_id.as_deref(); + messaging::send_dm(&app.db, &app.events, dm)? + } + }; + + // A delegation opens the same task here, so the recipient's + // `complete_task` and every budget work as if the sender were local. The + // hop count carries over, so the chain limit holds across machines. + let mut task_id = None; + if frame.kind == MessageKind::Task { + let (Some(spec), Some(linked)) = (&frame.task, &linked) else { + anyhow::bail!("a task needs a task frame and a bot sender"); + }; + let task = app.db.create_task( + &msg.id, + Some(&linked.id), + &to.id, + spec.deadline_at, + spec.hop_count, + &linked.id, + )?; + app.db.map_peer_task(&peer.id, &spec.id, &task.id)?; + task_id = Some(task.id); + } + app.db.map_peer_message(&peer.id, &frame.id, &msg.id)?; + Ok(Received { + message_id: msg.id, + task_id, + }) +} + +/// The linked bot standing in for the peer's sender, created on first +/// contact. It takes the sender's name, or `-` if a bot here +/// already has it. +fn linked_sender( + app: &Arc, + peer: &Peer, + remote: &RemoteBot, + project_id: &str, +) -> anyhow::Result { + if let Some(bot) = app.db.linked_bot(&peer.id, &remote.id, project_id)? { + return Ok(bot); + } + let mut name = bus::names::validate(&remote.name).map_err(anyhow::Error::msg)?; + if app.db.get_bot_by_name(project_id, &name)?.is_some() { + name = + bus::names::validate(&format!("{name}-{}", peer.name)).map_err(anyhow::Error::msg)?; + anyhow::ensure!( + app.db.get_bot_by_name(project_id, &name)?.is_none(), + "both '{}' and '{name}' are taken in this project", + remote.name + ); + } + let stand_in = RemoteBot { + name, + ..remote.clone() + }; + let bot = app.db.create_linked_bot(project_id, &stand_in, &peer.id)?; + tracing::info!(peer = %peer.name, bot = %bot.name, "linked bot created on first contact"); + app.events.push(Push::BotUpdated { bot: bot.clone() }); + Ok(bot) +} diff --git a/crates/gravityd/src/peer/socket.rs b/crates/gravityd/src/peer/socket.rs new file mode 100644 index 0000000..3d0dd4d --- /dev/null +++ b/crates/gravityd/src/peer/socket.rs @@ -0,0 +1,237 @@ +//! The two ends of a peer link: `/peer` accepts a dialing daemon, and a +//! dialer keeps a connection open to each peer this daemon was told to reach. +//! Both authenticate with a `peer_hello`, then hand the socket to the hub as a +//! pair of frame channels. + +use std::sync::Arc; +use std::time::Duration; + +use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; +use axum::extract::State; +use axum::response::IntoResponse; +use futures::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::protocol::WebSocketConfig; +use tokio_tungstenite::tungstenite::Message as TungMessage; + +use crate::app::AppState; + +/// Version of the frames exchanged on a peer link. +const PEER_PROTOCOL: u64 = 1; + +/// Largest frame on a peer link: a result whose artifacts are at the size +/// cap, base64-encoded, with room for the envelope. +const MAX_PEER_FRAME_BYTES: usize = 80 * 1024 * 1024; + +const HELLO_TIMEOUT: Duration = Duration::from_secs(10); +const MIN_REDIAL: Duration = Duration::from_secs(2); +const MAX_REDIAL: Duration = Duration::from_secs(60); + +pub async fn peer_handler( + State(app): State>, + upgrade: WebSocketUpgrade, +) -> axum::response::Response { + upgrade + .max_message_size(MAX_PEER_FRAME_BYTES) + .max_frame_size(MAX_PEER_FRAME_BYTES) + .on_upgrade(move |socket| accept(app, socket)) + .into_response() +} + +async fn accept(app: Arc, socket: WebSocket) { + let (mut sink, mut stream) = socket.split(); + let hello = match tokio::time::timeout(HELLO_TIMEOUT, stream.next()).await { + Ok(Some(Ok(WsMessage::Text(text)))) => serde_json::from_str::(&text).ok(), + _ => None, + }; + let peer_id = match hello.as_ref().map(|hello| authenticate(&app, hello)) { + Some(Ok(peer_id)) => peer_id, + Some(Err(e)) => { + tracing::warn!(error = %e, "peer handshake refused"); + let refusal = json!({ "type": "error", "message": e.to_string() }); + let _ = sink.send(WsMessage::Text(refusal.to_string())).await; + return; + } + None => return, + }; + let Ok(daemon_id) = app.db.daemon_id() else { + return; + }; + let ok = json!({ "type": "peer_hello_ok", "protocol": PEER_PROTOCOL, "daemon_id": daemon_id }); + if sink.send(WsMessage::Text(ok.to_string())).await.is_err() { + return; + } + + let (in_tx, in_rx) = mpsc::unbounded_channel(); + let (out_tx, mut out_rx) = mpsc::unbounded_channel::(); + let writer = tokio::spawn(async move { + while let Some(frame) = out_rx.recv().await { + if sink.send(WsMessage::Text(frame.to_string())).await.is_err() { + break; + } + } + }); + let reader = tokio::spawn(async move { + while let Some(Ok(frame)) = stream.next().await { + match frame { + WsMessage::Text(text) => { + if let Ok(value) = serde_json::from_str::(&text) { + if in_tx.send(value).is_err() { + break; + } + } + } + WsMessage::Close(_) => break, + _ => {} + } + } + }); + app.peers.serve(app.clone(), peer_id, in_rx, out_tx).await; + reader.abort(); + writer.abort(); +} + +/// The peer a `peer_hello` authenticates as. +fn authenticate(app: &AppState, hello: &Value) -> anyhow::Result { + anyhow::ensure!( + hello.get("type").and_then(Value::as_str) == Some("peer_hello"), + "expected peer_hello" + ); + let protocol = hello.get("protocol").and_then(Value::as_u64).unwrap_or(0); + anyhow::ensure!( + protocol == PEER_PROTOCOL, + "this daemon speaks peer protocol {PEER_PROTOCOL}, not {protocol}" + ); + let token = hello.get("token").and_then(Value::as_str).unwrap_or(""); + let daemon_id = hello + .get("daemon_id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + .ok_or_else(|| anyhow::anyhow!("peer_hello carries no daemon_id"))?; + let peer_id = app + .secrets + .peer_for_token(token) + .ok_or_else(|| anyhow::anyhow!("invalid peer token"))?; + let peer = app + .db + .get_peer(&peer_id)? + .filter(|p| p.revoked_at.is_none()) + .ok_or_else(|| anyhow::anyhow!("peer revoked"))?; + anyhow::ensure!( + app.db.bind_peer_daemon(&peer.id, daemon_id)?, + "this peer token belongs to another daemon" + ); + Ok(peer.id) +} + +/// Starts a dialer for every peer this daemon reaches out to. +pub fn spawn_dialers(app: &Arc) { + match app.db.list_peers() { + Ok(peers) => { + for peer in peers { + if peer.url.is_some() && peer.revoked_at.is_none() { + spawn_dialer(app.clone(), peer.id); + } + } + } + Err(e) => tracing::warn!(error = %e, "could not list peers to dial"), + } +} + +/// Keeps a link open to one peer until it is revoked. +pub fn spawn_dialer(app: Arc, peer_id: String) { + tokio::spawn(async move { + let mut wait = MIN_REDIAL; + loop { + let peer = match app.db.get_peer(&peer_id) { + Ok(Some(peer)) if peer.revoked_at.is_none() => peer, + _ => return, + }; + let (Some(url), Some(token)) = (peer.url.clone(), app.secrets.peer_token(&peer_id)) + else { + return; + }; + match dial(&app, &peer_id, &url, &token).await { + Ok(()) => wait = MIN_REDIAL, + Err(e) => { + tracing::debug!(peer = %peer.name, error = %format!("{e:#}"), "peer dial failed"); + } + } + tokio::time::sleep(wait).await; + wait = (wait * 2).min(MAX_REDIAL); + } + }); +} + +/// One connection attempt, served until it drops. `Ok` once the link was up. +async fn dial(app: &Arc, peer_id: &str, url: &str, token: &str) -> anyhow::Result<()> { + let config = WebSocketConfig { + max_message_size: Some(MAX_PEER_FRAME_BYTES), + max_frame_size: Some(MAX_PEER_FRAME_BYTES), + ..WebSocketConfig::default() + }; + let (socket, _) = tokio_tungstenite::connect_async_with_config(url, Some(config), true).await?; + let (mut sink, mut stream) = socket.split(); + let hello = json!({ + "type": "peer_hello", + "protocol": PEER_PROTOCOL, + "token": token, + "daemon_id": app.db.daemon_id()?, + }); + sink.send(TungMessage::Text(hello.to_string())).await?; + let reply = match tokio::time::timeout(HELLO_TIMEOUT, stream.next()).await { + Ok(Some(Ok(TungMessage::Text(text)))) => serde_json::from_str::(&text)?, + _ => anyhow::bail!("peer closed the link during the handshake"), + }; + if reply.get("type").and_then(Value::as_str) != Some("peer_hello_ok") { + let reason = reply + .get("message") + .and_then(Value::as_str) + .unwrap_or("handshake refused"); + anyhow::bail!("{reason}"); + } + let remote = reply + .get("daemon_id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("peer did not say which daemon it is"))?; + anyhow::ensure!( + app.db.bind_peer_daemon(peer_id, remote)?, + "the daemon at {url} is not the one this peer was paired with" + ); + + let (in_tx, in_rx) = mpsc::unbounded_channel(); + let (out_tx, mut out_rx) = mpsc::unbounded_channel::(); + let writer = tokio::spawn(async move { + while let Some(frame) = out_rx.recv().await { + if sink + .send(TungMessage::Text(frame.to_string())) + .await + .is_err() + { + break; + } + } + }); + let reader = tokio::spawn(async move { + while let Some(Ok(frame)) = stream.next().await { + match frame { + TungMessage::Text(text) => { + if let Ok(value) = serde_json::from_str::(&text) { + if in_tx.send(value).is_err() { + break; + } + } + } + TungMessage::Close(_) => break, + _ => {} + } + } + }); + app.peers + .serve(app.clone(), peer_id.to_string(), in_rx, out_tx) + .await; + reader.abort(); + writer.abort(); + Ok(()) +} diff --git a/crates/gravityd/src/secrets.rs b/crates/gravityd/src/secrets.rs index 48fbb58..7f3ec72 100644 --- a/crates/gravityd/src/secrets.rs +++ b/crates/gravityd/src/secrets.rs @@ -16,6 +16,9 @@ pub struct Secrets { bot_tokens: Mutex>, /// token -> device_id for device-scoped client credentials. device_tokens: Mutex>, + /// token -> peer_id for the `/peer` link. The same token serves both + /// ends: the listening daemon accepts it, the dialing daemon presents it. + peer_tokens: Mutex>, client_token: String, } @@ -47,6 +50,7 @@ impl Secrets { let mut bot_tokens = HashMap::new(); let mut device_tokens = HashMap::new(); + let mut peer_tokens = HashMap::new(); for entry in fs::read_dir(dir)? { let entry = entry?; let name = entry.file_name().to_string_lossy().to_string(); @@ -66,6 +70,14 @@ impl Secrets { if !token.is_empty() { device_tokens.insert(token, device_id.to_string()); } + } else if let Some(peer_id) = name + .strip_prefix("peer-") + .and_then(|s| s.strip_suffix(".token")) + { + let token = fs::read_to_string(entry.path())?.trim().to_string(); + if !token.is_empty() { + peer_tokens.insert(token, peer_id.to_string()); + } } } @@ -73,6 +85,7 @@ impl Secrets { dir: dir.to_path_buf(), bot_tokens: Mutex::new(bot_tokens), device_tokens: Mutex::new(device_tokens), + peer_tokens: Mutex::new(peer_tokens), client_token, }) } @@ -158,6 +171,49 @@ impl Secrets { } } +impl Secrets { + /// Issue a fresh link token for a peer that will dial this daemon. + pub fn issue_peer_token(&self, peer_id: &str) -> anyhow::Result { + let token = random_token(); + self.store_peer_token(peer_id, &token)?; + Ok(token) + } + + /// Keep the token from a peer's invite, presented when dialing it. + pub fn store_peer_token(&self, peer_id: &str, token: &str) -> anyhow::Result<()> { + let mut map = self.peer_tokens.lock().unwrap_or_else(|e| e.into_inner()); + map.retain(|_, id| id.as_str() != peer_id); + write_secret(&self.dir.join(format!("peer-{peer_id}.token")), token)?; + map.insert(token.to_string(), peer_id.to_string()); + Ok(()) + } + + pub fn peer_for_token(&self, token: &str) -> Option { + let map = self.peer_tokens.lock().unwrap_or_else(|e| e.into_inner()); + map.iter() + .find(|(t, _)| constant_time_eq(t.as_bytes(), token.as_bytes())) + .map(|(_, id)| id.clone()) + } + + pub fn peer_token(&self, peer_id: &str) -> Option { + let map = self.peer_tokens.lock().unwrap_or_else(|e| e.into_inner()); + map.iter() + .find(|(_, id)| id.as_str() == peer_id) + .map(|(t, _)| t.clone()) + } + + /// Delete a revoked peer's token so neither end can use the link again. + pub fn remove_peer_token(&self, peer_id: &str) -> anyhow::Result<()> { + let mut map = self.peer_tokens.lock().unwrap_or_else(|e| e.into_inner()); + map.retain(|_, id| id.as_str() != peer_id); + let path = self.dir.join(format!("peer-{peer_id}.token")); + if path.exists() { + fs::remove_file(path)?; + } + Ok(()) + } +} + fn constant_time_eq(a: &[u8], b: &[u8]) -> bool { if a.len() != b.len() { return false; diff --git a/crates/gravityd/src/server.rs b/crates/gravityd/src/server.rs index e2845ca..a5d5ebe 100644 --- a/crates/gravityd/src/server.rs +++ b/crates/gravityd/src/server.rs @@ -25,6 +25,7 @@ pub fn router(app: Arc) -> Router { .route("/ws", get(crate::ws::ws_handler)) .route("/mcp", post(crate::mcp::mcp_handler)) .route("/hook", post(crate::mcp::hook_handler)) + .route("/peer", get(crate::peer::peer_handler)) .with_state(app) } @@ -72,7 +73,10 @@ pub fn spawn_workers(app: &Arc) { }); } + crate::peer::spawn_dialers(app); + let worker = DeliveryWorker { + app: app.clone(), db: app.db.clone(), supervisor: app.supervisor.clone(), events: app.events.clone(), diff --git a/crates/gravityd/src/supervisor/lifecycle.rs b/crates/gravityd/src/supervisor/lifecycle.rs index 98328cf..007cf59 100644 --- a/crates/gravityd/src/supervisor/lifecycle.rs +++ b/crates/gravityd/src/supervisor/lifecycle.rs @@ -9,6 +9,9 @@ use super::*; impl Supervisor { pub fn start_bot(&self, bot_id: &str) -> anyhow::Result<()> { let bot = self.inner.db.get_bot(bot_id)?.context("bot not found")?; + if bot.is_linked() { + return Ok(()); // runs on its peer; nothing to start here + } // Creation and the supervision tick can both start a new bot. Held // until this call returns, so a second start finds the claim and @@ -228,7 +231,7 @@ impl Supervisor { return; } }; - for bot in bots { + for bot in bots.into_iter().filter(|b| !b.is_linked()) { if !self.wants_start(&bot.id) { continue; } diff --git a/crates/gravityd/src/ws.rs b/crates/gravityd/src/ws.rs index 72c0b7d..7187eb9 100644 --- a/crates/gravityd/src/ws.rs +++ b/crates/gravityd/src/ws.rs @@ -23,6 +23,7 @@ mod decisions_publish; mod dispatch; mod entities; mod messaging; +mod peers; mod routines; mod runtime; mod terminal; diff --git a/crates/gravityd/src/ws/dispatch.rs b/crates/gravityd/src/ws/dispatch.rs index 658f2e0..0a70511 100644 --- a/crates/gravityd/src/ws/dispatch.rs +++ b/crates/gravityd/src/ws/dispatch.rs @@ -30,6 +30,8 @@ const READ_ONLY: &[&str] = &[ "get_decision", "list_tags", "count_pending_decisions", + "list_peers", + "list_peer_bots", ]; /// Requests that exercise the owner's ruling authority. @@ -130,6 +132,12 @@ impl Conn { "rename_tag" => self.rename_tag(&req_id, req), "delete_tag" => self.delete_tag(&req_id, req), "set_project_lead" => self.set_project_lead(&req_id, req), + "list_peers" => self.list_peers(&req_id), + "create_peer_invite" => self.create_peer_invite(&req_id, req), + "add_peer" => self.add_peer(&req_id, req), + "revoke_peer" => self.revoke_peer(&req_id, req), + "list_peer_bots" => self.list_peer_bots(&req_id, req), + "link_peer_bot" => self.link_peer_bot(&req_id, req), other => { self.reply_err( &req_id, diff --git a/crates/gravityd/src/ws/peers.rs b/crates/gravityd/src/ws/peers.rs new file mode 100644 index 0000000..350b152 --- /dev/null +++ b/crates/gravityd/src/ws/peers.rs @@ -0,0 +1,197 @@ +//! Peer requests: pairing with another daemon, and linking its bots into a +//! project here. See `docs/peer-bots.md`. + +use std::net::SocketAddr; +use std::sync::Arc; + +use bus::RemoteBot; +use serde_json::{json, Value}; + +use crate::app::AppState; +use crate::events::Push; + +use super::Conn; + +impl Conn { + pub(super) fn list_peers(&self, req_id: &Value) -> anyhow::Result<()> { + let peers: Vec = self + .app + .db + .list_peers()? + .into_iter() + .map(|peer| peer_view(&self.app, &peer)) + .collect(); + self.send(json!({ "type": "peers", "req_id": req_id, "peers": peers })); + Ok(()) + } + + /// Pairing, listening side: registers the machine that will dial in and + /// returns the code to paste into it. The code holds the token, so it is + /// shown once, like a device token. + pub(super) fn create_peer_invite(&self, req_id: &Value, req: &Value) -> anyhow::Result<()> { + let name = peer_name(&self.app, Self::str_field(req, "name")?)?; + let url = match req.get("url").and_then(Value::as_str) { + Some(url) => url.to_string(), + None => listening_url(&self.app)?, + }; + let peer = self.app.db.create_peer(&name, None)?; + let token = self.app.secrets.issue_peer_token(&peer.id)?; + self.send(json!({ + "type": "peer", "req_id": req_id, + "peer": peer_view(&self.app, &peer), + "invite": format!("{url}#{token}") + })); + Ok(()) + } + + /// Pairing, dialing side: stores the other daemon's invite and starts + /// keeping a link open to it. + pub(super) fn add_peer(&self, req_id: &Value, req: &Value) -> anyhow::Result<()> { + let name = peer_name(&self.app, Self::str_field(req, "name")?)?; + let invite = Self::str_field(req, "invite")?.trim(); + let (url, token) = invite + .rsplit_once('#') + .filter(|(url, token)| { + (url.starts_with("ws://") || url.starts_with("wss://")) && !token.is_empty() + }) + .ok_or_else(|| { + anyhow::anyhow!("not a peer invite: expected ws://host:port/peer#token") + })?; + let peer = self.app.db.create_peer(&name, Some(url))?; + self.app.secrets.store_peer_token(&peer.id, token)?; + crate::peer::spawn_dialer(self.app.clone(), peer.id.clone()); + self.send(json!({ "type": "peer", "req_id": req_id, "peer": peer_view(&self.app, &peer) })); + Ok(()) + } + + pub(super) fn revoke_peer(&self, req_id: &Value, req: &Value) -> anyhow::Result<()> { + let peer_id = Self::str_field(req, "peer_id")?; + if !self.app.db.revoke_peer(peer_id)? { + self.reply_err(req_id, "not_found", "peer not found or already revoked"); + return Ok(()); + } + self.app.secrets.remove_peer_token(peer_id)?; + self.app.peers.disconnect(peer_id); + let peer = self + .app + .db + .get_peer(peer_id)? + .ok_or_else(|| anyhow::anyhow!("peer vanished"))?; + self.send(json!({ "type": "peer", "req_id": req_id, "peer": peer_view(&self.app, &peer) })); + Ok(()) + } + + /// The bots running on a peer, for choosing one to link. + pub(super) fn list_peer_bots(&self, req_id: &Value, req: &Value) -> anyhow::Result<()> { + let peer_id = Self::str_field(req, "peer_id")?.to_string(); + let (app, out, req_id) = (self.app.clone(), self.out.clone(), req_id.clone()); + tokio::spawn(async move { + let reply = match app + .peers + .request(&peer_id, json!({ "type": "list_bots" })) + .await + { + Ok(result) => json!({ + "type": "peer_bots", "req_id": req_id, "peer_id": peer_id, + "bots": result.get("bots").cloned().unwrap_or_else(|| json!([])) + }), + Err(e) => error(&req_id, "unavailable", &e.to_string()), + }; + let _ = out.send(reply); + }); + Ok(()) + } + + /// Stands a peer's bot in as a linked bot in a project here. + pub(super) fn link_peer_bot(&self, req_id: &Value, req: &Value) -> anyhow::Result<()> { + let peer_id = Self::str_field(req, "peer_id")?.to_string(); + let remote_bot_id = Self::str_field(req, "remote_bot_id")?.to_string(); + let project_id = Self::str_field(req, "project_id")?.to_string(); + let (app, out, req_id) = (self.app.clone(), self.out.clone(), req_id.clone()); + tokio::spawn(async move { + let reply = match link(&app, &peer_id, &remote_bot_id, &project_id).await { + Ok(bot) => json!({ + "type": "bot", "req_id": req_id, "bot": super::views::bot_view(&app, &bot) + }), + Err(e) => error(&req_id, "conflict", &format!("{e:#}")), + }; + let _ = out.send(reply); + }); + Ok(()) + } +} + +async fn link( + app: &Arc, + peer_id: &str, + remote_bot_id: &str, + project_id: &str, +) -> anyhow::Result { + let project = app + .db + .get_project(project_id)? + .filter(|p| p.deleted_at.is_none()) + .ok_or_else(|| anyhow::anyhow!("project not found"))?; + if let Some(bot) = app.db.linked_bot(peer_id, remote_bot_id, &project.id)? { + return Ok(bot); + } + let result = app + .peers + .request(peer_id, json!({ "type": "link", "bot_id": remote_bot_id })) + .await + .map_err(|e| anyhow::anyhow!("{e}"))?; + let remote: RemoteBot = serde_json::from_value(result["bot"].clone())?; + // Bots address each other by name, so a clash would make one of the two + // unreachable. The owner renames one and links again. + anyhow::ensure!( + app.db.get_bot_by_name(&project.id, &remote.name)?.is_none(), + "a bot named '{}' already exists in this project; rename it first", + remote.name + ); + let bot = app.db.create_linked_bot(&project.id, &remote, peer_id)?; + app.events.push(Push::BotUpdated { bot: bot.clone() }); + Ok(bot) +} + +fn peer_name(app: &AppState, raw: &str) -> anyhow::Result { + let name = bus::names::validate_project(raw).map_err(anyhow::Error::msg)?; + anyhow::ensure!( + app.db.get_peer_by_name(&name)?.is_none(), + "a peer named '{name}' already exists" + ); + Ok(name) +} + +/// The URL another daemon dials to reach this one: its first non-loopback +/// bind address, which is the Tailscale interface in the documented setup. +fn listening_url(app: &AppState) -> anyhow::Result { + let ip = app + .cfg + .bind + .iter() + .find(|ip| !ip.is_loopback()) + .ok_or_else(|| { + anyhow::anyhow!( + "this daemon only listens on localhost; add its Tailscale address to \ + `bind` in gravityd.toml and restart it, or pass the url to dial" + ) + })?; + Ok(format!("ws://{}/peer", SocketAddr::new(*ip, app.cfg.port))) +} + +pub(crate) fn peer_view(app: &AppState, peer: &bus::Peer) -> Value { + json!({ + "id": peer.id, + "name": peer.name, + "url": peer.url, + "daemon_id": peer.daemon_id, + "online": app.peers.is_online(&peer.id), + "created_at": peer.created_at.to_rfc3339(), + "last_seen_at": peer.last_seen_at.map(|t| t.to_rfc3339()), + "revoked_at": peer.revoked_at.map(|t| t.to_rfc3339()) + }) +} + +fn error(req_id: &Value, code: &str, message: &str) -> Value { + json!({ "type": "error", "req_id": req_id, "code": code, "message": message }) +} diff --git a/crates/gravityd/src/ws/views.rs b/crates/gravityd/src/ws/views.rs index 612d5af..3b1881e 100644 --- a/crates/gravityd/src/ws/views.rs +++ b/crates/gravityd/src/ws/views.rs @@ -24,9 +24,31 @@ pub(crate) fn project_view(project: &bus::Project) -> Value { /// A bot as clients see it: the stored row plus the runtime fields the /// supervisor and the delivery tables own. pub(crate) fn bot_view(app: &AppState, bot: &bus::Bot) -> Value { - let (state, reason) = app.supervisor.state(&bot.id); + let (mut state, mut reason) = app.supervisor.state(&bot.id); let unread = app.db.unread_count(&bot.id).unwrap_or(0); + // A linked bot has no session here; what matters is whether its machine + // is reachable. + let peer = bot + .peer_id + .as_deref() + .and_then(|id| app.db.get_peer(id).ok().flatten()); + if let Some(peer) = &peer { + let online = app.peers.is_online(&peer.id); + state = if online { + bus::BotState::Ready + } else { + bus::BotState::Stopped + }; + reason = format!( + "runs on {}{}", + peer.name, + if online { "" } else { " (offline)" } + ); + } json!({ + "peer": peer.as_ref().map(|p| json!({ + "id": p.id, "name": p.name, "online": app.peers.is_online(&p.id) + })), "id": bot.id, "project_id": bot.project_id, // Archived rows carry a tombstoned name so the original is free to diff --git a/crates/gravityd/tests/peer_bots.rs b/crates/gravityd/tests/peer_bots.rs new file mode 100644 index 0000000..283d5f5 --- /dev/null +++ b/crates/gravityd/tests/peer_bots.rs @@ -0,0 +1,322 @@ +//! Peer bots: two daemons paired over a peer link, a lead on one delegating +//! to a bot on the other. Both daemons run in this process on the runtime +//! double, so the whole path — pairing, linking, forwarding, task mirroring, +//! artifact transfer — runs for real without a second machine. + +mod common; + +use std::time::Duration; + +use bus::TaskState; +use common::tasks::{drain_until, error_text}; +use common::*; +use serde_json::{json, Value}; + +struct Team { + mac: TestDaemon, + win: TestDaemon, + mac_client: WsClient, + /// The Windows daemon's id for the peer row standing for the Mac. + win_peer_id: String, + /// The Mac's linked bot standing in for `windev`. + linked_windev: String, + lead: McpClient, + windev: McpClient, + windev_id: String, +} + +async fn wait_until(what: &str, mut check: impl FnMut() -> bool) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while !check() { + assert!(tokio::time::Instant::now() < deadline, "timed out: {what}"); + tokio::time::sleep(Duration::from_millis(25)).await; + } +} + +async fn project(c: &mut WsClient, name: &str) -> String { + let created = c + .request(json!({"type": "create_project", "name": name})) + .await; + created["project"]["id"].as_str().expect("pid").to_string() +} + +/// A lead on "the Mac" and a Windows developer on "the PC", paired, with the +/// developer linked into the lead's project. +async fn team() -> Team { + let mac = spawn_daemon().await; + let win = spawn_daemon().await; + let mut mac_client = WsClient::connect(&mac).await; + let mut win_client = WsClient::connect(&win).await; + + let mac_project = project(&mut mac_client, "app").await; + let lead = create_bot(&mut mac_client, &mac_project, "lead").await; + let win_project = project(&mut win_client, "app").await; + let windev = create_bot(&mut win_client, &win_project, "windev").await; + let windev_id = windev["id"].as_str().expect("id").to_string(); + + let invite = win_client + .request(json!({ + "type": "create_peer_invite", "name": "mac", + "url": format!("ws://{}/peer", win.addr) + })) + .await; + assert_eq!(invite["type"], "peer", "{invite}"); + let win_peer_id = invite["peer"]["id"].as_str().expect("peer id").to_string(); + let added = mac_client + .request(json!({ + "type": "add_peer", "name": "win", "invite": invite["invite"] + })) + .await; + assert_eq!(added["type"], "peer", "{added}"); + let mac_peer_id = added["peer"]["id"].as_str().expect("peer id").to_string(); + wait_until("the link comes up", || { + mac.app.peers.is_online(&mac_peer_id) + }) + .await; + + let bots = mac_client + .request(json!({"type": "list_peer_bots", "peer_id": mac_peer_id})) + .await; + assert_eq!(bots["bots"][0]["name"], "windev", "{bots}"); + let linked = mac_client + .request(json!({ + "type": "link_peer_bot", "peer_id": mac_peer_id, + "remote_bot_id": windev_id, "project_id": mac_project + })) + .await; + assert_eq!(linked["type"], "bot", "{linked}"); + assert_eq!(linked["bot"]["peer"]["name"], "win"); + + let lead_token = mac + .app + .secrets + .bot_token(lead["id"].as_str().expect("id")) + .expect("token"); + let windev_token = win.app.secrets.bot_token(&windev_id).expect("token"); + Team { + lead: McpClient::new(&mac, &lead_token), + windev: McpClient::new(&win, &windev_token), + linked_windev: linked["bot"]["id"].as_str().expect("id").to_string(), + mac, + win, + mac_client, + win_peer_id, + windev_id, + } +} + +fn find<'a>(messages: &'a [Value], needle: &str) -> &'a Value { + messages + .iter() + .find(|m| m["body"].as_str().is_some_and(|b| b.contains(needle))) + .unwrap_or_else(|| panic!("no message containing {needle:?} in {messages:?}")) +} + +#[tokio::test] +async fn a_task_crosses_to_the_peer_and_its_result_and_files_come_back() { + let mut t = team().await; + + let bots = t.lead.call("list_bots", json!({})).await; + let listed = bots["bots"] + .as_array() + .expect("bots") + .iter() + .find(|b| b["name"] == "windev") + .expect("windev listed"); + assert_eq!(listed["machine"], "win"); + assert_eq!(listed["online"], true); + + let sent = t + .lead + .call( + "send_message", + json!({"to": "windev", "kind": "task", "body": "build the windows installer"}), + ) + .await; + let mac_task = sent["task_id"].as_str().expect("task id").to_string(); + + // On the PC the task arrives from a linked "lead", with a task of its own. + let inbox = drain_until(&mut t.windev, "build the windows installer").await; + let task_msg = find(&inbox, "build the windows installer"); + assert_eq!(task_msg["from"], "lead"); + let win_task = task_msg["task_id"] + .as_str() + .expect("mirrored task") + .to_string(); + + let windev = t + .win + .app + .db + .get_bot(&t.windev_id) + .expect("db") + .expect("bot"); + let out = std::path::Path::new(&windev.workspace_path).join("out"); + std::fs::create_dir_all(&out).expect("mkdir"); + std::fs::write(out.join("setup.txt"), "installer bytes").expect("write"); + t.windev + .call( + "complete_task", + json!({ + "task_id": win_task, "result": "installer built", + "artifacts": ["out/setup.txt", "../../../../secrets/client.token"] + }), + ) + .await; + + // Back on the Mac: the lead's task is closed by the peer's result, and + // the file it listed is on this disk. + let inbox = drain_until(&mut t.lead, "installer built").await; + let done = find(&inbox, "installer built"); + assert_eq!(done["kind"], "done"); + let body = done["body"].as_str().expect("body"); + let local = body + .lines() + .find_map(|l| l.strip_prefix("- ").filter(|p| p.ends_with("setup.txt"))) + .unwrap_or_else(|| panic!("no transferred file in {body}")); + assert_eq!( + std::fs::read_to_string(local).expect("artifact"), + "installer bytes" + ); + // A path outside the bot's own files is listed, never sent. + assert!(body.contains("not transferred"), "{body}"); + let task = t.mac.app.db.get_task(&mac_task).expect("db").expect("task"); + assert_eq!(task.state, TaskState::Done); + + // The lead's session saw the sender's machine in the envelope. + let lead_id = t + .mac + .app + .db + .get_task(&mac_task) + .unwrap() + .unwrap() + .from_bot_id + .unwrap(); + wait_until("the result reaches the lead's session", || { + terminal(&t.mac, &lead_id).contains("from WINDEV @ WIN · done") + }) + .await; +} + +#[tokio::test] +async fn cancelling_on_one_side_closes_the_task_on_the_other() { + let mut t = team().await; + let sent = t + .lead + .call( + "send_message", + json!({"to": "windev", "kind": "task", "body": "port the updater"}), + ) + .await; + let inbox = drain_until(&mut t.windev, "port the updater").await; + let win_task = find(&inbox, "port the updater")["task_id"] + .as_str() + .expect("task id") + .to_string(); + + t.lead + .call( + "cancel_task", + json!({"task_id": sent["task_id"], "reason": "not needed"}), + ) + .await; + drain_until(&mut t.windev, "was cancelled").await; + let task = t.win.app.db.get_task(&win_task).expect("db").expect("task"); + assert_eq!(task.state, TaskState::Cancelled); + let refused = t + .windev + .call_raw("complete_task", json!({"task_id": win_task, "result": "x"})) + .await; + assert!(error_text(&refused).contains("already cancelled")); +} + +#[tokio::test] +async fn the_owner_can_message_a_linked_bot_from_the_other_machine() { + let mut t = team().await; + let sent = t + .mac_client + .request(json!({ + "type": "send_user_message", "to_bot_id": t.linked_windev, + "body": "which SDK are you on?" + })) + .await; + assert_eq!(sent["type"], "message", "{sent}"); + let inbox = drain_until(&mut t.windev, "which SDK").await; + let chat = find(&inbox, "which SDK"); + assert_eq!(chat["from"], "user"); + assert_eq!(chat["kind"], "chat"); +} + +#[tokio::test] +async fn a_revoked_peer_is_cut_off() { + let mut t = team().await; + let mut win_client = WsClient::connect(&t.win).await; + let revoked = win_client + .request(json!({"type": "revoke_peer", "peer_id": t.win_peer_id})) + .await; + assert_eq!(revoked["type"], "peer", "{revoked}"); + let mac = &t.mac; + wait_until("the Mac sees the link drop", || { + mac.app + .db + .list_peers() + .expect("peers") + .iter() + .all(|p| !mac.app.peers.is_online(&p.id)) + }) + .await; + + // Messages wait rather than fail while the peer is unreachable. + t.lead + .call( + "send_message", + json!({"to": "windev", "kind": "note", "body": "ping"}), + ) + .await; + tokio::time::sleep(Duration::from_millis(400)).await; + let queued = t + .mac + .app + .db + .list_deliveries(Some(&t.linked_windev), None) + .expect("deliveries"); + assert!( + queued + .iter() + .all(|d| d.state != bus::DeliveryState::Delivered), + "{queued:?}" + ); +} + +#[tokio::test] +async fn the_peer_bot_can_ask_back_on_an_open_task() { + let mut t = team().await; + t.lead + .call( + "send_message", + json!({"to": "windev", "kind": "task", "body": "sign the binaries"}), + ) + .await; + drain_until(&mut t.windev, "sign the binaries").await; + t.windev + .call( + "send_message", + json!({"to": "lead", "kind": "reply", "body": "which certificate?"}), + ) + .await; + let inbox = drain_until(&mut t.lead, "which certificate?").await; + let question = find(&inbox, "which certificate?"); + assert_eq!(question["kind"], "reply"); + assert_eq!(question["from"], "windev"); + + // Handing the work back up the chain is still a loop across machines. + let refused = t + .windev + .call_raw( + "send_message", + json!({"to": "lead", "kind": "task", "body": "you do it"}), + ) + .await; + assert!(error_text(&refused).contains("loop")); +} diff --git a/docs/architecture.md b/docs/architecture.md index bbf4cc9..a0da928 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -106,7 +106,9 @@ an Agent SDK adapter are not implemented runtime choices. Messages travel through the durable bus. A user can send a message through the control plane; a bot uses MCP tools such as `send_message`, `check_inbox`, and -`complete_task`. Bot-to-bot addressing is scoped to a project. +`complete_task`. Bot-to-bot addressing is scoped to a project. A project can +include linked bots that run on a paired peer daemon; see +[peer bots](peer-bots.md). The delivery path is: diff --git a/docs/peer-bots.md b/docs/peer-bots.md index b40de06..d831ff8 100644 --- a/docs/peer-bots.md +++ b/docs/peer-bots.md @@ -1,6 +1,9 @@ # Peer bots: one team across two machines -Status: design, not yet implemented. +Status: the daemon side (pairing, linking, forwarding, task mirroring and +artifact transfer) is implemented and covered by `crates/gravityd/tests/peer_bots.rs`. +The desktop UI is not built yet; pair and link with `gravityd peer …` until +it is. Requests are listed in [protocol.md](protocol.md). A team can span two Gravity daemons. The motivating case is an app built for macOS and Windows: a lead and a Mac developer run on the Mac, a Windows @@ -12,6 +15,30 @@ Each bot still runs, with its own terminal and runtime, on the machine that owns it. What crosses machines is the bus: messages, tasks, results, and the artifacts those results attach. +## Setting it up + +On the Windows PC, add its Tailscale address to `bind` in `gravityd.toml` +(next to `127.0.0.1`) and restart the daemon. Then create an invite for the +Mac: + +```sh +gravityd peer invite mac +``` + +On the Mac, add the PC with that invite, check that the link is up, and link +the Windows bot into the project the lead works in: + +```sh +gravityd peer add win-pc "ws://100.x.y.z:49777/peer#…" +gravityd peer list +gravityd peer bots win-pc +gravityd peer link win-pc windev --project my-app +``` + +The lead can now `send_message(to: "windev", kind: "task", …)`. The first +message creates a linked `lead` in the Windows bot's project, so its answers +come back without any setup on the PC. + ## Assumptions - Both daemons belong to one owner and reach each other over Tailscale. Like diff --git a/docs/protocol.md b/docs/protocol.md index a1f254c..dc377ac 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -83,6 +83,12 @@ Codes: `auth_failed`, `unsupported_version`, `not_found`, `invalid_request`, | `list_devices` | – | `devices` | | `create_device` | `name, capabilities` (array of `"read"`/`"control"`/`"approve"`) | `device` (includes `token`, shown once) | | `revoke_device` | `device_id` | `device` | +| `list_peers` | – | `peers` | +| `create_peer_invite` | `name, url?` (defaults to the first non-loopback bind address) | `peer` plus `invite` (holds the link token, shown once) | +| `add_peer` | `name, invite` | `peer`; the daemon starts dialing it | +| `revoke_peer` | `peer_id` | `peer` | +| `list_peer_bots` | `peer_id` | `peer_bots` (`bots`: id, name, description, avatar, runtime, project) | +| `link_peer_bot` | `peer_id, remote_bot_id, project_id` | `bot` (a linked bot; `peer` is set on it) | | `list_decisions` | `project_id?, state?, tag?, bot_id?, query?, before?, limit?` | `decisions` | | `get_decision` | `decision_id` | `decision` (with comments, tags, notifications) | | `count_pending_decisions` | – | `pending_decisions` | diff --git a/third-party/inventory.json b/third-party/inventory.json index 37b2ce0..51fec10 100644 --- a/third-party/inventory.json +++ b/third-party/inventory.json @@ -4,9 +4,9 @@ "x86_64-pc-windows-msvc" ], "inputs": { - "Cargo.lock": "ed089d7f431124cb093bc49f5c54a6ca4cb0af573ce8be2f6386475d742fca84", + "Cargo.lock": "2d03468c3b96ec984adfa2e515c1f8a20a6fa7dc39b3b2753b350085647f75c9", "Cargo.toml": "17f46fbe0ce2a8fab2e1b26a1b10435ac1968ae6f8150db33bb1e05c25fb7491", - "crates/gravityd/Cargo.toml": "94c7560d2df9a9f0c377739a0f858b6af099db9ff5de3789f6e1876e0169ad8e", + "crates/gravityd/Cargo.toml": "e948b84372298a104eba4ea6dd4e76cd75e2c16278587dae0a51368943d4c2a7", "crates/bus/Cargo.toml": "c4c4f51131d78f8ae8cb6e96fcb918c245998d9c1e01312f513fb56dfc452910", "apps/desktop/src-tauri/Cargo.lock": "7c73bbce32b88bd02862bcf73cea344288764824065d4379787d9c0de7991474", "apps/desktop/src-tauri/Cargo.toml": "421dc0d9bcf49e5d93eb01a987ef7b77b393e1e383be7eac64cabde2daf53702",