lean_ctx/core/context_os/

context_bus.rs

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use chrono::{DateTime, Utc};
use rusqlite::{Connection, params};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::broadcast;

use super::observation::{
    LocalObservationPayload, LocalObservationSource, LocalObservationV1,
    ObservationPersistenceError, decode_row, lock_observation, validate_identity,
};
use crate::core::events::LeanCtxEvent;
use crate::core::ocla_bus::{OclaBusRecord, OclaEvent};

const MAX_READ_CONNS: usize = 4;

fn encode_target_agents(target_agents: Option<&[String]>) -> Option<String> {
    target_agents.map(|targets| serde_json::to_string(targets).unwrap_or_else(|_| "[]".to_string()))
}

fn decode_target_agents(raw: Option<String>) -> Option<Vec<String>> {
    match raw {
        None => None,
        Some(value) => match serde_json::from_str::<Vec<String>>(&value) {
            Ok(targets) => Some(targets),
            Err(error) => {
                // A malformed directed-list must never turn into a broadcast.
                tracing::warn!("invalid directed-agent list in context event: {error}");
                Some(Vec::new())
            }
        },
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ContextEventKindV1 {
    ToolCallRecorded,
    SessionMutated,
    KnowledgeRemembered,
    ArtifactStored,
    GraphBuilt,
    ProofAdded,
    ProvenanceCheckpoint,
    ProvenanceEdit,
    HealthChanged,
}

impl ContextEventKindV1 {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::ToolCallRecorded => "tool_call_recorded",
            Self::SessionMutated => "session_mutated",
            Self::KnowledgeRemembered => "knowledge_remembered",
            Self::ArtifactStored => "artifact_stored",
            Self::GraphBuilt => "graph_built",
            Self::ProofAdded => "proof_added",
            Self::ProvenanceCheckpoint => "provenance_checkpoint",
            Self::ProvenanceEdit => "provenance_edit",
            Self::HealthChanged => "health_changed",
        }
    }

    pub fn parse(s: &str) -> Self {
        match s.trim().to_lowercase().as_str() {
            "tool_call_recorded" => Self::ToolCallRecorded,
            "session_mutated" => Self::SessionMutated,
            "knowledge_remembered" => Self::KnowledgeRemembered,
            "artifact_stored" => Self::ArtifactStored,
            "graph_built" => Self::GraphBuilt,
            "proof_added" => Self::ProofAdded,
            "provenance_checkpoint" => Self::ProvenanceCheckpoint,
            "provenance_edit" => Self::ProvenanceEdit,
            "health_changed" => Self::HealthChanged,
            other => {
                tracing::warn!(
                    "unknown ContextEventKind '{other}', defaulting to ToolCallRecorded"
                );
                Self::ToolCallRecorded
            }
        }
    }

    /// Classifies the consistency requirement for this event kind.
    ///
    /// - `Local`: Agent-local, never shared (tool reads, cache hits).
    /// - `Eventual`: Broadcast via bus, other agents see it "soon" (knowledge, artifacts).
    /// - `Strong`: Critical decisions that require acknowledgment before proceeding.
    pub fn consistency_level(&self) -> ConsistencyLevel {
        match self {
            Self::ToolCallRecorded | Self::GraphBuilt => ConsistencyLevel::Local,
            Self::KnowledgeRemembered
            | Self::ArtifactStored
            | Self::ProvenanceEdit
            | Self::HealthChanged => ConsistencyLevel::Eventual,
            Self::SessionMutated | Self::ProofAdded | Self::ProvenanceCheckpoint => {
                ConsistencyLevel::Strong
            }
        }
    }
}

/// Consistency requirement for shared context events.
/// Ordered from least to most strict for filtering comparisons.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConsistencyLevel {
    /// Agent-local, authoritative: session task, local cache, current file set.
    Local = 0,
    /// Shared, eventually consistent: knowledge facts, gotchas, artifact refs.
    Eventual = 1,
    /// Shared, strongly consistent: workspace config, critical decisions.
    Strong = 2,
}

impl ConsistencyLevel {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Local => "local",
            Self::Eventual => "eventual",
            Self::Strong => "strong",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContextEventV1 {
    pub id: i64,
    pub workspace_id: String,
    pub channel_id: String,
    pub kind: String,
    pub actor: Option<String>,
    pub timestamp: DateTime<Utc>,
    pub version: i64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent_id: Option<i64>,
    pub consistency_level: String,
    pub payload: Value,
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pub target_agents: Option<Vec<String>>,
}

impl ContextEventV1 {
    pub fn consistency(&self) -> ConsistencyLevel {
        ContextEventKindV1::parse(&self.kind).consistency_level()
    }

    pub fn is_visible_to_agent(&self, agent_id: &str) -> bool {
        match &self.target_agents {
            None => true,
            Some(targets) => targets.iter().any(|t| t == agent_id),
        }
    }
}

/// Filter for selective event subscriptions.
/// All fields are optional; `None` means "accept all". Directed events are
/// included by default; callers without an agent identity can opt into the
/// broadcast-only view with [`TopicFilter::broadcast_only`].
#[derive(Debug, Clone)]
pub struct TopicFilter {
    pub kinds: Option<Vec<ContextEventKindV1>>,
    pub actors: Option<Vec<String>>,
    pub min_consistency: Option<ConsistencyLevel>,
    pub agent_id: Option<String>,
    pub include_directed: bool,
}

impl Default for TopicFilter {
    fn default() -> Self {
        Self {
            kinds: None,
            actors: None,
            min_consistency: None,
            agent_id: None,
            include_directed: true,
        }
    }
}

impl TopicFilter {
    /// Convenience constructor: filter by event kind strings.
    pub fn kinds(kind_strs: &[&str]) -> Self {
        Self {
            kinds: Some(
                kind_strs
                    .iter()
                    .map(|s| ContextEventKindV1::parse(s))
                    .collect(),
            ),
            ..Self::default()
        }
    }

