From 4f40ba9f693aef244e02954492eaf2208841957e Mon Sep 17 00:00:00 2001 From: Shiv Rossi Date: Sat, 10 Oct 2026 08:36:06 -0600 Subject: [PATCH 1/4] feat(herdr): add protocol 19 bridge daemon --- Cargo.lock | 19 + Cargo.toml | 1 + README.md | 5 + contrib/systemd/herdr-iris-bridge.service | 18 + crates/iris-core/src/ingest.rs | 45 +- crates/iris-herdr-bridge/Cargo.toml | 27 + crates/iris-herdr-bridge/src/config.rs | 53 + crates/iris-herdr-bridge/src/delivery.rs | 181 ++ crates/iris-herdr-bridge/src/lib.rs | 54 + crates/iris-herdr-bridge/src/main.rs | 15 + crates/iris-herdr-bridge/src/protocol.rs | 1622 +++++++++++++++++ crates/iris-herdr-bridge/src/spool.rs | 1140 ++++++++++++ .../herdr-v0.8.0-pane-agent-status-event.json | 1 + .../herdr-v0.8.0-pane-created-event.json | 1 + .../herdr-v0.8.0-subscription-ack.json | 1 + .../herdr-v0.8.0-workspace-list-response.json | 1 + .../herdr-v0.8.0-workspace-renamed-event.json | 1 + crates/iris-providers/src/herdr/mod.rs | 257 ++- crates/iris-server/src/routes.rs | 40 +- crates/iris-storage/src/ingest.rs | 2 +- docs/providers/herdr-bridge.md | 63 + 21 files changed, 3511 insertions(+), 36 deletions(-) create mode 100644 contrib/systemd/herdr-iris-bridge.service create mode 100644 crates/iris-herdr-bridge/Cargo.toml create mode 100644 crates/iris-herdr-bridge/src/config.rs create mode 100644 crates/iris-herdr-bridge/src/delivery.rs create mode 100644 crates/iris-herdr-bridge/src/lib.rs create mode 100644 crates/iris-herdr-bridge/src/main.rs create mode 100644 crates/iris-herdr-bridge/src/protocol.rs create mode 100644 crates/iris-herdr-bridge/src/spool.rs create mode 100644 crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-agent-status-event.json create mode 100644 crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-created-event.json create mode 100644 crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-subscription-ack.json create mode 100644 crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-list-response.json create mode 100644 crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-renamed-event.json create mode 100644 docs/providers/herdr-bridge.md diff --git a/Cargo.lock b/Cargo.lock index 465a81f..4f85335 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1059,6 +1059,25 @@ dependencies = [ "uuid", ] +[[package]] +name = "iris-herdr-bridge" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "iris-core", + "iris-providers", + "reqwest", + "serde", + "serde_json", + "tempfile", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", + "wiremock", +] + [[package]] name = "iris-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index bce679d..4dbc70d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ members = [ "crates/iris-cli", "crates/iris-mcp", "crates/iris-codegen", + "crates/iris-herdr-bridge", ] [workspace.package] diff --git a/README.md b/README.md index d4711f5..a0ba85f 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,11 @@ Point Iris at your messaging sources. Query all of them through one API — via - **Self-hostable**: MIT licensed, no cloud dependencies, runs anywhere Rust runs. - **Extensible**: Adding a provider = implementing one trait. +Herdr workstation events can be bridged to Iris's existing authenticated +`POST /ingest` contract by the separate `iris-herdr-bridge` daemon. See +[`docs/providers/herdr-bridge.md`](docs/providers/herdr-bridge.md) for its +protocol-19, bounded-spool, retry, and manual systemd installation contract. + ## Quick Start ```bash diff --git a/contrib/systemd/herdr-iris-bridge.service b/contrib/systemd/herdr-iris-bridge.service new file mode 100644 index 0000000..c6253bb --- /dev/null +++ b/contrib/systemd/herdr-iris-bridge.service @@ -0,0 +1,18 @@ +[Unit] +Description=Herdr to Iris durable event bridge +After=default.target + +[Service] +Type=simple +EnvironmentFile=%h/.config/herdr-iris-bridge/env +ExecStart=%h/.local/bin/herdr-iris-bridge +Restart=on-failure +RestartSec=2s +UMask=0077 +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=read-only +ReadWritePaths=%h/.local/state/herdr-iris-bridge + +[Install] +WantedBy=default.target diff --git a/crates/iris-core/src/ingest.rs b/crates/iris-core/src/ingest.rs index d8f99b0..822fc74 100644 --- a/crates/iris-core/src/ingest.rs +++ b/crates/iris-core/src/ingest.rs @@ -24,8 +24,12 @@ pub enum IngestMutation { UpsertThread(Thread), /// Mark a source-backed thread archived without deleting its history. ArchiveThread { source: String, source_id: String }, - /// Append a normalized message. Replays are controlled by the batch key. - AppendMessage(Message), + /// Append a normalized message. The nested field keeps the mutation tag + /// distinct from `Message::kind` in the JSON representation. + AppendMessage { + /// Normalized message to append to its source-backed thread. + message: Message, + }, } /// A source cursor committed only when the complete batch commits. @@ -128,6 +132,7 @@ fn canonicalize(value: serde_json::Value) -> serde_json::Value { mod tests { use chrono::TimeZone; use serde_json::json; + use uuid::Uuid; use super::*; @@ -182,4 +187,40 @@ mod tests { second.canonical_hash().unwrap() ); } + + #[test] + fn append_message_mutation_round_trips_without_colliding_with_message_kind() { + let sender = Contact { + id: Uuid::from_u128(1), + source: "herdr".into(), + provider_instance: None, + source_id: "bridge".into(), + display_name: Some("Bridge".into()), + avatar_url: None, + metadata: json!({}), + }; + let mutation = IngestMutation::AppendMessage { + message: Message { + id: Uuid::from_u128(2), + thread_id: Uuid::from_u128(3), + source: "herdr".into(), + source_id: "event-1".into(), + sender, + kind: crate::MessageKind::System, + body: "heartbeat".into(), + attachments: Vec::new(), + timestamp: Utc.timestamp_opt(1_700_000_000, 0).unwrap(), + is_outbound: false, + metadata: json!({}), + }, + }; + + let value = serde_json::to_value(&mutation).unwrap(); + assert_eq!(value["kind"], "append_message"); + assert_eq!(value["message"]["kind"], "system"); + assert_eq!( + serde_json::from_value::(value).unwrap(), + mutation + ); + } } diff --git a/crates/iris-herdr-bridge/Cargo.toml b/crates/iris-herdr-bridge/Cargo.toml new file mode 100644 index 0000000..61e3e09 --- /dev/null +++ b/crates/iris-herdr-bridge/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "iris-herdr-bridge" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +description = "Durable Herdr v0.8.0 event bridge into Iris ingest" + +[ lints ] +workspace = true + +[dependencies] +anyhow = { workspace = true } +chrono = { workspace = true } +iris-core = { path = "../iris-core" } +iris-providers = { path = "../iris-providers" } +reqwest = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +uuid = { workspace = true } + +[dev-dependencies] +tempfile = "3" +wiremock = { workspace = true } diff --git a/crates/iris-herdr-bridge/src/config.rs b/crates/iris-herdr-bridge/src/config.rs new file mode 100644 index 0000000..4058258 --- /dev/null +++ b/crates/iris-herdr-bridge/src/config.rs @@ -0,0 +1,53 @@ +use std::path::PathBuf; + +use anyhow::{Context as _, bail}; + +const DEFAULT_INGEST_URL: &str = "http://100.66.233.79:9876/ingest"; + +/// Runtime configuration loaded from the systemd environment file. +/// +/// The token deliberately has no `Debug` implementation and is never written +/// to the spool or logs. +pub struct Config { + /// Herdr's local API Unix socket. + pub socket_path: PathBuf, + /// Existing source-agnostic Iris ingest endpoint. + pub ingest_url: reqwest::Url, + /// Bearer credential loaded from the systemd environment file. + pub ingest_token: String, + /// Private durable outbox directory. + pub state_dir: PathBuf, +} + +impl Config { + /// Load and validate the runtime configuration from environment variables. + pub fn from_env() -> anyhow::Result { + let home = std::env::var_os("HOME") + .map(PathBuf::from) + .context("HOME is not set")?; + let socket_path = std::env::var_os("HERDR_SOCKET") + .map_or_else(|| home.join(".config/herdr/herdr.sock"), PathBuf::from); + let state_dir = std::env::var_os("HERDR_BRIDGE_STATE_DIR").map_or_else( + || home.join(".local/state/herdr-iris-bridge"), + PathBuf::from, + ); + let ingest_url = reqwest::Url::parse(DEFAULT_INGEST_URL) + .context("the approved Iris ingest URL is invalid")?; + let ingest_token = + std::env::var("IRIS_INGEST_TOKEN").context("IRIS_INGEST_TOKEN is not set")?; + if ingest_token.trim().is_empty() + || ingest_token.trim() != ingest_token + || ingest_token.contains('\r') + || ingest_token.contains('\n') + { + bail!("IRIS_INGEST_TOKEN must be a non-empty single-line bearer token"); + } + + Ok(Self { + socket_path, + ingest_url, + ingest_token, + state_dir, + }) + } +} diff --git a/crates/iris-herdr-bridge/src/delivery.rs b/crates/iris-herdr-bridge/src/delivery.rs new file mode 100644 index 0000000..0ea96d7 --- /dev/null +++ b/crates/iris-herdr-bridge/src/delivery.rs @@ -0,0 +1,181 @@ +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use reqwest::{Client, StatusCode}; +use tokio::sync::Mutex; +use tracing::{error, info, warn}; + +use crate::spool::{BatchRecord, Spool}; + +const POLL_INTERVAL: Duration = Duration::from_millis(250); +const INITIAL_BACKOFF: Duration = Duration::from_secs(1); +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +/// Seal due event batches and enqueue an idle heartbeat independently of HTTP. +pub async fn run_sealer(spool: Arc>) -> anyhow::Result<()> { + let mut interval = tokio::time::interval(POLL_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + let now = Utc::now(); + let mut spool = spool.lock().await; + if spool.maybe_append_heartbeat(now)? { + info!("idle bridge heartbeat durably queued"); + } + spool.flush_due(now)?; + } +} + +/// Retry the oldest durable batch until it is accepted by Iris. +pub async fn run_delivery( + client: Client, + ingest_url: reqwest::Url, + token: String, + spool: Arc>, +) -> anyhow::Result<()> { + let mut backoff = INITIAL_BACKOFF; + loop { + let batch = spool.lock().await.oldest_batch()?; + + let Some(batch) = batch else { + tokio::time::sleep(POLL_INTERVAL).await; + continue; + }; + + match submit_batch(&client, &ingest_url, &token, &batch).await { + Ok(status) if status.is_success() => { + spool + .lock() + .await + .remove_batch(batch.sequence, batch.event_count)?; + info!( + batch_sequence = batch.sequence, + event_count = batch.event_count, + "Herdr batch flush delivered to Iris" + ); + backoff = INITIAL_BACKOFF; + } + Ok(status) => { + if status.is_client_error() { + error!( + http_status = status.as_u16(), + event_count = batch.event_count, + "Iris rejected a durable batch; exact payload retained for retry" + ); + } else { + warn!( + http_status = status.as_u16(), + event_count = batch.event_count, + "Iris unavailable; exact durable batch retained for retry" + ); + } + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + } + Err(()) => { + warn!( + event_count = batch.event_count, + "Iris request failed; exact durable batch retained for retry" + ); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + } + } + } +} + +async fn submit_batch( + client: &Client, + ingest_url: &reqwest::Url, + token: &str, + batch: &BatchRecord, +) -> Result { + let response = client + .post(ingest_url.clone()) + .bearer_auth(token) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(batch.body.clone()) + .send() + .await + .map_err(|_| ())?; + Ok(response.status()) +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::Duration as ChronoDuration; + use serde_json::json; + use tempfile::tempdir; + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + #[tokio::test] + async fn retries_after_restart_reuse_the_same_identity_and_exact_request_body() { + let dir = tempdir().unwrap(); + let state_dir = dir.path().join("state"); + let mut spool = Spool::open(state_dir.clone()).unwrap(); + let now = Utc::now(); + spool + .append_upstream_event(json!({"event":"future_event", "data":{}}), now) + .unwrap(); + spool.flush_due(now + ChronoDuration::seconds(5)).unwrap(); + let first_record = spool.oldest_batch().unwrap().unwrap(); + let expected_body = first_record.body.clone(); + let server = MockServer::start().await; + let token = "synthetic-test-token"; + let client = Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(); + + Mock::given(method("POST")) + .and(path("/ingest")) + .and(header("authorization", format!("Bearer {token}"))) + .respond_with(ResponseTemplate::new(503)) + .expect(1) + .mount(&server) + .await; + let first_status = submit_batch( + &client, + &reqwest::Url::parse(&format!("{}/ingest", server.uri())).unwrap(), + token, + &first_record, + ) + .await + .unwrap(); + assert_eq!(first_status, StatusCode::SERVICE_UNAVAILABLE); + let first_request = server.received_requests().await.unwrap().remove(0); + assert_eq!(first_request.body, expected_body.as_bytes()); + + drop(spool); + let mut reopened = Spool::open(state_dir).unwrap(); + let retry_record = reopened.oldest_batch().unwrap().unwrap(); + assert_eq!(retry_record.body, expected_body); + assert_eq!(retry_record.sequence, first_record.sequence); + server.reset().await; + Mock::given(method("POST")) + .and(path("/ingest")) + .and(header("authorization", format!("Bearer {token}"))) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + let retry_status = submit_batch( + &client, + &reqwest::Url::parse(&format!("{}/ingest", server.uri())).unwrap(), + token, + &retry_record, + ) + .await + .unwrap(); + assert_eq!(retry_status, StatusCode::OK); + let retry_request = server.received_requests().await.unwrap().remove(0); + assert_eq!(retry_request.body, expected_body.as_bytes()); + reopened + .remove_batch(retry_record.sequence, retry_record.event_count) + .unwrap(); + assert!(reopened.oldest_batch().unwrap().is_none()); + } +} diff --git a/crates/iris-herdr-bridge/src/lib.rs b/crates/iris-herdr-bridge/src/lib.rs new file mode 100644 index 0000000..10cc771 --- /dev/null +++ b/crates/iris-herdr-bridge/src/lib.rs @@ -0,0 +1,54 @@ +//! Hosted implementation of the bounded Herdr v0.8.0/protocol-19 → Iris ingest bridge. +//! +//! The upstream socket has no stable event cursor or event IDs. Each captured +//! event therefore gets a bridge-owned ID before it is durably spooled. Retries +//! preserve the exact already-spooled request; a new Herdr subscription may +//! replay retained upstream events, and those possible duplicates are accepted. + +mod config; +mod delivery; +mod protocol; +mod spool; + +pub use config::Config; + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Context as _; +use tokio::sync::Mutex; +use tracing::info; + +use crate::spool::Spool; + +const HTTP_TIMEOUT: Duration = Duration::from_secs(15); + +/// Run the source reader and durable delivery worker until either fails. +pub async fn run(config: Config) -> anyhow::Result<()> { + let spool = Arc::new(Mutex::new( + Spool::open(config.state_dir.clone()).context("open durable bridge spool")?, + )); + let client = reqwest::Client::builder() + .timeout(HTTP_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .build() + .context("create HTTP client")?; + info!("Herdr bridge started (protocol 19)"); + + tokio::select! { + result = protocol::run_source(config.socket_path, Arc::clone(&spool)) => { + result.context("Herdr source loop stopped") + } + result = delivery::run_sealer(Arc::clone(&spool)) => { + result.context("batch sealer stopped") + } + result = delivery::run_delivery( + client, + config.ingest_url, + config.ingest_token, + Arc::clone(&spool), + ) => { + result.context("Iris delivery loop stopped") + } + } +} diff --git a/crates/iris-herdr-bridge/src/main.rs b/crates/iris-herdr-bridge/src/main.rs new file mode 100644 index 0000000..f61c2c2 --- /dev/null +++ b/crates/iris-herdr-bridge/src/main.rs @@ -0,0 +1,15 @@ +use anyhow::Context as _; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_target(false) + .init(); + + let config = iris_herdr_bridge::Config::from_env().context("invalid bridge configuration")?; + iris_herdr_bridge::run(config).await +} diff --git a/crates/iris-herdr-bridge/src/protocol.rs b/crates/iris-herdr-bridge/src/protocol.rs new file mode 100644 index 0000000..a28df59 --- /dev/null +++ b/crates/iris-herdr-bridge/src/protocol.rs @@ -0,0 +1,1622 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use serde_json::{Map, Value, json}; +use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; +use tokio::net::UnixStream; +use tokio::sync::Mutex; +use tracing::{info, warn}; +use uuid::Uuid; + +use crate::spool::Spool; + +const EXPECTED_HERDR_VERSION: &str = "0.8.0"; +const EXPECTED_PROTOCOL: u64 = 19; +const MAX_WIRE_LINE_BYTES: usize = 4 * 1024 * 1024; +const RECONNECT_DELAY: Duration = Duration::from_secs(2); + +/// Recover forever from socket/server failures at the approved two-second +/// interval. A planned pane-set refresh reconnects immediately to minimize the +/// interval without pane-scoped status subscriptions. No socket payload, event +/// content, path, or credential is included in logs. +pub async fn run_source(socket_path: PathBuf, spool: Arc>) -> anyhow::Result<()> { + loop { + let delay = match consume_subscription(&socket_path, &spool).await { + Ok(SubscriptionEnd::PeerClosed) => { + info!("Herdr subscription ended; reconnecting"); + RECONNECT_DELAY + } + Ok(SubscriptionEnd::PaneSetChanged) => { + info!("Herdr pane set changed; refreshing scoped status subscriptions"); + reconnect_delay(SubscriptionEnd::PaneSetChanged) + } + Err(error) => { + warn!( + reason = error.class, + "Herdr subscription unavailable; retrying" + ); + RECONNECT_DELAY + } + }; + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SubscriptionEnd { + PeerClosed, + PaneSetChanged, +} + +#[derive(Debug)] +struct SourceFailure { + class: &'static str, +} + +impl SourceFailure { + const fn new(class: &'static str) -> Self { + Self { class } + } +} + +const fn reconnect_delay(end: SubscriptionEnd) -> Duration { + match end { + SubscriptionEnd::PeerClosed => RECONNECT_DELAY, + SubscriptionEnd::PaneSetChanged => Duration::ZERO, + } +} + +async fn consume_subscription( + socket_path: &Path, + spool: &Arc>, +) -> Result { + verify_protocol(socket_path).await?; + let mut pane_infos = list_panes(socket_path).await?; + let mut workspace_snapshots = list_workspace_snapshots(socket_path).await?; + resolve_unmapped_events(spool, &pane_infos, &workspace_snapshots).await?; + let pane_ids = pane_infos.keys().cloned().collect::>(); + let request_id = format!("herdr-bridge-{}", Uuid::new_v4()); + let request = subscription_request(&request_id, &pane_ids); + let stream = UnixStream::connect(socket_path) + .await + .map_err(|_| SourceFailure::new("socket_connect"))?; + let mut reader = send_and_ack(stream, &request_id, &request).await?; + + // Herdr v0.8.0's unfiltered pane-status subscription reports transitions + // after its baseline but does not emit the pane's current state. Refresh + // snapshots after the ACK, then durably capture one current-status + // observation per pane so startup and reconnects establish a usable read + // model without narrowing the ongoing subscription to one status value. + let current_panes = list_panes(socket_path).await?; + let current_pane_ids = current_panes.keys().cloned().collect::>(); + let current_workspace_snapshots = list_workspace_snapshots(socket_path).await?; + append_status_snapshots(spool, ¤t_panes, ¤t_workspace_snapshots).await?; + if current_pane_ids != pane_ids { + return Ok(SubscriptionEnd::PaneSetChanged); + } + pane_infos = current_panes; + workspace_snapshots = current_workspace_snapshots; + + loop { + let line = read_line(&mut reader) + .await + .map_err(|_| SourceFailure::new("subscription_read"))?; + let Some(line) = line else { + return Ok(SubscriptionEnd::PeerClosed); + }; + let mut payload: Value = + serde_json::from_slice(&line).map_err(|_| SourceFailure::new("subscription_json"))?; + if payload.get("error").is_some() { + return Err(SourceFailure::new("subscription_error")); + } + let event_kind = payload + .get("event") + .and_then(Value::as_str) + .ok_or_else(|| SourceFailure::new("subscription_event_shape"))? + .replace('.', "_"); + if !payload.get("data").is_some_and(Value::is_object) { + return Err(SourceFailure::new("subscription_event_shape")); + } + + // Persist a bounded normalized projection before per-event IPC. Keep + // only fields used by the Iris mapper; raw pane titles, paths, token + // maps, terminal data, and unknown source fields never enter the spool. + sanitize_event_payload(&mut payload)?; + let received_at = Utc::now(); + let sequence = spool + .lock() + .await + .append_unmapped_upstream_event(payload.clone(), received_at) + .map_err(|_| SourceFailure::new("spool_write"))?; + // Event subscriptions can replay retained history and Herdr polls + // independent event kinds without a total source order. Refresh the + // authoritative workspace snapshot before normalizing any event that + // can upsert a thread; never let an old event payload regress it. + if refresh_workspace_snapshot(&event_kind) { + workspace_snapshots = list_workspace_snapshots(socket_path).await?; + } + let current_panes = if refresh_pane_snapshot(&event_kind) { + list_panes(socket_path).await? + } else { + pane_infos.clone() + }; + enrich_workspace_snapshots(&mut payload, &workspace_snapshots, ¤t_panes); + spool + .lock() + .await + .mark_event_ready(sequence, payload) + .map_err(|_| SourceFailure::new("spool_write"))?; + // A subscription starts by replaying retained events. A replayed + // pane.created/pane.moved payload may describe a pane that no longer + // exists, so pane.list—not the event payload—is authoritative for this + // cache. Otherwise the same retained stale frame can force reconnects + // indefinitely. + let current_pane_ids = current_panes.keys().cloned().collect::>(); + let pane_change = changed_pane(&event_kind, &pane_ids, ¤t_pane_ids); + if pane_change { + return Ok(SubscriptionEnd::PaneSetChanged); + } + pane_infos = current_panes; + } +} + +async fn resolve_unmapped_events( + spool: &Arc>, + pane_infos: &BTreeMap, + workspace_snapshots: &BTreeMap, +) -> Result<(), SourceFailure> { + let pending = spool + .lock() + .await + .unmapped_event_payloads() + .map_err(|_| SourceFailure::new("spool_read"))?; + for (sequence, _received_at, mut payload) in pending { + let is_status_snapshot = payload + .pointer("/bridge/status_snapshot") + .and_then(Value::as_bool) + == Some(true); + if is_status_snapshot { + if let Some(pane_id) = payload.pointer("/data/pane_id").and_then(Value::as_str) + && let Some(pane_info) = pane_infos.get(pane_id) + && let Some(agent_status) = pane_info.get("agent_status") + { + payload["data"]["agent_status"] = agent_status.clone(); + } + enrich_status_event(&mut payload, pane_infos); + } + enrich_workspace_snapshots(&mut payload, workspace_snapshots, pane_infos); + strip_tokens(&mut payload); + spool + .lock() + .await + .mark_event_ready(sequence, payload) + .map_err(|_| SourceFailure::new("spool_write"))?; + } + Ok(()) +} + +async fn append_status_snapshots( + spool: &Arc>, + pane_infos: &BTreeMap, + workspace_snapshots: &BTreeMap, +) -> Result<(), SourceFailure> { + for (pane_id, pane_info) in pane_infos { + let agent_status = pane_info + .get("agent_status") + .and_then(Value::as_str) + .ok_or_else(|| SourceFailure::new("pane_list_shape"))?; + let mut payload = json!({ + "event": "pane.agent_status_changed", + "data": {"pane_id": pane_id, "agent_status": agent_status} + }); + sanitize_event_payload(&mut payload)?; + payload["bridge"] = json!({"status_snapshot": true}); + let received_at = Utc::now(); + let sequence = spool + .lock() + .await + .append_unmapped_upstream_event(payload.clone(), received_at) + .map_err(|_| SourceFailure::new("spool_write"))?; + enrich_status_event(&mut payload, pane_infos); + enrich_workspace_snapshots(&mut payload, workspace_snapshots, pane_infos); + spool + .lock() + .await + .mark_event_ready(sequence, payload) + .map_err(|_| SourceFailure::new("spool_write"))?; + } + Ok(()) +} + +async fn verify_protocol(socket_path: &Path) -> Result<(), SourceFailure> { + let response = one_shot_request( + socket_path, + json!({"id": "herdr-bridge-ping", "method": "ping", "params": {}}), + ) + .await?; + let version = response.pointer("/result/version").and_then(Value::as_str); + let protocol = response.pointer("/result/protocol").and_then(Value::as_u64); + if response.pointer("/result/type").and_then(Value::as_str) != Some("pong") + || version != Some(EXPECTED_HERDR_VERSION) + || protocol != Some(EXPECTED_PROTOCOL) + { + return Err(SourceFailure::new("unsupported_herdr_version")); + } + Ok(()) +} + +async fn list_panes(socket_path: &Path) -> Result, SourceFailure> { + let response = one_shot_request( + socket_path, + json!({"id": "herdr-bridge-pane-list", "method": "pane.list", "params": {}}), + ) + .await?; + let panes = response + .pointer("/result/panes") + .and_then(Value::as_array) + .ok_or_else(|| SourceFailure::new("pane_list_shape"))?; + let mut panes_by_id = BTreeMap::new(); + for pane in panes { + let id = pane + .get("pane_id") + .and_then(Value::as_str) + .ok_or_else(|| SourceFailure::new("pane_list_shape"))?; + if pane.get("agent_status").and_then(Value::as_str).is_none() { + return Err(SourceFailure::new("pane_list_shape")); + } + if panes_by_id.insert(id.to_owned(), pane.clone()).is_some() { + return Err(SourceFailure::new("pane_list_duplicate_id")); + } + } + Ok(panes_by_id) +} + +async fn list_workspace_snapshots( + socket_path: &Path, +) -> Result, SourceFailure> { + let response = one_shot_request( + socket_path, + json!({ + "id": "herdr-bridge-workspace-list", + "method": "workspace.list", + "params": {} + }), + ) + .await?; + if response.pointer("/result/type").and_then(Value::as_str) != Some("workspace_list") { + return Err(SourceFailure::new("workspace_list_shape")); + } + let workspaces = response + .pointer("/result/workspaces") + .and_then(Value::as_array) + .ok_or_else(|| SourceFailure::new("workspace_list_shape"))?; + let mut snapshots = BTreeMap::new(); + for workspace in workspaces { + let Some(workspace_id) = workspace.get("workspace_id").and_then(Value::as_str) else { + return Err(SourceFailure::new("workspace_list_shape")); + }; + if workspace.get("label").and_then(Value::as_str).is_none() { + return Err(SourceFailure::new("workspace_list_shape")); + } + let mut snapshot = workspace.clone(); + strip_tokens(&mut snapshot); + if snapshots + .insert(workspace_id.to_owned(), snapshot) + .is_some() + { + return Err(SourceFailure::new("workspace_list_duplicate_id")); + } + } + Ok(snapshots) +} + +fn sanitize_event_payload(payload: &mut Value) -> Result<(), SourceFailure> { + let event = payload + .get("event") + .and_then(Value::as_str) + .ok_or_else(|| SourceFailure::new("subscription_event_shape"))?; + let data = payload + .get("data") + .and_then(Value::as_object) + .ok_or_else(|| SourceFailure::new("subscription_event_shape"))?; + let mut normalized = Map::new(); + for field in [ + "workspace_id", + "workspace_label", + "closed_workspace_id", + "previous_workspace_id", + "pane_id", + "tab_id", + "agent_status", + "agent", + "display_agent", + "host", + "label", + ] { + if let Some(value) = data.get(field).filter(|value| value.is_string()) { + normalized.insert(field.to_owned(), value.clone()); + } + } + for (field, allowed) in [ + ("workspace", &["workspace_id", "label"][..]), + ("created_workspace", &["workspace_id", "label"][..]), + ("tab", &["workspace_id", "tab_id"][..]), + ("pane", &["workspace_id", "pane_id"][..]), + ] { + if let Some(object) = data.get(field).and_then(Value::as_object) { + let mut nested = Map::new(); + for key in allowed { + if let Some(value) = object.get(*key).filter(|value| value.is_string()) { + nested.insert((*key).to_owned(), value.clone()); + } + } + normalized.insert(field.to_owned(), Value::Object(nested)); + } + } + if let Some(ids) = data.get("workspace_ids").and_then(Value::as_array) { + normalized.insert( + "workspace_ids".to_owned(), + Value::Array( + ids.iter() + .filter(|value| value.is_string()) + .cloned() + .collect(), + ), + ); + } + for field in ["workspace_labels", "state_labels"] { + if let Some(labels) = data.get(field).and_then(Value::as_object) { + let labels = labels + .iter() + .filter(|(_, value)| value.is_string()) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(); + normalized.insert(field.to_owned(), Value::Object(labels)); + } + } + *payload = json!({"event": event, "data": normalized}); + Ok(()) +} + +fn enrich_status_event(payload: &mut Value, pane_infos: &BTreeMap) { + let Some(data) = payload.get_mut("data").and_then(Value::as_object_mut) else { + return; + }; + let Some(pane_id) = data.get("pane_id").and_then(Value::as_str) else { + return; + }; + let Some(pane_info) = pane_infos.get(pane_id) else { + return; + }; + let safe_pane = normalized_pane_snapshot(pane_info); + let Some(pane) = safe_pane.as_object() else { + return; + }; + for field in ["workspace_id", "agent", "display_agent", "state_labels"] { + if !data.contains_key(field) + && let Some(value) = pane.get(field) + { + data.insert(field.to_owned(), value.clone()); + } + } +} + +async fn one_shot_request(socket_path: &Path, request: Value) -> Result { + let expected_id = request + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| SourceFailure::new("request_id_missing"))?; + let mut stream = UnixStream::connect(socket_path) + .await + .map_err(|_| SourceFailure::new("socket_connect"))?; + write_json_line(&mut stream, &request) + .await + .map_err(|_| SourceFailure::new("request_write"))?; + let mut reader = BufReader::new(stream); + let line = read_line(&mut reader) + .await + .map_err(|_| SourceFailure::new("response_read"))? + .ok_or_else(|| SourceFailure::new("empty_response"))?; + let response: Value = + serde_json::from_slice(&line).map_err(|_| SourceFailure::new("response_json"))?; + if response.get("id").and_then(Value::as_str) != Some(expected_id) { + return Err(SourceFailure::new("response_id_mismatch")); + } + if response.get("error").is_some() { + return Err(SourceFailure::new("request_rejected")); + } + Ok(response) +} + +async fn send_and_ack( + mut stream: UnixStream, + request_id: &str, + request: &Value, +) -> Result, SourceFailure> { + write_json_line(&mut stream, request) + .await + .map_err(|_| SourceFailure::new("request_write"))?; + let mut reader = BufReader::new(stream); + let line = read_line(&mut reader) + .await + .map_err(|_| SourceFailure::new("ack_read"))? + .ok_or_else(|| SourceFailure::new("empty_ack"))?; + let ack: Value = serde_json::from_slice(&line).map_err(|_| SourceFailure::new("ack_json"))?; + if ack.get("id").and_then(Value::as_str) != Some(request_id) + || ack.pointer("/result/type").and_then(Value::as_str) != Some("subscription_started") + { + return Err(SourceFailure::new("subscription_ack_mismatch")); + } + Ok(reader) +} + +async fn read_line(reader: &mut R) -> io::Result>> +where + R: tokio::io::AsyncBufRead + Unpin, +{ + let mut line = Vec::new(); + loop { + let available = reader.fill_buf().await?; + if available.is_empty() { + if line.is_empty() { + return Ok(None); + } + break; + } + let newline = available.iter().position(|byte| *byte == b'\n'); + let consumed = newline.map_or(available.len(), |index| index + 1); + if line.len().saturating_add(consumed) > MAX_WIRE_LINE_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "wire line exceeds bound", + )); + } + line.extend_from_slice(&available[..consumed]); + reader.consume(consumed); + if newline.is_some() { + break; + } + } + if line.last() == Some(&b'\n') { + line.pop(); + } + if line.last() == Some(&b'\r') { + line.pop(); + } + if line.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "empty JSON line", + )); + } + Ok(Some(line)) +} + +async fn write_json_line(writer: &mut W, value: &Value) -> io::Result<()> +where + W: tokio::io::AsyncWrite + Unpin, +{ + let bytes = serde_json::to_vec(value).map_err(io::Error::other)?; + writer.write_all(&bytes).await?; + writer.write_all(b"\n").await?; + writer.flush().await +} + +fn subscription_request(request_id: &str, pane_ids: &BTreeSet) -> Value { + let mut subscriptions = [ + "workspace.created", + "workspace.updated", + "workspace.metadata_updated", + "workspace.renamed", + "workspace.moved", + "workspace.reordered", + "workspace.closed", + "workspace.focused", + "worktree.created", + "worktree.opened", + "worktree.removed", + "tab.created", + "tab.closed", + "tab.focused", + "tab.renamed", + "tab.moved", + "pane.created", + "pane.closed", + "pane.updated", + "pane.focused", + "pane.moved", + "pane.exited", + "pane.agent_detected", + "layout.updated", + ] + .into_iter() + .map(|kind| json!({"type": kind})) + .collect::>(); + subscriptions.extend( + pane_ids + .iter() + .map(|pane_id| json!({"type": "pane.agent_status_changed", "pane_id": pane_id})), + ); + json!({ + "id": request_id, + "method": "events.subscribe", + "params": {"subscriptions": subscriptions} + }) +} + +fn changed_pane(kind: &str, subscribed: &BTreeSet, current: &BTreeSet) -> bool { + match kind { + "pane_created" | "pane_moved" | "pane_closed" => current != subscribed, + _ => false, + } +} + +fn refresh_pane_snapshot(kind: &str) -> bool { + matches!( + kind, + "pane_created" + | "pane_closed" + | "pane_updated" + | "pane_moved" + | "pane_exited" + | "pane_agent_detected" + | "pane_agent_status_changed" + ) +} + +fn refresh_workspace_snapshot(kind: &str) -> bool { + matches!( + kind, + "workspace_created" + | "workspace_updated" + | "workspace_metadata_updated" + | "workspace_renamed" + | "workspace_moved" + | "workspace_reordered" + | "workspace_closed" + | "workspace_focused" + | "tab_created" + | "tab_closed" + | "tab_focused" + | "tab_renamed" + | "tab_moved" + | "pane_created" + | "pane_updated" + | "pane_focused" + | "pane_moved" + | "pane_exited" + | "pane_agent_detected" + | "pane_agent_status_changed" + ) +} + +fn workspace_ids_in(data: &Map) -> BTreeSet { + let mut ids = BTreeSet::new(); + for value in [ + data.get("workspace_id"), + data.get("workspace") + .and_then(|item| item.get("workspace_id")), + data.get("pane").and_then(|item| item.get("workspace_id")), + data.get("tab").and_then(|item| item.get("workspace_id")), + data.get("created_workspace") + .and_then(|item| item.get("workspace_id")), + data.get("closed_workspace_id"), + data.get("previous_workspace_id"), + ] + .into_iter() + .flatten() + .filter_map(Value::as_str) + { + ids.insert(value.to_owned()); + } + if let Some(workspace_ids) = data.get("workspace_ids").and_then(Value::as_array) { + ids.extend( + workspace_ids + .iter() + .filter_map(Value::as_str) + .map(ToOwned::to_owned), + ); + } + ids +} + +fn primary_workspace_id(data: &Map) -> Option<&str> { + data.get("workspace_id") + .and_then(Value::as_str) + .or_else(|| { + data.get("workspace") + .and_then(|item| item.get("workspace_id")) + .and_then(Value::as_str) + }) + .or_else(|| { + data.get("pane") + .and_then(|item| item.get("workspace_id")) + .and_then(Value::as_str) + }) + .or_else(|| { + data.get("tab") + .and_then(|item| item.get("workspace_id")) + .and_then(Value::as_str) + }) + .or_else(|| { + data.get("created_workspace") + .and_then(|item| item.get("workspace_id")) + .and_then(Value::as_str) + }) +} + +fn normalized_pane_snapshot(value: &Value) -> Value { + let Some(pane) = value.as_object() else { + return Value::Object(Map::new()); + }; + let mut normalized = Map::new(); + for field in [ + "pane_id", + "workspace_id", + "tab_id", + "agent", + "display_agent", + "agent_status", + ] { + if let Some(value) = pane.get(field).filter(|value| value.is_string()) { + normalized.insert(field.to_owned(), value.clone()); + } + } + if let Some(value) = pane.get("focused").filter(|value| value.is_boolean()) { + normalized.insert("focused".to_owned(), value.clone()); + } + if let Some(labels) = pane.get("state_labels").and_then(Value::as_object) { + normalized.insert( + "state_labels".to_owned(), + Value::Object( + labels + .iter() + .filter(|(_, value)| value.is_string()) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + ), + ); + } + Value::Object(normalized) +} + +fn normalized_workspace_snapshot( + workspace_id: &str, + value: &Value, + pane_infos: &BTreeMap, +) -> Value { + let Some(workspace) = value.as_object() else { + return Value::Object(Map::new()); + }; + let mut normalized = Map::new(); + for field in ["workspace_id", "label", "active_tab_id", "agent_status"] { + if let Some(value) = workspace.get(field).filter(|value| value.is_string()) { + normalized.insert(field.to_owned(), value.clone()); + } + } + for field in ["number", "pane_count", "tab_count"] { + if let Some(value) = workspace.get(field).filter(|value| value.is_number()) { + normalized.insert(field.to_owned(), value.clone()); + } + } + if let Some(value) = workspace.get("focused").filter(|value| value.is_boolean()) { + normalized.insert("focused".to_owned(), value.clone()); + } + let panes = pane_infos + .values() + .filter(|pane| pane.get("workspace_id").and_then(Value::as_str) == Some(workspace_id)) + .map(normalized_pane_snapshot) + .collect(); + normalized.insert("panes".to_owned(), Value::Array(panes)); + Value::Object(normalized) +} + +fn enrich_workspace_snapshots( + payload: &mut Value, + workspace_snapshots: &BTreeMap, + pane_infos: &BTreeMap, +) { + let Some(data) = payload.get("data").and_then(Value::as_object) else { + return; + }; + let ids = workspace_ids_in(data); + let primary_id = primary_workspace_id(data).map(ToOwned::to_owned); + let mut current_snapshots = Map::new(); + let mut current_labels = Map::new(); + for id in ids { + let Some(snapshot) = workspace_snapshots.get(&id) else { + continue; + }; + let snapshot = normalized_workspace_snapshot(&id, snapshot, pane_infos); + if let Some(label) = snapshot.get("label").and_then(Value::as_str) { + current_labels.insert(id.clone(), Value::String(label.to_owned())); + } + current_snapshots.insert(id, snapshot.clone()); + } + + if let Some(data) = payload.get_mut("data").and_then(Value::as_object_mut) { + data.remove("workspace_label"); + data.remove("workspace_labels"); + if let Some(label) = primary_id.as_deref().and_then(|id| current_labels.get(id)) { + data.insert("workspace_label".to_owned(), label.clone()); + } + if !current_labels.is_empty() { + data.insert("workspace_labels".to_owned(), Value::Object(current_labels)); + } + } + if let Some(payload) = payload.as_object_mut() { + let bridge = payload + .entry("bridge") + .or_insert_with(|| Value::Object(Map::new())); + if let Some(bridge) = bridge.as_object_mut() { + bridge.insert( + "workspace_snapshots".to_owned(), + Value::Object(current_snapshots), + ); + } + } +} + +fn strip_tokens(value: &mut Value) { + match value { + Value::Object(object) => { + object.retain(|key, _| !key.eq_ignore_ascii_case("tokens")); + for value in object.values_mut() { + strip_tokens(value); + } + } + Value::Array(values) => values.iter_mut().for_each(strip_tokens), + _ => {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeSet; + + use tempfile::tempdir; + use tokio::net::UnixListener; + + use crate::spool::Spool; + + const ACK: &str = include_str!("../tests/fixtures/herdr-v0.8.0-subscription-ack.json"); + const STATUS_EVENT: &str = + include_str!("../tests/fixtures/herdr-v0.8.0-pane-agent-status-event.json"); + const CLOSED_PANE_CREATED_EVENT: &str = + include_str!("../tests/fixtures/herdr-v0.8.0-pane-created-event.json"); + const WORKSPACE_LIST_RESPONSE: &str = + include_str!("../tests/fixtures/herdr-v0.8.0-workspace-list-response.json"); + + async fn read_request(stream: &mut UnixStream) -> Value { + let mut reader = BufReader::new(stream); + let line = read_line(&mut reader).await.unwrap().unwrap(); + serde_json::from_slice(&line).unwrap() + } + + async fn respond_workspace_list(listener: &UnixListener) { + respond_workspace_list_with_status(listener, "working").await; + } + + async fn respond_workspace_list_with_status(listener: &UnixListener, status: &str) { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_request(&mut stream).await; + assert_eq!(request["method"], "workspace.list"); + let mut response: Value = serde_json::from_str(WORKSPACE_LIST_RESPONSE).unwrap(); + response["id"] = request["id"].clone(); + response["result"]["workspaces"][0]["agent_status"] = json!(status); + super::write_json_line(&mut stream, &response) + .await + .unwrap(); + } + + async fn respond_pane_list(listener: &UnixListener, panes: Value) { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_request(&mut stream).await; + assert_eq!(request["method"], "pane.list"); + super::write_json_line( + &mut stream, + &json!({ + "id": request["id"], + "result": {"type": "pane_list", "panes": panes} + }), + ) + .await + .unwrap(); + } + + fn synthetic_pane() -> Value { + json!({ + "pane_id": "pane-1", + "terminal_id": "terminal-1", + "workspace_id": "workspace-1", + "tab_id": "tab-1", + "focused": true, + "agent": "claude", + "title": "fixture task", + "display_agent": "Claude", + "agent_status": "working", + "state_labels": {"phase": "working"}, + "tokens": {}, + "revision": 1 + }) + } + + fn synthetic_idle_pane() -> Value { + json!({ + "pane_id": "pane-1", + "terminal_id": "terminal-1", + "workspace_id": "workspace-1", + "tab_id": "tab-1", + "focused": true, + "agent": "claude", + "display_agent": "Claude", + "agent_status": "idle", + "tokens": {}, + "revision": 2 + }) + } + + #[tokio::test] + async fn read_line_rejects_oversized_frames_at_the_configured_bound() { + let mut frame = vec![b'x'; MAX_WIRE_LINE_BYTES + 1]; + frame.push(b'\n'); + let mut reader = BufReader::new(frame.as_slice()); + let error = read_line(&mut reader).await.unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert_eq!(error.to_string(), "wire line exceeds bound"); + } + + #[test] + fn source_event_projection_discards_raw_pane_details_before_spooling() { + let mut payload = json!({ + "event": "pane.agent_status_changed", + "data": { + "pane_id": "pane-1", + "workspace_id": "workspace-1", + "agent_status": "idle", + "agent": "claude", + "display_agent": "Claude", + "state_labels": {"phase": "idle", "bad": 7}, + "title": "private pane title", + "cwd": "/private/worktree", + "tokens": {"access": "fixture-only-do-not-forward"}, + "private_payload": "fixture-only-do-not-forward" + }, + "sequence": 999 + }); + sanitize_event_payload(&mut payload).unwrap(); + assert_eq!( + payload, + json!({ + "event": "pane.agent_status_changed", + "data": { + "pane_id": "pane-1", + "workspace_id": "workspace-1", + "agent_status": "idle", + "agent": "claude", + "display_agent": "Claude", + "state_labels": {"phase": "idle"} + } + }) + ); + assert!(!payload.to_string().contains("private")); + assert!(!payload.to_string().contains("tokens")); + } + + #[test] + fn pane_move_projection_keeps_both_workspace_ids_and_drops_private_pane_data() { + let mut payload = json!({ + "event": "pane.moved", + "data": { + "previous_workspace_id": "workspace-old", + "pane": { + "workspace_id": "workspace-new", + "pane_id": "pane-1", + "title": "private pane title", + "cwd": "/private/worktree", + "tokens": {"access": "fixture-only-do-not-forward"} + } + } + }); + + sanitize_event_payload(&mut payload).unwrap(); + + let data = payload["data"].as_object().unwrap(); + assert_eq!(data["previous_workspace_id"], "workspace-old"); + assert_eq!(data["pane"]["workspace_id"], "workspace-new"); + assert_eq!( + workspace_ids_in(data), + BTreeSet::from(["workspace-new".to_owned(), "workspace-old".to_owned()]) + ); + let workspace_snapshots = BTreeMap::from([ + ( + "workspace-old".to_owned(), + json!({"workspace_id": "workspace-old", "label": "Old"}), + ), + ( + "workspace-new".to_owned(), + json!({"workspace_id": "workspace-new", "label": "New"}), + ), + ]); + let pane_infos = BTreeMap::from([( + "pane-1".to_owned(), + json!({ + "workspace_id": "workspace-new", + "pane_id": "pane-1", + "tab_id": "tab-1", + "agent_status": "working" + }), + )]); + enrich_workspace_snapshots(&mut payload, &workspace_snapshots, &pane_infos); + let snapshots = payload["bridge"]["workspace_snapshots"] + .as_object() + .unwrap(); + assert_eq!(snapshots.len(), 2); + assert_eq!(snapshots["workspace-new"]["panes"][0]["pane_id"], "pane-1"); + assert_eq!(snapshots["workspace-old"]["panes"], json!([])); + assert!(!payload.to_string().contains("private")); + assert!(!payload.to_string().contains("tokens")); + } + + #[test] + fn synthetic_status_snapshot_enrichment_uses_only_normalized_fields() { + let mut event = json!({ + "event": "pane.agent_status_changed", + "data": {"pane_id": "pane-1", "agent_status": "working"} + }); + let pane_infos = BTreeMap::from([( + "pane-1".to_owned(), + json!({ + "workspace_id": "workspace-1", + "agent": "claude", + "display_agent": "Claude", + "title": "private pane title", + "state_labels": {"phase": "working"} + }), + )]); + + enrich_status_event(&mut event, &pane_infos); + + assert_eq!(event["data"]["workspace_id"], "workspace-1"); + assert_eq!(event["data"]["agent"], "claude"); + assert_eq!(event["data"]["display_agent"], "Claude"); + assert_eq!(event["data"]["state_labels"]["phase"], "working"); + assert!(event["data"].get("title").is_none()); + } + + #[test] + fn stale_workspace_event_uses_the_authoritative_snapshot() { + let snapshots = BTreeMap::from([( + "workspace-1".to_owned(), + json!({ + "workspace_id": "workspace-1", + "label": "Current label", + "pane_count": 3, + "tokens": {"synthetic_private_value": "fixture-only-do-not-forward"} + }), + )]); + let mut rename = json!({ + "event": "workspace_renamed", + "data": {"workspace_id": "workspace-1", "label": "Stale retained label"} + }); + + enrich_workspace_snapshots(&mut rename, &snapshots, &BTreeMap::new()); + strip_tokens(&mut rename); + + assert_eq!(rename["data"]["workspace_label"], "Current label"); + assert_eq!( + rename["bridge"]["workspace_snapshots"]["workspace-1"]["label"], + "Current label" + ); + assert!(!rename.to_string().contains("fixture-only-do-not-forward")); + assert!( + rename["bridge"]["workspace_snapshots"]["workspace-1"] + .get("worktree") + .is_none() + ); + } + + #[test] + fn workspace_snapshot_refresh_covers_all_thread_mutation_events() { + for kind in [ + "workspace_created", + "workspace_renamed", + "workspace_closed", + "tab_created", + "pane_created", + "pane_agent_status_changed", + ] { + assert!(refresh_workspace_snapshot(kind), "{kind}"); + } + assert!(refresh_pane_snapshot("pane_agent_status_changed")); + for kind in [ + "pane_output_changed", + "pane_scroll_changed", + "worktree_opened", + ] { + assert!(!refresh_workspace_snapshot(kind), "{kind}"); + } + } + + #[tokio::test] + async fn durably_captured_unmapped_event_recovers_from_current_startup_snapshots() { + let dir = tempdir().unwrap(); + let spool = Arc::new(Mutex::new(Spool::open(dir.path().join("state")).unwrap())); + let received_at = Utc::now() - chrono::Duration::seconds(10); + let raw_event = json!({ + "event": "workspace.renamed", + "data": {"workspace_id": "workspace-1", "workspace_label": "stale label"} + }); + spool + .lock() + .await + .append_unmapped_upstream_event(raw_event, received_at) + .unwrap(); + let pane_infos = BTreeMap::new(); + let workspace_snapshots = BTreeMap::from([( + "workspace-1".to_owned(), + json!({ + "workspace_id": "workspace-1", + "label": "authoritative current label", + "tokens": {"fixture_only": "do-not-forward"} + }), + )]); + + resolve_unmapped_events(&spool, &pane_infos, &workspace_snapshots) + .await + .unwrap(); + let mut spool_guard = spool.lock().await; + assert!(spool_guard.unmapped_event_payloads().unwrap().is_empty()); + assert!(spool_guard.flush_due(Utc::now()).unwrap()); + let batch = spool_guard.oldest_batch().unwrap().unwrap(); + assert!(batch.body.contains("authoritative current label")); + assert!(!batch.body.contains("stale label")); + assert!(!batch.body.contains("do-not-forward")); + drop(spool_guard); + } + + #[tokio::test] + async fn unmapped_source_status_recovery_preserves_event_context_and_omitted_labels() { + let dir = tempdir().unwrap(); + let spool = Arc::new(Mutex::new(Spool::open(dir.path().join("state")).unwrap())); + let mut source_event = json!({ + "event": "pane.agent_status_changed", + "data": { + "pane_id": "pane-1", + "workspace_id": "workspace-at-event", + "agent_status": "working", + "agent": "event-agent", + "display_agent": "Event Agent", + "title": "private title", + "tokens": {"fixture": "do-not-forward"} + } + }); + sanitize_event_payload(&mut source_event).unwrap(); + spool + .lock() + .await + .append_unmapped_upstream_event(source_event, Utc::now()) + .unwrap(); + let pane_infos = BTreeMap::from([( + "pane-1".to_owned(), + json!({ + "pane_id": "pane-1", + "workspace_id": "workspace-current", + "agent_status": "idle", + "agent": "current-agent", + "display_agent": "Current Agent", + "state_labels": {"phase": "current"} + }), + )]); + let workspace_snapshots = BTreeMap::from([ + ( + "workspace-at-event".to_owned(), + json!({"workspace_id": "workspace-at-event", "label": "At event"}), + ), + ( + "workspace-current".to_owned(), + json!({"workspace_id": "workspace-current", "label": "Current"}), + ), + ]); + + resolve_unmapped_events(&spool, &pane_infos, &workspace_snapshots) + .await + .unwrap(); + + let mut spool = spool.lock().await; + assert!(spool.unmapped_event_payloads().unwrap().is_empty()); + assert!( + spool + .flush_due(Utc::now() + chrono::Duration::seconds(5)) + .unwrap() + ); + let batch = spool.oldest_batch().unwrap().unwrap(); + drop(spool); + let value: Value = serde_json::from_str(&batch.body).unwrap(); + let messages = value["mutations"] + .as_array() + .unwrap() + .iter() + .filter(|mutation| mutation["kind"] == "append_message") + .collect::>(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0]["message"]["body"], "event-agent: working"); + assert_eq!( + messages[0]["message"]["metadata"]["state_labels"], + json!({}) + ); + assert!(!batch.body.contains("current-agent")); + assert!(!batch.body.contains("private title")); + assert!(!batch.body.contains("do-not-forward")); + } + + #[tokio::test] + async fn interrupted_status_snapshot_reconciles_to_current_pane_state() { + let dir = tempdir().unwrap(); + let spool = Arc::new(Mutex::new(Spool::open(dir.path().join("state")).unwrap())); + spool + .lock() + .await + .append_unmapped_upstream_event( + json!({ + "event": "pane.agent_status_changed", + "data": {"pane_id": "pane-1", "agent_status": "working"}, + "bridge": {"status_snapshot": true} + }), + Utc::now() - chrono::Duration::seconds(10), + ) + .unwrap(); + let pane_infos = BTreeMap::from([("pane-1".to_owned(), synthetic_idle_pane())]); + let workspace_snapshots = BTreeMap::from([( + "workspace-1".to_owned(), + json!({"workspace_id": "workspace-1", "label": "Iris Pilot"}), + )]); + + resolve_unmapped_events(&spool, &pane_infos, &workspace_snapshots) + .await + .unwrap(); + let mut spool = spool.lock().await; + assert!(spool.unmapped_event_payloads().unwrap().is_empty()); + assert!(spool.flush_due(Utc::now()).unwrap()); + let batch = spool.oldest_batch().unwrap().unwrap(); + drop(spool); + let value: Value = serde_json::from_str(&batch.body).unwrap(); + let messages = value["mutations"] + .as_array() + .unwrap() + .iter() + .filter(|mutation| mutation["kind"] == "append_message") + .collect::>(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0]["message"]["body"], "claude: idle"); + assert!(!batch.body.contains("private pane title")); + assert!(!batch.body.contains("/private/")); + } + + #[tokio::test] + async fn source_derived_ack_stream_and_eof_are_consumed_and_spooled() { + let dir = tempdir().unwrap(); + let socket_path = dir.path().join("herdr.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_request(&mut stream).await; + assert_eq!(request["method"], "events.subscribe"); + assert_eq!(request["id"], "fixture-subscription"); + assert!( + request["params"]["subscriptions"] + .as_array() + .unwrap() + .iter() + .any(|item| { + item["type"] == "pane.agent_status_changed" && item["pane_id"] == "pane-1" + }) + ); + let ack = ACK.replace("fixture-id", "fixture-subscription"); + stream.write_all(ack.as_bytes()).await.unwrap(); + stream.write_all(b"\n").await.unwrap(); + stream.write_all(STATUS_EVENT.as_bytes()).await.unwrap(); + stream.write_all(b"\n").await.unwrap(); + stream.shutdown().await.unwrap(); + }); + + let stream = UnixStream::connect(&socket_path).await.unwrap(); + let request_id = "fixture-subscription"; + let pane_ids = BTreeSet::from(["pane-1".to_owned()]); + let request = subscription_request(request_id, &pane_ids); + let mut reader = send_and_ack(stream, request_id, &request).await.unwrap(); + let line = read_line(&mut reader).await.unwrap().unwrap(); + let mut event: Value = serde_json::from_slice(&line).unwrap(); + sanitize_event_payload(&mut event).unwrap(); + let spool = Arc::new(Mutex::new(Spool::open(dir.path().join("state")).unwrap())); + spool + .lock() + .await + .append_upstream_event(event, Utc::now()) + .unwrap(); + assert!(read_line(&mut reader).await.unwrap().is_none()); + server.await.unwrap(); + assert_eq!(spool.lock().await.pending_event_count(), 1); + let event_files = std::fs::read_dir(dir.path().join("state")) + .unwrap() + .filter_map(std::result::Result::ok) + .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "json")) + .count(); + assert_eq!(event_files, 2, "event plus durable state metadata"); + } + + #[tokio::test] + async fn new_subscription_captures_current_pane_status_without_a_transition() { + let dir = tempdir().unwrap(); + let socket_path = dir.path().join("herdr.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + let server = tokio::spawn(async move { + let (mut ping_stream, _) = listener.accept().await.unwrap(); + let ping = read_request(&mut ping_stream).await; + super::write_json_line( + &mut ping_stream, + &json!({ + "id": ping["id"], + "result": {"type": "pong", "version": "0.8.0", "protocol": 19} + }), + ) + .await + .unwrap(); + drop(ping_stream); + + respond_pane_list(&listener, json!([synthetic_pane()])).await; + respond_workspace_list(&listener).await; + + let (mut event_stream, _) = listener.accept().await.unwrap(); + let subscribe = read_request(&mut event_stream).await; + let status_subscription = subscribe["params"]["subscriptions"] + .as_array() + .unwrap() + .iter() + .find(|item| { + item["type"] == "pane.agent_status_changed" && item["pane_id"] == "pane-1" + }) + .unwrap(); + assert!(status_subscription.get("agent_status").is_none()); + let ack = ACK.replace("fixture-id", subscribe["id"].as_str().unwrap()); + event_stream.write_all(ack.as_bytes()).await.unwrap(); + event_stream.write_all(b"\n").await.unwrap(); + + // No transition is sent: the current pane snapshot must still + // produce a normalized status observation on connect. + respond_pane_list(&listener, json!([synthetic_pane()])).await; + respond_workspace_list(&listener).await; + event_stream.shutdown().await.unwrap(); + }); + + let spool = Arc::new(Mutex::new(Spool::open(dir.path().join("state")).unwrap())); + assert_eq!( + consume_subscription(&socket_path, &spool).await.unwrap(), + SubscriptionEnd::PeerClosed + ); + server.await.unwrap(); + let mut spool = spool.lock().await; + assert_eq!(spool.pending_event_count(), 1); + assert!( + spool + .flush_due(Utc::now() + chrono::Duration::seconds(6)) + .unwrap() + ); + let batch_body = spool.oldest_batch().unwrap().unwrap().body; + drop(spool); + let value: Value = serde_json::from_str(&batch_body).unwrap(); + let messages = value["mutations"] + .as_array() + .unwrap() + .iter() + .filter(|mutation| mutation["kind"] == "append_message") + .collect::>(); + assert_eq!(messages.len(), 1); + assert!( + messages[0]["message"]["body"] + .as_str() + .is_some_and(|body| body.contains("claude: working")) + ); + } + + #[allow(clippy::too_many_lines)] + #[tokio::test] + async fn protocol_19_transition_is_spooled_and_reconnect_snapshots_current_status() { + let dir = tempdir().unwrap(); + let socket_path = dir.path().join("herdr.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + let server = tokio::spawn(async move { + for attempt in 0..2 { + let current_pane = if attempt == 0 { + synthetic_pane() + } else { + synthetic_idle_pane() + }; + let (mut ping_stream, _) = listener.accept().await.unwrap(); + let ping_request = read_request(&mut ping_stream).await; + assert_eq!(ping_request["method"], "ping"); + super::write_json_line( + &mut ping_stream, + &json!({ + "id": ping_request["id"], + "result": {"type": "pong", "version": "0.8.0", "protocol": 19} + }), + ) + .await + .unwrap(); + drop(ping_stream); + + respond_pane_list(&listener, json!([current_pane.clone()])).await; + let workspace_status = if attempt == 0 { "working" } else { "idle" }; + respond_workspace_list_with_status(&listener, workspace_status).await; + + let (mut event_stream, _) = listener.accept().await.unwrap(); + let subscribe_request = read_request(&mut event_stream).await; + assert_eq!(subscribe_request["method"], "events.subscribe"); + assert!( + subscribe_request["params"]["subscriptions"] + .as_array() + .unwrap() + .iter() + .any(|item| { + item["type"] == "pane.agent_status_changed" + && item["pane_id"] == "pane-1" + }) + ); + let ack = ACK.replace("fixture-id", subscribe_request["id"].as_str().unwrap()); + event_stream.write_all(ack.as_bytes()).await.unwrap(); + event_stream.write_all(&[10]).await.unwrap(); + if attempt == 0 { + event_stream + .write_all(STATUS_EVENT.as_bytes()) + .await + .unwrap(); + event_stream.write_all(&[10]).await.unwrap(); + } + event_stream.shutdown().await.unwrap(); + respond_pane_list(&listener, json!([current_pane])).await; + let current_status = if attempt == 0 { "working" } else { "idle" }; + respond_workspace_list_with_status(&listener, current_status).await; + if attempt == 0 { + // The genuine status transition refreshes workspace and + // pane snapshots before its sanitized record is sealed. + respond_workspace_list_with_status(&listener, "idle").await; + respond_pane_list(&listener, json!([synthetic_idle_pane()])).await; + } + } + }); + + let state_dir = dir.path().join("state"); + let spool = Arc::new(Mutex::new(Spool::open(state_dir.clone()).unwrap())); + assert_eq!( + consume_subscription(&socket_path, &spool).await.unwrap(), + SubscriptionEnd::PeerClosed + ); + drop(spool); + let spool = Arc::new(Mutex::new(Spool::open(state_dir.clone()).unwrap())); + assert_eq!( + consume_subscription(&socket_path, &spool).await.unwrap(), + SubscriptionEnd::PeerClosed + ); + server.await.unwrap(); + let mut spool = spool.lock().await; + assert_eq!(spool.pending_event_count(), 3); + let stored_events = std::fs::read_dir(&state_dir) + .unwrap() + .filter_map(std::result::Result::ok) + .filter(|entry| entry.file_name().to_string_lossy().starts_with("event-")) + .map(|entry| { + serde_json::from_slice::(&std::fs::read(entry.path()).unwrap()).unwrap() + }) + .collect::>(); + assert_eq!(stored_events.len(), 3); + assert!(stored_events.iter().all(|event| { + !event.to_string().contains("fixture-only-do-not-forward") + && !event.to_string().contains("tokens") + && !event + .to_string() + .contains("private-pane-title-do-not-forward") + && !event.to_string().contains("/private/") + && !event.to_string().contains("worktree") + })); + assert!( + spool + .flush_due(Utc::now() + chrono::Duration::seconds(5)) + .unwrap() + ); + let batch = spool.oldest_batch().unwrap().unwrap(); + let value: Value = serde_json::from_str(&batch.body).unwrap(); + assert_eq!(value["audit"]["metadata"]["dropped_count"], 0); + assert_eq!( + value["audit"]["metadata"]["suppressed_thread_upsert_count"], + 0 + ); + assert!( + value["mutations"] + .as_array() + .unwrap() + .iter() + .any(|mutation| { + mutation["kind"] == "upsert_thread" && mutation["title"] == "Iris Pilot" + }) + ); + assert!(!batch.body.contains("fixture-only-do-not-forward")); + assert!(!batch.body.contains("private-pane-title-do-not-forward")); + assert!(!batch.body.contains("/private/")); + assert!(!batch.body.contains("worktree")); + let threads = value["mutations"] + .as_array() + .unwrap() + .iter() + .filter(|mutation| mutation["kind"] == "upsert_thread") + .collect::>(); + assert!(!threads.is_empty()); + assert!(threads.iter().all(|thread| { + thread["metadata"]["panes"][0]["title"].is_null() + && thread["metadata"]["panes"][0]["pane_id"] == "pane-1" + && thread["metadata"]["panes"][0]["agent_status"].is_string() + })); + let messages = value["mutations"] + .as_array() + .unwrap() + .iter() + .filter(|mutation| mutation["kind"] == "append_message") + .collect::>(); + assert_eq!(messages.len(), 3); + let idle_messages = messages + .iter() + .filter(|mutation| mutation["message"]["body"] == "claude: idle") + .collect::>(); + assert_eq!(idle_messages.len(), 2); + assert_ne!( + idle_messages[0]["message"]["source_id"], + idle_messages[1]["message"]["source_id"] + ); + assert!( + messages.iter().any(|mutation| { + mutation["message"]["body"] == "claude: working (phase=working)" + }) + ); + assert!( + messages + .iter() + .all(|mutation| { mutation["message"]["metadata"].get("data").is_none() }) + ); + drop(spool); + } + + #[tokio::test] + async fn retained_pane_created_for_a_closed_pane_does_not_trigger_refresh() { + let dir = tempdir().unwrap(); + let socket_path = dir.path().join("herdr.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + let server = tokio::spawn(async move { + let (mut ping_stream, _) = listener.accept().await.unwrap(); + let ping = read_request(&mut ping_stream).await; + super::write_json_line( + &mut ping_stream, + &json!({ + "id": ping["id"], + "result": {"type": "pong", "version": "0.8.0", "protocol": 19} + }), + ) + .await + .unwrap(); + drop(ping_stream); + + let (mut panes_stream, _) = listener.accept().await.unwrap(); + let panes = read_request(&mut panes_stream).await; + assert_eq!(panes["method"], "pane.list"); + super::write_json_line( + &mut panes_stream, + &json!({"id": panes["id"], "result": {"type": "pane_list", "panes": []}}), + ) + .await + .unwrap(); + drop(panes_stream); + + respond_workspace_list(&listener).await; + + let (mut event_stream, _) = listener.accept().await.unwrap(); + let subscribe = read_request(&mut event_stream).await; + assert_eq!(subscribe["method"], "events.subscribe"); + let ack = ACK.replace("fixture-id", subscribe["id"].as_str().unwrap()); + event_stream.write_all(ack.as_bytes()).await.unwrap(); + event_stream.write_all(b"\n").await.unwrap(); + respond_pane_list(&listener, json!([])).await; + respond_workspace_list(&listener).await; + event_stream + .write_all(CLOSED_PANE_CREATED_EVENT.trim_end().as_bytes()) + .await + .unwrap(); + event_stream.write_all(b"\n").await.unwrap(); + respond_workspace_list(&listener).await; + + // The retained pane.created frame is stale. The bridge must ask + // for current workspace metadata and pane.list before deciding + // whether to refresh the pane-scoped subscription. + let (mut refresh_stream, _) = listener.accept().await.unwrap(); + let refresh = read_request(&mut refresh_stream).await; + assert_eq!(refresh["method"], "pane.list"); + super::write_json_line( + &mut refresh_stream, + &json!({ + "id": refresh["id"], + "result": {"type": "pane_list", "panes": []} + }), + ) + .await + .unwrap(); + drop(refresh_stream); + event_stream.shutdown().await.unwrap(); + }); + + let spool = Arc::new(Mutex::new(Spool::open(dir.path().join("state")).unwrap())); + assert_eq!( + consume_subscription(&socket_path, &spool).await.unwrap(), + SubscriptionEnd::PeerClosed + ); + server.await.unwrap(); + assert_eq!(spool.lock().await.pending_event_count(), 1); + } + + #[test] + fn planned_pane_refresh_is_immediate_but_peer_close_retries_after_two_seconds() { + assert_eq!( + reconnect_delay(SubscriptionEnd::PaneSetChanged), + Duration::ZERO + ); + assert_eq!( + reconnect_delay(SubscriptionEnd::PeerClosed), + RECONNECT_DELAY + ); + } + + #[test] + fn newly_created_pane_requests_subscription_refresh() { + let subscribed = BTreeSet::from(["old-pane".to_owned()]); + let current = BTreeSet::from(["old-pane".to_owned(), "new-pane".to_owned()]); + assert!(changed_pane("pane_created", &subscribed, ¤t)); + assert!(!changed_pane("pane_created", ¤t, ¤t)); + assert!(!changed_pane( + "pane_created", + &subscribed, + &BTreeSet::from(["old-pane".to_owned()]) + )); + } + + #[test] + fn cross_workspace_pane_move_keeps_the_same_pane_id_subscription() { + let pane_id = BTreeSet::from(["pane-1".to_owned()]); + assert!(!changed_pane("pane_moved", &pane_id, &pane_id)); + let replacement_id = BTreeSet::from(["replacement-pane".to_owned()]); + assert!(changed_pane("pane_moved", &pane_id, &replacement_id)); + } + + #[test] + fn every_protocol_event_subscription_is_type_scoped_and_status_is_pane_scoped() { + let request = subscription_request("request", &BTreeSet::from(["pane-1".to_owned()])); + let subscriptions = request["params"]["subscriptions"].as_array().unwrap(); + assert!( + subscriptions + .iter() + .any(|entry| entry["type"] == "workspace.created") + ); + assert!(subscriptions.iter().any(|entry| { + entry["type"] == "pane.agent_status_changed" && entry["pane_id"] == "pane-1" + })); + assert!( + subscriptions + .iter() + .filter(|entry| entry["type"] == "pane.agent_status_changed") + .all(|entry| entry.get("agent_status").is_none()) + ); + } +} diff --git a/crates/iris-herdr-bridge/src/spool.rs b/crates/iris-herdr-bridge/src/spool.rs new file mode 100644 index 0000000..adaddd3 --- /dev/null +++ b/crates/iris-herdr-bridge/src/spool.rs @@ -0,0 +1,1140 @@ +use std::collections::HashSet; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Duration, Utc}; +use iris_core::{ + AuditAction, AuditEvent, Contact, IngestBatch, IngestMutation, Message, MessageKind, Thread, +}; +use iris_providers::herdr::{HerdrEvent, map_ingest_batch, map_workspace_snapshot}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use uuid::Uuid; + +const MAX_PENDING_EVENTS: usize = 10_000; +const MAX_BATCH_EVENTS: usize = 20; +const MAX_BATCH_BODY_BYTES: usize = 1024 * 1024; +const MAX_BATCH_AGE: Duration = Duration::seconds(5); +const HEARTBEAT_INTERVAL: Duration = Duration::hours(1); +const HEARTBEAT_THREAD_ID: Uuid = Uuid::from_u128(0x0b94_4f64_67a5_5f3b_87d9_119a_9b80_2a31); +const HEARTBEAT_CONTACT_ID: Uuid = Uuid::from_u128(0x18d9_7a44_2a83_58a7_98c5_d79d_19a6_30d4); + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct EventRecord { + sequence: u64, + event_id: String, + received_at: DateTime, + #[serde(default = "mapping_ready_by_default")] + mapping_ready: bool, + payload: Value, +} + +const fn mapping_ready_by_default() -> bool { + true +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +/// Immutable HTTP request persisted before the first delivery attempt. +pub struct BatchRecord { + /// Monotonic sequence used to order pending batch files. + pub sequence: u64, + /// Number of source events represented by this batch. + pub event_count: usize, + event_sequences: Vec, + /// Exact serialized `IngestBatch` body reused across retries/restarts. + pub body: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct BatchIndex { + event_count: usize, + event_sequences: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct SpoolState { + next_sequence: u64, + #[serde(default)] + pending_event_count: usize, + last_upstream_event_at: DateTime, + last_heartbeat_at: Option>, +} + +/// Filesystem-backed bounded outbox. Every request body is persisted exactly +/// once before any network attempt and reused byte-for-byte until accepted. +pub struct Spool { + directory: PathBuf, + state: SpoolState, + max_events: usize, + blocked_oversized_event: Option, + has_warned_oversized_event: bool, +} + +impl Spool { + /// Open a private outbox and reconcile interrupted batch commits. + pub fn open(directory: PathBuf) -> io::Result { + Self::open_with_limit(directory, MAX_PENDING_EVENTS) + } + + fn open_with_limit(directory: PathBuf, max_events: usize) -> io::Result { + fs::create_dir_all(&directory)?; + set_private_directory(&directory)?; + remove_abandoned_temps(&directory)?; + let now = Utc::now(); + let state_path = directory.join("bridge-state.json"); + let mut state = match fs::read(&state_path) { + Ok(bytes) => serde_json::from_slice::(&bytes) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?, + Err(error) if error.kind() == io::ErrorKind::NotFound => SpoolState { + next_sequence: 1, + pending_event_count: 0, + last_upstream_event_at: now, + last_heartbeat_at: None, + }, + Err(error) => return Err(error), + }; + let max_on_disk = max_sequence(&directory)?; + state.next_sequence = state.next_sequence.max(max_on_disk.saturating_add(1)); + let mut spool = Self { + directory, + state, + max_events, + blocked_oversized_event: None, + has_warned_oversized_event: false, + }; + spool.reconcile_batched_events()?; + spool.state.pending_event_count = spool.count_pending_events()?; + spool.enforce_bound()?; + spool.persist_state()?; + Ok(spool) + } + + /// Durably append an upstream event that needs no further protocol enrichment. + #[cfg(test)] + pub fn append_upstream_event( + &mut self, + payload: Value, + received_at: DateTime, + ) -> io::Result<()> { + self.append_event(payload, received_at, true).map(|_| ()) + } + + /// Capture an upstream event durably before protocol enrichment that can fail. + pub(crate) fn append_unmapped_upstream_event( + &mut self, + payload: Value, + received_at: DateTime, + ) -> io::Result { + self.append_event(payload, received_at, false) + } + + fn append_event( + &mut self, + payload: Value, + received_at: DateTime, + mapping_ready: bool, + ) -> io::Result { + let sequence = self.allocate_sequence(); + let record = EventRecord { + sequence, + event_id: Uuid::new_v4().to_string(), + received_at, + mapping_ready, + payload, + }; + self.write_event(&record)?; + self.state.pending_event_count = self.state.pending_event_count.saturating_add(1); + self.state.last_upstream_event_at = received_at; + self.persist_state()?; + self.enforce_bound()?; + Ok(sequence) + } + + pub(crate) fn unmapped_event_payloads(&self) -> io::Result, Value)>> { + Ok(self + .event_records()? + .into_iter() + .filter(|event| !event.mapping_ready) + .map(|event| (event.sequence, event.received_at, event.payload)) + .collect()) + } + + pub(crate) fn mark_event_ready(&self, sequence: u64, payload: Value) -> io::Result<()> { + let path = self.event_path(sequence); + let mut record: EventRecord = read_json(&path)?; + if record.sequence != sequence { + return Err(io::Error::other("spool event sequence mismatch")); + } + record.payload = payload; + record.mapping_ready = true; + self.write_event(&record) + } + + /// Enqueue the synthetic health message after an hour without source events. + pub fn maybe_append_heartbeat(&mut self, now: DateTime) -> io::Result { + let idle_since = self + .state + .last_heartbeat_at + .unwrap_or(self.state.last_upstream_event_at) + .max(self.state.last_upstream_event_at); + if now.signed_duration_since(idle_since) < HEARTBEAT_INTERVAL { + return Ok(false); + } + let sequence = self.allocate_sequence(); + let payload = json!({ + "event": "bridge_heartbeat", + "data": {"bridge": "herdr-iris-bridge", "received_at": now} + }); + self.write_event(&EventRecord { + sequence, + event_id: Uuid::new_v4().to_string(), + received_at: now, + mapping_ready: true, + payload, + })?; + self.state.pending_event_count = self.state.pending_event_count.saturating_add(1); + self.state.last_heartbeat_at = Some(now); + self.persist_state()?; + self.enforce_bound()?; + Ok(true) + } + + /// Persist a due immutable batch and remove its source records only after the + /// batch file is durable. Startup reconciles a crash between these steps. + pub fn flush_due(&mut self, now: DateTime) -> io::Result { + let all_events = self.event_records()?; + let ready_count = all_events + .iter() + .take_while(|event| event.mapping_ready) + .count(); + let events = &all_events[..ready_count]; + let Some(oldest) = events.first() else { + return Ok(false); + }; + if self.blocked_oversized_event == Some(oldest.sequence) { + return Ok(false); + } + self.blocked_oversized_event = None; + let due = events.len() >= MAX_BATCH_EVENTS + || now.signed_duration_since(oldest.received_at) >= MAX_BATCH_AGE; + if !due { + return Ok(false); + } + let mut selected = events + .iter() + .take(MAX_BATCH_EVENTS) + .cloned() + .collect::>(); + let sequence = self.state.next_sequence; + let batch = loop { + let candidate = build_batch(&selected, sequence)?; + if candidate.body.len() <= MAX_BATCH_BODY_BYTES { + break candidate; + } + if selected.len() > 1 { + selected.pop(); + continue; + } + let oversized = selected + .first() + .ok_or_else(|| io::Error::other("due event batch unexpectedly empty"))?; + if !self.has_warned_oversized_event { + tracing::warn!( + source_event_sequence = oversized.sequence, + body_limit_bytes = MAX_BATCH_BODY_BYTES, + "Herdr event exceeds Iris's ingest body limit; retaining it at the spool head" + ); + self.has_warned_oversized_event = true; + } + self.blocked_oversized_event = Some(oversized.sequence); + return Ok(false); + }; + let file_path = self.batch_path(batch.sequence); + write_atomic( + &file_path, + &serde_json::to_vec(&batch).map_err(io::Error::other)?, + )?; + let index = BatchIndex { + event_count: batch.event_count, + event_sequences: batch.event_sequences.clone(), + }; + write_atomic( + &self.batch_index_path(batch.sequence), + &serde_json::to_vec(&index).map_err(io::Error::other)?, + )?; + self.state.next_sequence = self.state.next_sequence.saturating_add(1); + self.persist_state()?; + for event in &selected { + remove_if_exists(&self.event_path(event.sequence))?; + } + sync_directory(&self.directory)?; + Ok(true) + } + + /// Read the oldest immutable request without changing its serialized body. + pub fn oldest_batch(&self) -> io::Result> { + let Some((_, path)) = self.batch_paths()?.into_iter().next() else { + return Ok(None); + }; + read_json(&path).map(Some) + } + + /// Remove a durable request only after Iris acknowledges a success status. + pub fn remove_batch(&mut self, sequence: u64, event_count: usize) -> io::Result<()> { + let batch_path = self.batch_path(sequence); + match fs::remove_file(&batch_path) { + Ok(()) => { + remove_if_exists(&self.batch_index_path(sequence))?; + self.state.pending_event_count = + self.state.pending_event_count.saturating_sub(event_count); + sync_directory(&self.directory)?; + self.persist_state() + } + Err(error) if error.kind() == io::ErrorKind::NotFound => { + remove_if_exists(&self.batch_index_path(sequence))?; + sync_directory(&self.directory) + } + Err(error) => Err(error), + } + } + + const fn allocate_sequence(&mut self) -> u64 { + let sequence = self.state.next_sequence; + self.state.next_sequence = self.state.next_sequence.saturating_add(1); + sequence + } + + fn write_event(&self, event: &EventRecord) -> io::Result<()> { + let bytes = serde_json::to_vec(event).map_err(io::Error::other)?; + write_atomic(&self.event_path(event.sequence), &bytes) + } + + fn persist_state(&self) -> io::Result<()> { + let bytes = serde_json::to_vec(&self.state).map_err(io::Error::other)?; + write_atomic(&self.directory.join("bridge-state.json"), &bytes) + } + + fn event_path(&self, sequence: u64) -> PathBuf { + self.directory.join(format!("event-{sequence:020}.json")) + } + + fn batch_path(&self, sequence: u64) -> PathBuf { + self.directory.join(format!("batch-{sequence:020}.json")) + } + + fn batch_index_path(&self, sequence: u64) -> PathBuf { + self.directory + .join(format!("batch-{sequence:020}.index.json")) + } + + fn event_records(&self) -> io::Result> { + let mut records = Vec::new(); + for (_, path) in self.event_paths()? { + records.push(read_json(&path)?); + } + records.sort_by_key(|record: &EventRecord| record.sequence); + Ok(records) + } + + fn event_paths(&self) -> io::Result> { + numbered_paths(&self.directory, "event-") + } + + fn batch_paths(&self) -> io::Result> { + numbered_paths(&self.directory, "batch-") + } + + fn batch_index_paths(&self) -> io::Result> { + let mut paths = Vec::new(); + for entry in fs::read_dir(&self.directory)? { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_string_lossy(); + let Some(sequence) = name + .strip_prefix("batch-") + .and_then(|name| name.strip_suffix(".index.json")) + .and_then(|sequence| sequence.parse::().ok()) + else { + continue; + }; + paths.push((sequence, entry.path())); + } + paths.sort_by_key(|(sequence, _)| *sequence); + Ok(paths) + } + + fn batch_index_for(&self, sequence: u64, batch_path: &Path) -> io::Result { + let index_path = self.batch_index_path(sequence); + let index = match read_json::(&index_path) { + Ok(index) => index, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + // A crash may occur after the immutable batch file lands but + // before its compact index. Rebuild the index once from it. + let batch: BatchRecord = read_json(batch_path)?; + let index = BatchIndex { + event_count: batch.event_count, + event_sequences: batch.event_sequences, + }; + write_atomic( + &index_path, + &serde_json::to_vec(&index).map_err(io::Error::other)?, + )?; + index + } + Err(error) => return Err(error), + }; + if index.event_count == 0 || index.event_count != index.event_sequences.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Herdr batch index has inconsistent event metadata", + )); + } + Ok(index) + } + + fn count_pending_events(&self) -> io::Result { + let mut count = self.event_paths()?.len(); + for (sequence, path) in self.batch_paths()? { + count = count.saturating_add(self.batch_index_for(sequence, &path)?.event_count); + } + Ok(count) + } + + fn reconcile_batched_events(&self) -> io::Result<()> { + let batch_paths = self.batch_paths()?; + let mut batch_sequences = HashSet::new(); + let mut already_batched = HashSet::new(); + for (sequence, path) in &batch_paths { + batch_sequences.insert(*sequence); + already_batched.extend(self.batch_index_for(*sequence, path)?.event_sequences); + } + for (sequence, path) in self.batch_index_paths()? { + if !batch_sequences.contains(&sequence) { + remove_if_exists(&path)?; + } + } + for (sequence, path) in self.event_paths()? { + if already_batched.contains(&sequence) { + remove_if_exists(&path)?; + } + } + sync_directory(&self.directory) + } + + fn enforce_bound(&mut self) -> io::Result { + let mut dropped = 0; + while self.state.pending_event_count > self.max_events { + let oldest_event = self.event_paths()?.into_iter().next(); + let oldest_batch = if let Some((batch_sequence, batch_path)) = + self.batch_paths()?.into_iter().next() + { + let index = self.batch_index_for(batch_sequence, &batch_path)?; + let oldest_event_sequence = + index.event_sequences.iter().copied().min().ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + "Herdr batch index has no source event sequences", + ) + })?; + Some((oldest_event_sequence, batch_sequence, batch_path, index)) + } else { + None + }; + match (oldest_event, oldest_batch) { + (Some((event_sequence, event_path)), Some((batch_event_sequence, _, _, _))) + if event_sequence < batch_event_sequence => + { + remove_if_exists(&event_path)?; + self.state.pending_event_count = + self.state.pending_event_count.saturating_sub(1); + dropped += 1; + } + (_, Some((_, batch_sequence, batch_path, index))) => { + remove_if_exists(&batch_path)?; + remove_if_exists(&self.batch_index_path(batch_sequence))?; + self.state.pending_event_count = self + .state + .pending_event_count + .saturating_sub(index.event_count); + dropped += index.event_count; + } + (Some((_, event_path)), None) => { + remove_if_exists(&event_path)?; + self.state.pending_event_count = + self.state.pending_event_count.saturating_sub(1); + dropped += 1; + } + (None, None) => { + self.state.pending_event_count = 0; + } + } + } + if dropped > 0 { + tracing::warn!( + dropped_events = dropped, + "bounded Herdr spool evicted oldest queued events" + ); + sync_directory(&self.directory)?; + self.persist_state()?; + } + Ok(dropped) + } + + #[cfg(test)] + /// Count all source events still queued or held by durable batches. + pub const fn pending_event_count(&self) -> usize { + self.state.pending_event_count + } +} + +fn build_batch(events: &[EventRecord], sequence: u64) -> io::Result { + let batch_id = Uuid::new_v4().to_string(); + let replay_key = format!("herdr:bridge-batch:{batch_id}"); + let event_ids = events + .iter() + .map(|event| event.event_id.clone()) + .collect::>(); + let event_sequences = events + .iter() + .map(|event| event.sequence) + .collect::>(); + let mut mutations = Vec::new(); + let mut dropped_count = 0; + let mut suppressed_thread_upsert_count = 0; + let mut missing_workspace_snapshot_event_count = 0; + + for event in events { + if event.payload.get("event").and_then(Value::as_str) == Some("bridge_heartbeat") { + mutations.extend(heartbeat_mutations(&event.event_id, event.received_at)); + continue; + } + let mapped = map_ingest_batch(&HerdrEvent { + event_id: &event.event_id, + payload: &event.payload, + received_at: event.received_at, + }) + .map_err(io::Error::other)?; + dropped_count += mapped + .audit + .as_ref() + .and_then(|audit| audit.metadata.get("dropped_count")) + .and_then(Value::as_u64) + .unwrap_or(0); + let mut event_mutations = mapped.mutations; + let (suppressed_threads, missing_snapshot) = + enrich_thread_snapshots(&event.payload, &mut event_mutations, event.received_at); + suppressed_thread_upsert_count += suppressed_threads; + if missing_snapshot { + dropped_count += 1; + missing_workspace_snapshot_event_count += 1; + } + mutations.extend(event_mutations); + } + + let audit = AuditEvent { + action: AuditAction::Normalize, + provider: "herdr".to_owned(), + source_id: Some(replay_key.clone()), + timestamp: events[0].received_at, + metadata: json!({ + "event_count": events.len(), + "mutation_count": mutations.len(), + "dropped_count": dropped_count, + "suppressed_thread_upsert_count": suppressed_thread_upsert_count, + "missing_workspace_snapshot_event_count": missing_workspace_snapshot_event_count, + "bridge_event_ids": event_ids, + }), + }; + let batch = IngestBatch { + source: "herdr".to_owned(), + replay_key, + mutations, + cursor: None, + audit: Some(audit), + }; + let body = serde_json::to_string(&batch).map_err(io::Error::other)?; + Ok(BatchRecord { + sequence, + event_count: events.len(), + event_sequences, + body, + }) +} + +fn enrich_thread_snapshots( + payload: &Value, + mutations: &mut Vec, + received_at: DateTime, +) -> (usize, bool) { + let snapshots = payload + .pointer("/bridge/workspace_snapshots") + .and_then(Value::as_object); + let event_kind = payload + .get("event") + .and_then(Value::as_str) + .unwrap_or_default(); + if event_kind == "pane_agent_status_changed" { + let workspace_id = payload + .pointer("/data/workspace_id") + .and_then(Value::as_str); + let current_thread = workspace_id + .and_then(|id| snapshots.and_then(|items| items.get(id))) + .and_then(|snapshot| map_workspace_snapshot(snapshot, received_at)); + if current_thread.is_none() { + mutations.clear(); + return (1, true); + } + } + + let mut normalized = Vec::with_capacity(mutations.len()); + let mut suppressed = 0; + for mutation in mutations.drain(..) { + match mutation { + IngestMutation::UpsertThread(thread) => { + let current_thread = snapshots + .and_then(|items| items.get(&thread.source_id)) + .and_then(|snapshot| map_workspace_snapshot(snapshot, received_at)) + .filter(|snapshot| snapshot.source_id == thread.source_id); + if let Some(snapshot) = current_thread { + normalized.push(IngestMutation::UpsertThread(snapshot)); + } else { + suppressed += 1; + } + } + IngestMutation::ArchiveThread { source, source_id } => { + let current_thread = snapshots + .and_then(|items| items.get(&source_id)) + .and_then(|snapshot| map_workspace_snapshot(snapshot, received_at)) + .filter(|snapshot| snapshot.source_id == source_id); + if let Some(snapshot) = current_thread { + // A retained close event can be stale relative to the + // authoritative workspace list. Restore, don't archive. + normalized.push(IngestMutation::UpsertThread(snapshot)); + } else { + normalized.push(IngestMutation::ArchiveThread { source, source_id }); + } + } + other => normalized.push(other), + } + } + *mutations = normalized; + (suppressed, false) +} + +fn heartbeat_mutations(event_id: &str, at: DateTime) -> Vec { + let contact = Contact { + id: HEARTBEAT_CONTACT_ID, + source: "herdr".to_owned(), + provider_instance: None, + source_id: "bridge".to_owned(), + display_name: Some("Herdr bridge".to_owned()), + avatar_url: None, + metadata: json!({"role": "bridge_health"}), + }; + let thread = Thread { + id: HEARTBEAT_THREAD_ID, + source: "herdr".to_owned(), + provider_instance: None, + source_id: "bridge-heartbeat".to_owned(), + title: Some("Herdr bridge health".to_owned()), + participants: vec![contact.clone()], + last_message_at: at, + unread_count: Some(0), + metadata: json!({"kind": "bridge_heartbeat"}), + }; + let message = Message { + id: Uuid::new_v5(&HEARTBEAT_THREAD_ID, event_id.as_bytes()), + thread_id: HEARTBEAT_THREAD_ID, + source: "herdr".to_owned(), + source_id: format!("bridge-heartbeat:{event_id}"), + sender: contact.clone(), + kind: MessageKind::System, + body: "Herdr bridge heartbeat".to_owned(), + attachments: Vec::new(), + timestamp: at, + is_outbound: false, + metadata: json!({"event": "bridge_heartbeat", "bridge_event_id": event_id}), + }; + vec![ + IngestMutation::UpsertContact(contact), + IngestMutation::UpsertThread(thread), + IngestMutation::AppendMessage { message }, + ] +} + +fn numbered_paths(directory: &Path, prefix: &str) -> io::Result> { + let mut result = Vec::new(); + for entry in fs::read_dir(directory)? { + let entry = entry?; + let path = entry.path(); + let Some(name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + if let Some(sequence) = name + .strip_prefix(prefix) + .and_then(|name| name.strip_suffix(".json")) + .and_then(|digits| digits.parse::().ok()) + { + result.push((sequence, path)); + } + } + result.sort_by_key(|(sequence, _)| *sequence); + Ok(result) +} + +fn max_sequence(directory: &Path) -> io::Result { + let mut max = 0; + for prefix in ["event-", "batch-"] { + for (sequence, _) in numbered_paths(directory, prefix)? { + max = max.max(sequence); + } + } + Ok(max) +} + +fn read_json Deserialize<'de>>(path: &Path) -> io::Result { + let bytes = fs::read(path)?; + serde_json::from_slice(&bytes) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) +} + +fn write_atomic(path: &Path, bytes: &[u8]) -> io::Result<()> { + let parent = path + .parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "spool path has no parent"))?; + let name = path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("record"); + let temp = parent.join(format!(".{name}.{}.tmp", Uuid::new_v4())); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(0o600); + } + let write_result = (|| { + let mut file = options.open(&temp)?; + file.write_all(bytes)?; + file.sync_all()?; + fs::rename(&temp, path)?; + sync_directory(parent) + })(); + if write_result.is_err() { + let _ = fs::remove_file(&temp); + } + write_result +} + +fn set_private_directory(directory: &Path) -> io::Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(directory, fs::Permissions::from_mode(0o700))?; + } + Ok(()) +} + +fn remove_abandoned_temps(directory: &Path) -> io::Result<()> { + for entry in fs::read_dir(directory)? { + let entry = entry?; + let Some(name) = entry.file_name().to_str().map(ToOwned::to_owned) else { + continue; + }; + if name.starts_with('.') + && Path::new(&name) + .extension() + .is_some_and(|extension| extension.eq_ignore_ascii_case("tmp")) + { + remove_if_exists(&entry.path())?; + } + } + sync_directory(directory) +} + +fn remove_if_exists(path: &Path) -> io::Result<()> { + match fs::remove_file(path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +fn sync_directory(directory: &Path) -> io::Result<()> { + File::open(directory)?.sync_all() +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + use tempfile::tempdir; + + fn unknown_event() -> Value { + json!({"event": "future_event", "data": {"safe": true}}) + } + + fn pane_status_event(label_length: usize) -> Value { + let mut event: Value = serde_json::from_str(include_str!( + "../tests/fixtures/herdr-v0.8.0-pane-agent-status-event.json" + )) + .unwrap(); + event["data"]["workspace_id"] = json!("workspace-1"); + event["data"]["agent"] = json!("claude"); + event["data"]["display_agent"] = json!("Claude"); + event["data"]["state_labels"]["detail"] = json!("x".repeat(label_length)); + event["data"]["workspace_label"] = json!("Synthetic workspace"); + event["bridge"]["workspace_snapshots"]["workspace-1"] = json!({ + "workspace_id": "workspace-1", + "number": 1, + "label": "Synthetic workspace", + "focused": true, + "pane_count": 1, + "tab_count": 1, + "active_tab_id": "tab-1", + "agent_status": "working", + "revision": 7 + }); + event + } + + #[test] + fn batches_at_twenty_and_after_five_seconds_and_preserves_body_on_reopen() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 100).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 10, 9, 12, 0, 0).unwrap(); + for index in 0..19 { + let mut event = unknown_event(); + event["data"]["index"] = json!(index); + spool.append_upstream_event(event, start).unwrap(); + } + assert!(!spool.flush_due(start + Duration::seconds(4)).unwrap()); + spool + .append_upstream_event(unknown_event(), start + Duration::seconds(4)) + .unwrap(); + assert!(spool.flush_due(start + Duration::seconds(4)).unwrap()); + let first = spool.oldest_batch().unwrap().unwrap(); + assert_eq!(first.event_count, 20); + assert_eq!(spool.pending_event_count(), 20); + fs::remove_file(spool.batch_index_path(first.sequence)).unwrap(); + drop(spool); + + let reopened = Spool::open_with_limit(dir.path().join("spool"), 100).unwrap(); + let restored = reopened.oldest_batch().unwrap().unwrap(); + assert_eq!(first.body, restored.body); + assert_eq!(first.sequence, restored.sequence); + + let mut reopened = reopened; + let mut after_reopen = unknown_event(); + after_reopen["data"]["index"] = json!(20); + reopened.append_upstream_event(after_reopen, start).unwrap(); + assert!(reopened.flush_due(start + Duration::seconds(5)).unwrap()); + assert_eq!(reopened.pending_event_count(), 21); + } + + #[test] + fn bounded_spool_drops_oldest_records_only() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 3).unwrap(); + let now = Utc::now(); + for index in 0..4 { + spool + .append_upstream_event(json!({"event":"future", "data":{"n": index}}), now) + .unwrap(); + } + assert_eq!(spool.pending_event_count(), 3); + let events = spool.event_records().unwrap(); + assert_eq!(events.len(), 3); + assert_eq!(events[0].payload["data"]["n"], 1); + } + + #[test] + fn thread_upserts_use_allowlisted_authoritative_workspace_snapshot() { + let received_at = Utc::now(); + let snapshot = json!({ + "workspace_id": "workspace-1", + "number": 1, + "label": "Current label", + "focused": true, + "pane_count": 6, + "tab_count": 2, + "active_tab_id": "tab-2", + "agent_status": "working", + "revision": 7, + "worktree": {"repo_root": "/private/repo"}, + "panes": [{ + "pane_id": "pane-1", + "workspace_id": "workspace-1", + "tab_id": "tab-2", + "agent_status": "working", + "title": "private pane title", + "cwd": "/private/repo" + }], + "tokens": {"synthetic_private_value": "fixture-only-do-not-forward"} + }); + let payload = json!({ + "event": "pane_created", + "data": { + "pane": {"workspace_id": "workspace-1", "pane_id": "pane-1"}, + "workspace_label": "Stale label" + }, + "bridge": {"workspace_snapshots": {"workspace-1": snapshot}} + }); + let mapped = map_ingest_batch(&HerdrEvent { + event_id: "pane-created-snapshot", + payload: &payload, + received_at, + }) + .unwrap(); + let mut mutations = mapped.mutations; + assert_eq!( + enrich_thread_snapshots(&payload, &mut mutations, received_at), + (0, false) + ); + let [IngestMutation::UpsertThread(thread)] = mutations.as_slice() else { + panic!("expected one complete workspace thread upsert: {mutations:?}"); + }; + assert_eq!(thread.title.as_deref(), Some("Current label")); + assert_eq!(thread.metadata["pane_count"], 6); + assert_eq!(thread.metadata["tab_count"], 2); + assert!(thread.metadata.get("revision").is_none()); + assert_eq!(thread.metadata["panes"][0]["pane_id"], "pane-1"); + assert_eq!(thread.metadata["panes"][0]["tab_id"], "tab-2"); + assert!(thread.metadata["panes"][0].get("title").is_none()); + assert!(thread.metadata["worktree"].is_null()); + assert!(!thread.metadata.to_string().contains("/private/repo")); + assert!(thread.metadata.get("tokens").is_none()); + + let status_payload = pane_status_event(8); + let mapped = map_ingest_batch(&HerdrEvent { + event_id: "status-with-full-snapshot", + payload: &status_payload, + received_at, + }) + .unwrap(); + let mut status_mutations = mapped.mutations; + assert_eq!( + enrich_thread_snapshots(&status_payload, &mut status_mutations, received_at), + (0, false) + ); + let status_thread = status_mutations.iter().find_map(|mutation| match mutation { + IngestMutation::UpsertThread(thread) => Some(thread), + _ => None, + }); + assert_eq!(status_thread.unwrap().metadata["pane_count"], 1); + assert!( + status_mutations + .iter() + .any(|mutation| matches!(mutation, IngestMutation::AppendMessage { .. })) + ); + } + + #[test] + fn status_without_an_authoritative_workspace_does_not_orphan_a_message() { + let received_at = Utc::now(); + let payload = json!({ + "event": "pane_agent_status_changed", + "data": { + "workspace_id": "workspace-closed", + "pane_id": "pane-1", + "agent_status": "working" + }, + "bridge": {"workspace_snapshots": {}} + }); + let mapped = map_ingest_batch(&HerdrEvent { + event_id: "stale-status", + payload: &payload, + received_at, + }) + .unwrap(); + let mut mutations = mapped.mutations; + assert_eq!( + enrich_thread_snapshots(&payload, &mut mutations, received_at), + (1, true) + ); + assert!(mutations.is_empty()); + } + + #[test] + fn batch_serialization_splits_at_iris_ingest_body_limit() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 10).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 10, 9, 12, 0, 0).unwrap(); + for _ in 0..2 { + spool + .append_upstream_event(pane_status_event(300_000), start) + .unwrap(); + } + + assert!(spool.flush_due(start + Duration::seconds(5)).unwrap()); + let first = spool.oldest_batch().unwrap().unwrap(); + assert_eq!(first.event_count, 1); + assert!(first.body.len() <= MAX_BATCH_BODY_BYTES); + assert_eq!(spool.pending_event_count(), 2); + + assert!(spool.flush_due(start + Duration::seconds(5)).unwrap()); + let second = spool + .batch_paths() + .unwrap() + .into_iter() + .nth(1) + .map(|(_, path)| read_json::(&path).unwrap()) + .unwrap(); + assert_eq!(second.event_count, 1); + assert!(second.body.len() <= MAX_BATCH_BODY_BYTES); + assert_ne!(first.sequence, second.sequence); + assert_eq!(spool.pending_event_count(), 2); + } + + #[test] + fn overflow_compares_batches_by_their_oldest_source_event() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 2).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 10, 9, 12, 0, 0).unwrap(); + for _ in 0..2 { + spool + .append_upstream_event(pane_status_event(300_000), start) + .unwrap(); + } + + assert!(spool.flush_due(start + Duration::seconds(5)).unwrap()); + let first = spool.oldest_batch().unwrap().unwrap(); + assert_eq!(first.event_count, 1); + assert_eq!(first.event_sequences, vec![1]); + assert_eq!( + spool.event_records().unwrap()[0].sequence, + 2, + "the later event remains queued while the older event is batched" + ); + + // Model overflow after the first event has been sealed. The batch's + // filename sequence is 3, but its source event is sequence 1; event 2 + // must survive because it is newer than the event inside the batch. + spool.max_events = 1; + assert_eq!(spool.enforce_bound().unwrap(), 1); + assert!(spool.batch_paths().unwrap().is_empty()); + let remaining = spool.event_records().unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].sequence, 2); + assert_eq!(spool.pending_event_count(), 1); + } + + #[test] + fn unmapped_event_blocks_later_sealing_until_durable_enrichment() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 10).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 10, 9, 12, 0, 0).unwrap(); + let first_sequence = spool + .append_unmapped_upstream_event(unknown_event(), start) + .unwrap(); + spool + .append_upstream_event(unknown_event(), start + Duration::seconds(1)) + .unwrap(); + + assert!(!spool.flush_due(start + Duration::seconds(5)).unwrap()); + let records = spool.event_records().unwrap(); + assert_eq!(records.len(), 2); + assert!(!records[0].mapping_ready); + assert!(records[1].mapping_ready); + + let mut enriched = unknown_event(); + enriched["bridge"] = json!({"workspace_snapshots": []}); + spool.mark_event_ready(first_sequence, enriched).unwrap(); + assert!(spool.flush_due(start + Duration::seconds(5)).unwrap()); + assert!(spool.event_records().unwrap().is_empty()); + assert_eq!(spool.oldest_batch().unwrap().unwrap().event_count, 2); + } + + #[test] + fn oversized_event_is_retained_at_the_head_instead_of_silently_dropped() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 10).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 10, 9, 12, 0, 0).unwrap(); + spool + .append_upstream_event(pane_status_event(600_000), start) + .unwrap(); + spool + .append_upstream_event(unknown_event(), start + Duration::seconds(1)) + .unwrap(); + + let events = spool.event_records().unwrap(); + assert!(build_batch(&events[..1], 3).unwrap().body.len() > MAX_BATCH_BODY_BYTES); + assert!(!spool.flush_due(start + Duration::seconds(5)).unwrap()); + assert!(!spool.flush_due(start + Duration::seconds(6)).unwrap()); + assert!(spool.oldest_batch().unwrap().is_none()); + let retained = spool.event_records().unwrap(); + assert_eq!(retained.len(), 2); + assert_eq!(retained[0].sequence, 1); + assert_eq!(retained[1].sequence, 2); + assert_eq!( + retained[0].payload["data"]["state_labels"]["detail"] + .as_str() + .unwrap() + .len(), + 600_000 + ); + assert_eq!(spool.pending_event_count(), 2); + } + + #[test] + fn idle_heartbeat_creates_a_normalized_freshness_message() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 20).unwrap(); + let start = Utc.with_ymd_and_hms(2026, 10, 9, 12, 0, 0).unwrap(); + spool.state.last_upstream_event_at = start; + assert!( + !spool + .maybe_append_heartbeat(start + Duration::minutes(59)) + .unwrap() + ); + let heartbeat_at = start + Duration::hours(1); + assert!(spool.maybe_append_heartbeat(heartbeat_at).unwrap()); + assert!( + spool + .flush_due(heartbeat_at + Duration::seconds(5)) + .unwrap() + ); + let batch = spool.oldest_batch().unwrap().unwrap(); + let value: Value = serde_json::from_str(&batch.body).unwrap(); + assert_eq!(value["audit"]["metadata"]["event_count"], 1); + let mutations = value["mutations"].as_array().unwrap(); + assert_eq!(mutations.len(), 3); + assert!( + mutations.iter().any(|item| { + item["kind"] == "append_message" + && item["message"]["kind"] == "system" + && item["message"]["body"] == "Herdr bridge heartbeat" + }), + "unexpected heartbeat batch: {value}" + ); + } + + #[test] + fn protocol_19_payload_maps_into_existing_ingest_mutations_without_raw_metadata_leaks() { + let dir = tempdir().unwrap(); + let mut spool = Spool::open_with_limit(dir.path().join("spool"), 20).unwrap(); + let mut payload: Value = serde_json::from_str(include_str!( + "../tests/fixtures/herdr-v0.8.0-workspace-renamed-event.json" + )) + .unwrap(); + payload["bridge"]["workspace_snapshots"]["workspace-1"] = json!({ + "workspace_id": "workspace-1", + "number": 1, + "label": "Current workspace", + "pane_count": 1, + "tab_count": 1, + "active_tab_id": "tab-1", + "agent_status": "working" + }); + spool.append_upstream_event(payload, Utc::now()).unwrap(); + spool.flush_due(Utc::now() + Duration::seconds(5)).unwrap(); + let batch = spool.oldest_batch().unwrap().unwrap(); + let value: Value = serde_json::from_str(&batch.body).unwrap(); + assert_eq!(value["source"], "herdr"); + assert_eq!(value["mutations"][0]["kind"], "upsert_thread"); + assert!(value["audit"]["metadata"].get("bridge_event_ids").is_some()); + assert!(value["audit"]["metadata"].get("raw_payload").is_none()); + } +} diff --git a/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-agent-status-event.json b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-agent-status-event.json new file mode 100644 index 0000000..4dc565b --- /dev/null +++ b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-agent-status-event.json @@ -0,0 +1 @@ +{"event":"pane.agent_status_changed","data":{"pane_id":"pane-1","workspace_id":"workspace-1","agent_status":"idle","agent":"claude","display_agent":"Claude","title":"private-pane-title-do-not-forward","cwd":"/private/herdr/worktree","tokens":{"access":"fixture-only-do-not-forward"},"private_payload":"fixture-only-do-not-forward"}} \ No newline at end of file diff --git a/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-created-event.json b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-created-event.json new file mode 100644 index 0000000..69b5395 --- /dev/null +++ b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-pane-created-event.json @@ -0,0 +1 @@ +{"event":"pane_created","data":{"pane":{"pane_id":"closed-pane","terminal_id":"terminal-1","workspace_id":"workspace-1","tab_id":"tab-1","focused":false,"agent_status":"unknown","tokens":{},"revision":1}}} diff --git a/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-subscription-ack.json b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-subscription-ack.json new file mode 100644 index 0000000..3d59b42 --- /dev/null +++ b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-subscription-ack.json @@ -0,0 +1 @@ +{"id":"fixture-id","result":{"type":"subscription_started"}} \ No newline at end of file diff --git a/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-list-response.json b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-list-response.json new file mode 100644 index 0000000..ac02f5f --- /dev/null +++ b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-list-response.json @@ -0,0 +1 @@ +{"id":"fixture-workspace-list","result":{"type":"workspace_list","workspaces":[{"workspace_id":"workspace-1","number":1,"label":"Iris Pilot","focused":true,"pane_count":1,"tab_count":1,"active_tab_id":"tab-1","agent_status":"working","tokens":{"synthetic_private_value":"fixture-only-do-not-forward"},"worktree":{"repo_name":"private-repo","repo_root":"/private/repo","checkout_path":"/private/herdr/worktree"}}]}} diff --git a/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-renamed-event.json b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-renamed-event.json new file mode 100644 index 0000000..5bbb92a --- /dev/null +++ b/crates/iris-herdr-bridge/tests/fixtures/herdr-v0.8.0-workspace-renamed-event.json @@ -0,0 +1 @@ +{"event":"workspace_renamed","data":{"type":"workspace_renamed","workspace_id":"workspace-1","label":"fixture workspace"}} \ No newline at end of file diff --git a/crates/iris-providers/src/herdr/mod.rs b/crates/iris-providers/src/herdr/mod.rs index 97c66fb..5da2477 100644 --- a/crates/iris-providers/src/herdr/mod.rs +++ b/crates/iris-providers/src/herdr/mod.rs @@ -115,9 +115,9 @@ pub fn map_event(event: &HerdrEvent<'_>) -> HerdrMapping { .map(|id| { HerdrIntent::UpsertThread(thread_with_metadata( id, - None, + workspace_label_for(data, id), event.received_at, - json!({"event": kind, "data": canonical_value(&Value::Object(data.clone()))}), + json!({"event": kind}), )) }) .collect() @@ -130,9 +130,9 @@ pub fn map_event(event: &HerdrEvent<'_>) -> HerdrMapping { .map(|id| { vec![HerdrIntent::UpsertThread(thread_with_metadata( id, - None, + workspace_label(data), event.received_at, - json!({"event": kind, "data": canonical_value(&Value::Object(data.clone()))}), + json!({"event": kind}), ))] }) .unwrap_or_else(|| dropped(&kind, "missing workspace_id")), @@ -148,9 +148,9 @@ pub fn map_event(event: &HerdrEvent<'_>) -> HerdrMapping { .map(|id| { vec![HerdrIntent::UpsertThread(thread_with_metadata( id, - None, + workspace_label(data), event.received_at, - json!({"event": kind, "data": canonical_value(&Value::Object(data.clone()))}), + json!({"event": kind}), ))] }) .unwrap_or_else(|| dropped(&kind, "missing workspace_id")), @@ -201,7 +201,7 @@ pub fn map_ingest_batch(event: &HerdrEvent<'_>) -> Result { source: SOURCE.to_owned(), source_id, }), - HerdrIntent::AppendMessage(message) => Some(IngestMutation::AppendMessage(message)), + HerdrIntent::AppendMessage(message) => Some(IngestMutation::AppendMessage { message }), HerdrIntent::Dropped { .. } => None, }) .collect::>(); @@ -266,16 +266,15 @@ fn map_agent_status( "pane_id": pane_id, "agent_status": status, "state_labels": labels, - "data": canonical_value(&Value::Object(data.clone())), }), }; vec![ HerdrIntent::UpsertContact(contact), HerdrIntent::UpsertThread(thread_with_metadata( workspace_id, - None, + workspace_label(data), event.received_at, - json!({"event": kind, "data": canonical_value(&Value::Object(data.clone()))}), + json!({"event": kind}), )), HerdrIntent::AppendMessage(message), ] @@ -298,12 +297,25 @@ fn map_pane_moved( source_id: closed_workspace_id.to_owned(), }); } - if let Some(workspace_id) = workspace_id_from_pane_event(data) { + let current_workspace_id = workspace_id_from_pane_event(data); + let closed_workspace_id = data.get("closed_workspace_id").and_then(Value::as_str); + if let Some(previous_workspace_id) = data.get("previous_workspace_id").and_then(Value::as_str) + && Some(previous_workspace_id) != current_workspace_id + && Some(previous_workspace_id) != closed_workspace_id + { intents.push(HerdrIntent::UpsertThread(thread_with_metadata( - workspace_id, + previous_workspace_id, None, received_at, - json!({"event": kind, "data": canonical_value(&Value::Object(data.clone()))}), + json!({"event": kind}), + ))); + } + if let Some(workspace_id) = current_workspace_id { + intents.push(HerdrIntent::UpsertThread(thread_with_metadata( + workspace_id, + workspace_label(data), + received_at, + json!({"event": kind}), ))); } if intents.is_empty() { @@ -320,10 +332,87 @@ fn workspace_thread(value: &Value, received_at: DateTime) -> Option id, workspace.get("label").and_then(Value::as_str), received_at, - Value::Object(sorted_object(Some(workspace))), + normalized_workspace_metadata(workspace), )) } +fn normalized_workspace_metadata(workspace: &Map) -> Value { + let mut metadata = Map::new(); + for field in ["workspace_id", "label", "active_tab_id", "agent_status"] { + if let Some(value) = workspace.get(field).filter(|value| value.is_string()) { + metadata.insert(field.to_owned(), value.clone()); + } + } + for field in ["number", "pane_count", "tab_count"] { + if let Some(value) = workspace.get(field).filter(|value| value.is_number()) { + metadata.insert(field.to_owned(), value.clone()); + } + } + if let Some(value) = workspace.get("focused").filter(|value| value.is_boolean()) { + metadata.insert("focused".to_owned(), value.clone()); + } + if let Some(panes) = workspace.get("panes").and_then(Value::as_array) { + let panes = panes + .iter() + .filter_map(normalized_pane_metadata) + .collect::>(); + metadata.insert("panes".to_owned(), Value::Array(panes)); + } + Value::Object(metadata) +} + +fn normalized_pane_metadata(value: &Value) -> Option { + let pane = value.as_object()?; + let mut metadata = Map::new(); + for field in [ + "pane_id", + "workspace_id", + "tab_id", + "agent", + "display_agent", + "agent_status", + ] { + if let Some(value) = pane.get(field).filter(|value| value.is_string()) { + metadata.insert(field.to_owned(), value.clone()); + } + } + if let Some(value) = pane.get("focused").filter(|value| value.is_boolean()) { + metadata.insert("focused".to_owned(), value.clone()); + } + if let Some(labels) = pane.get("state_labels").and_then(Value::as_object) { + metadata.insert( + "state_labels".to_owned(), + Value::Object( + labels + .iter() + .filter(|(_, value)| value.is_string()) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + ), + ); + } + Some(Value::Object(metadata)) +} + +/// Maps an authoritative workspace snapshot to a complete Iris thread record. +/// +/// The bridge uses this when replacing partial event-derived thread updates so +/// storage's replace-on-upsert semantics cannot erase workspace metadata. +pub fn map_workspace_snapshot(value: &Value, received_at: DateTime) -> Option { + workspace_thread(value, received_at) +} + +fn workspace_label(data: &Map) -> Option<&str> { + data.get("workspace_label").and_then(Value::as_str) +} + +fn workspace_label_for<'a>(data: &'a Map, workspace_id: &str) -> Option<&'a str> { + data.get("workspace_labels") + .and_then(|labels| labels.get(workspace_id)) + .and_then(Value::as_str) + .or_else(|| workspace_label(data)) +} + fn thread_with_metadata( id: &str, title: Option<&str>, @@ -412,19 +501,6 @@ fn sorted_object(object: Option<&Map>) -> Map { .collect() } -fn canonical_value(value: &Value) -> Value { - match value { - Value::Object(object) => Value::Object( - object - .iter() - .map(|(key, value)| (key.clone(), canonical_value(value))) - .collect(), - ), - Value::Array(values) => Value::Array(values.iter().map(canonical_value).collect()), - value => value.clone(), - } -} - fn labels_suffix(labels: &Map) -> String { if labels.is_empty() { String::new() @@ -511,7 +587,7 @@ mod tests { )); assert!(matches!( batch.mutations[2], - IngestMutation::AppendMessage(_) + IngestMutation::AppendMessage { .. } )); } @@ -548,13 +624,14 @@ mod tests { } #[test] - fn maps_agent_status_with_deterministic_contact_thread_and_message() { + fn maps_agent_status_without_copying_the_source_payload_into_metadata() { let mapping = map( "evt-1", "pane_agent_status_changed", json!({ "workspace_id": "workspace-1", "pane_id": "pane-2", "agent": "codex", - "agent_status": "working", "state_labels": {"branch": "main", "task": "COD-438"} + "agent_status": "working", "state_labels": {"branch": "main", "task": "COD-438"}, + "title": "private pane title", "private_payload": "fixture-only-do-not-forward" }), ); assert_eq!(mapping.dedupe_key, "herdr:evt-1"); @@ -566,6 +643,102 @@ mod tests { assert_eq!(message.sender.source_id, "unknown-host:codex"); assert_eq!(message.source_id, "evt-1"); assert_eq!(message.thread_id, thread_uuid("workspace-1")); + assert!(message.metadata.get("data").is_none()); + assert!(!message.metadata.to_string().contains("private pane title")); + assert!( + !message + .metadata + .to_string() + .contains("fixture-only-do-not-forward") + ); + } + + #[test] + fn enriched_workspace_label_survives_partial_agent_status_update() { + let mapping = map( + "evt-labeled", + "pane_agent_status_changed", + json!({ + "workspace_id": "workspace-1", "workspace_label": "Iris Pilot", + "pane_id": "pane-2", "agent": "codex", "agent_status": "working" + }), + ); + let HerdrIntent::UpsertThread(thread) = &mapping.intents[1] else { + panic!("thread intent") + }; + assert_eq!(thread.title.as_deref(), Some("Iris Pilot")); + } + + #[test] + fn workspace_thread_metadata_contains_only_normalized_summary_fields() { + let mapping = map( + "evt-workspace", + "workspace_created", + json!({ + "workspace": { + "workspace_id": "workspace-1", + "label": "Iris Pilot", + "tokens": {"access": "fixture-only-do-not-forward"}, + "worktree": {"repo_root": "/private/repo"}, + "private_payload": "fixture-only-do-not-forward" + } + }), + ); + let HerdrIntent::UpsertThread(thread) = &mapping.intents[0] else { + panic!("thread intent") + }; + assert_eq!(thread.title.as_deref(), Some("Iris Pilot")); + assert!( + !thread + .metadata + .to_string() + .contains("fixture-only-do-not-forward") + ); + assert!(thread.metadata.get("tokens").is_none()); + assert!(thread.metadata.get("worktree").is_none()); + assert_eq!(thread.metadata["workspace_id"], "workspace-1"); + assert_eq!(thread.metadata["label"], "Iris Pilot"); + } + + #[test] + fn workspace_snapshot_keeps_safe_pane_summary_but_drops_raw_details() { + let snapshot = json!({ + "workspace_id": "workspace-1", + "label": "Iris Pilot", + "number": 1, + "focused": true, + "pane_count": 1, + "tab_count": 1, + "active_tab_id": "tab-1", + "agent_status": "working", + "worktree": {"repo_root": "/private/repo"}, + "tokens": {"access": "fixture-only-do-not-forward"}, + "panes": [{ + "pane_id": "pane-1", + "workspace_id": "workspace-1", + "tab_id": "tab-1", + "agent": "codex", + "agent_status": "working", + "state_labels": {"phase": "test"}, + "title": "private pane title", + "cwd": "/private/repo", + "tokens": {"access": "fixture-only-do-not-forward"} + }] + }); + let thread = + map_workspace_snapshot(&snapshot, Utc.timestamp_opt(1_700_000_000, 0).unwrap()) + .unwrap(); + assert_eq!(thread.metadata["panes"][0]["pane_id"], "pane-1"); + assert_eq!(thread.metadata["panes"][0]["tab_id"], "tab-1"); + assert_eq!(thread.metadata["panes"][0]["agent_status"], "working"); + assert!(!thread.metadata.to_string().contains("private pane title")); + assert!(!thread.metadata.to_string().contains("/private/repo")); + assert!( + !thread + .metadata + .to_string() + .contains("fixture-only-do-not-forward") + ); } #[test] @@ -611,7 +784,7 @@ mod tests { "move-1", "pane_moved", json!({ - "closed_workspace_id":"old", "created_workspace":{"workspace_id":"new","label":"New"}, + "previous_workspace_id":"old", "closed_workspace_id":"old", "created_workspace":{"workspace_id":"new","label":"New"}, "pane":{"workspace_id":"current","pane_id":"p"} }), ); @@ -626,6 +799,28 @@ mod tests { ); } + #[test] + fn pane_moved_upserts_previous_and_current_open_workspaces() { + let mapping = map( + "move-active-1", + "pane_moved", + json!({ + "previous_workspace_id": "old", + "pane": {"workspace_id": "current", "pane_id": "p"} + }), + ); + + let upserted_workspaces = mapping + .intents + .iter() + .filter_map(|intent| match intent { + HerdrIntent::UpsertThread(thread) => Some(thread.source_id.as_str()), + _ => None, + }) + .collect::>(); + assert_eq!(upserted_workspaces, ["old", "current"]); + } + #[test] fn status_without_pane_identity_is_dropped() { let mapping = map( diff --git a/crates/iris-server/src/routes.rs b/crates/iris-server/src/routes.rs index 5bb7eb0..c3e0594 100644 --- a/crates/iris-server/src/routes.rs +++ b/crates/iris-server/src/routes.rs @@ -775,8 +775,8 @@ mod tests { use chrono::{TimeZone, Utc}; use iris_core::{ AttachmentContent, AttachmentRef, AttachmentStore, AuditAction, AuditEvent, AuditLog, - IngestBatch, IngestCursor, IngestOutcome, IngestStore, IrisError, MessageKind, - ProviderMetadata, RealtimeState, RealtimeStatus, + IngestBatch, IngestCursor, IngestMutation, IngestOutcome, IngestStore, IrisError, + MessageKind, ProviderMetadata, RealtimeState, RealtimeStatus, }; use pretty_assertions::assert_eq; use std::collections::BTreeMap; @@ -1276,6 +1276,42 @@ mod tests { assert_eq!(conflict_response.status(), StatusCode::CONFLICT); } + #[tokio::test] + async fn ingest_route_accepts_append_message_mutation_and_idempotent_retry() { + let temp = tempfile::tempdir().unwrap(); + let store: Arc = + Arc::new(iris_storage::LocalFsIngestStore::new(temp.path())); + let mut batch = ingest_batch("message-event-1"); + batch.mutations.push(IngestMutation::AppendMessage { + message: message(Uuid::new_v4(), "alpha", 3, "bridge message"), + }); + let body = serde_json::to_vec(&batch).unwrap(); + + let applied = router(ingest_state(store.clone(), "secret")) + .oneshot( + Request::post("/ingest") + .header(header::CONTENT_TYPE, "application/json") + .header(header::AUTHORIZATION, "Bearer secret") + .body(Body::from(body.clone())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(applied.status(), StatusCode::CREATED); + + let retried = router(ingest_state(store, "secret")) + .oneshot( + Request::post("/ingest") + .header(header::CONTENT_TYPE, "application/json") + .header(header::AUTHORIZATION, "Bearer secret") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(retried.status(), StatusCode::OK); + } + #[tokio::test] async fn ingest_route_rejects_unknown_sources_and_unavailable_configuration() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/iris-storage/src/ingest.rs b/crates/iris-storage/src/ingest.rs index 88d24fc..f50ae33 100644 --- a/crates/iris-storage/src/ingest.rs +++ b/crates/iris-storage/src/ingest.rs @@ -100,7 +100,7 @@ impl LocalFsIngestStore { .or_default() .insert(source_id, true); } - IngestMutation::AppendMessage(message) => state.messages.push(message), + IngestMutation::AppendMessage { message } => state.messages.push(message), } } if let Some(cursor) = batch.cursor { diff --git a/docs/providers/herdr-bridge.md b/docs/providers/herdr-bridge.md new file mode 100644 index 0000000..eb4ff7f --- /dev/null +++ b/docs/providers/herdr-bridge.md @@ -0,0 +1,63 @@ +# Herdr → Iris bridge + +`iris-herdr-bridge` is a separate hosted daemon for the installed Herdr **0.8.0 / protocol 19**. It subscribes to Herdr's local newline-delimited JSON Unix socket, maps events with the existing pure Iris Herdr mapper, and writes source-agnostic `IngestBatch` requests to Iris `POST /ingest`. It does not add a provider-specific Iris route or change generated surfaces. + +The existing `AppendMessage` mutation's internally tagged newtype collided with `Message::kind`, making message batches impossible to deserialize. This change represents that existing mutation as `{ "kind": "append_message", "message": { ... } }`, with a core round-trip test and an authenticated-router/retry test. The `/ingest` route, OpenAPI source, and generated public surfaces are unchanged. + +The process verifies the exact Herdr version/protocol on every connection. It exits neither on a missing socket nor on an Iris outage: socket failures reconnect every two seconds, while a planned pane-set refresh reconnects immediately so a new pane's status subscription is not delayed. HTTP retries use capped exponential backoff. + +The bridge refreshes `workspace.list` for each workspace/tab/pane/status event that can mutate a thread and refreshes `pane.list` for pane/status events before normalizing them. Herdr v0.8.0's unfiltered pane-status subscription emits transitions after its baseline but no initial current-status event; an optional `agent_status` filter would also suppress later transitions to other statuses. After each successful subscription acknowledgement, the bridge refreshes pane/workspace snapshots and durably captures one synthetic current-status observation per active pane before reading the stream. This establishes current status on startup/reconnect while leaving the ongoing subscription unfiltered. The observation can repeat after reconnect; it is deliberately not payload-deduplicated. Live transition frames carry their own workspace ID, status, and optional agent/display/state-label context; those source-time fields are preserved, and omitted optional fields are not borrowed from a later pane snapshot. Synthetic startup/reconnect observations are enriched only from the fresh pane snapshot used to create them. Cross-workspace `pane.moved` events preserve both previous and current workspace IDs and attach snapshots for both, so the old open workspace removes the moved pane while the new one includes it. Workspace thread metadata is a fixed allowlist of workspace summary fields plus current pane IDs, tab IDs, agent/status, and state labels; raw workspace/pane objects are not forwarded. Source event frames are projected to an allowlist before they enter the spool; titles, paths, terminal contents, token maps, and unknown fields are excluded. Before delivery, every thread upsert is replaced with the current normalized snapshot so Iris's replace-on-upsert behavior cannot erase workspace/tab/pane metadata. If no active snapshot exists, stale thread upserts are suppressed (and stale status events are dropped as a whole rather than creating orphan messages). + +## Delivery and recovery contract + +- Protocol 19 events have no upstream event ID, global sequence, or resume cursor. Herdr polls its independent subscriptions in request order, so frames from different event kinds/panes are not guaranteed to be source-chronological; the bridge preserves wire-arrival order and cannot reconstruct a total source order. Each event receives a bridge-owned UUID when durably captured. Non-status subscriptions may replay retained history; the status subscription starts at its current sequence and emits future transitions. Reconnect status snapshots may repeat an identical status message; this is accepted. The bridge does not deduplicate by payload or claim exactly-once upstream recovery, because identical events can be legitimate. +- The bridge persists each event before acknowledging it to its own delivery loop. It converts up to 20 events into one immutable request and flushes when the batch reaches 20 events or its oldest event is five seconds old. +- Each source frame is projected to an explicit event/data allowlist and durably persisted with `mapping_ready=false` before per-event IPC. Raw titles, paths, terminal details, token maps, arbitrary envelope fields, and unknown payload fields are not stored. After fresh workspace/pane snapshot lookups succeed, the normalized record is atomically rewritten as ready. If a lookup fails, that normalized record blocks later ordered sealing; the next connection resolves it from fresh startup snapshots before subscribing. This avoids relying on Herdr's bounded replay buffer to recover an event already read from the socket. +- The complete serialized `IngestBatch`, including its stable replay key, is atomically persisted and synced before any HTTP request. Retried/restarted delivery submits the exact same request body and identity, allowing Iris's transactional ingest replay check to make an uncertain HTTP retry idempotent. Only normalized event projections and allowlisted workspace/pane summaries remain in the private spool. The bridge never logs or copies the raw Herdr event into Iris message metadata or the audit record. +- Herdr v0.8.0 retains at most 512 events. If the bridge is disconnected long enough for older events to rotate out before it reconnects, protocol 19 supplies no resume cursor or loss marker; that upstream offline loss cannot be repaired by this bridge. +- The local spool is mode `0700`, its files are mode `0600`, and it is bounded to 10,000 queued events (including events in pending batches). On overflow it drops the oldest queued event/batch and logs only the count. An Iris outage longer than the spool can absorb therefore causes explicitly logged local loss. +- Iris `/ingest` accepts request bodies up to 1 MiB. If one normalized source event alone exceeds that limit, the bridge retains it as the oldest spool record and blocks sealing later events rather than retiring it as a successful audit-only delivery. It emits one content-free warning per process; if the queue eventually reaches its approved 10,000-event bound, normal oldest-first spool eviction applies and logs only the aggregate loss count. Reducing the source event or changing the Iris request limit requires a separately reviewed change. +- When no upstream event has arrived for an hour, the bridge enqueues a synthetic `bridge_heartbeat` as a system message in the dedicated Herdr bridge-health thread. This is ordinary authenticated ingest, not a Herdr event. +- Request/response bodies, bearer tokens, workspace names, terminal contents, and event IDs are not written to logs. Logs contain only aggregate loss counts, the local spool sequence and configured body limit for a blocked oversized record, protocol error classes, and HTTP status codes. + +## Build and isolated verification + +From the repository root: + +```sh +cargo test -p iris-herdr-bridge --locked +cargo build --locked -p iris-herdr-bridge +``` + +The crate tests use Herdr v0.8.0/protocol-19 subscribe-ack, workspace-list, pane-created, and pane-status fixtures, a synthetic Unix socket, and a loopback HTTP mock. Fixtures include synthetic tokens, private titles, and worktree paths and verify those values never reach the durable spool or Iris metadata. Tests do not connect to endver, install secrets, or call the configured Iris host. + +## User-unit installation (manual/local acceptance only) + +Build and install the binary on the target host after reviewing the PR: + +```sh +cargo build --release --locked -p iris-herdr-bridge +install -d -m 0700 "$HOME/.local/bin" "$HOME/.config/herdr-iris-bridge" \ + "$HOME/.local/state/herdr-iris-bridge" "$HOME/.config/systemd/user" +install -m 0755 target/release/iris-herdr-bridge "$HOME/.local/bin/iris-herdr-bridge" +install -m 0644 contrib/systemd/herdr-iris-bridge.service \ + "$HOME/.config/systemd/user/herdr-iris-bridge.service" +``` + +Create `$HOME/.config/herdr-iris-bridge/env` in a trusted local session. Never commit this file or copy its contents into issue comments/logs. The bridge defaults `HERDR_SOCKET` to `$HOME/.config/herdr/herdr.sock`; omit that variable for the default. If an alternate socket path is necessary, use a literal absolute path: systemd `EnvironmentFile` entries are not shell-expanded, so `$HOME` would be passed literally. + +```sh +IRIS_INGEST_TOKEN="" +``` + +Then restrict permissions and start the user unit: + +```sh +chmod 0600 "$HOME/.config/herdr-iris-bridge/env" +systemctl --user daemon-reload +systemctl --user enable --now herdr-iris-bridge.service +systemctl --user status herdr-iris-bridge.service +journalctl --user -u herdr-iris-bridge.service +``` + +The Iris endpoint is deliberately fixed to `http://100.66.233.79:9876/ingest`, the approved ticket target; only the bearer credential is provisioned locally. The environment file must contain a non-empty single-line token. The service unit has no secret value and does not install itself. For local acceptance, separately verify service restart, Herdr restart/reconnect, batch-flush journal lines, and real Iris delivery. Hosted fixtures/build tests are not evidence that installation, credentials, live delivery, or deployment occurred. From 056e1338358d4411c293f98dd374d5d5e5974252 Mon Sep 17 00:00:00 2001 From: Shiv Rossi Date: Sat, 10 Oct 2026 08:47:00 -0600 Subject: [PATCH 2/4] test(herdr): fix strict clippy assertions --- crates/iris-herdr-bridge/src/protocol.rs | 8 ++++---- crates/iris-herdr-bridge/src/spool.rs | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/crates/iris-herdr-bridge/src/protocol.rs b/crates/iris-herdr-bridge/src/protocol.rs index a28df59..a3c4df3 100644 --- a/crates/iris-herdr-bridge/src/protocol.rs +++ b/crates/iris-herdr-bridge/src/protocol.rs @@ -1073,7 +1073,7 @@ mod tests { .await .unwrap(); let mut spool_guard = spool.lock().await; - assert!(spool_guard.unmapped_event_payloads().unwrap().is_empty()); + assert_eq!(spool_guard.unmapped_event_payloads().unwrap(), Vec::new()); assert!(spool_guard.flush_due(Utc::now()).unwrap()); let batch = spool_guard.oldest_batch().unwrap().unwrap(); assert!(batch.body.contains("authoritative current label")); @@ -1131,7 +1131,7 @@ mod tests { .unwrap(); let mut spool = spool.lock().await; - assert!(spool.unmapped_event_payloads().unwrap().is_empty()); + assert_eq!(spool.unmapped_event_payloads().unwrap(), Vec::new()); assert!( spool .flush_due(Utc::now() + chrono::Duration::seconds(5)) @@ -1183,7 +1183,7 @@ mod tests { .await .unwrap(); let mut spool = spool.lock().await; - assert!(spool.unmapped_event_payloads().unwrap().is_empty()); + assert_eq!(spool.unmapped_event_payloads().unwrap(), Vec::new()); assert!(spool.flush_due(Utc::now()).unwrap()); let batch = spool.oldest_batch().unwrap().unwrap(); drop(spool); @@ -1457,7 +1457,7 @@ mod tests { .iter() .filter(|mutation| mutation["kind"] == "upsert_thread") .collect::>(); - assert!(!threads.is_empty()); + assert_ne!(threads, Vec::<&Value>::new()); assert!(threads.iter().all(|thread| { thread["metadata"]["panes"][0]["title"].is_null() && thread["metadata"]["panes"][0]["pane_id"] == "pane-1" diff --git a/crates/iris-herdr-bridge/src/spool.rs b/crates/iris-herdr-bridge/src/spool.rs index adaddd3..0423591 100644 --- a/crates/iris-herdr-bridge/src/spool.rs +++ b/crates/iris-herdr-bridge/src/spool.rs @@ -954,7 +954,7 @@ mod tests { enrich_thread_snapshots(&payload, &mut mutations, received_at), (1, true) ); - assert!(mutations.is_empty()); + assert_eq!(mutations, Vec::new()); } #[test] @@ -1014,7 +1014,7 @@ mod tests { // must survive because it is newer than the event inside the batch. spool.max_events = 1; assert_eq!(spool.enforce_bound().unwrap(), 1); - assert!(spool.batch_paths().unwrap().is_empty()); + assert_eq!(spool.batch_paths().unwrap(), Vec::new()); let remaining = spool.event_records().unwrap(); assert_eq!(remaining.len(), 1); assert_eq!(remaining[0].sequence, 2); From 9fb7bd8928cec189a8c2c55dda56bfe5f0ceb34c Mon Sep 17 00:00:00 2001 From: Shiv Rossi Date: Tue, 13 Oct 2026 13:57:59 -0600 Subject: [PATCH 3/4] fix(herdr): align systemd ExecStart with installed binary name --- contrib/systemd/herdr-iris-bridge.service | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/contrib/systemd/herdr-iris-bridge.service b/contrib/systemd/herdr-iris-bridge.service index c6253bb..ce16847 100644 --- a/contrib/systemd/herdr-iris-bridge.service +++ b/contrib/systemd/herdr-iris-bridge.service @@ -5,7 +5,7 @@ After=default.target [Service] Type=simple EnvironmentFile=%h/.config/herdr-iris-bridge/env -ExecStart=%h/.local/bin/herdr-iris-bridge +ExecStart=%h/.local/bin/iris-herdr-bridge Restart=on-failure RestartSec=2s UMask=0077 From deb74f632d9392dc656d0d8c58fb25ad5bc4e658 Mon Sep 17 00:00:00 2001 From: Shiv Rossi Date: Tue, 13 Oct 2026 14:03:35 -0600 Subject: [PATCH 4/4] fix(deps): bump quinn-proto to 0.11.19 for RUSTSEC-2026-0338..0342 --- Cargo.lock | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4f85335..569ee1d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1559,9 +1559,9 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.16" +version = "0.11.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" +checksum = "0e750cca55fe4f0439a15d0bb529da9651e79993e8e72c61a899a36d462befbe" dependencies = [ "bytes", "getrandom 0.4.3", @@ -2078,7 +2078,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2",