clawft_kernel/
causal_state_fold.rs

1//! Fold-replay for causal node lifecycle state (ADR-067 P1-graph / WEFT-629).
2//!
3//! [`set_node_state`](crate::causal::CausalGraph::set_node_state) emits a
4//! [`causal.node.state`](crate::chain::EVENT_KIND_CAUSAL_NODE_STATE) chain event
5//! on every real transition. Folding those events (optionally plus
6//! `causal.node.add` / `causal.node.remove`) reconstructs
7//! `node_id → state` at any chain sequence without requiring a full graph
8//! snapshot.
9//!
10//! This is the **chain-event fold** half of P1-graph. ADR-067 also describes a
11//! per-commit snapshot-diff path (more faithful for un-instrumented mutations);
12//! both can coexist — fold is the small kernel ask named by WEFT-629.
13
14use std::collections::HashMap;
15
16use crate::causal::NodeId;
17use crate::chain::{
18    EVENT_KIND_CAUSAL_NODE_ADD, EVENT_KIND_CAUSAL_NODE_REMOVE, EVENT_KIND_CAUSAL_NODE_STATE,
19    ChainEvent, ChainManager,
20};
21
22/// A single parsed `causal.node.state` transition.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct NodeStateTransition {
25    /// Chain sequence of the event.
26    pub sequence: u64,
27    /// Causal node id.
28    pub node_id: NodeId,
29    /// Previous state tag (`None` if the node had no `metadata.state` yet).
30    pub from: Option<String>,
31    /// New state tag.
32    pub to: String,
33}
34
35/// Parse a chain event into a [`NodeStateTransition`] when kind matches.
36pub fn parse_node_state_event(ev: &ChainEvent) -> Option<NodeStateTransition> {
37    if ev.kind != EVENT_KIND_CAUSAL_NODE_STATE {
38        return None;
39    }
40    let payload = ev.payload.as_ref()?;
41    let node_id = payload.get("node_id")?.as_u64()?;
42    let to = payload.get("to")?.as_str()?.to_string();
43    let from = payload
44        .get("from")
45        .and_then(|v| {
46            if v.is_null() {
47                None
48            } else {
49                v.as_str().map(|s| s.to_string())
50            }
51        });
52    Some(NodeStateTransition {
53        sequence: ev.sequence,
54        node_id,
55        from,
56        to,
57    })
58}
59
60/// Fold a sequence of chain events into `node_id → current state`.
61///
62/// Rules:
63/// - `causal.node.add` seeds the node with optional `state` from payload, else
64///   leaves it unmapped until a state event arrives.
65/// - `causal.node.state` sets `to` (last write wins in event order).
66/// - `causal.node.remove` drops the node from the map.
67///
68/// Events of other kinds are ignored. Order is the fold order — pass chain
69/// order (ascending sequence).
70pub fn fold_node_states<'a, I>(events: I) -> HashMap<NodeId, String>
71where
72    I: IntoIterator<Item = &'a ChainEvent>,
73{
74    let mut map: HashMap<NodeId, String> = HashMap::new();
75    for ev in events {
76        match ev.kind.as_str() {
77            k if k == EVENT_KIND_CAUSAL_NODE_ADD => {
78                if let Some(payload) = ev.payload.as_ref()
79                    && let Some(id) = payload.get("node_id").and_then(|v| v.as_u64())
80                    && let Some(state) = payload.get("state").and_then(|v| v.as_str())
81                {
82                    map.insert(id, state.to_string());
83                }
84            }
85            k if k == EVENT_KIND_CAUSAL_NODE_STATE => {
86                if let Some(t) = parse_node_state_event(ev) {
87                    map.insert(t.node_id, t.to);
88                }
89            }
90            k if k == EVENT_KIND_CAUSAL_NODE_REMOVE => {
91                if let Some(payload) = ev.payload.as_ref()
92                    && let Some(id) = payload.get("node_id").and_then(|v| v.as_u64())
93                {
94                    map.remove(&id);
95                }
96            }
97            _ => {}
98        }
99    }
100    map
101}
102
103/// Fold all events on a [`ChainManager`] (via [`ChainManager::tail`]).
104pub fn fold_node_states_from_chain(cm: &ChainManager) -> HashMap<NodeId, String> {
105    let events = cm.tail(cm.len());
106    fold_node_states(&events)
107}
108
109/// Fold only events with `sequence <= at_seq` (scrub to chain time T).
110pub fn fold_node_states_until(cm: &ChainManager, at_seq: u64) -> HashMap<NodeId, String> {
111    let events = cm.tail(cm.len());
112    let filtered: Vec<&ChainEvent> = events.iter().filter(|e| e.sequence <= at_seq).collect();
113    fold_node_states(filtered)
114}
115
116// ---------------------------------------------------------------------------
117// Tests
118// ---------------------------------------------------------------------------
119
120#[cfg(test)]
121mod tests {
122    use super::*;
123    use crate::causal::CausalGraph;
124    use std::sync::Arc;
125
126    fn make_event(seq: u64, kind: &str, payload: serde_json::Value) -> ChainEvent {
127        // Build via a real chain so hashes/fields are valid.
128        let cm = ChainManager::new(99, 1000);
129        // Burn sequences up to `seq` if needed by appending dummies first.
130        // Simpler: just append once and patch sequence for pure fold tests.
131        let mut ev = cm.append("test", kind, Some(payload));
132        // For pure unit tests we only need kind + payload + sequence.
133        ev.sequence = seq;
134        ev
135    }
136
137    #[test]
138    fn parse_node_state_event_round_trip_fields() {
139        let ev = make_event(
140            7,
141            EVENT_KIND_CAUSAL_NODE_STATE,
142            serde_json::json!({
143                "node_id": 42u64,
144                "from": "frontier",
145                "to": "committed",
146            }),
147        );
148        let t = parse_node_state_event(&ev).expect("parse");
149        assert_eq!(t.node_id, 42);
150        assert_eq!(t.from.as_deref(), Some("frontier"));
151        assert_eq!(t.to, "committed");
152    }
153
154    #[test]
155    fn fold_applies_last_write_wins() {
156        let events = vec![
157            make_event(
158                1,
159                EVENT_KIND_CAUSAL_NODE_STATE,
160                serde_json::json!({"node_id": 1u64, "from": null, "to": "speculative"}),
161            ),
162            make_event(
163                2,
164                EVENT_KIND_CAUSAL_NODE_STATE,
165                serde_json::json!({"node_id": 1u64, "from": "speculative", "to": "frontier"}),
166            ),
167            make_event(
168                3,
169                EVENT_KIND_CAUSAL_NODE_STATE,
170                serde_json::json!({"node_id": 1u64, "from": "frontier", "to": "committed"}),
171            ),
172            make_event(
173                4,
174                EVENT_KIND_CAUSAL_NODE_STATE,
175                serde_json::json!({"node_id": 2u64, "from": null, "to": "frontier"}),
176            ),
177        ];
178        let map = fold_node_states(&events);
179        assert_eq!(map.get(&1).map(String::as_str), Some("committed"));
180        assert_eq!(map.get(&2).map(String::as_str), Some("frontier"));
181    }
182
183    #[test]
184    fn fold_remove_drops_node() {
185        let events = vec![
186            make_event(
187                1,
188                EVENT_KIND_CAUSAL_NODE_STATE,
189                serde_json::json!({"node_id": 5u64, "from": null, "to": "committed"}),
190            ),
191            make_event(
192                2,
193                EVENT_KIND_CAUSAL_NODE_REMOVE,
194                serde_json::json!({"node_id": 5u64, "label": "x"}),
195            ),
196        ];
197        let map = fold_node_states(&events);
198        assert!(map.is_empty());
199    }
200
201    /// End-to-end: CausalGraph.set_node_state emits chain events; fold reconstructs.
202    #[test]
203    fn emit_and_fold_round_trip() {
204        let cm = Arc::new(ChainManager::new(1, 1000));
205        let mut g = CausalGraph::new();
206        g.set_chain_manager(Arc::clone(&cm));
207
208        let id = g.add_node("turn-1".into(), serde_json::json!({"conv_id": "c1"}));
209        assert!(g.set_node_state(id, "frontier"));
210        assert!(g.set_node_state(id, "committed"));
211        assert!(g.set_node_state(id, "stale"));
212        // No-op: same state must not emit another event.
213        let len_before = cm.len();
214        assert!(g.set_node_state(id, "stale"));
215        assert_eq!(cm.len(), len_before, "identical state must not append");
216
217        let map = fold_node_states_from_chain(&cm);
218        assert_eq!(
219            map.get(&id).map(String::as_str),
220            Some("stale"),
221            "fold must recover final lifecycle state"
222        );
223
224        // Live graph metadata agrees with the fold.
225        let node = g.get_node(id).expect("node");
226        assert_eq!(
227            node.metadata.get("state").and_then(|v| v.as_str()),
228            Some("stale")
229        );
230
231        // Scrub to the sequence of the *first* state event → should be "frontier".
232        let all = cm.tail(cm.len());
233        let first_state_seq = all
234            .iter()
235            .find(|e| e.kind == EVENT_KIND_CAUSAL_NODE_STATE)
236            .map(|e| e.sequence)
237            .expect("at least one state event");
238        let at_first = fold_node_states_until(&cm, first_state_seq);
239        assert_eq!(
240            at_first.get(&id).map(String::as_str),
241            Some("frontier"),
242            "scrub to first state event"
243        );
244    }
245
246    #[test]
247    fn set_node_state_without_chain_manager_still_writes_metadata() {
248        let g = CausalGraph::new();
249        let id = g.add_node("n".into(), serde_json::json!({}));
250        assert!(g.set_node_state(id, "committed"));
251        assert_eq!(
252            g.get_node(id)
253                .unwrap()
254                .metadata
255                .get("state")
256                .and_then(|v| v.as_str()),
257            Some("committed")
258        );
259    }
260}
clawft_kernel/causal_state_fold.rs

clawft_kernel/
causal_state_fold.rs