    /// Filter that accepts broadcast events while excluding agent-directed
    /// events for callers that have no authenticated agent identity.
    pub fn broadcast_only() -> Self {
        Self {
            include_directed: false,
            ..Self::default()
        }
    }

    pub fn matches(&self, event: &ContextEventV1) -> bool {
        if !self.include_directed && event.target_agents.is_some() {
            return false;
        }
        if let Some(ref kinds) = self.kinds {
            let parsed = ContextEventKindV1::parse(&event.kind);
            if !kinds.contains(&parsed) {
                return false;
            }
        }
        if let Some(ref actors) = self.actors {
            match &event.actor {
                Some(actor) if actors.iter().any(|a| a == actor) => {}
                Some(_) | None => return false,
            }
        }
        if let Some(min) = self.min_consistency
            && event.consistency() < min
        {
            return false;
        }
        if let Some(ref aid) = self.agent_id
            && !event.is_visible_to_agent(aid)
        {
            return false;
        }
        true
    }
}

fn event_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ContextEventV1> {
    let ts_str: String = row.get(5)?;
    let ts = DateTime::parse_from_rfc3339(&ts_str)
        .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc));
    let payload_str: String = row.get(6)?;
    let payload: Value = serde_json::from_str(&payload_str).unwrap_or(Value::Null);
    let kind_str: String = row.get(3)?;
    let target_agents_json: Option<String> = row.get(9)?;
    let cl = ContextEventKindV1::parse(&kind_str)
        .consistency_level()
        .as_str()
        .to_string();
    Ok(ContextEventV1 {
        id: row.get(0)?,
        workspace_id: row.get(1)?,
        channel_id: row.get(2)?,
        kind: kind_str,
        actor: row.get::<_, Option<String>>(4)?,
        timestamp: ts,
        version: row.get::<_, i64>(7).unwrap_or(0),
        parent_id: row.get::<_, Option<i64>>(8).ok().flatten(),
        consistency_level: cl,
        payload,
        target_agents: decode_target_agents(target_agents_json),
    })
}

#[derive(Clone)]
pub struct ContextBus {
    inner: Arc<Inner>,
}

const STREAM_CHANNEL_SIZE: usize = 256;
const MAX_SUBSCRIBERS_PER_CHANNEL: usize = 64;
/// Bound the write-side version cache. Each entry is re-derivable from the DB via
/// `MAX(version)`, so when the map exceeds this (e.g. a client cycling workspace/channel
/// ids) it is simply cleared — costing at most one extra `MAX()` query per active channel.
const MAX_VERSION_CACHE_ENTRIES: usize = 4096;

struct Inner {
    write_conn: Mutex<Connection>,
    read_pool: Mutex<Vec<Connection>>,
    streams: Mutex<HashMap<String, broadcast::Sender<ContextEventV1>>>,
    version_cache: Mutex<HashMap<String, i64>>,
    db_path: PathBuf,
}

impl Inner {
    fn open_read_conn(path: &PathBuf) -> Connection {
        Self::try_open_read_conn(path).expect("open read context-os db")
    }

    fn try_open_read_conn(path: &PathBuf) -> Result<Connection, ObservationPersistenceError> {
        let conn = Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?;
        conn.busy_timeout(std::time::Duration::from_secs(5))?;
        conn.execute_batch("PRAGMA query_only=ON;")?;
        Ok(conn)
    }

    fn try_take_read_conn(&self) -> Result<Connection, ObservationPersistenceError> {
        let cached = lock_observation(&self.read_pool)?.pop();
        match cached {
            Some(conn) => Ok(conn),
            None => Self::try_open_read_conn(&self.db_path),
        }
    }

    fn take_read_conn(&self) -> Connection {
        self.read_pool
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .pop()
            .unwrap_or_else(|| Self::open_read_conn(&self.db_path))
    }

    fn return_read_conn(&self, conn: Connection) {
        let mut pool = self
            .read_pool
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if pool.len() < MAX_READ_CONNS {
            pool.push(conn);
        }
    }

    fn stream_key(workspace_id: &str, channel_id: &str) -> String {
        format!("{workspace_id}\0{channel_id}")
    }

    fn next_version(&self, workspace_id: &str, channel_id: &str) -> i64 {
        let key = Self::stream_key(workspace_id, channel_id);

        {
            let mut cache = self
                .version_cache
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if let Some(v) = cache.get_mut(&key) {
                *v += 1;
                return *v;
            }
        }

        let conn = self.take_read_conn();
        let v: i64 = conn
            .query_row(
                "SELECT COALESCE(MAX(version), 0) FROM context_events WHERE workspace_id = ?1 AND channel_id = ?2",
                params![workspace_id, channel_id],
                |row| row.get(0),
            )
            .unwrap_or(0);
        self.return_read_conn(conn);

        let mut cache = self
            .version_cache
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if cache.len() > MAX_VERSION_CACHE_ENTRIES {
            // Re-derivable from the DB; drop the whole cache rather than grow unbounded.
            cache.clear();
        }
        let entry = cache.entry(key).or_insert(v);
        *entry = (*entry).max(v) + 1;
        *entry
    }
}

impl Default for ContextBus {
    fn default() -> Self {
        Self::new()
    }
}

impl ContextBus {
    pub fn new() -> Self {
        let path = default_db_path();
        Self::open_at(path)
    }

