lean_ctx/core/context_os/
observation.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Private, durable observations owned by the ContextBus.
3
4use crate::core::context_admission::stores::StoreAdmission;
5use crate::core::events::LeanCtxEvent;
6use crate::core::ocla_bus::OclaEvent;
7use std::fmt;
8use std::sync::{Mutex, MutexGuard, OnceLock, TryLockError};
9use std::time::{Duration, Instant};
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum LocalObservationSource {
13    Ocla,
14    Legacy,
15}
16
17impl LocalObservationSource {
18    pub fn as_str(self) -> &'static str {
19        match self {
20            Self::Ocla => "ocla",
21            Self::Legacy => "legacy",
22        }
23    }
24
25    pub fn parse(value: &str) -> Result<Self, ObservationPersistenceError> {
26        match value {
27            "ocla" => Ok(Self::Ocla),
28            "legacy" => Ok(Self::Legacy),
29            other => Err(ObservationPersistenceError::InvalidRow(format!(
30                "unknown local observation source {other:?}"
31            ))),
32        }
33    }
34}
35
36#[derive(Debug)]
37pub enum ObservationPersistenceError {
38    Io(std::io::Error),
39    LockPoisoned,
40    LockTimeout,
41    IdentityConflict,
42    Serialization(String),
43    Sqlite(String),
44    InvalidRow(String),
45    /// Storage admission would not keep the payload as it is: not persisted.
46    Withheld,
47}
48
49#[derive(Debug, Clone, PartialEq)]
50pub enum LocalObservationPayload {
51    Ocla(OclaEvent),
52    Legacy(LeanCtxEvent),
53}
54
55#[derive(Debug, Clone, PartialEq)]
56pub struct LocalObservationV1 {
57    /// SQLite's local cursor; this is not the producer's identity.
58    pub cursor_id: i64,
59    /// A producer instance distinguishes IDs reused after a process restart.
60    pub producer_instance: String,
61    /// The original producer ID, retained as u64 independently of SQLite.
62    pub source_id: u64,
63    /// OCLA IDs use decimal milliseconds; legacy timestamps remain verbatim.
64    pub source_timestamp: String,
65    pub source: LocalObservationSource,
66    pub payload: LocalObservationPayload,
67}
68
69impl fmt::Display for ObservationPersistenceError {
70    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
71        match self {
72            Self::Io(error) => write!(f, "context bus storage unavailable: {error}"),
73            Self::LockPoisoned => f.write_str("context bus observation lock poisoned"),
74            Self::LockTimeout => f.write_str("context bus observation lock timed out"),
75            Self::IdentityConflict => {
76                f.write_str("observation identity reused with different data")
77            }
78            Self::Serialization(error) => write!(f, "observation serialization failed: {error}"),
79            Self::Sqlite(error) => write!(f, "observation sqlite operation failed: {error}"),
80            Self::InvalidRow(error) => write!(f, "invalid local observation row: {error}"),
81            Self::Withheld => f.write_str("observation withheld by the context gateway"),
82        }
83    }
84}
85
86// ─── Admission at the write, current source rights at the use (G04, E3) ─────
87//
88// An observation is a typed record of an operation of this process. It is not
89// text copied out of a source, but it carries strings its producer filled in
90// (a path, a reason, a detail) and — in five event kinds — names the file the
91// operation was about. Its `path` field is the source reference on record;
92// nothing is added to the row, and the table format is unchanged.
93
94/// Tool identity the file an observation names is re-read for.
95const SOURCE_TOOL: &str = "ctx_read";
96
97/// Every string value of a payload, decoded, one per line: what admission
98/// inspects (the content a reader will see — not JSON escapes and not field
99/// names next to numbers).
100fn payload_text(payload: &serde_json::Value) -> String {
101    fn collect<'a>(value: &'a serde_json::Value, leaves: &mut Vec<&'a str>) {
102        match value {
103            serde_json::Value::String(text) => leaves.push(text),
104            serde_json::Value::Array(items) => {
105                for item in items {
106                    collect(item, leaves);
107                }
108            }
109            serde_json::Value::Object(fields) => {
110                fields.values().for_each(|field| collect(field, leaves));
111            }
112            _ => {}
113        }
114    }
115    let mut leaves = Vec::new();
116    collect(payload, &mut leaves);
117    leaves.join("\n")
118}
119
120/// The file a payload names: the first string field called `path`.
121fn payload_path(payload: &serde_json::Value) -> Option<&str> {
122    let fields = payload.as_object()?;
123    fields
124        .get("path")
125        .and_then(serde_json::Value::as_str)
126        .or_else(|| fields.values().find_map(payload_path))
127}
128
129/// Storage admission of a payload before it becomes a durable row (E3): its
130/// strings pass the current gateway policy (`StoreAdmission`, as every derived
131/// store) and are kept only verbatim. A payload admission would withhold,
132/// restrict or rewrite is not persisted; the refusal carries no value.
133pub(crate) fn admit_payload(payload_json: &str) -> Result<(), ObservationPersistenceError> {
134    let text = payload_text(&serde_json::from_str(payload_json)?);
135    if kept_verbatim(&admission_snapshot(), &text) {
136        Ok(())
137    } else {
138        Err(ObservationPersistenceError::Withheld)
139    }
140}
141
142/// The current gateway policy as a snapshot that is not observed: a producer
143/// may hold a store lock, and observing a policy change drops resident
144/// stores. Payloads are kept only verbatim, so they are decided by the
145/// built-in floor (`StoreAdmission::built_in_floor`): a semantic finding in a
146/// path or a tool name could only drop the observation, never mask it.
147fn admission_snapshot() -> StoreAdmission {
148    StoreAdmission::with_policy(
149        crate::core::context_admission::AdmissionPolicy::from_config(
150            &crate::core::config::Config::load_arc(),
151        ),
152    )
153    .built_in_floor()
154}
155
156/// The ONE payload rule, for the write and for every use: the strings of a
157/// payload may be persisted — and shown — only if admission keeps them
158/// verbatim NOW. The project policy decides on the original text, then the
159/// gateway detectors, then the project policy again on what they left
160/// (`StoreAdmission::admit_shell_command`: that exact composition; nothing in
161/// it is specific to a shell). A masked variant is never stored and never
162/// returned.
163fn kept_verbatim(admission: &StoreAdmission, text: &str) -> bool {
164    text.is_empty() || admission.admit_shell_command(text).as_deref() == Some(text)
165}
166
167/// The serve-time check of ONE read of the observation store, for a surface
168/// that shows observations — paths, names, free text — to a host or a model.
169/// Built per call and dropped with it; counts and timestamps need no gate.
170///
171/// - An observation that names a file, read by a reader with a project: the
172///   file is read NOW through the admitted rooted reader in that project
173///   (`SourceGate`: path jail, role, project policy, gateway; bounded, over
174///   the bound refused). Gone, outside the project, withheld or restricted:
175///   not shown.
176/// - An observation without a file, or a reader without a project: no source
177///   authority on record — the originless rule (`serve_stored` without a
178///   binding): withheld while protection is active in any layer, re-admitted
179///   under today's content rules in Community developer mode.
180///
181/// BOTH conditions hold for every observation that leaves: the source right
182/// above AND the payload rule of the write side ([`kept_verbatim`]) under
183/// the policy of NOW. A row stored before admission existed, or before a
184/// rule was tightened, is not legitimised by an admitted file.
185///
186/// One view per gate, taken in [`Self::for_project`]: one gateway snapshot
187/// for every payload, one reading of the protection state for the originless
188/// rule. Nothing is loaded per observation.
189pub(crate) struct ObservationGate {
190    files: Option<crate::core::bm25_index::source_gate::SourceGate>,
191    admission: StoreAdmission,
192    /// `stored::source_authority_required()` when the gate was built.
193    protected: bool,
194    withheld: usize,
195}
196
197impl ObservationGate {
198    /// The gate of a reader working in `project`; without one, the project of
199    /// the running request, if any.
200    pub(crate) fn for_project(project: Option<&std::path::Path>) -> Self {
201        use crate::core::bm25_index::source_gate::SourceGate;
202        let request = crate::core::admitted_sources::request_root();
203        // ONE gateway snapshot for the payload checks and the source reads
204        // of this gate: the file gate decides under the same rules.
205        let admission = admission_snapshot();
206        let root = match project {
207            Some(root) => Some(root),
208            None if request.is_empty() => None,
209            None => Some(std::path::Path::new(&request)),
210        };
211        let files = root.map(|root| SourceGate::with_admission(root, SOURCE_TOOL, &admission));
212        Self {
213            files,
214            admission,
215            protected: crate::core::context_admission::stored::source_authority_required(),
216            withheld: 0,
217        }
218    }
219
220    /// Whether `observation` (an event or a decoded payload) may be shown now.
221    pub(crate) fn serves<T: serde::Serialize>(&mut self, observation: &T) -> bool {
222        let served = serde_json::to_value(observation).is_ok_and(|payload| {
223            // The payload first, in both branches: every string value, the
224            // set the write side walks, under today's rules.
225            if !kept_verbatim(&self.admission, &payload_text(&payload)) {
226                return false;
227            }
228            match (payload_path(&payload), self.files.as_mut()) {
229                (Some(path), Some(files)) => files.serves(path, ""),
230                // The originless rule: no source authority on record, so
231                // nothing is shown under protection; without protection the
232                // payload check above is the re-admission.
233                _ => !self.protected,
234            }
235        });
236        if !served {
237            self.withheld += 1;
238        }
239        served
240    }
241
242    /// Whether the gateway rules configured NOW are still the ones this gate
243    /// took its snapshot of: the digests are compared (`StoreAdmission::digest`,
244    /// the form of the importer's `gateway_rules_unchanged`). Asked once, when
245    /// the last observation was decided; `false` means the source reads of
246    /// this call may have run under other rules than its payload checks.
247    pub(crate) fn gateway_unchanged(&self) -> bool {
248        self.admission.digest() == admission_snapshot().digest()
249    }
250
251    /// Observations this gate refused. A count only: no path, no text.
252    pub(crate) fn withheld(&self) -> usize {
253        self.withheld
254    }
255}
256
257impl std::error::Error for ObservationPersistenceError {}
258
259impl From<serde_json::Error> for ObservationPersistenceError {
260    fn from(error: serde_json::Error) -> Self {
261        Self::Serialization(error.to_string())
262    }
263}
264
265impl From<rusqlite::Error> for ObservationPersistenceError {
266    fn from(error: rusqlite::Error) -> Self {
267        Self::Sqlite(error.to_string())
268    }
269}
270
271impl From<std::io::Error> for ObservationPersistenceError {
272    fn from(error: std::io::Error) -> Self {
273        Self::Io(error)
274    }
275}
276
277pub(crate) fn validate_identity(
278    producer_instance: &str,
279    source_id: u64,
280) -> Result<(), ObservationPersistenceError> {
281    if source_id == 0
282        || !uuid::Uuid::parse_str(producer_instance)
283            .is_ok_and(|id| id.to_string() == producer_instance && !id.is_nil())
284    {
285        return Err(ObservationPersistenceError::InvalidRow(
286            "invalid producer instance or source ID".into(),
287        ));
288    }
289    Ok(())
290}
291
292/// Validate persisted metadata before either reader projects a payload.
293///
294/// Returns `Ok(None)` for a legacy record whose event kind this build retired
295/// (see `events::is_retired_event_record`): such rows are valid history from an
296/// older release and are skipped instead of failing the whole projection.
297pub(crate) fn decode_row(
298    row: &rusqlite::Row<'_>,
299) -> Result<Option<LocalObservationV1>, ObservationPersistenceError> {
300    if row.get::<_, i64>(6)? != 1 {
301        return Err(ObservationPersistenceError::InvalidRow(
302            "unsupported local observation schema".into(),
303        ));
304    }
305    let cursor_id: i64 = row.get(0)?;
306    let source = LocalObservationSource::parse(row.get::<_, String>(1)?.as_str())?;
307    let source_id_raw: String = row.get(2)?;
308    let source_id = source_id_raw.parse::<u64>().map_err(|_| {
309        ObservationPersistenceError::InvalidRow("invalid source ID encoding".into())
310    })?;
311    let producer_instance: String = row.get(5)?;
312    validate_identity(&producer_instance, source_id)?;
313    if cursor_id <= 0 || source_id.to_string() != source_id_raw {
314        return Err(ObservationPersistenceError::InvalidRow(
315            "noncanonical observation identity".into(),
316        ));
317    }
318    let source_timestamp: String = row.get(3)?;
319    let payload_json: String = row.get(4)?;
320    let payload = match source {
321        LocalObservationSource::Ocla => {
322            if !source_timestamp
323                .parse::<u64>()
324                .is_ok_and(|value| value.to_string() == source_timestamp)
325            {
326                return Err(ObservationPersistenceError::InvalidRow(
327                    "invalid OCLA timestamp encoding".into(),
328                ));
329            }
330            LocalObservationPayload::Ocla(serde_json::from_str(&payload_json)?)
331        }
332        LocalObservationSource::Legacy => {
333            if crate::core::events::is_retired_event_record(&payload_json) {
334                return Ok(None);
335            }
336            let event: LeanCtxEvent = serde_json::from_str(&payload_json)?;
337            if event.id != source_id || event.timestamp != source_timestamp {
338                return Err(ObservationPersistenceError::InvalidRow(
339                    "legacy payload contradicts observation identity or timestamp".into(),
340                ));
341            }
342            LocalObservationPayload::Legacy(event)
343        }
344    };
345    Ok(Some(LocalObservationV1 {
346        cursor_id,
347        producer_instance,
348        source_id,
349        source_timestamp,
350        source,
351        payload,
352    }))
353}
354
355pub(crate) fn producer_instance() -> &'static str {
356    static INSTANCE: OnceLock<String> = OnceLock::new();
357    INSTANCE.get_or_init(|| uuid::Uuid::new_v4().to_string())
358}
359
360/// Bound contention and reject poisoned state; callers never recover a guard.
361pub(crate) fn lock_observation<T>(
362    mutex: &Mutex<T>,
363) -> Result<MutexGuard<'_, T>, ObservationPersistenceError> {
364    let started = Instant::now();
365    loop {
366        match mutex.try_lock() {
367            Ok(guard) => return Ok(guard),
368            Err(TryLockError::Poisoned(_)) => {
369                return Err(ObservationPersistenceError::LockPoisoned);
370            }
371            Err(TryLockError::WouldBlock) => {
372                if started.elapsed() >= Duration::from_secs(5) {
373                    return Err(ObservationPersistenceError::LockTimeout);
374                }
375                std::thread::sleep(Duration::from_millis(1));
376            }
377        }
378    }
379}
380
381#[cfg(test)]
382mod tests {
383    use super::*;
384    use crate::core::context_os::context_bus::ContextBus;
385    use crate::core::events::EventKind;
386
387    const AWS_KEY: &str = concat!("AK", "IAIOSFODNN7EXAMPLE");
388
389    fn event(id: u64, kind: EventKind) -> LeanCtxEvent {
390        LeanCtxEvent {
391            id,
392            timestamp: "2026-01-01T00:00:00.000".to_owned(),
393            kind,
394        }
395    }
396
397    /// G04/E3: an observation that names a file is shown while the admitted
398    /// reader returns that file and no longer once it is withdrawn; a payload
399    /// storage admission would rewrite never becomes a row, and the refusal
400    /// carries nothing of it.
401    #[test]
402    fn an_observation_is_shown_only_while_the_file_it_names_is_admitted() {
403        let _data = crate::core::data_dir::isolated_data_dir();
404        let community = crate::core::policy::runtime::TestPolicyOverride::set(None);
405        let project = tempfile::tempdir().expect("project");
406        let root = project.path().canonicalize().expect("canonical root");
407        let store = tempfile::tempdir().expect("store");
408        let bus = ContextBus::open_at(store.path().join("context-bus.db"));
409        std::fs::write(root.join("billing.rs"), "pub fn settle() {}\n").expect("source");
410        let hit = event(
411            1,
412            EventKind::CacheHit {
413                path: "billing.rs".to_owned(),
414                saved_tokens: 10,
415            },
416        );
417        bus.append_legacy_observation(producer_instance(), &hit)
418            .expect("an admitted payload is stored");
419        let shown = || -> usize {
420            let mut gate = ObservationGate::for_project(Some(root.as_path()));
421            bus.read_legacy_observations(10)
422                .expect("rows")
423                .iter()
424                .filter(|stored| gate.serves(*stored))
425                .count()
426        };
427        assert_eq!(shown(), 1);
428
429        std::fs::remove_file(root.join("billing.rs")).expect("withdraw");
430        assert_eq!(shown(), 0);
431        let mut gate = ObservationGate::for_project(Some(root.as_path()));
432        assert!(!gate.serves(&hit));
433        assert_eq!(gate.withheld(), 1);
434
435        let keyed = event(
436            2,
437            EventKind::VerificationWarning {
438                warning_kind: "drift".to_owned(),
439                detail: format!("deploy token {AWS_KEY}"),
440                severity: "warn".to_owned(),
441            },
442        );
443        let refused = bus
444            .append_legacy_observation(producer_instance(), &keyed)
445            .expect_err("a payload admission would mask is not persisted");
446        assert!(matches!(refused, ObservationPersistenceError::Withheld));
447        assert!(!refused.to_string().contains(AWS_KEY));
448        assert_eq!(bus.read_legacy_observations(10).expect("rows").len(), 1);
449
450        // Both conditions, also for a row stored before a rule existed: two
451        // observations name the same admitted file; one carries a string
452        // today's policy refuses. An admitted file does not legitimise it.
453        std::fs::write(root.join("ledger.rs"), "pub fn post() {}\n").expect("source");
454        let call = |id: u64, mode: &str| {
455            event(
456                id,
457                EventKind::ToolCall {
458                    tool: "ctx_read".to_owned(),
459                    tokens_original: 10,
460                    tokens_saved: 5,
461                    mode: Some(mode.to_owned()),
462                    duration_ms: 1,
463                    path: Some("ledger.rs".to_owned()),
464                },
465            )
466        };
467        for stored in [call(3, "map"), call(4, "CONFIDENTIAL")] {
468            bus.append_legacy_observation(producer_instance(), &stored)
469                .expect("stored under the looser rule");
470        }
471        drop(community);
472        let policy = crate::core::policy::load(
473            "name = \"observation-test\"\nversion = \"1.0.0\"\ndescription = \"test\"\n\
474             [filters]\nclassification = \"block\"\nblocked_labels = [\"CONFIDENTIAL\"]\n",
475        )
476        .expect("policy");
477        let _tightened = crate::core::policy::runtime::TestPolicyOverride::set(Some(policy));
478        let mut gate = ObservationGate::for_project(Some(root.as_path()));
479        let shown: Vec<u64> = bus
480            .read_legacy_observations(10)
481            .expect("rows")
482            .iter()
483            .filter(|stored| gate.serves(*stored))
484            .map(|stored| stored.id)
485            .collect();
486        assert_eq!(shown, [3]);
487    }
488}
lean_ctx/core/context_os/observation.rs

lean_ctx/core/context_os/
observation.rs
