perspt_sdk/
workgraph.rs

1//! Mutable, revisioned work graph (PSP-8 System 4).
2//!
3//! Each graph revision is acyclic, but the session as a whole may add, retire,
4//! split, merge, update, or requeue nodes as verifier evidence changes. Updating
5//! a node creates a new [`WorkNode`] generation rather than mutating it in place;
6//! retired generations remain in the ledger but are no longer executable.
7
8use std::collections::{BTreeSet, HashMap, HashSet};
9
10use serde::{Deserialize, Serialize};
11
12use crate::error::{Result, SdkError};
13use crate::residual::ResidualEventRef;
14
15/// The kind of work a node performs.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum NodeClass {
19    Explore,
20    Plan,
21    Implement,
22    Verify,
23    Test,
24    Integrate,
25    Repair,
26    Interface,
27}
28
29/// Execution state of a node generation.
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
31#[serde(rename_all = "snake_case")]
32pub enum WorkNodeState {
33    /// Waiting on dependencies, sensors, or leases.
34    Pending,
35    /// Eligible for dispatch.
36    Ready,
37    /// Currently executing.
38    Running,
39    /// Accepted (hard pass or descent).
40    Stable,
41    /// Stopped with a residual certificate.
42    Stopped { certificate_id: String },
43    /// Superseded by a newer generation.
44    Retired { reason: String },
45    /// Blocked on a missing or degraded required sensor.
46    BlockedOnSensor { sensor: String },
47}
48
49/// Typed edge semantics (PSP-8 System 4).
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
51#[serde(rename_all = "snake_case")]
52pub enum EdgeKind {
53    /// Destination reads a file produced by source.
54    RequiresArtifact,
55    /// Destination relies on a sealed signature/schema/symbol.
56    RequiresInterface,
57    /// Destination test node validates source implementation node.
58    Tests,
59    /// Destination reconciles a cross-node/domain/adapter boundary.
60    Integrates,
61    /// Nodes touch non-commuting durable state and must serialize.
62    ConflictsWith,
63    /// Graph-rewrite lineage edge for audit.
64    DerivedFrom,
65    /// Node cannot execute until a required sensor exists or is downgraded.
66    BlocksOnSensor,
67}
68
69impl EdgeKind {
70    /// Whether this edge imposes an execution-ordering dependency (the source
71    /// must reach a stable state before the destination is ready). `DerivedFrom`
72    /// is audit-only and `ConflictsWith` is handled by footprint serialization,
73    /// not readiness.
74    pub fn is_dependency(self) -> bool {
75        matches!(
76            self,
77            EdgeKind::RequiresArtifact
78                | EdgeKind::RequiresInterface
79                | EdgeKind::Tests
80                | EdgeKind::Integrates
81        )
82    }
83}
84
85/// A directed edge in a graph revision.
86#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
87pub struct WorkEdge {
88    pub src: String,
89    pub dst: String,
90    pub kind: EdgeKind,
91}
92
93impl WorkEdge {
94    pub fn new(src: impl Into<String>, dst: impl Into<String>, kind: EdgeKind) -> Self {
95        Self {
96            src: src.into(),
97            dst: dst.into(),
98            kind,
99        }
100    }
101}
102
103/// An immutable incarnation of a node (PSP-8 `WorkNode`).
104#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
105pub struct WorkNode {
106    pub node_id: String,
107    pub generation: u32,
108    pub goal: String,
109    pub node_class: NodeClass,
110    pub owner_domains: Vec<String>,
111    pub output_targets: Vec<String>,
112    /// Required sensors that gate readiness (mapped from `BlocksOnSensor`).
113    pub required_sensors: Vec<String>,
114    pub state: WorkNodeState,
115}
116
117impl WorkNode {
118    pub fn new(node_id: impl Into<String>, goal: impl Into<String>, node_class: NodeClass) -> Self {
119        Self {
120            node_id: node_id.into(),
121            generation: 0,
122            goal: goal.into(),
123            node_class,
124            owner_domains: Vec::new(),
125            output_targets: Vec::new(),
126            required_sensors: Vec::new(),
127            state: WorkNodeState::Pending,
128        }
129    }
130
131    pub fn with_outputs(mut self, outputs: Vec<String>) -> Self {
132        self.output_targets = outputs;
133        self
134    }
135
136    /// Create the next generation of this node, resetting it to pending.
137    pub fn next_generation(&self) -> Self {
138        Self {
139            generation: self.generation + 1,
140            state: WorkNodeState::Pending,
141            ..self.clone()
142        }
143    }
144
145    pub fn is_terminal(&self) -> bool {
146        matches!(
147            self.state,
148            WorkNodeState::Stable | WorkNodeState::Stopped { .. } | WorkNodeState::Retired { .. }
149        )
150    }
151
152    pub fn is_accepted(&self) -> bool {
153        matches!(self.state, WorkNodeState::Stable)
154    }
155}
156
157/// Why a graph revision was produced.
158#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
159#[serde(rename_all = "snake_case")]
160pub enum GraphRevisionReason {
161    InitialPlan,
162    ExecutionUpdate,
163    LocalRepair,
164    ScopeExpansion,
165    UserEdit,
166    Replan,
167}
168
169/// Result of validating a revision before activation.
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct GraphValidationReport {
172    pub acyclic: bool,
173    /// A topological order over dependency edges, if acyclic.
174    pub topo_order: Vec<String>,
175    pub dangling_edges: Vec<WorkEdge>,
176}
177
178/// A durable version of the work graph (PSP-8 `WorkGraphRevision`).
179#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
180pub struct WorkGraphRevision {
181    pub revision_id: String,
182    pub sequence: u32,
183    pub parent_revision_id: Option<String>,
184    pub reason: GraphRevisionReason,
185    pub nodes: Vec<WorkNode>,
186    pub edges: Vec<WorkEdge>,
187    pub validation: GraphValidationReport,
188    pub evidence: Vec<ResidualEventRef>,
189}
190
191/// One deterministic edit applied to an immutable graph revision. Coding work
192/// is represented by an evolving sequence of acyclic snapshots, not by the
193/// fiction that the initial DAG is complete.
194#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
195#[serde(tag = "edit", rename_all = "snake_case")]
196pub enum GraphEdit {
197    AddNode { node: WorkNode },
198    ReplaceNode { node: WorkNode },
199    RetireNode { node_id: String, reason: String },
200    AddEdge { edge: WorkEdge },
201    RemoveEdge { edge: WorkEdge },
202}
203
204impl WorkGraphRevision {
205    /// Build and validate a revision. Returns an error if it is not acyclic over
206    /// dependency edges; every revision SHALL validate acyclicity before
207    /// activation (PSP-8 System 4).
208    pub fn build(
209        sequence: u32,
210        parent_revision_id: Option<String>,
211        reason: GraphRevisionReason,
212        nodes: Vec<WorkNode>,
213        edges: Vec<WorkEdge>,
214    ) -> Result<Self> {
215        let validation = validate(&nodes, &edges)?;
216        if !validation.acyclic {
217            return Err(SdkError::Domain("graph revision is not acyclic".into()));
218        }
219        if !validation.dangling_edges.is_empty() {
220            return Err(SdkError::Domain(format!(
221                "graph revision has {} dangling edge(s)",
222                validation.dangling_edges.len()
223            )));
224        }
225        Ok(Self {
226            revision_id: uuid::Uuid::new_v4().to_string(),
227            sequence,
228            parent_revision_id,
229            reason,
230            nodes,
231            edges,
232            validation,
233            evidence: Vec::new(),
234        })
235    }
236
237    pub fn node(&self, node_id: &str) -> Option<&WorkNode> {
238        self.nodes.iter().find(|n| n.node_id == node_id)
239    }
240
241    /// Dependency predecessors of a node (sources of dependency edges into it).
242    pub fn dependencies_of(&self, node_id: &str) -> Vec<&str> {
243        self.edges
244            .iter()
245            .filter(|e| e.dst == node_id && e.kind.is_dependency())
246            .map(|e| e.src.as_str())
247            .collect()
248    }
249
250    /// Nodes that conflict with the given node via an explicit `ConflictsWith`
251    /// edge (in addition to footprint-derived conflicts handled by the
252    /// scheduler).
253    pub fn explicit_conflicts_of(&self, node_id: &str) -> Vec<&str> {
254        self.edges
255            .iter()
256            .filter(|e| e.kind == EdgeKind::ConflictsWith && (e.src == node_id || e.dst == node_id))
257            .map(|e| {
258                if e.src == node_id {
259                    e.dst.as_str()
260                } else {
261                    e.src.as_str()
262                }
263            })
264            .collect()
265    }
266
267    /// Apply evidence-backed edits and validate the complete next snapshot
268    /// before returning it. The current revision remains usable if any edit
269    /// would introduce a dangling edge, duplicate node, stale generation, or
270    /// dependency cycle.
271    pub fn revise(
272        &self,
273        reason: GraphRevisionReason,
274        edits: &[GraphEdit],
275        evidence: Vec<ResidualEventRef>,
276    ) -> Result<Self> {
277        if edits.is_empty() {
278            return Err(SdkError::Domain("graph revision has no edits".into()));
279        }
280        let mut nodes = self.nodes.clone();
281        let mut edges = self.edges.clone();
282        for edit in edits {
283            match edit {
284                GraphEdit::AddNode { node } => add_node(&mut nodes, node)?,
285                GraphEdit::ReplaceNode { node } => {
286                    let existing = nodes
287                        .iter_mut()
288                        .find(|existing| existing.node_id == node.node_id)
289                        .ok_or_else(|| {
290                            SdkError::Domain(format!("unknown graph node {:?}", node.node_id))
291                        })?;
292                    if node.generation < existing.generation {
293                        return Err(SdkError::Domain(format!(
294                            "stale generation {} for node {:?} at generation {}",
295                            node.generation, node.node_id, existing.generation
296                        )));
297                    }
298                    // A generation names one immutable incarnation: content may
299                    // only change under a strictly newer generation.
300                    if node.generation == existing.generation && *existing != *node {
301                        return Err(SdkError::Domain(format!(
302                            "node {:?} changed without a new generation",
303                            node.node_id
304                        )));
305                    }
306                    *existing = node.clone();
307                }
308                GraphEdit::RetireNode { node_id, reason } => {
309                    let node = nodes
310                        .iter_mut()
311                        .find(|node| &node.node_id == node_id)
312                        .ok_or_else(|| {
313                            SdkError::Domain(format!("unknown graph node {node_id:?}"))
314                        })?;
315                    node.state = WorkNodeState::Retired {
316                        reason: reason.clone(),
317                    };
318                    // A retired node never runs, so leaving its edges in place
319                    // would deadlock every dependent (the scheduler requires
320                    // accepted dependencies). Retirement releases its edges.
321                    edges.retain(|edge| edge.src != *node_id && edge.dst != *node_id);
322                }
323                GraphEdit::AddEdge { edge } => {
324                    if !edges.contains(edge) {
325                        edges.push(edge.clone());
326                    }
327                }
328                GraphEdit::RemoveEdge { edge } => edges.retain(|existing| existing != edge),
329            }
330        }
331        let mut revision = Self::build(
332            self.sequence.saturating_add(1),
333            Some(self.revision_id.clone()),
334            reason,
335            nodes,
336            edges,
337        )?;
338        revision.evidence = evidence;
339        Ok(revision)
340    }
341}
342
343fn add_node(nodes: &mut Vec<WorkNode>, node: &WorkNode) -> Result<()> {
344    if nodes
345        .iter()
346        .any(|existing| existing.node_id == node.node_id)
347    {
348        return Err(SdkError::Domain(format!(
349            "graph already contains node {:?}",
350            node.node_id
351        )));
352    }
353    nodes.push(node.clone());
354    Ok(())
355}
356
357/// Validate a node/edge set: check that all edge endpoints exist and that the
358/// dependency subgraph is acyclic (Kahn's algorithm).
359pub fn validate(nodes: &[WorkNode], edges: &[WorkEdge]) -> Result<GraphValidationReport> {
360    let ids: HashSet<&str> = nodes.iter().map(|n| n.node_id.as_str()).collect();
361    let dangling: Vec<WorkEdge> = edges
362        .iter()
363        .filter(|e| !ids.contains(e.src.as_str()) || !ids.contains(e.dst.as_str()))
364        .cloned()
365        .collect();
366
367    // Kahn's algorithm over dependency edges only.
368    let mut indegree: HashMap<&str, usize> =
369        nodes.iter().map(|n| (n.node_id.as_str(), 0)).collect();
370    let mut adj: HashMap<&str, Vec<&str>> = HashMap::new();
371    for e in edges.iter().filter(|e| e.kind.is_dependency()) {
372        if !ids.contains(e.src.as_str()) || !ids.contains(e.dst.as_str()) {
373            continue;
374        }
375        adj.entry(e.src.as_str()).or_default().push(e.dst.as_str());
376        *indegree.get_mut(e.dst.as_str()).unwrap() += 1;
377    }
378
379    let mut queue: BTreeSet<&str> = indegree
380        .iter()
381        .filter(|(_, &d)| d == 0)
382        .map(|(&n, _)| n)
383        .collect();
384    let mut topo_order = Vec::new();
385    while let Some(&n) = queue.iter().next() {
386        queue.remove(n);
387        topo_order.push(n.to_string());
388        if let Some(succs) = adj.get(n) {
389            for &s in succs {
390                let d = indegree.get_mut(s).unwrap();
391                *d -= 1;
392                if *d == 0 {
393                    queue.insert(s);
394                }
395            }
396        }
397    }
398
399    let acyclic = topo_order.len() == nodes.len();
400    Ok(GraphValidationReport {
401        acyclic,
402        topo_order: if acyclic { topo_order } else { Vec::new() },
403        dangling_edges: dangling,
404    })
405}
406
407#[cfg(test)]
408mod tests {
409    use super::*;
410
411    fn node(id: &str) -> WorkNode {
412        WorkNode::new(id, format!("goal {id}"), NodeClass::Implement)
413    }
414
415    #[test]
416    fn acyclic_graph_validates_with_topo_order() {
417        let nodes = vec![node("a"), node("b"), node("c")];
418        let edges = vec![
419            WorkEdge::new("a", "b", EdgeKind::RequiresArtifact),
420            WorkEdge::new("b", "c", EdgeKind::RequiresInterface),
421        ];
422        let rev = WorkGraphRevision::build(0, None, GraphRevisionReason::InitialPlan, nodes, edges)
423            .unwrap();
424        assert!(rev.validation.acyclic);
425        assert_eq!(rev.validation.topo_order, vec!["a", "b", "c"]);
426    }
427
428    #[test]
429    fn cyclic_graph_is_rejected() {
430        let nodes = vec![node("a"), node("b")];
431        let edges = vec![
432            WorkEdge::new("a", "b", EdgeKind::RequiresArtifact),
433            WorkEdge::new("b", "a", EdgeKind::RequiresArtifact),
434        ];
435        assert!(
436            WorkGraphRevision::build(0, None, GraphRevisionReason::InitialPlan, nodes, edges)
437                .is_err()
438        );
439    }
440
441    #[test]
442    fn dangling_edge_is_rejected() {
443        let nodes = vec![node("a")];
444        let edges = vec![WorkEdge::new("a", "ghost", EdgeKind::RequiresArtifact)];
445        assert!(
446            WorkGraphRevision::build(0, None, GraphRevisionReason::InitialPlan, nodes, edges)
447                .is_err()
448        );
449    }
450
451    #[test]
452    fn conflicts_with_does_not_create_dependency_cycle() {
453        // ConflictsWith is bidirectional but is not a dependency edge, so a pair
454        // of conflicting nodes is still a valid acyclic revision.
455        let nodes = vec![node("a"), node("b")];
456        let edges = vec![WorkEdge::new("a", "b", EdgeKind::ConflictsWith)];
457        let rev = WorkGraphRevision::build(0, None, GraphRevisionReason::InitialPlan, nodes, edges)
458            .unwrap();
459        assert!(rev.validation.acyclic);
460        assert_eq!(rev.explicit_conflicts_of("a"), vec!["b"]);
461    }
462
463    #[test]
464    fn next_generation_resets_state() {
465        let mut n = node("a");
466        n.state = WorkNodeState::Stable;
467        let g1 = n.next_generation();
468        assert_eq!(g1.generation, 1);
469        assert_eq!(g1.state, WorkNodeState::Pending);
470    }
471
472    #[test]
473    fn dependencies_exclude_audit_and_conflict_edges() {
474        let nodes = vec![node("a"), node("b")];
475        let edges = vec![
476            WorkEdge::new("a", "b", EdgeKind::DerivedFrom),
477            WorkEdge::new("a", "b", EdgeKind::ConflictsWith),
478        ];
479        let rev = WorkGraphRevision::build(0, None, GraphRevisionReason::LocalRepair, nodes, edges)
480            .unwrap();
481        assert!(rev.dependencies_of("b").is_empty());
482    }
483
484    #[test]
485    fn evolving_revisions_can_add_work_after_execution_starts() {
486        let original = WorkGraphRevision::build(
487            0,
488            None,
489            GraphRevisionReason::InitialPlan,
490            vec![node("a")],
491            vec![],
492        )
493        .unwrap();
494        let revised = original
495            .revise(
496                GraphRevisionReason::LocalRepair,
497                &[
498                    GraphEdit::AddNode { node: node("b") },
499                    GraphEdit::AddEdge {
500                        edge: WorkEdge::new("a", "b", EdgeKind::RequiresArtifact),
501                    },
502                ],
503                vec![],
504            )
505            .unwrap();
506        assert_eq!(revised.parent_revision_id, Some(original.revision_id));
507        assert_eq!(revised.sequence, 1);
508        assert_eq!(revised.validation.topo_order, vec!["a", "b"]);
509    }
510
511    #[test]
512    fn a_failed_dynamic_revision_leaves_the_parent_valid() {
513        let original = WorkGraphRevision::build(
514            0,
515            None,
516            GraphRevisionReason::InitialPlan,
517            vec![node("a"), node("b")],
518            vec![WorkEdge::new("a", "b", EdgeKind::RequiresArtifact)],
519        )
520        .unwrap();
521        let result = original.revise(
522            GraphRevisionReason::LocalRepair,
523            &[GraphEdit::AddEdge {
524                edge: WorkEdge::new("b", "a", EdgeKind::RequiresArtifact),
525            }],
526            vec![],
527        );
528        assert!(result.is_err());
529        assert!(original.validation.acyclic);
530    }
531}
perspt_sdk/workgraph.rs

perspt_sdk/
workgraph.rs
