netsky_core/

envelope.rs

//! Shared agent-bus envelope: the on-disk shape, filename convention, and
//! atomic-with-create-new write used by every producer on the bus. Consumers
//! read envelopes with their own deserializer; writers MUST go through this
//! module so every inbox file agrees on filename shape.
//!
//! Filename: `{nanos:020}-{pid:010}-{rand_hex:08}-{seq:010}-from-{from}.json`.
//! The nanos+pid+rand+seq quartet gives cross-process uniqueness that a
//! process-local `AtomicU64` alone cannot: two fresh `netsky channel send`
//! processes both start at `seq=0`, and a wall-clock nanosecond collision
//! between them would clobber each other's envelope under the old shape.
//!
//! Ordering is best-effort: consumers sort filenames lexicographically,
//! which orders by nanos first. Clock skew between producers or concurrent
//! writers can reorder causal order. Consumers that need causal ordering
//! must embed a sequence number in the envelope body.
//!
//! Write flow: serialize → write `.<name>.<attempt>.tmp` → hard-link to
//! `<name>` (create-new semantics; fails on collision) → unlink tmp.
//! Retries with a fresh name on collision; returns AlreadyExists after 8
//! failed attempts so a wedged inbox cannot spin forever.

use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use serde::{Deserialize, Serialize};

const CHANNEL_WRAPPER_TOKENS: &[&str] = &["</channel>", "<channel source="];

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Envelope {
    pub from: String,
    pub text: String,
    pub ts: String,
}

/// `from`/`target` id validator: `agent` prefix + non-empty lowercase-alnum
/// suffix. Mirrored in the MCP agent source so filenames and inbox names
/// round-trip under a single rule.
pub fn valid_agent_id(id: &str) -> bool {
    match id.strip_prefix("agent") {
        Some(suffix) => {
            !suffix.is_empty()
                && suffix
                    .chars()
                    .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
        }
        None => false,
    }
}

/// Build the canonical inbox filename. `from` is trusted to have already
/// passed [`valid_agent_id`] — callers that have not validated must do so
/// before constructing envelopes.
pub fn build_filename(from: &str) -> String {
    static SEQ: AtomicU64 = AtomicU64::new(0);
    let seq = SEQ.fetch_add(1, Ordering::Relaxed);
    let nanos = chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0);
    let pid = std::process::id();
    let rand_hex: u32 = rand::random();
    format!("{nanos:020}-{pid:010}-{rand_hex:08x}-{seq:010}-from-{from}.json")
}

/// Atomically write `env` into `inbox`. Loops up to 8 times on filename
/// collision (AlreadyExists) with a fresh name each attempt. Returns the
/// final path on success.
pub fn write_envelope(inbox: &Path, env: &Envelope) -> io::Result<PathBuf> {
    fs::create_dir_all(inbox)?;
    let bytes =
        serde_json::to_vec(env).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
    let mut last_err: Option<io::Error> = None;
    for attempt in 0..8 {
        let name = build_filename(&env.from);
        let final_path = inbox.join(&name);
        let tmp_path = inbox.join(format!(".{name}.{attempt}.tmp"));
        fs::write(&tmp_path, &bytes)?;
        match fs::hard_link(&tmp_path, &final_path) {
            Ok(()) => {
                let _ = fs::remove_file(&tmp_path);
                return Ok(final_path);
            }
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
                let _ = fs::remove_file(&tmp_path);
                last_err = Some(e);
                continue;
            }
            Err(e) => {
                let _ = fs::remove_file(&tmp_path);
                return Err(e);
            }
        }
    }
    Err(last_err.unwrap_or_else(|| {
        io::Error::new(
            io::ErrorKind::AlreadyExists,
            "exhausted 8 attempts to create unique envelope filename",
        )
    }))
}

/// Reject content that would break the shell-facing `<channel>` framing.
pub fn body_contains_wrapper_tokens(body: &str) -> bool {
    CHANNEL_WRAPPER_TOKENS
        .iter()
        .any(|needle| body.contains(needle))
}

/// Validate the shared agent-bus envelope shape before a drain prints it
/// into channel framing.
pub fn validate_bus_envelope(env: &Envelope) -> Result<(), String> {
    if !valid_agent_id(&env.from) {
        return Err(format!(
            "invalid from {:?} (expected agent<lowercase-alnum>)",
            env.from
        ));
    }
    if chrono::DateTime::parse_from_rfc3339(&env.ts).is_err() {
        return Err(format!("invalid ts {:?} (expected RFC3339)", env.ts));
    }
    if body_contains_wrapper_tokens(&env.text) {
        return Err("body contains <channel> wrapper token; framing-break rejected".to_string());
    }
    Ok(())
}

/// Escape the channel body defensively before it is wrapped in XML-like
/// framing and shown to a model.
pub fn xml_escape_body(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        match c {
            '&' => out.push_str("&amp;"),
            '<' => out.push_str("&lt;"),
            '>' => out.push_str("&gt;"),
            '"' => out.push_str("&quot;"),
            '\'' => out.push_str("&apos;"),
            other => out.push(other),
        }
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn valid_agent_id_accepts_canonical_shapes() {
        assert!(valid_agent_id("agent0"));
        assert!(valid_agent_id("agent42"));
        assert!(valid_agent_id("agentinfinity"));
    }

    #[test]
    fn valid_agent_id_rejects_junk() {
        assert!(!valid_agent_id(""));
        assert!(!valid_agent_id("agent"));
        assert!(!valid_agent_id("AGENT0"));
        assert!(!valid_agent_id("agent-7"));
        assert!(!valid_agent_id("bob"));
        assert!(!valid_agent_id("agent 0"));
    }

    #[test]
    fn build_filename_shape() {
        let n = build_filename("agent7");
        assert!(n.ends_with("-from-agent7.json"));
        let parts: Vec<&str> = n.splitn(5, '-').collect();
        assert_eq!(parts.len(), 5);
        assert_eq!(parts[0].len(), 20);
        assert_eq!(parts[1].len(), 10);
        assert_eq!(parts[2].len(), 8);
    }

    #[test]
    fn build_filename_is_unique_within_process() {
        let mut seen = std::collections::HashSet::new();
        for _ in 0..1000 {
            assert!(seen.insert(build_filename("agent0")));
        }
    }

    #[test]
    fn write_envelope_round_trips() {
        let tmp = tempfile::tempdir().unwrap();
        let env = Envelope {
            from: "agent2".to_string(),
            text: "hello".to_string(),
            ts: "2026-04-15T21:00:00Z".to_string(),
        };
        let path = write_envelope(tmp.path(), &env).unwrap();
        let raw = fs::read_to_string(&path).unwrap();
        let round: Envelope = serde_json::from_str(&raw).unwrap();
        assert_eq!(round.from, env.from);
        assert_eq!(round.text, env.text);
    }

    #[test]
    fn write_envelope_rejects_name_collision() {
        // Pre-populate the inbox with a file whose hard-link target we
        // force to collide. We cannot cheaply force the AtomicU64+rand to
        // repeat, so assert the happy path here and exercise the collision
        // branch via a unit on the retry loop.
        let tmp = tempfile::tempdir().unwrap();
        let env = Envelope {
            from: "agent2".to_string(),
            text: "hi".to_string(),
            ts: "2026-04-15T21:00:00Z".to_string(),
        };
        let a = write_envelope(tmp.path(), &env).unwrap();
        let b = write_envelope(tmp.path(), &env).unwrap();
        assert_ne!(a, b);
    }
}