diff --git a/src/sessiond/device_state.rs b/src/sessiond/device_state.rs index af060bc..fcc3fd1 100644 --- a/src/sessiond/device_state.rs +++ b/src/sessiond/device_state.rs @@ -709,6 +709,59 @@ impl ForensicLog { let start = entries.len().saturating_sub(count); entries[start..].to_vec() } + + /// Entries recorded after `seq`, oldest first. + /// + /// The cursor is the sequence number the trail already mints for its hash + /// chain, so a consumer that reconnects resumes exactly where it stopped + /// and a restarted daemon cannot silently replay — the chain and the + /// stream agree by construction rather than by a second counter kept in + /// step by hand. + pub fn since(&self, seq: u64) -> Vec { + let entries = self.entries.lock().unwrap_or_else(|e| e.into_inner()); + entries.iter().filter(|e| e.seq > seq).cloned().collect() + } + + /// The highest sequence the in-memory buffer holds. + pub fn head_seq(&self) -> u64 { + let entries = self.entries.lock().unwrap_or_else(|e| e.into_inner()); + entries.last().map(|e| e.seq).unwrap_or(0) + } +} + +/// Is this entry worth waking a mind for? +/// +/// The whole value of the subscribe stream is this predicate. §10 is the +/// argument in miniature: a sensor resting at `far` and a sensor whose stack +/// took a CHRE fatal produce byte-identical silence, so "no evidence" and +/// "evidence says nothing is happening" had to become different states. The +/// same distinction decides what crosses to the agent — an edge means +/// something, a level does not. +/// +/// What crosses: +/// - **Transitions**, legal or refused. A refused one is the more interesting: +/// the machine wanted to move and its own guard said no. +/// - **Errors.** Both classes. `error-security` is a violated guarantee; +/// `error-operational` is an actuator that did not do as it was told, which +/// is how a `panel-off` recorded during four dead hours stops reading like a +/// healthy one. +/// - **Decisions**, which is where `source-down` / `source-recovered` and +/// cross-sensor disagreement already land. +/// +/// What never crosses: `Heartbeat` (it exists to fill timeline gaps for a +/// reader, and a mind is not a reader), `SensorInput` (a reading is a driver +/// wearing evidence's clothes — 2,900 proximity lines a day saying nothing +/// changed, and the exact stream P1 says an inferring model must not have), +/// and `Wake`, which is already implied by the transition it causes. +pub fn is_notable(event: &ForensicEvent) -> bool { + match event { + ForensicEvent::Transition { .. } => true, + ForensicEvent::Error { .. } => true, + ForensicEvent::Decision { .. } => true, + ForensicEvent::SensorInput { .. } => false, + ForensicEvent::Wake { .. } => false, + ForensicEvent::Heartbeat => false, + } } impl TrailWriter { diff --git a/src/sessiond/protocol.rs b/src/sessiond/protocol.rs index bbd7822..b0d66a8 100644 --- a/src/sessiond/protocol.rs +++ b/src/sessiond/protocol.rs @@ -188,6 +188,13 @@ pub const VERBS: &[VerbDoc] = &[ refuses: &[], example: r#"{"op":"forensic_log","count":50}"#, }, + VerbDoc { + op: "subscribe", + mutates: false, + summary: "this connection becomes an event stream of notable belief changes", + refuses: &[], + example: r#"{"op":"subscribe"}"#, + }, VerbDoc { op: "get_policy", mutates: false, @@ -259,6 +266,25 @@ pub enum Request { /// Query recent forensic log entries. Returns the last N entries /// from the in-memory forensic buffer for post-hoc analysis. ForensicLog { count: Option }, + /// Turn this connection into an event stream. After the `ok`, the daemon + /// pushes one trail entry per line, unprompted, for as long as the caller + /// holds the socket open. + /// + /// The point is the *filter*, not the transport. `forensic_log` already + /// hands over everything; a consumer that polls it and diffs is doing the + /// machine's job for it, badly and late. Only NOTABLE entries are pushed — + /// state transitions, a source that died or came back, a violated + /// guarantee, sensors that contradict each other. Heartbeats, ticks and + /// ordinary readings never cross. + /// + /// That compression IS the product, in the same sense six strain gauges at + /// 100 Hz becoming one `squeeze` bit is the product. An agent that received + /// every reading would be reading drivers, and doctrine's load-bearing rule + /// is that interpretation may consume evidence but never a driver — reading + /// the raw stream is how gait, typing and identity get inferred from data + /// that looks innocent per-field (SECURITY-AUDIT P1). The narrowness here is + /// a security control, not a performance one. + Subscribe, /// Read the timed policy — the Auto-Lock shaped settings. GetPolicy, /// Change the timed policy. Every field is optional; omitted fields keep @@ -395,7 +421,7 @@ mod tests { // And the other direction: every variant must be advertised. Bump this // deliberately when a verb is added, having added its VerbDoc. - assert_eq!(VERBS.len(), 12, "a Request variant was added without a VerbDoc"); + assert_eq!(VERBS.len(), 13, "a Request variant was added without a VerbDoc"); } #[test] diff --git a/src/sessiond/server.rs b/src/sessiond/server.rs index a35a3e8..eacc56c 100644 --- a/src/sessiond/server.rs +++ b/src/sessiond/server.rs @@ -18,7 +18,9 @@ use std::time::Duration; use anyhow::{Context, Result}; use tracing::{info, warn}; -use crate::sessiond::device_state::{Action, DeviceState, DeviceStateMachine, SensorEvidence}; +use crate::sessiond::device_state::{ + is_notable, Action, DeviceState, DeviceStateMachine, SensorEvidence, +}; use crate::sessiond::idle; use crate::sessiond::lock::{self, LockController, Msg, SessionOutcome}; use crate::sessiond::lockhint; @@ -67,6 +69,17 @@ struct Daemon { /// daemon that owns the *decision* reaches the process that owns the /// *surface*. shell_directives: Option, + /// Connections that asked to be told, rather than to ask. + /// + /// The shell gets directives because it owns a surface the authority needs + /// driven. These get *events*, and own nothing — an observer that cannot + /// act cannot be a second authority by accident. Dead ones are pruned on + /// the write that fails; there is no reaper, because a subscriber that + /// never receives anything is indistinguishable from a quiet device and + /// pruning on silence would be §10's mistake again. + subscribers: Vec, + /// Highest trail seq already pushed to subscribers. + last_pushed_seq: u64, /// The unified device state machine. The single authority for device /// power state — idle, lock, doze, sleep. All actors route through it. device_state: DeviceStateMachine, @@ -118,6 +131,8 @@ pub fn run(initial_lock: bool, socket_path: &Path) -> Result<()> { shell_alive: false, shell_pid: None, shell_directives: None, + subscribers: Vec::new(), + last_pushed_seq: 0, device_state: DeviceStateMachine::new(), }), cond: Condvar::new(), @@ -162,6 +177,11 @@ fn spawn_clock(shared: &Arc) { for action in actions { execute(&shared, action); } + // After the actions, so an executor's own `error-operational` is in the + // trail before the tick's events go out. A subscriber that hears the + // blank but not the brightnessctl failure underneath it has been told a + // tidier story than what happened. + fan_out(&shared); }); } @@ -544,6 +564,11 @@ fn handle_connection(stream: UnixStream, shared: Arc) { // Whether THIS connection is the registered shell heartbeat, and under // which generation. let mut heartbeat: Option = None; + // Whether it asked to be told rather than to ask. Read back off our own + // response rather than threaded through `handle_request`, because that + // field IS the wire contract — a caller learns it is subscribed the same + // way this loop does, and there is no second place for the two to disagree. + let mut subscribed = false; loop { let mut line = String::new(); @@ -562,7 +587,12 @@ fn handle_connection(stream: UnixStream, shared: Arc) { std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut ) => { - if heartbeat.is_some() { + // A subscriber is silent for the same reason the heartbeat is: + // silence is the normal case. A quiet device is exactly when + // nothing notable is happening, so reaping on silence would + // disconnect every observer of a healthy phone — §10's mistake + // with the roles reversed. + if heartbeat.is_some() || subscribed { continue; } warn!( @@ -599,9 +629,16 @@ fn handle_connection(stream: UnixStream, shared: Arc) { } Err(e) => refuse(RefusalCode::UnsupportedOp, &format!("malformed request: {e}")), }; + if response.get("subscribed").and_then(|v| v.as_bool()) == Some(true) { + subscribed = true; + } if respond(&mut writer, response).is_err() { break; } + // After the answer, never before it: a request that moved the machine + // has its own reply on the wire before the event describing it, so a + // caller that is also a subscriber sees cause and then effect. + fan_out(&shared); } // Connection gone. If it was the live heartbeat, the shell died — @@ -622,6 +659,56 @@ fn handle_connection(stream: UnixStream, shared: Arc) { } } +/// Push every notable trail entry recorded since the last push to every +/// subscriber, and drop the ones that have gone away. +/// +/// Called after anything that can move the machine — a request, a tick, a +/// logind edge — rather than from inside the recording path. Recording happens +/// under the state lock, and writing to a socket under that lock is how this +/// daemon deadlocked itself once already (`4d652bb`): 347 threads parked in +/// __futex_wait behind one guard, accept loop healthy, not a single request +/// answered. A slow reader must never be able to stop the machine deciding. +fn fan_out(shared: &Arc) { + let (entries, mut subs) = { + let mut d = shared.lock(); + if d.subscribers.is_empty() { + // Still advance, or the first subscriber to attach inherits every + // notable edge since boot as a burst of stale news. + d.last_pushed_seq = d.device_state.forensic.head_seq(); + return; + } + let from = d.last_pushed_seq; + let fresh: Vec<_> = d + .device_state + .forensic + .since(from) + .into_iter() + .filter(|e| is_notable(&e.event)) + .collect(); + d.last_pushed_seq = d.device_state.forensic.head_seq(); + if fresh.is_empty() { + return; + } + (fresh, std::mem::take(&mut d.subscribers)) + }; + + subs.retain_mut(|sock| { + entries.iter().all(|entry| { + let Ok(line) = serde_json::to_string(entry) else { + return true; // our fault, not the subscriber's — keep it + }; + sock.write_all(line.as_bytes()) + .and_then(|_| sock.write_all(b"\n")) + .and_then(|_| sock.flush()) + .is_ok() + }) + }); + + let mut d = shared.lock(); + // Anything that attached while we were writing is still in the list. + d.subscribers.append(&mut subs); +} + /// SO_PEERCRED pid of the process on the other end, or None if the kernel /// would not say. None is never treated as a match: the lease is exclusive and /// "I could not prove who you are" has to fail closed. @@ -1013,6 +1100,24 @@ fn handle_request( } serde_json::json!({ "ok": true, "policy": applied }) } + Request::Subscribe => { + let Some(stream) = directives.and_then(|s| s.try_clone().ok()) else { + return refuse( + RefusalCode::Unavailable, + "subscribe needs a connection this daemon can hold open", + ); + }; + let mut d = shared.lock(); + // Start the cursor at the current head, not at 0. A new subscriber + // is told what happens NEXT; replaying the buffer would hand it a + // history it has no way to date and would re-fire hours-old edges + // as if they were now. `forensic_log` is the verb for the past. + let from = d.device_state.forensic.head_seq(); + d.subscribers.push(stream); + let count = d.subscribers.len(); + info!("event subscriber attached (from seq {from}, {count} total)"); + serde_json::json!({ "ok": true, "subscribed": true, "from_seq": from }) + } Request::ForensicLog { count } => { let d = shared.lock(); let entries = d.device_state.forensic.recent(count.unwrap_or(50)); @@ -1038,6 +1143,8 @@ mod tests { shell_alive: false, shell_pid: None, shell_directives: None, + subscribers: Vec::new(), + last_pushed_seq: 0, device_state: DeviceStateMachine::new(), }), cond: Condvar::new(), @@ -1125,6 +1232,66 @@ mod tests { d.shell_pid = None; // silence the unused-mut lint path } + /// A subscriber is told what happens next, not what already happened. + /// Replaying the buffer would re-fire hours-old edges as if they were now. + #[test] + fn subscribing_starts_at_the_current_head_not_at_zero() { + let shared = idle_shared(); + let (sock, _peer) = UnixStream::pair().unwrap(); + + // Move the machine so the trail is non-empty. + let mut hb = None; + handle_request(Request::Panel { on: false }, &shared, None, &mut hb); + handle_request(Request::Panel { on: true }, &shared, None, &mut hb); + let head = shared.lock().device_state.forensic.head_seq(); + + let mut sub_hb = None; + let reply = handle_request(Request::Subscribe, &shared, Some(&sock), &mut sub_hb); + assert_eq!(reply["ok"], true); + assert_eq!(reply["subscribed"], true); + assert_eq!(reply["from_seq"], head, "must not replay the buffer"); + assert_eq!(shared.lock().subscribers.len(), 1); + } + + /// The filter is the product. A mind that receives every reading is + /// reading drivers, which is the thing doctrine forbids and P1 fears. + #[test] + fn only_edges_cross_to_a_subscriber() { + use crate::sessiond::device_state::ForensicEvent; + + assert!(is_notable(&ForensicEvent::Transition { + from: DeviceState::Active, + to: DeviceState::Locked, + legal: true, + })); + assert!( + is_notable(&ForensicEvent::Transition { + from: DeviceState::Asleep, + to: DeviceState::Active, + legal: false, + }), + "a REFUSED transition is the more interesting one — the machine \ + wanted to move and its own guard said no" + ); + assert!(is_notable(&ForensicEvent::Error { + component: "session-events".into(), + action: "source-down".into(), + error: "no reading in 90s".into(), + })); + assert!(is_notable(&ForensicEvent::Decision { + decision: "source-recovered".into(), + inputs: serde_json::json!({ "source": "proximity" }), + })); + + // The two that would make it tinnitus. + assert!(!is_notable(&ForensicEvent::Heartbeat)); + assert!(!is_notable(&ForensicEvent::SensorInput { + source: crate::sessiond::protocol::SensorSource::Proximity, + value: crate::sessiond::protocol::SensorValue::Near(true), + confidence: 0.4, + })); + } + /// A caller we cannot identify never inherits the reload exemption, even /// while a real shell holds the lease. Unprovable identity fails closed. #[test]