//! Subagent terminals: the subject each subagent's rows go to, and the //! stream that keeps them. //! //! An agent's subagent daemon publishes each subagent's classified terminal //! rows on `$SWARM.term..sub.`, under the agent's own queue //! credential. The queue grants that credential publish on this subject //! family and no `JetStream` subject. //! //! The stream lets the swarm list an agent's subagents: subagents //! are spawned on demand under caller-chosen names, and the set of subjects //! the stream holds is the list. `swarm-controller` creates one stream per //! agent with the fixed config //! `swarm_queue_client::subagent_term::open_or_create` spells, and takes the //! list from its subject counts ([`crate::subagent_term::subagent_names`]). A //! reader attaching to one subagent's terminal replays that subagent's newest //! rows from it. //! //! The name and the subject live here because they must agree across the //! publisher in `hive-subagent-mcp`, `swarm-controller`, the queue grant in //! `swarm-nats.nix`, whose `{agent}` template spells the same subjects, and //! the reader's consumer grant in `swarm-nats-auth`. use crate::TERM_SUBJECT_PREFIX; /// The token between the agent and the subagent name. A subagent subject has /// two more tokens than the agent's own `$SWARM.term.`, so neither /// family's subscriber receives the other's rows. const SUBAGENT_TOKEN: &str = "sub"; /// How long the stream keeps a row. The subagent list is the subjects with a /// row inside this window, so a subagent drops off the list this long after /// its last output. pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); /// The stream holding `agent`'s subagent rows: `term-sub-`. #[must_use] pub fn stream_name(agent: &str) -> String { format!("term-sub-{agent}") } /// The subject `subagent`'s rows are published on. #[must_use] pub fn subject(agent: &str, subagent: &str) -> String { format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.{subagent}") } /// Every [`subject`] at once: the reader's subscribe grant, which has no /// roster to enumerate agents or subagents from. #[must_use] pub fn subject_wildcard() -> String { subject("*", "*") } /// The reader's grant to create a replay consumer on one subagent's rows: /// the extended create form of [`crate::agent_term::consumer_create_grant`], /// on any stream and under any consumer name, with a [`subject`] as the /// filter tokens the server checks the payload against. The last `*` also /// matches a literal `*` or `>`, so it admits `$SWARM.term..sub.>` /// too, and nothing outside [`subject_wildcard`]'s family. #[must_use] pub fn consumer_create_grant() -> String { format!("$JS.API.CONSUMER.CREATE.*.*.{}", subject_wildcard()) } /// Every subject [`stream_name`]'s stream captures, which is also the publish /// grant's shape. #[must_use] pub fn stream_subjects(agent: &str) -> String { format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.>") } /// The subagent a stream subject names, or `None` for a subject that is not /// exactly one subagent token under `agent`'s family. #[must_use] pub fn subagent_from_subject<'a>(agent: &str, subject: &'a str) -> Option<&'a str> { let name = subject .strip_prefix(TERM_SUBJECT_PREFIX)? .strip_prefix('.')? .strip_prefix(agent)? .strip_prefix('.')? .strip_prefix(SUBAGENT_TOKEN)? .strip_prefix('.')?; (!name.is_empty() && !name.contains('.')).then_some(name) } /// The subagents named by a stream's subjects, sorted and deduplicated. /// Subjects outside `agent`'s family are skipped. #[must_use] pub fn subagent_names<'a>(agent: &str, subjects: impl IntoIterator) -> Vec { let mut names: Vec = subjects .into_iter() .filter_map(|s| subagent_from_subject(agent, s)) .map(str::to_owned) .collect(); names.sort_unstable(); names.dedup(); names } #[cfg(test)] mod tests { use super::*; /// The names the grant in `swarm-nats.nix` spells with `{agent}`. #[test] fn names_match_the_queue_grant() { assert_eq!(stream_name("iris"), "term-sub-iris"); assert_eq!(subject("iris", "scout"), "$SWARM.term.iris.sub.scout"); assert_eq!(stream_subjects("iris"), "$SWARM.term.iris.sub.>"); assert_eq!( consumer_create_grant(), "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*.sub.*" ); } #[test] fn a_subject_reads_back_as_its_subagent() { assert_eq!( subagent_from_subject("iris", &subject("iris", "scout")), Some("scout") ); } /// The agent's own terminal, another agent's subagent, and anything /// deeper than one token are not subagents of `iris`. #[test] fn other_subjects_name_no_subagent() { for s in [ "$SWARM.term.iris", "$SWARM.term.h1.iris", "$SWARM.term.iris.sub", "$SWARM.term.iris.sub.", "$SWARM.term.iris.sub.a.b", "$SWARM.term.iris-2.sub.scout", "$SWARM.term.argus.sub.scout", "$SWARM.agent-state.iris.sub.scout", ] { assert_eq!(subagent_from_subject("iris", s), None, "{s}"); } } #[test] fn the_list_is_sorted_unique_and_scoped_to_the_agent() { let subjects = [ "$SWARM.term.iris.sub.zed", "$SWARM.term.iris.sub.alpha", "$SWARM.term.argus.sub.other", "$SWARM.term.iris.sub.alpha", ]; assert_eq!(subagent_names("iris", subjects), ["alpha", "zed"]); } }