    /// Open a ContextBus backed by the given SQLite path. Useful for tests
    /// that need an isolated database to avoid contention on Windows CI.
    pub fn open_at(path: PathBuf) -> Self {
        Self::try_open_at(path).expect("open context-os db")
    }

    /// Open without panicking or silently substituting a different data directory.
    pub fn try_open_at(path: PathBuf) -> Result<Self, ObservationPersistenceError> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let conn = Connection::open(&path)?;
        conn.busy_timeout(std::time::Duration::from_secs(5))?;
        conn.execute_batch(
            "PRAGMA journal_mode=WAL;
             CREATE TABLE IF NOT EXISTS context_events (
               id INTEGER PRIMARY KEY AUTOINCREMENT,
               workspace_id TEXT NOT NULL,
               channel_id TEXT NOT NULL,
               kind TEXT NOT NULL,
               actor TEXT,
               timestamp TEXT NOT NULL,
               payload_json TEXT NOT NULL,
               version INTEGER NOT NULL DEFAULT 0,
               parent_id INTEGER,
               delivery_id TEXT,
               target_agents_json TEXT
             );
             CREATE INDEX IF NOT EXISTS idx_context_events_stream
               ON context_events(workspace_id, channel_id, id);
             CREATE TABLE IF NOT EXISTS context_local_observations (
               id INTEGER PRIMARY KEY AUTOINCREMENT,
               schema_version INTEGER NOT NULL CHECK(schema_version = 1),
               source_kind TEXT NOT NULL,
               producer_instance TEXT NOT NULL,
               source_id TEXT NOT NULL,
               source_timestamp TEXT NOT NULL,
               payload_json TEXT NOT NULL,
               UNIQUE(source_kind, producer_instance, source_id)
             );
             CREATE INDEX IF NOT EXISTS idx_context_local_observations_source
               ON context_local_observations(source_kind, id);",
        )?;

        let _ = conn.execute_batch(
            "ALTER TABLE context_events ADD COLUMN version INTEGER NOT NULL DEFAULT 0;",
        );
        let _ = conn.execute_batch("ALTER TABLE context_events ADD COLUMN parent_id INTEGER;");
        let _ = conn.execute_batch("ALTER TABLE context_events ADD COLUMN delivery_id TEXT;");
        let _ =
            conn.execute_batch("ALTER TABLE context_events ADD COLUMN target_agents_json TEXT;");
        conn.execute_batch(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_context_events_delivery
               ON context_events(workspace_id, channel_id, delivery_id)
               WHERE delivery_id IS NOT NULL;",
        )?;

