sessiond: subscribe — the body tells the mind when a belief changes

A held connection receives one trail entry per notable edge: transitions,
source-down/recovered, violated guarantees. Readings, wakes and heartbeats
never cross — the filter is the point, not the transport.

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<ForensicEntry> {
+        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 {

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<usize> },
+    /// 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]

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<UnixStream>,
+    /// 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<UnixStream>,
+    /// 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<Shared>) {
         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<Shared>) {
     // Whether THIS connection is the registered shell heartbeat, and under
     // which generation.
     let mut heartbeat: Option<u64> = 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<Shared>) {
                     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<Shared>) {
             }
             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<Shared>) {
     }
 }
 
+/// 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<Shared>) {
+    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]