lean_ctx/core/context_os/
context_bus.rs

1use std::collections::HashMap;
2use std::path::PathBuf;
3use std::sync::{Arc, Mutex};
4
5use chrono::{DateTime, Utc};
6use rusqlite::{Connection, params};
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use tokio::sync::broadcast;
10
11const MAX_READ_CONNS: usize = 4;
12
13#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
14#[serde(rename_all = "snake_case")]
15pub enum ContextEventKindV1 {
16    ToolCallRecorded,
17    SessionMutated,
18    KnowledgeRemembered,
19    ArtifactStored,
20    GraphBuilt,
21    ProofAdded,
22}
23
24impl ContextEventKindV1 {
25    pub fn as_str(&self) -> &'static str {
26        match self {
27            Self::ToolCallRecorded => "tool_call_recorded",
28            Self::SessionMutated => "session_mutated",
29            Self::KnowledgeRemembered => "knowledge_remembered",
30            Self::ArtifactStored => "artifact_stored",
31            Self::GraphBuilt => "graph_built",
32            Self::ProofAdded => "proof_added",
33        }
34    }
35
36    pub fn parse(s: &str) -> Self {
37        match s.trim().to_lowercase().as_str() {
38            "tool_call_recorded" => Self::ToolCallRecorded,
39            "session_mutated" => Self::SessionMutated,
40            "knowledge_remembered" => Self::KnowledgeRemembered,
41            "artifact_stored" => Self::ArtifactStored,
42            "graph_built" => Self::GraphBuilt,
43            "proof_added" => Self::ProofAdded,
44            other => {
45                tracing::warn!(
46                    "unknown ContextEventKind '{other}', defaulting to ToolCallRecorded"
47                );
48                Self::ToolCallRecorded
49            }
50        }
51    }
52
53    /// Classifies the consistency requirement for this event kind.
54    ///
55    /// - `Local`: Agent-local, never shared (tool reads, cache hits).
56    /// - `Eventual`: Broadcast via bus, other agents see it "soon" (knowledge, artifacts).
57    /// - `Strong`: Critical decisions that require acknowledgment before proceeding.
58    pub fn consistency_level(&self) -> ConsistencyLevel {
59        match self {
60            Self::ToolCallRecorded | Self::GraphBuilt => ConsistencyLevel::Local,
61            Self::KnowledgeRemembered | Self::ArtifactStored => ConsistencyLevel::Eventual,
62            Self::SessionMutated | Self::ProofAdded => ConsistencyLevel::Strong,
63        }
64    }
65}
66
67/// Consistency requirement for shared context events.
68/// Ordered from least to most strict for filtering comparisons.
69#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
70#[serde(rename_all = "snake_case")]
71pub enum ConsistencyLevel {
72    /// Agent-local, authoritative: session task, local cache, current file set.
73    Local = 0,
74    /// Shared, eventually consistent: knowledge facts, gotchas, artifact refs.
75    Eventual = 1,
76    /// Shared, strongly consistent: workspace config, critical decisions.
77    Strong = 2,
78}
79
80impl ConsistencyLevel {
81    pub fn as_str(&self) -> &'static str {
82        match self {
83            Self::Local => "local",
84            Self::Eventual => "eventual",
85            Self::Strong => "strong",
86        }
87    }
88}
89
90#[derive(Debug, Clone, Serialize, Deserialize)]
91#[serde(rename_all = "camelCase")]
92pub struct ContextEventV1 {
93    pub id: i64,
94    pub workspace_id: String,
95    pub channel_id: String,
96    pub kind: String,
97    pub actor: Option<String>,
98    pub timestamp: DateTime<Utc>,
99    pub version: i64,
100    #[serde(skip_serializing_if = "Option::is_none")]
101    pub parent_id: Option<i64>,
102    pub consistency_level: String,
103    pub payload: Value,
104    #[serde(skip_serializing_if = "Option::is_none", default)]
105    pub target_agents: Option<Vec<String>>,
106}
107
108impl ContextEventV1 {
109    pub fn consistency(&self) -> ConsistencyLevel {
110        ContextEventKindV1::parse(&self.kind).consistency_level()
111    }
112
113    pub fn is_visible_to_agent(&self, agent_id: &str) -> bool {
114        match &self.target_agents {
115            None => true,
116            Some(targets) => targets.iter().any(|t| t == agent_id),
117        }
118    }
119}
120
121/// Filter for selective event subscriptions.
122/// All fields are optional; `None` means "accept all".
123#[derive(Debug, Clone, Default)]
124pub struct TopicFilter {
125    pub kinds: Option<Vec<ContextEventKindV1>>,
126    pub actors: Option<Vec<String>>,
127    pub min_consistency: Option<ConsistencyLevel>,
128    pub agent_id: Option<String>,
129}
130
131impl TopicFilter {
132    /// Convenience constructor: filter by event kind strings.
133    pub fn kinds(kind_strs: &[&str]) -> Self {
134        Self {
135            kinds: Some(
136                kind_strs
137                    .iter()
138                    .map(|s| ContextEventKindV1::parse(s))
139                    .collect(),
140            ),
141            ..Self::default()
142        }
143    }
144
145    pub fn matches(&self, event: &ContextEventV1) -> bool {
146        if let Some(ref kinds) = self.kinds {
147            let parsed = ContextEventKindV1::parse(&event.kind);
148            if !kinds.contains(&parsed) {
149                return false;
150            }
151        }
152        if let Some(ref actors) = self.actors {
153            match &event.actor {
154                Some(actor) if actors.iter().any(|a| a == actor) => {}
155                Some(_) | None => return false,
156            }
157        }
158        if let Some(min) = self.min_consistency
159            && event.consistency() < min
160        {
161            return false;
162        }
163        if let Some(ref aid) = self.agent_id
164            && !event.is_visible_to_agent(aid)
165        {
166            return false;
167        }
168        true
169    }
170}
171
172fn event_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ContextEventV1> {
173    let ts_str: String = row.get(5)?;
174    let ts = DateTime::parse_from_rfc3339(&ts_str)
175        .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc));
176    let payload_str: String = row.get(6)?;
177    let payload: Value = serde_json::from_str(&payload_str).unwrap_or(Value::Null);
178    let kind_str: String = row.get(3)?;
179    let cl = ContextEventKindV1::parse(&kind_str)
180        .consistency_level()
181        .as_str()
182        .to_string();
183    Ok(ContextEventV1 {
184        id: row.get(0)?,
185        workspace_id: row.get(1)?,
186        channel_id: row.get(2)?,
187        kind: kind_str,
188        actor: row.get::<_, Option<String>>(4)?,
189        timestamp: ts,
190        version: row.get::<_, i64>(7).unwrap_or(0),
191        parent_id: row.get::<_, Option<i64>>(8).ok().flatten(),
192        consistency_level: cl,
193        payload,
194        target_agents: None,
195    })
196}
197
198#[derive(Clone)]
199pub struct ContextBus {
200    inner: Arc<Inner>,
201}
202
203const STREAM_CHANNEL_SIZE: usize = 256;
204const MAX_SUBSCRIBERS_PER_CHANNEL: usize = 64;
205/// Bound the write-side version cache. Each entry is re-derivable from the DB via
206/// `MAX(version)`, so when the map exceeds this (e.g. a client cycling workspace/channel
207/// ids) it is simply cleared — costing at most one extra `MAX()` query per active channel.
208const MAX_VERSION_CACHE_ENTRIES: usize = 4096;
209
210struct Inner {
211    write_conn: Mutex<Connection>,
212    read_pool: Mutex<Vec<Connection>>,
213    streams: Mutex<HashMap<String, broadcast::Sender<ContextEventV1>>>,
214    version_cache: Mutex<HashMap<String, i64>>,
215    db_path: PathBuf,
216}
217
218impl Inner {
219    fn open_read_conn(path: &PathBuf) -> Connection {
220        // The process-global runtime (context_os::runtime) captures its DB path
221        // once, from LEAN_CTX_DATA_DIR. Under `cargo test`, a parallel
222        // `isolated_data_dir` can delete that directory after the runtime bound
223        // to it, so a lazily-opened read connection would hit a missing dir.
224        // Recreating the parent keeps opens infallible (matches `open_at`).
225        if let Some(parent) = path.parent() {
226            let _ = std::fs::create_dir_all(parent);
227        }
228        let conn = Connection::open(path).expect("open read context-os db");
229        let _ = conn.busy_timeout(std::time::Duration::from_secs(5));
230        let _ = conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA query_only=ON;");
231        conn
232    }
233
234    fn take_read_conn(&self) -> Connection {
235        self.read_pool
236            .lock()
237            .unwrap_or_else(std::sync::PoisonError::into_inner)
238            .pop()
239            .unwrap_or_else(|| Self::open_read_conn(&self.db_path))
240    }
241
242    fn return_read_conn(&self, conn: Connection) {
243        let mut pool = self
244            .read_pool
245            .lock()
246            .unwrap_or_else(std::sync::PoisonError::into_inner);
247        if pool.len() < MAX_READ_CONNS {
248            pool.push(conn);
249        }
250    }
251
252    fn stream_key(workspace_id: &str, channel_id: &str) -> String {
253        format!("{workspace_id}\0{channel_id}")
254    }
255
256    fn next_version(&self, workspace_id: &str, channel_id: &str) -> i64 {
257        let key = Self::stream_key(workspace_id, channel_id);
258
259        {
260            let mut cache = self
261                .version_cache
262                .lock()
263                .unwrap_or_else(std::sync::PoisonError::into_inner);
264            if let Some(v) = cache.get_mut(&key) {
265                *v += 1;
266                return *v;
267            }
268        }
269
270        let conn = self.take_read_conn();
271        let v: i64 = conn
272            .query_row(
273                "SELECT COALESCE(MAX(version), 0) FROM context_events WHERE workspace_id = ?1 AND channel_id = ?2",
274                params![workspace_id, channel_id],
275                |row| row.get(0),
276            )
277            .unwrap_or(0);
278        self.return_read_conn(conn);
279
280        let mut cache = self
281            .version_cache
282            .lock()
283            .unwrap_or_else(std::sync::PoisonError::into_inner);
284        if cache.len() > MAX_VERSION_CACHE_ENTRIES {
285            // Re-derivable from the DB; drop the whole cache rather than grow unbounded.
286            cache.clear();
287        }
288        let entry = cache.entry(key).or_insert(v);
289        *entry = (*entry).max(v) + 1;
290        *entry
291    }
292}
293
294impl Default for ContextBus {
295    fn default() -> Self {
296        Self::new()
297    }
298}
299
300impl ContextBus {
301    pub fn new() -> Self {
302        let path = default_db_path();
303        Self::open_at(path)
304    }
305
306    fn open_at(path: PathBuf) -> Self {
307        if let Some(parent) = path.parent() {
308            let _ = std::fs::create_dir_all(parent);
309        }
310        let conn = Connection::open(&path).expect("open context-os db");
311        let _ = conn.busy_timeout(std::time::Duration::from_secs(5));
312        conn.execute_batch(
313            "PRAGMA journal_mode=WAL;
314             CREATE TABLE IF NOT EXISTS context_events (
315               id INTEGER PRIMARY KEY AUTOINCREMENT,
316               workspace_id TEXT NOT NULL,
317               channel_id TEXT NOT NULL,
318               kind TEXT NOT NULL,
319               actor TEXT,
320               timestamp TEXT NOT NULL,
321               payload_json TEXT NOT NULL,
322               version INTEGER NOT NULL DEFAULT 0,
323               parent_id INTEGER
324             );
325             CREATE INDEX IF NOT EXISTS idx_context_events_stream
326               ON context_events(workspace_id, channel_id, id);",
327        )
328        .expect("init context-os db");
329
330        let _ = conn.execute_batch(
331            "ALTER TABLE context_events ADD COLUMN version INTEGER NOT NULL DEFAULT 0;",
332        );
333        let _ = conn.execute_batch("ALTER TABLE context_events ADD COLUMN parent_id INTEGER;");
334
335        let _ = conn.execute_batch(
336            "CREATE VIRTUAL TABLE IF NOT EXISTS context_events_fts USING fts5(
337               payload_text,
338               content=context_events,
339               content_rowid=id
340             );",
341        );
342
343        let mut read_conns = Vec::with_capacity(MAX_READ_CONNS);
344        for _ in 0..MAX_READ_CONNS {
345            read_conns.push(Inner::open_read_conn(&path));
346        }
347
348        Self {
349            inner: Arc::new(Inner {
350                write_conn: Mutex::new(conn),
351                read_pool: Mutex::new(read_conns),
352                streams: Mutex::new(HashMap::new()),
353                version_cache: Mutex::new(HashMap::new()),
354                db_path: path,
355            }),
356        }
357    }
358
359    pub fn subscribe(
360        &self,
361        workspace_id: &str,
362        channel_id: &str,
363    ) -> Option<broadcast::Receiver<ContextEventV1>> {
364        let key = Inner::stream_key(workspace_id, channel_id);
365        let mut streams = self
366            .inner
367            .streams
368            .lock()
369            .unwrap_or_else(std::sync::PoisonError::into_inner);
370        // Reap senders left behind by departed clients (receiver_count() == 0). A removed
371        // key is transparently recreated below; the DB is the durability source so this
372        // loses no deliverable events. Bounds `streams` to ~live connection count even
373        // under client-cycled workspace/channel ids.
374        streams.retain(|_, tx| tx.receiver_count() > 0);
375        let tx = streams
376            .entry(key)
377            .or_insert_with(|| broadcast::channel(STREAM_CHANNEL_SIZE).0);
378        if tx.receiver_count() >= MAX_SUBSCRIBERS_PER_CHANNEL {
379            tracing::warn!(
380                "SSE subscriber cap ({MAX_SUBSCRIBERS_PER_CHANNEL}) reached for {workspace_id}/{channel_id} — rejecting"
381            );
382            return None;
383        }
384        Some(tx.subscribe())
385    }
386
387    /// Subscribe with a filter — only events matching the filter are delivered.
388    /// Returns `(Receiver, TopicFilter)` for use in filtered receive loops.
389    pub fn subscribe_filtered(
390        &self,
391        workspace_id: &str,
392        channel_id: &str,
393        filter: TopicFilter,
394    ) -> Option<FilteredSubscription> {
395        let rx = self.subscribe(workspace_id, channel_id)?;
396        Some(FilteredSubscription { rx, filter })
397    }
398
399    pub fn append(
400        &self,
401        workspace_id: &str,
402        channel_id: &str,
403        kind: &ContextEventKindV1,
404        actor: Option<&str>,
405        payload: Value,
406    ) -> Option<ContextEventV1> {
407        self.append_with_parent(workspace_id, channel_id, kind, actor, payload, None)
408    }
409
410    pub fn append_with_parent(
411        &self,
412        workspace_id: &str,
413        channel_id: &str,
414        kind: &ContextEventKindV1,
415        actor: Option<&str>,
416        payload: Value,
417        parent_id: Option<i64>,
418    ) -> Option<ContextEventV1> {
419        let ev = self.insert_event(
420            workspace_id,
421            channel_id,
422            kind,
423            actor,
424            payload,
425            parent_id,
426            None,
427        )?;
428        self.broadcast_event(&ev);
429        Some(ev)
430    }
431
432    /// Append an event directed at specific agents only.
433    /// Only subscribers whose `TopicFilter.agent_id` matches a target will see it.
434    pub fn append_directed(
435        &self,
436        workspace_id: &str,
437        channel_id: &str,
438        kind: &ContextEventKindV1,
439        actor: Option<&str>,
440        payload: Value,
441        target_agents: Vec<String>,
442    ) -> Option<ContextEventV1> {
443        let ev = self.insert_event(
444            workspace_id,
445            channel_id,
446            kind,
447            actor,
448            payload,
449            None,
450            Some(target_agents),
451        )?;
452        self.broadcast_event(&ev);
453        Some(ev)
454    }
455
456    fn insert_event(
457        &self,
458        workspace_id: &str,
459        channel_id: &str,
460        kind: &ContextEventKindV1,
461        actor: Option<&str>,
462        payload: Value,
463        parent_id: Option<i64>,
464        target_agents: Option<Vec<String>>,
465    ) -> Option<ContextEventV1> {
466        let ts = Utc::now();
467        let payload_json = payload.to_string();
468
469        let (id, version) = {
470            let Ok(conn) = self.inner.write_conn.lock() else {
471                return None;
472            };
473            let version = self.inner.next_version(workspace_id, channel_id);
474
475            let result: Result<(i64, i64), rusqlite::Error> = conn
476                .execute_batch("BEGIN IMMEDIATE")
477                .and_then(|()| {
478                    conn.execute(
479                        "INSERT INTO context_events (workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id)
480                         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
481                        params![
482                            workspace_id,
483                            channel_id,
484                            kind.as_str(),
485                            actor.map(str::to_string),
486                            ts.to_rfc3339(),
487                            payload_json,
488                            version,
489                            parent_id,
490                        ],
491                    )?;
492                    let rowid = conn.last_insert_rowid();
493                    if let Err(e) = conn.execute(
494                        "INSERT INTO context_events_fts(rowid, payload_text) VALUES (?1, ?2)",
495                        params![rowid, payload_json],
496                    ) {
497                        tracing::warn!("FTS insert failed for event {rowid}: {e}");
498                    }
499                    conn.execute_batch("COMMIT")?;
500                    Ok((rowid, version))
501                });
502
503            match result {
504                Ok(pair) => pair,
505                Err(e) => {
506                    tracing::warn!("context bus append failed: {e}");
507                    let _ = conn.execute_batch("ROLLBACK");
508                    return None;
509                }
510            }
511        };
512
513        Some(ContextEventV1 {
514            id,
515            workspace_id: workspace_id.to_string(),
516            channel_id: channel_id.to_string(),
517            consistency_level: kind.consistency_level().as_str().to_string(),
518            kind: kind.as_str().to_string(),
519            actor: actor.map(str::to_string),
520            timestamp: ts,
521            version,
522            parent_id,
523            payload,
524            target_agents,
525        })
526    }
527
528    fn broadcast_event(&self, ev: &ContextEventV1) {
529        let key = Inner::stream_key(&ev.workspace_id, &ev.channel_id);
530        let tx = self
531            .inner
532            .streams
533            .lock()
534            .unwrap_or_else(std::sync::PoisonError::into_inner)
535            .get(&key)
536            .cloned();
537        if let Some(tx) = tx {
538            let _ = tx.send(ev.clone());
539        }
540    }
541
542    pub fn read(
543        &self,
544        workspace_id: &str,
545        channel_id: &str,
546        since: i64,
547        limit: usize,
548    ) -> Vec<ContextEventV1> {
549        let limit = limit.clamp(1, 1000) as i64;
550        let conn = self.inner.take_read_conn();
551        let result = (|| {
552            let mut stmt = conn.prepare(
553                "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id
554                 FROM context_events
555                 WHERE workspace_id = ?1 AND channel_id = ?2 AND id > ?3
556                 ORDER BY id ASC
557                 LIMIT ?4",
558            ).ok()?;
559            let rows = stmt
560                .query_map(
561                    params![workspace_id, channel_id, since, limit],
562                    event_from_row,
563                )
564                .ok()?;
565            Some(rows.flatten().collect::<Vec<_>>())
566        })();
567        self.inner.return_read_conn(conn);
568        result.unwrap_or_default()
569    }
570
571    /// Query recent events of a specific kind (for conflict detection).
572    pub fn recent_by_kind(
573        &self,
574        workspace_id: &str,
575        channel_id: &str,
576        kind: &str,
577        limit: usize,
578    ) -> Vec<ContextEventV1> {
579        let limit = limit.clamp(1, 100) as i64;
580        let conn = self.inner.take_read_conn();
581        let result = (|| {
582            let mut stmt = conn.prepare(
583                "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id
584                 FROM context_events
585                 WHERE workspace_id = ?1 AND channel_id = ?2 AND kind = ?3
586                 ORDER BY id DESC
587                 LIMIT ?4",
588            ).ok()?;
589            let rows = stmt
590                .query_map(
591                    params![workspace_id, channel_id, kind, limit],
592                    event_from_row,
593                )
594                .ok()?;
595            Some(rows.flatten().collect::<Vec<_>>())
596        })();
597        self.inner.return_read_conn(conn);
598        result.unwrap_or_default()
599    }
600
601    /// Full-text search over event payloads via FTS5.
602    pub fn search(
603        &self,
604        workspace_id: &str,
605        channel_id: Option<&str>,
606        query: &str,
607        limit: usize,
608    ) -> Vec<ContextEventV1> {
609        let limit = limit.clamp(1, 100) as i64;
610        let conn = self.inner.take_read_conn();
611        let result =
612            if let Some(ch) = channel_id {
613                (|| {
614                    let mut stmt = conn.prepare(
615                    "SELECT e.id, e.workspace_id, e.channel_id, e.kind, e.actor, e.timestamp,
616                            e.payload_json, e.version, e.parent_id
617                     FROM context_events e
618                     JOIN context_events_fts f ON e.id = f.rowid
619                     WHERE f.payload_text MATCH ?1 AND e.workspace_id = ?2 AND e.channel_id = ?3
620                     ORDER BY f.rank
621                     LIMIT ?4",
622                ).ok()?;
623                    let rows = stmt
624                        .query_map(params![query, workspace_id, ch, limit], event_from_row)
625                        .ok()?;
626                    Some(rows.flatten().collect::<Vec<_>>())
627                })()
628            } else {
629                (|| {
630                    let mut stmt = conn.prepare(
631                    "SELECT e.id, e.workspace_id, e.channel_id, e.kind, e.actor, e.timestamp,
632                            e.payload_json, e.version, e.parent_id
633                     FROM context_events e
634                     JOIN context_events_fts f ON e.id = f.rowid
635                     WHERE f.payload_text MATCH ?1 AND e.workspace_id = ?2
636                     ORDER BY f.rank
637                     LIMIT ?3",
638                ).ok()?;
639                    let rows = stmt
640                        .query_map(params![query, workspace_id, limit], event_from_row)
641                        .ok()?;
642                    Some(rows.flatten().collect::<Vec<_>>())
643                })()
644            };
645        self.inner.return_read_conn(conn);
646        result.unwrap_or_default()
647    }
648
649    /// Trace the causal lineage of an event by following parent_id chains.
650    /// Only returns events belonging to the given workspace (tenant isolation).
651    pub fn lineage(
652        &self,
653        event_id: i64,
654        workspace_id: &str,
655        max_depth: usize,
656    ) -> Vec<ContextEventV1> {
657        let max_depth = max_depth.clamp(1, 50);
658        let conn = self.inner.take_read_conn();
659        let mut chain = Vec::new();
660        let mut current_id = Some(event_id);
661
662        for _ in 0..max_depth {
663            let Some(id) = current_id else {
664                break;
665            };
666            let ev = conn.query_row(
667                "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id
668                 FROM context_events WHERE id = ?1 AND workspace_id = ?2",
669                params![id, workspace_id],
670                event_from_row,
671            );
672            match ev {
673                Ok(ev) => {
674                    current_id = ev.parent_id;
675                    chain.push(ev);
676                }
677                Err(_) => break,
678            }
679        }
680        self.inner.return_read_conn(conn);
681        chain
682    }
683
684    /// Returns the highest event id for a workspace/channel pair, or 0 if none.
685    pub fn latest_id(&self, workspace_id: &str, channel_id: &str) -> i64 {
686        let conn = self.inner.take_read_conn();
687        let result = conn
688            .query_row(
689                "SELECT COALESCE(MAX(id), 0) FROM context_events WHERE workspace_id = ?1 AND channel_id = ?2",
690                params![workspace_id, channel_id],
691                |row| row.get(0),
692            )
693            .unwrap_or(0);
694        self.inner.return_read_conn(conn);
695        result
696    }
697}
698
699/// A subscription wrapper that applies a [`TopicFilter`] to received events.
700pub struct FilteredSubscription {
701    pub rx: broadcast::Receiver<ContextEventV1>,
702    pub filter: TopicFilter,
703}
704
705impl FilteredSubscription {
706    /// Receive the next event that matches the filter.
707    /// Skips non-matching events silently.
708    pub async fn recv_filtered(&mut self) -> Result<ContextEventV1, broadcast::error::RecvError> {
709        loop {
710            let ev = self.rx.recv().await?;
711            if self.filter.matches(&ev) {
712                return Ok(ev);
713            }
714        }
715    }
716}
717
718fn default_db_path() -> PathBuf {
719    let data = crate::core::data_dir::lean_ctx_data_dir().unwrap_or_else(|_| PathBuf::from("."));
720    data.join("context-os").join("context-os.db")
721}
722
723#[cfg(test)]
724mod tests {
725    use super::*;
726    use tempfile::tempdir;
727
728    fn test_bus() -> (ContextBus, tempfile::TempDir) {
729        let td = tempdir().expect("tempdir");
730        let bus = ContextBus::open_at(td.path().join("test-context-os.db"));
731        (bus, td)
732    }
733
734    #[test]
735    fn append_and_read_roundtrip() {
736        let (bus, _td) = test_bus();
737        let ev = bus
738            .append(
739                "ws",
740                "ch",
741                &ContextEventKindV1::ToolCallRecorded,
742                Some("agent"),
743                serde_json::json!({"tool":"ctx_read"}),
744            )
745            .expect("append");
746        let got = bus.read("ws", "ch", ev.id - 1, 10);
747        assert!(got.iter().any(|e| e.id == ev.id));
748    }
749
750    #[test]
751    fn multi_client_concurrent_appends_have_deterministic_ordering() {
752        let (bus, _td) = test_bus();
753        let bus = Arc::new(bus);
754        let n_clients = 5;
755        let n_events_per_client = 20;
756        let ws = format!("ws-concurrent-{}", std::process::id());
757        let ch = format!("ch-concurrent-{}", std::process::id());
758
759        let mut handles = vec![];
760        for client_idx in 0..n_clients {
761            let bus = Arc::clone(&bus);
762            let ws = ws.clone();
763            let ch = ch.clone();
764            handles.push(std::thread::spawn(move || {
765                let agent = format!("agent-{client_idx}");
766                for event_idx in 0..n_events_per_client {
767                    bus.append(
768                        &ws,
769                        &ch,
770                        &ContextEventKindV1::ToolCallRecorded,
771                        Some(&agent),
772                        serde_json::json!({"client": client_idx, "seq": event_idx}),
773                    );
774                }
775            }));
776        }
777
778        for h in handles {
779            h.join().unwrap();
780        }
781
782        let all = bus.read(&ws, &ch, 0, 1000);
783        assert_eq!(
784            all.len(),
785            n_clients * n_events_per_client,
786            "all events should be persisted"
787        );
788
789        let ids: Vec<i64> = all.iter().map(|e| e.id).collect();
790        let mut sorted = ids.clone();
791        sorted.sort_unstable();
792        assert_eq!(ids, sorted, "events must be in strictly ascending ID order");
793
794        for win in ids.windows(2) {
795            assert!(
796                win[1] > win[0],
797                "IDs must be strictly monotonic (no gaps from concurrent access)"
798            );
799        }
800    }
801
802    #[test]
803    fn workspace_channel_isolation() {
804        let (bus, _td) = test_bus();
805        let pid = std::process::id();
806        let ws_a = format!("ws-iso-a-{pid}");
807        let ws_b = format!("ws-iso-b-{pid}");
808        let ws_c = format!("ws-iso-c-{pid}");
809        let ch1 = format!("ch-iso-1-{pid}");
810        let ch2 = format!("ch-iso-2-{pid}");
811
812        bus.append(
813            &ws_a,
814            &ch1,
815            &ContextEventKindV1::SessionMutated,
816            Some("agent-a"),
817            serde_json::json!({"ws":"a","ch":"1"}),
818        );
819        bus.append(
820            &ws_a,
821            &ch2,
822            &ContextEventKindV1::KnowledgeRemembered,
823            Some("agent-a"),
824            serde_json::json!({"ws":"a","ch":"2"}),
825        );
826        bus.append(
827            &ws_b,
828            &ch1,
829            &ContextEventKindV1::ArtifactStored,
830            Some("agent-b"),
831            serde_json::json!({"ws":"b","ch":"1"}),
832        );
833
834        let ws_a_ch_1 = bus.read(&ws_a, &ch1, 0, 100);
835        assert_eq!(ws_a_ch_1.len(), 1);
836        assert_eq!(ws_a_ch_1[0].kind, "session_mutated");
837
838        let ws_a_ch_2 = bus.read(&ws_a, &ch2, 0, 100);
839        assert_eq!(ws_a_ch_2.len(), 1);
840        assert_eq!(ws_a_ch_2[0].kind, "knowledge_remembered");
841
842        let ws_b_ch_1 = bus.read(&ws_b, &ch1, 0, 100);
843        assert_eq!(ws_b_ch_1.len(), 1);
844        assert_eq!(ws_b_ch_1[0].kind, "artifact_stored");
845
846        let ws_c_ch_1 = bus.read(&ws_c, &ch1, 0, 100);
847        assert!(ws_c_ch_1.is_empty(), "non-existent workspace returns empty");
848    }
849
850    #[test]
851    fn replay_from_cursor_returns_only_newer_events() {
852        let (bus, _td) = test_bus();
853        let pid = std::process::id();
854        let ws = &format!("ws-replay-{pid}");
855        let ch = &format!("ch-replay-{pid}");
856
857        let ev1 = bus
858            .append(
859                ws,
860                ch,
861                &ContextEventKindV1::ToolCallRecorded,
862                None,
863                serde_json::json!({"seq":1}),
864            )
865            .unwrap();
866        let ev2 = bus
867            .append(
868                ws,
869                ch,
870                &ContextEventKindV1::SessionMutated,
871                None,
872                serde_json::json!({"seq":2}),
873            )
874            .unwrap();
875        let _ev3 = bus
876            .append(
877                ws,
878                ch,
879                &ContextEventKindV1::GraphBuilt,
880                None,
881                serde_json::json!({"seq":3}),
882            )
883            .unwrap();
884
885        let from_cursor = bus.read(ws, ch, ev2.id, 100);
886        assert_eq!(from_cursor.len(), 1, "only events after cursor");
887        assert_eq!(from_cursor[0].kind, "graph_built");
888
889        let from_first = bus.read(ws, ch, ev1.id, 100);
890        assert_eq!(from_first.len(), 2, "events after first");
891
892        let from_zero = bus.read(ws, ch, 0, 100);
893        assert_eq!(from_zero.len(), 3, "all events from zero");
894    }
895
896    #[test]
897    fn broadcast_subscriber_receives_events() {
898        let (bus, _td) = test_bus();
899        let mut rx = bus.subscribe("ws", "ch").expect("subscribe should succeed");
900
901        let ev = bus
902            .append(
903                "ws",
904                "ch",
905                &ContextEventKindV1::ProofAdded,
906                Some("verifier"),
907                serde_json::json!({"proof":"hash"}),
908            )
909            .unwrap();
910
911        let received = rx.try_recv().expect("subscriber should receive event");
912        assert_eq!(received.id, ev.id);
913        assert_eq!(received.kind, "proof_added");
914        assert_eq!(received.actor.as_deref(), Some("verifier"));
915    }
916}
lean_ctx/core/context_os/context_bus.rs

lean_ctx/core/context_os/
context_bus.rs