        let _ = conn.execute_batch(
            "CREATE VIRTUAL TABLE IF NOT EXISTS context_events_fts USING fts5(
               payload_text,
               content=context_events,
               content_rowid=id
             );",
        );

        let mut read_conns = Vec::with_capacity(MAX_READ_CONNS);
        for _ in 0..MAX_READ_CONNS {
            read_conns.push(Inner::try_open_read_conn(&path)?);
        }

        Ok(Self {
            inner: Arc::new(Inner {
                write_conn: Mutex::new(conn),
                read_pool: Mutex::new(read_conns),
                streams: Mutex::new(HashMap::new()),
                version_cache: Mutex::new(HashMap::new()),
                db_path: path,
            }),
        })
    }

    #[cfg(test)]
    pub(crate) fn db_path(&self) -> &std::path::Path {
        &self.inner.db_path
    }

    pub fn subscribe(
        &self,
        workspace_id: &str,
        channel_id: &str,
    ) -> Option<broadcast::Receiver<ContextEventV1>> {
        let key = Inner::stream_key(workspace_id, channel_id);
        let mut streams = self
            .inner
            .streams
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Reap senders left behind by departed clients (receiver_count() == 0). A removed
        // key is transparently recreated below; the DB is the durability source so this
        // loses no deliverable events. Bounds `streams` to ~live connection count even
        // under client-cycled workspace/channel ids.
        streams.retain(|_, tx| tx.receiver_count() > 0);
        let tx = streams
            .entry(key)
            .or_insert_with(|| broadcast::channel(STREAM_CHANNEL_SIZE).0);
        if tx.receiver_count() >= MAX_SUBSCRIBERS_PER_CHANNEL {
            tracing::warn!(
                "SSE subscriber cap ({MAX_SUBSCRIBERS_PER_CHANNEL}) reached for {workspace_id}/{channel_id} — rejecting"
            );
            return None;
        }
        Some(tx.subscribe())
    }

    /// Subscribe with a filter — only events matching the filter are delivered.
    /// Returns `(Receiver, TopicFilter)` for use in filtered receive loops.
    pub fn subscribe_filtered(
        &self,
        workspace_id: &str,
        channel_id: &str,
        filter: TopicFilter,
    ) -> Option<FilteredSubscription> {
        let rx = self.subscribe(workspace_id, channel_id)?;
        Some(FilteredSubscription { rx, filter })
    }

    pub fn append(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &ContextEventKindV1,
        actor: Option<&str>,
        payload: Value,
    ) -> Option<ContextEventV1> {
        self.append_with_parent(workspace_id, channel_id, kind, actor, payload, None)
    }

    /// Persist an OCLA record in the ContextBus-owned private observation table.
    /// The table is deliberately outside `context_events`, so normal replay,
    /// search, lineage, and subscriptions cannot expose local observations.
    pub fn append_ocla_observation(
        &self,
        producer_instance: &str,
        record: &OclaBusRecord,
    ) -> Result<i64, ObservationPersistenceError> {
        let payload_json = serde_json::to_string(&record.event)?;
        // JSON maps non-finite numbers to null; require a readable typed payload.
        let _: OclaEvent = serde_json::from_str(&payload_json)?;
        super::observation::admit_payload(&payload_json)?;
        self.insert_local_observation(
            LocalObservationSource::Ocla,
            producer_instance,
            record.id,
            &record.timestamp_ms.to_string(),
            &payload_json,
        )
    }

    /// Persist a legacy event as a compatibility observation without appending
    /// to the immutable historical JSONL journal.
    pub fn append_legacy_observation(
        &self,
        producer_instance: &str,
        event: &LeanCtxEvent,
    ) -> Result<i64, ObservationPersistenceError> {
        let payload_json = serde_json::to_string(event)?;
        let _: LeanCtxEvent = serde_json::from_str(&payload_json)?;
        super::observation::admit_payload(&payload_json)?;
        self.insert_local_observation(
            LocalObservationSource::Legacy,
            producer_instance,
            event.id,
            &event.timestamp,
            &payload_json,
        )
    }

    fn insert_local_observation(
        &self,
        source: LocalObservationSource,
        producer_instance: &str,
        source_id: u64,
        source_timestamp: &str,
        payload_json: &str,
    ) -> Result<i64, ObservationPersistenceError> {
        use rusqlite::OptionalExtension;
        validate_identity(producer_instance, source_id)?;
        let mut conn = lock_observation(&self.inner.write_conn)?;
        let transaction =
            conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let existing: Option<(i64, String, String)> = transaction
            .query_row(
                "SELECT id, source_timestamp, payload_json FROM context_local_observations
             WHERE source_kind = ?1 AND producer_instance = ?2 AND source_id = ?3",
                params![source.as_str(), producer_instance, source_id.to_string()],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .optional()?;
        if let Some((cursor, timestamp, payload)) = existing {
            if timestamp != source_timestamp || payload != payload_json {
                return Err(ObservationPersistenceError::IdentityConflict);
            }
            transaction.commit()?;
            return Ok(cursor);
        }
        transaction.execute(
            "INSERT INTO context_local_observations
             (schema_version, source_kind, producer_instance, source_id, source_timestamp, payload_json)
             VALUES (1, ?1, ?2, ?3, ?4, ?5)",
            params![
                source.as_str(),
                producer_instance,
                source_id.to_string(),
                source_timestamp,
                payload_json
            ],
        )?;
        let cursor_id = transaction.last_insert_rowid();
        transaction.commit()?;
        Ok(cursor_id)
    }

    /// Read private observations for regression and compatibility readers.
    pub fn read_local_observations(
        &self,
        limit: usize,
    ) -> Result<Vec<LocalObservationV1>, ObservationPersistenceError> {
        let limit = limit.clamp(1, 10_000) as i64;
        let conn = self.inner.try_take_read_conn()?;
        let result = (|| {
            let mut stmt = conn.prepare(
                "SELECT id, source_kind, source_id, source_timestamp, payload_json,
                        producer_instance, schema_version
                 FROM context_local_observations ORDER BY id ASC LIMIT ?1",
            )?;
            let mut rows = stmt.query(params![limit])?;
            let mut observations = Vec::new();
            while let Some(row) = rows.next()? {
                if let Some(observation) = decode_row(row)? {
                    observations.push(observation);
                }
            }
            Ok(observations)
        })();
        self.return_observation_read_conn(conn)?;
        result
    }

    pub fn read_legacy_observations(
        &self,
        limit: usize,
    ) -> Result<Vec<LeanCtxEvent>, ObservationPersistenceError> {
        let limit = limit.clamp(1, 10_000) as i64;
        let conn = self.inner.try_take_read_conn()?;
        let result = (|| {
            let mut stmt = conn.prepare(
                "SELECT id, source_kind, source_id, source_timestamp, payload_json,
                            producer_instance, schema_version FROM context_local_observations
                 WHERE source_kind = 'legacy' ORDER BY id DESC LIMIT ?1",
            )?;
            let mut rows = stmt.query(params![limit])?;
            let mut events = Vec::new();
            while let Some(row) = rows.next()? {
                let Some(observation) = decode_row(row)? else {
                    continue;
                };
                let LocalObservationPayload::Legacy(event) = observation.payload else {
                    return Err(ObservationPersistenceError::InvalidRow(
                        "non-legacy observation in legacy projection".into(),
                    ));
                };
                events.push(event);
            }
            events.reverse();
            Ok(events)
        })();
        self.return_observation_read_conn(conn)?;
        result
    }

    fn return_observation_read_conn(
        &self,
        conn: Connection,
    ) -> Result<(), ObservationPersistenceError> {
        let mut pool = lock_observation(&self.inner.read_pool)?;
        if pool.len() < MAX_READ_CONNS {
            pool.push(conn);
        }
        Ok(())
    }

    pub fn append_with_parent(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &ContextEventKindV1,
        actor: Option<&str>,
        payload: Value,
        parent_id: Option<i64>,
    ) -> Option<ContextEventV1> {
        let ev = self.insert_event(
            workspace_id,
            channel_id,
            kind,
            actor,
            payload,
            parent_id,
            None,
        )?;
        self.broadcast_event(&ev);
        Some(ev)
    }

    /// Append an event directed at specific agents only.
    /// Only subscribers whose `TopicFilter.agent_id` matches a target will see it.
    pub fn append_directed(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &ContextEventKindV1,
        actor: Option<&str>,
        payload: Value,
        target_agents: Vec<String>,
    ) -> Option<ContextEventV1> {
        let ev = self.insert_event(
            workspace_id,
            channel_id,
            kind,
            actor,
            payload,
            None,
            Some(target_agents),
        )?;
        self.broadcast_event(&ev);
        Some(ev)
    }

    fn insert_event(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &ContextEventKindV1,
        actor: Option<&str>,
        payload: Value,
        parent_id: Option<i64>,
        target_agents: Option<Vec<String>>,
    ) -> Option<ContextEventV1> {
        let ts = Utc::now();
        let payload_json = payload.to_string();
        let target_agents_json = encode_target_agents(target_agents.as_deref());

        let (id, version) = {
            let Ok(conn) = self.inner.write_conn.lock() else {
                return None;
            };
            let version = self.inner.next_version(workspace_id, channel_id);

            let result: Result<(i64, i64), rusqlite::Error> = conn
                .execute_batch("BEGIN IMMEDIATE")
                .and_then(|()| {
                    conn.execute(
                        "INSERT INTO context_events (workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id, target_agents_json)
                         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                        params![
                            workspace_id,
                            channel_id,
                            kind.as_str(),
                            actor.map(str::to_string),
                            ts.to_rfc3339(),
                            payload_json,
                            version,
                            parent_id,
                            target_agents_json,
                        ],
                    )?;
                    let rowid = conn.last_insert_rowid();
                    if let Err(e) = conn.execute(
                        "INSERT INTO context_events_fts(rowid, payload_text) VALUES (?1, ?2)",
                        params![rowid, payload_json],
                    ) {
                        tracing::warn!("FTS insert failed for event {rowid}: {e}");
                    }
                    conn.execute_batch("COMMIT")?;
                    Ok((rowid, version))
                });

            match result {
                Ok(pair) => pair,
                Err(e) => {
                    tracing::warn!("context bus append failed: {e}");
                    let _ = conn.execute_batch("ROLLBACK");
                    return None;
                }
            }
        };

        Some(ContextEventV1 {
            id,
            workspace_id: workspace_id.to_string(),
            channel_id: channel_id.to_string(),
            consistency_level: kind.consistency_level().as_str().to_string(),
            kind: kind.as_str().to_string(),
            actor: actor.map(str::to_string),
            timestamp: ts,
            version,
            parent_id,
            payload,
            target_agents,
        })
    }

    fn broadcast_event(&self, ev: &ContextEventV1) {
        let key = Inner::stream_key(&ev.workspace_id, &ev.channel_id);
        let tx = self
            .inner
            .streams
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .get(&key)
            .cloned();
        if let Some(tx) = tx {
            let _ = tx.send(ev.clone());
        }
    }

    pub fn read(
        &self,
        workspace_id: &str,
        channel_id: &str,
        since: i64,
        limit: usize,
    ) -> Vec<ContextEventV1> {
        let limit = limit.clamp(1, 1000) as i64;
        let conn = self.inner.take_read_conn();
        let result = (|| {
            let mut stmt = conn.prepare(
                "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id, target_agents_json
                 FROM context_events
                  WHERE workspace_id = ?1 AND channel_id = ?2 AND id > ?3
                 ORDER BY id ASC
                 LIMIT ?4",
            ).ok()?;
            let rows = stmt
                .query_map(
                    params![workspace_id, channel_id, since, limit],
                    event_from_row,
                )
                .ok()?;
            Some(rows.flatten().collect::<Vec<_>>())
        })();
        self.inner.return_read_conn(conn);
        result.unwrap_or_default()
    }

    /// Atomically append at most one event for an authenticated delivery ID.
    ///
    /// `target_agents` scopes the event like [`Self::append_directed`]: when
    /// set, only subscribers and readers identifying as one of the targets see
    /// it, live and on replay.
    pub fn append_delivery_once(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &ContextEventKindV1,
        actor: Option<&str>,
        payload: Value,
        delivery_id: &str,
        target_agents: Option<Vec<String>>,
    ) -> Result<bool, String> {
        let ts = Utc::now();
        let payload_json = payload.to_string();
        let target_agents_json = encode_target_agents(target_agents.as_deref());
        let (id, version) = {
            let conn = self
                .inner
                .write_conn
                .lock()
                .map_err(|_| "context bus write lock poisoned".to_string())?;
            conn.execute_batch("BEGIN IMMEDIATE")
                .map_err(|error| error.to_string())?;
            let version = self.inner.next_version(workspace_id, channel_id);
            let inserted = conn
                .execute(
                    "INSERT OR IGNORE INTO context_events (workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id, delivery_id, target_agents_json)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8, ?9)",
                    params![workspace_id, channel_id, kind.as_str(), actor, ts.to_rfc3339(), payload_json, version, delivery_id, target_agents_json],
                )
                .map_err(|error| error.to_string());
            let inserted = match inserted {
                Ok(value) => value,
                Err(error) => {
                    let _ = conn.execute_batch("ROLLBACK");
                    return Err(error);
                }
            };
            if inserted == 0 {
                conn.execute_batch("COMMIT")
                    .map_err(|error| error.to_string())?;
                return Ok(false);
            }
            let rowid = conn.last_insert_rowid();
            if let Err(error) = conn.execute(
                "INSERT INTO context_events_fts(rowid, payload_text) VALUES (?1, ?2)",
                params![rowid, payload_json],
            ) {
                tracing::warn!("FTS insert failed for event {rowid}: {error}");
            }
            conn.execute_batch("COMMIT")
                .map_err(|error| error.to_string())?;
            (rowid, version)
        };
        self.broadcast_event(&ContextEventV1 {
            id,
            workspace_id: workspace_id.to_string(),
            channel_id: channel_id.to_string(),
            consistency_level: kind.consistency_level().as_str().to_string(),
            kind: kind.as_str().to_string(),
            actor: actor.map(str::to_string),
            timestamp: ts,
            version,
            parent_id: None,
            payload,
            target_agents,
        });
        Ok(true)
    }

    /// Query recent events of a specific kind (for conflict detection).
    pub fn recent_by_kind(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &str,
        limit: usize,
    ) -> Vec<ContextEventV1> {
        let limit = limit.clamp(1, 100) as i64;
        let conn = self.inner.take_read_conn();
        let result = (|| {
            let mut stmt = conn.prepare(
                 "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id, target_agents_json
                  FROM context_events
                 WHERE workspace_id = ?1 AND channel_id = ?2 AND kind = ?3
                 ORDER BY id DESC
                 LIMIT ?4",
            ).ok()?;
            let rows = stmt
                .query_map(
                    params![workspace_id, channel_id, kind, limit],
                    event_from_row,
                )
                .ok()?;
            Some(rows.flatten().collect::<Vec<_>>())
        })();
        self.inner.return_read_conn(conn);
        result.unwrap_or_default()
    }

    /// Full-text search over event payloads via FTS5.
    pub fn search(
        &self,
        workspace_id: &str,
        channel_id: Option<&str>,
        query: &str,
        limit: usize,
    ) -> Vec<ContextEventV1> {
        let limit = limit.clamp(1, 100) as i64;
        let conn = self.inner.take_read_conn();
        let result =
            if let Some(ch) = channel_id {
                (|| {
                    let mut stmt = conn.prepare(
                    "SELECT e.id, e.workspace_id, e.channel_id, e.kind, e.actor, e.timestamp,
                            e.payload_json, e.version, e.parent_id, e.target_agents_json
                     FROM context_events e
                     JOIN context_events_fts f ON e.id = f.rowid
                     WHERE f.payload_text MATCH ?1 AND e.workspace_id = ?2 AND e.channel_id = ?3
                     ORDER BY f.rank
                     LIMIT ?4",
                ).ok()?;
                    let rows = stmt
                        .query_map(params![query, workspace_id, ch, limit], event_from_row)
                        .ok()?;
                    Some(rows.flatten().collect::<Vec<_>>())
                })()
            } else {
                (|| {
                    let mut stmt = conn.prepare(
                    "SELECT e.id, e.workspace_id, e.channel_id, e.kind, e.actor, e.timestamp,
                            e.payload_json, e.version, e.parent_id, e.target_agents_json
                     FROM context_events e
                     JOIN context_events_fts f ON e.id = f.rowid
                     WHERE f.payload_text MATCH ?1 AND e.workspace_id = ?2
                     ORDER BY f.rank
                     LIMIT ?3",
                ).ok()?;
                    let rows = stmt
                        .query_map(params![query, workspace_id, limit], event_from_row)
                        .ok()?;
                    Some(rows.flatten().collect::<Vec<_>>())
                })()
            };
        self.inner.return_read_conn(conn);
        result.unwrap_or_default()
    }

    /// Trace the causal lineage of an event by following parent_id chains.
    /// Only returns events belonging to the given workspace (tenant isolation).
    pub fn lineage(
        &self,
        event_id: i64,
        workspace_id: &str,
        max_depth: usize,
    ) -> Vec<ContextEventV1> {
        let max_depth = max_depth.clamp(1, 50);
        let conn = self.inner.take_read_conn();
        let mut chain = Vec::new();
        let mut current_id = Some(event_id);

        for _ in 0..max_depth {
            let Some(id) = current_id else {
                break;
            };
            let ev = conn.query_row(
                "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json, version, parent_id, target_agents_json
                 FROM context_events WHERE id = ?1 AND workspace_id = ?2",
                params![id, workspace_id],
                event_from_row,
            );
            match ev {
                Ok(ev) => {
                    current_id = ev.parent_id;
                    chain.push(ev);
                }
                Err(_) => break,
            }
        }
        self.inner.return_read_conn(conn);
        chain
    }

    /// Returns the highest event id for a workspace/channel pair, or 0 if none.
    pub fn latest_id(&self, workspace_id: &str, channel_id: &str) -> i64 {
        let conn = self.inner.take_read_conn();
        let result = conn
            .query_row(
                "SELECT COALESCE(MAX(id), 0) FROM context_events WHERE workspace_id = ?1 AND channel_id = ?2",
                params![workspace_id, channel_id],
                |row| row.get(0),
            )
            .unwrap_or(0);
        self.inner.return_read_conn(conn);
        result
    }
}

