oino_core/
topics.rs

1//! Frontend topic feed (`_oino/topics`, modes §4): the sticky bus topics extensions publish
2//! (an embedder's `acme:long-horizon:state`, `acme:workflow:status`, `acme:bg:tasks`,
3//! `acme:fusion:status`, `acme:kanboard:status`, …) plus the core's own `oino:git` topic,
4//! mirrored to subscribed frontends as [`crate::events::OinoEvent::Topic`].
5//!
6//! The extension bus itself is served by the WASM host; it mirrors every publish into
7//! [`TopicStore::publish`] through [`crate::extensions::host::HostOp::BusPublish`]. The store
8//! keeps the last value of sticky topics per `(cwd, agent dir)` services, so a frontend that
9//! subscribes late receives the current state as a snapshot.
10
11use std::path::Path;
12use std::sync::{Mutex, MutexGuard, PoisonError};
13
14use indexmap::IndexMap;
15use oino_ai::JsonValue;
16use serde::{Deserialize, Serialize};
17
18/// Topic of the core's git summary of the session cwd (`{branch, head, dirty, changed}`).
19pub const GIT_TOPIC: &str = "oino:git";
20
21/// Prefix of the core's settings-change topics: `oino:settings:<namespace>`.
22pub const SETTINGS_TOPIC_PREFIX: &str = "oino:settings:";
23
24/// Topic announcing a write to extension settings namespace `namespace`
25/// (`{namespace, keys, scope, extensionId?}`).
26#[must_use]
27pub fn settings_topic(namespace: &str) -> String {
28    format!("{SETTINGS_TOPIC_PREFIX}{namespace}")
29}
30
31/// Publisher name of core-produced topics.
32pub const CORE_PUBLISHER: &str = "oino.core";
33
34/// One topic value as frontends see it.
35#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
36#[serde(rename_all = "camelCase")]
37pub struct TopicValue {
38    /// Topic name.
39    pub topic: String,
40    /// Payload.
41    pub data: JsonValue,
42    /// Published sticky (kept for late subscribers).
43    pub sticky: bool,
44    /// Publishing extension id (`oino.core` for core topics).
45    pub publisher: String,
46}
47
48/// `pattern` matches `topic`: exact, `prefix*` or `*` (the bus pattern syntax).
49#[must_use]
50pub fn topic_matches(pattern: &str, topic: &str) -> bool {
51    pattern == "*"
52        || pattern == topic
53        || pattern
54            .strip_suffix('*')
55            .is_some_and(|prefix| topic.starts_with(prefix))
56}
57
58/// Whether any of `patterns` matches `topic`.
59#[must_use]
60pub fn any_matches(patterns: &[String], topic: &str) -> bool {
61    patterns.iter().any(|p| topic_matches(p, topic))
62}
63
64/// Last values of sticky topics.
65#[derive(Debug, Default)]
66pub struct TopicStore {
67    sticky: Mutex<IndexMap<String, TopicValue>>,
68}
69
70fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
71    mutex.lock().unwrap_or_else(PoisonError::into_inner)
72}
73
74impl TopicStore {
75    /// Records a publish; returns the value to fan out, `None` when a sticky value did not
76    /// change (no event needed).
77    pub fn publish(
78        &self,
79        publisher: &str,
80        topic: &str,
81        data: JsonValue,
82        sticky: bool,
83    ) -> Option<TopicValue> {
84        let value = TopicValue {
85            topic: topic.to_owned(),
86            data,
87            sticky,
88            publisher: publisher.to_owned(),
89        };
90        if sticky {
91            let mut map = lock(&self.sticky);
92            if map.get(topic).is_some_and(|old| old.data == value.data) {
93                return None;
94            }
95            map.insert(topic.to_owned(), value.clone());
96        }
97        Some(value)
98    }
99
100    /// Last sticky value of `topic`.
101    #[must_use]
102    pub fn get(&self, topic: &str) -> Option<TopicValue> {
103        lock(&self.sticky).get(topic).cloned()
104    }
105
106    /// Sticky values matching any of `patterns`, in first-publish order.
107    #[must_use]
108    pub fn snapshot(&self, patterns: &[String]) -> Vec<TopicValue> {
109        lock(&self.sticky)
110            .values()
111            .filter(|v| any_matches(patterns, &v.topic))
112            .cloned()
113            .collect()
114    }
115}
116
117/// Git summary of `cwd` for [`GIT_TOPIC`]: `{branch, head, dirty, changed}`, `null` outside a
118/// repository or without `git`.
119#[must_use]
120pub fn git_info(cwd: &Path) -> JsonValue {
121    let output = std::process::Command::new("git")
122        .args([
123            "status",
124            "--porcelain=v2",
125            "--branch",
126            "--untracked-files=normal",
127        ])
128        .current_dir(cwd)
129        .stdin(std::process::Stdio::null())
130        .stderr(std::process::Stdio::null())
131        .output();
132    match output {
133        Ok(output) if output.status.success() => {
134            parse_git_status(&String::from_utf8_lossy(&output.stdout))
135        }
136        _ => JsonValue::Null,
137    }
138}
139
140/// Parses `git status --porcelain=v2 --branch`.
141fn parse_git_status(text: &str) -> JsonValue {
142    let mut branch = None;
143    let mut head = None;
144    let (mut ahead, mut behind) = (0_i64, 0_i64);
145    let mut changed = 0_u64;
146    for line in text.lines() {
147        if let Some(rest) = line.strip_prefix("# branch.head ") {
148            branch = (rest != "(detached)").then(|| rest.to_owned());
149        } else if let Some(rest) = line.strip_prefix("# branch.oid ") {
150            head = (rest != "(initial)").then(|| rest.chars().take(7).collect::<String>());
151        } else if let Some(rest) = line.strip_prefix("# branch.ab ") {
152            for part in rest.split_whitespace() {
153                if let Some(n) = part.strip_prefix('+') {
154                    ahead = n.parse().unwrap_or(0);
155                } else if let Some(n) = part.strip_prefix('-') {
156                    behind = n.parse().unwrap_or(0);
157                }
158            }
159        } else if !line.starts_with('#') && !line.is_empty() {
160            changed += 1;
161        }
162    }
163    serde_json::json!({
164        "branch": branch,
165        "head": head,
166        "dirty": changed > 0,
167        "changed": changed,
168        "ahead": ahead,
169        "behind": behind,
170    })
171}
172
173#[cfg(test)]
174mod tests {
175    use super::*;
176    use serde_json::json;
177
178    #[test]
179    fn patterns() {
180        assert!(topic_matches("acme:*", "acme:bg:tasks"));
181        assert!(topic_matches("*", "x"));
182        assert!(topic_matches("a", "a"));
183        assert!(!topic_matches("a", "ab"));
184        assert!(any_matches(&["x".into(), "oino:*".into()], GIT_TOPIC));
185    }
186
187    #[test]
188    fn sticky_values_dedupe_and_snapshot() {
189        let store = TopicStore::default();
190        assert!(store.publish("e", "a:x", json!(1), true).is_some());
191        assert!(store.publish("e", "a:x", json!(1), true).is_none());
192        assert!(store.publish("e", "a:x", json!(2), true).is_some());
193        assert!(store.publish("e", "b:once", json!(1), false).is_some());
194        assert!(store.publish("e", "b:once", json!(1), false).is_some());
195        assert_eq!(store.snapshot(&["*".into()]).len(), 1);
196        assert_eq!(store.get("a:x").unwrap().data, json!(2));
197        assert!(store.get("b:once").is_none());
198        assert_eq!(store.snapshot(&["b:*".into()]), Vec::new());
199    }
200
201    #[test]
202    fn git_status_parse() {
203        let v = parse_git_status(
204            "# branch.oid 0123456789abcdef\n# branch.head main\n# branch.ab +2 -1\n1 .M N... 100644 100644 100644 a b f.rs\n? new.txt\n",
205        );
206        assert_eq!(
207            v,
208            json!({"branch": "main", "head": "0123456", "dirty": true, "changed": 2, "ahead": 2, "behind": 1})
209        );
210        let clean = parse_git_status("# branch.oid (initial)\n# branch.head (detached)\n");
211        assert_eq!(clean["dirty"], json!(false));
212        assert_eq!(clean["branch"], JsonValue::Null);
213    }
214}
oino_core/topics.rs

oino_core/
topics.rs