/// A subscription wrapper that applies a [`TopicFilter`] to received events.
pub struct FilteredSubscription {
    pub rx: broadcast::Receiver<ContextEventV1>,
    pub filter: TopicFilter,
}

impl FilteredSubscription {
    /// Receive the next event that matches the filter.
    /// Skips non-matching events silently.
    pub async fn recv_filtered(&mut self) -> Result<ContextEventV1, broadcast::error::RecvError> {
        loop {
            let ev = self.rx.recv().await?;
            if self.filter.matches(&ev) {
                return Ok(ev);
            }
        }
    }
}

fn default_db_path() -> PathBuf {
    let data = crate::core::data_dir::lean_ctx_data_dir().unwrap_or_else(|_| PathBuf::from("."));
    data.join("context-os").join("context-os.db")
}

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

    fn test_bus() -> (ContextBus, tempfile::TempDir) {
        let td = tempdir().expect("tempdir");
        let bus = ContextBus::open_at(td.path().join("test-context-os.db"));
        (bus, td)
    }

    #[test]
    fn append_and_read_roundtrip() {
        let (bus, _td) = test_bus();
        let ev = bus
            .append(
                "ws",
                "ch",
                &ContextEventKindV1::ToolCallRecorded,
                Some("agent"),
                serde_json::json!({"tool":"ctx_read"}),
            )
            .expect("append");
        let got = bus.read("ws", "ch", ev.id - 1, 10);
        assert!(got.iter().any(|e| e.id == ev.id));
    }

    #[test]
    fn authenticated_delivery_is_unique_across_bus_instances() {
        let td = tempdir().expect("tempdir");
        let path = td.path().join("delivery-context-os.db");
        let buses = [ContextBus::open_at(path.clone()), ContextBus::open_at(path)];
        let barrier = Arc::new(std::sync::Barrier::new(2));
        let handles: Vec<_> = buses
            .into_iter()
            .map(|bus| {
                let barrier = Arc::clone(&barrier);
                std::thread::spawn(move || {
                    barrier.wait();
                    bus.append_delivery_once(
                        "ws",
                        "a2a",
                        &ContextEventKindV1::SessionMutated,
                        Some("agent"),
                        serde_json::json!({"delivery_id":"delivery-1"}),
                        "delivery-1",
                        None,
                    )
                    .expect("delivery append")
                })
            })
            .collect();
        let inserted: Vec<bool> = handles
            .into_iter()
            .map(|handle| handle.join().expect("thread"))
            .collect();
        assert_eq!(inserted.iter().filter(|value| **value).count(), 1);

        let bus = ContextBus::open_at(td.path().join("delivery-context-os.db"));
        assert_eq!(bus.read("ws", "a2a", 0, 10).len(), 1);
    }

    #[test]
    fn directed_delivery_reaches_only_its_target_live_and_on_replay() {
        let td = tempdir().expect("tempdir");
        let path = td.path().join("directed-delivery-context-os.db");
        let bus = ContextBus::open_at(path.clone());
        let target_filter = || TopicFilter {
            agent_id: Some("agent-a".to_string()),
            ..TopicFilter::default()
        };
        let other_filter = || TopicFilter {
            agent_id: Some("agent-b".to_string()),
            ..TopicFilter::default()
        };
        let mut target_rx = bus.subscribe("ws", "a2a").expect("subscribe");
        let targets = vec!["agent-a".to_string()];
        assert!(
            bus.append_delivery_once(
                "ws",
                "a2a",
                &ContextEventKindV1::SessionMutated,
                Some("sender"),
                serde_json::json!({"delivery_id":"directed-1"}),
                "directed-1",
                Some(targets.clone()),
            )
            .expect("delivery append")
        );
        let live = target_rx.try_recv().expect("live delivery event");
        assert_eq!(live.target_agents, Some(targets.clone()));
        assert!(target_filter().matches(&live));
        assert!(!other_filter().matches(&live));
        assert!(!TopicFilter::broadcast_only().matches(&live));
        drop(bus);

        let reopened = ContextBus::open_at(path);
        let replayed = reopened.read("ws", "a2a", 0, 10);
        assert_eq!(replayed.len(), 1);
        assert_eq!(replayed[0].target_agents, Some(targets));
        assert!(target_filter().matches(&replayed[0]));
        assert!(!other_filter().matches(&replayed[0]));
        assert!(!TopicFilter::broadcast_only().matches(&replayed[0]));
    }

    #[test]
    fn multi_client_concurrent_appends_have_deterministic_ordering() {
        let (bus, _td) = test_bus();
        let bus = Arc::new(bus);
        let n_clients = 5;
        let n_events_per_client = 20;
        let ws = format!("ws-concurrent-{}", std::process::id());
        let ch = format!("ch-concurrent-{}", std::process::id());

        let mut handles = vec![];
        for client_idx in 0..n_clients {
            let bus = Arc::clone(&bus);
            let ws = ws.clone();
            let ch = ch.clone();
            handles.push(std::thread::spawn(move || {
                let agent = format!("agent-{client_idx}");
                for event_idx in 0..n_events_per_client {
                    bus.append(
                        &ws,
                        &ch,
                        &ContextEventKindV1::ToolCallRecorded,
                        Some(&agent),
                        serde_json::json!({"client": client_idx, "seq": event_idx}),
                    );
                }
            }));
        }

        for h in handles {
            h.join().unwrap();
        }

        let all = bus.read(&ws, &ch, 0, 1000);
        assert_eq!(
            all.len(),
            n_clients * n_events_per_client,
            "all events should be persisted"
        );

        let ids: Vec<i64> = all.iter().map(|e| e.id).collect();
        let mut sorted = ids.clone();
        sorted.sort_unstable();
        assert_eq!(ids, sorted, "events must be in strictly ascending ID order");

        for win in ids.windows(2) {
            assert!(
                win[1] > win[0],
                "IDs must be strictly monotonic (no gaps from concurrent access)"
            );
        }
    }

    #[test]
    fn workspace_channel_isolation() {
        let (bus, _td) = test_bus();
        let pid = std::process::id();
        let ws_a = format!("ws-iso-a-{pid}");
        let ws_b = format!("ws-iso-b-{pid}");
        let ws_c = format!("ws-iso-c-{pid}");
        let ch1 = format!("ch-iso-1-{pid}");
        let ch2 = format!("ch-iso-2-{pid}");

        bus.append(
            &ws_a,
            &ch1,
            &ContextEventKindV1::SessionMutated,
            Some("agent-a"),
            serde_json::json!({"ws":"a","ch":"1"}),
        );
        bus.append(
            &ws_a,
            &ch2,
            &ContextEventKindV1::KnowledgeRemembered,
            Some("agent-a"),
            serde_json::json!({"ws":"a","ch":"2"}),
        );
        bus.append(
            &ws_b,
            &ch1,
            &ContextEventKindV1::ArtifactStored,
            Some("agent-b"),
            serde_json::json!({"ws":"b","ch":"1"}),
        );

        let ws_a_ch_1 = bus.read(&ws_a, &ch1, 0, 100);
        assert_eq!(ws_a_ch_1.len(), 1);
        assert_eq!(ws_a_ch_1[0].kind, "session_mutated");

        let ws_a_ch_2 = bus.read(&ws_a, &ch2, 0, 100);
        assert_eq!(ws_a_ch_2.len(), 1);
        assert_eq!(ws_a_ch_2[0].kind, "knowledge_remembered");

        let ws_b_ch_1 = bus.read(&ws_b, &ch1, 0, 100);
        assert_eq!(ws_b_ch_1.len(), 1);
        assert_eq!(ws_b_ch_1[0].kind, "artifact_stored");

        let ws_c_ch_1 = bus.read(&ws_c, &ch1, 0, 100);
        assert!(ws_c_ch_1.is_empty(), "non-existent workspace returns empty");
    }

    #[test]
    fn replay_from_cursor_returns_only_newer_events() {
        let (bus, _td) = test_bus();
        let pid = std::process::id();
        let ws = &format!("ws-replay-{pid}");
        let ch = &format!("ch-replay-{pid}");

        let ev1 = bus
            .append(
                ws,
                ch,
                &ContextEventKindV1::ToolCallRecorded,
                None,
                serde_json::json!({"seq":1}),
            )
            .unwrap();
        let ev2 = bus
            .append(
                ws,
                ch,
                &ContextEventKindV1::SessionMutated,
                None,
                serde_json::json!({"seq":2}),
            )
            .unwrap();
        let _ev3 = bus
            .append(
                ws,
                ch,
                &ContextEventKindV1::GraphBuilt,
                None,
                serde_json::json!({"seq":3}),
            )
            .unwrap();

        let from_cursor = bus.read(ws, ch, ev2.id, 100);
        assert_eq!(from_cursor.len(), 1, "only events after cursor");
        assert_eq!(from_cursor[0].kind, "graph_built");

        let from_first = bus.read(ws, ch, ev1.id, 100);
        assert_eq!(from_first.len(), 2, "events after first");

        let from_zero = bus.read(ws, ch, 0, 100);
        assert_eq!(from_zero.len(), 3, "all events from zero");
    }

    #[test]
    fn directed_visibility_survives_restart_and_filters_other_agents() {
        let td = tempdir().expect("tempdir");
        let path = td.path().join("directed-context-os.db");
        let bus = ContextBus::open_at(path.clone());
        let targets = vec!["agent-a".to_string()];
        let event = bus
            .append_directed(
                "ws",
                "ch",
                &ContextEventKindV1::SessionMutated,
                Some("lead"),
                serde_json::json!({"decision":"keep"}),
                targets.clone(),
            )
            .expect("directed event");
        assert_eq!(event.target_agents, Some(targets.clone()));
        drop(bus);

        let reopened = ContextBus::open_at(path);
        let replayed = reopened.read("ws", "ch", 0, 10);
        assert_eq!(replayed.len(), 1);
        assert_eq!(replayed[0].target_agents, Some(targets));
        assert!(
            TopicFilter {
                agent_id: Some("agent-a".to_string()),
                ..TopicFilter::default()
            }
            .matches(&replayed[0])
        );
        assert!(
            !TopicFilter {
                agent_id: Some("agent-b".to_string()),
                ..TopicFilter::default()
            }
            .matches(&replayed[0])
        );
    }

    #[test]
    fn legacy_schema_migrates_without_turning_events_into_directed_events() {
        let td = tempdir().expect("tempdir");
        let path = td.path().join("legacy-context-os.db");
        {
            let conn = Connection::open(&path).expect("legacy db");
            conn.execute_batch(
                "CREATE TABLE context_events (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   workspace_id TEXT NOT NULL,
                   channel_id TEXT NOT NULL,
                   kind TEXT NOT NULL,
                   actor TEXT,
                   timestamp TEXT NOT NULL,
                   payload_json TEXT NOT NULL,
                   version INTEGER NOT NULL DEFAULT 0,
                   parent_id INTEGER,
                   delivery_id TEXT
                 );
                 INSERT INTO context_events
                   (workspace_id, channel_id, kind, timestamp, payload_json)
                 VALUES ('ws', 'ch', 'knowledge_remembered',
                         '2026-01-01T00:00:00Z', '{}');",
            )
            .expect("legacy schema");
        }

        let bus = ContextBus::open_at(path);
        let replayed = bus.read("ws", "ch", 0, 10);
        assert_eq!(replayed.len(), 1);
        assert_eq!(replayed[0].target_agents, None);
    }

    #[test]
    fn broadcast_subscriber_receives_events() {
        let (bus, _td) = test_bus();
        let mut rx = bus.subscribe("ws", "ch").expect("subscribe should succeed");

        let ev = bus
            .append(
                "ws",
                "ch",
                &ContextEventKindV1::ProofAdded,
                Some("verifier"),
                serde_json::json!({"proof":"hash"}),
            )
            .unwrap();

        let received = rx.try_recv().expect("subscriber should receive event");
        assert_eq!(received.id, ev.id);
        assert_eq!(received.kind, "proof_added");
        assert_eq!(received.actor.as_deref(), Some("verifier"));
    }

    #[test]
    fn event_kind_parser_roundtrips_all_v1_kinds() {
        for kind in [
            ContextEventKindV1::ToolCallRecorded,
            ContextEventKindV1::SessionMutated,
            ContextEventKindV1::KnowledgeRemembered,
            ContextEventKindV1::ArtifactStored,
            ContextEventKindV1::GraphBuilt,
            ContextEventKindV1::ProofAdded,
            ContextEventKindV1::ProvenanceCheckpoint,
            ContextEventKindV1::ProvenanceEdit,
            ContextEventKindV1::HealthChanged,
        ] {
            assert_eq!(ContextEventKindV1::parse(kind.as_str()), kind);
        }
    }
}
// SPDX-License-Identifier: Apache-2.0