feat(kr0ki-core): Phase 5 — DecisionNode/ForkNode/JoinNode/MergeNode in ProcessChart

Cargo.toml

@@ -42,7 +42,7 @@ license = "MIT"
 repository = "https://github.com/PromptExecution/ledgrrr"
 
 [workspace.dependencies]
-ufo-types = { git = "https://github.com/promptexecution/ufo-types", rev = "1846e6f" }
+ufo-types = { git = "https://github.com/promptexecution/ufo-types", rev = "9f883e6" }
 rusqlite = { version = "=0.33.0", features = ["bundled", "hooks"] }
 rust_xlsxwriter = "0.94.0"
 serde = { version = "1", features = ["derive"] }
@@ -86,7 +86,6 @@ agentmesh = "=4.0.0"
 agentmesh-mcp = "=4.0.0"
 beankeeper = { git = "https://github.com/PromptExecution/beankeeper-b00t.git", branch = "feat/serde-feature", features = ["serde"] }
 ofx-rs = "0.2"
-lazily = { version = "0.59", features = ["statechart", "thread-safe"] }
 
 [workspace.lints.rust]
 unsafe_code = "deny"

crates/kr0ki-core/src/chart.rs

@@ -18,11 +18,23 @@ impl ProcessChart {
     /// Build a `ProcessChart` from a flat slice of `(ElementId, ElementKind)`
     /// element declarations and a slice of `Relation`s.
     ///
-    /// Mapping rules:
-    /// - `ActionDefinition` → compound root state (first one wins as root)
-    /// - `ActionUsage`       → atomic leaf state; parent = its `ActionDefinition`
-    ///   owner if a `FeatureMembership` relation is present, else root
-    /// - `Succession`        → `source.on("next", target)` transition
+    /// ## Mapping rules (Phase 1–5)
+    ///
+    /// | SysML-v2 element    | lazily state        | Events / guards                                        |
+    /// |---------------------|---------------------|--------------------------------------------------------|
+    /// | `ActionDefinition`  | compound root       | initial = first owned usage                            |
+    /// | `ActionUsage`       | atomic leaf         | `on("next", target)` per outgoing Succession           |
+    /// | `MergeNode`         | atomic              | `on("next", target)` — fan-in pass-through             |
+    /// | `DecisionNode`      | atomic              | `on_guarded("next", t, "choose:<t>")` per branch       |
+    /// | `ForkNode`          | atomic              | `on_guarded("next", t, "fork:<t>")` per branch         |
+    /// | `JoinNode`          | atomic              | `on("next", target)` — fan-out re-convergence          |
+    ///
+    /// **Phase 6 note:** True concurrent execution for ForkNode/JoinNode requires
+    /// lazily parallel-region construction and topology detection (detecting the
+    /// fork-join pair as a `StateBuilder::parallel` with per-branch child regions
+    /// and final states). Phase 5 models Fork/Join as guarded-branching atomic
+    /// states so the topology is representable and testable without full SCXML
+    /// parallel semantics.
     pub fn from_sysml(
         elements: &[(ElementId, ElementKind)],
         relations: &[Relation],
@@ -39,7 +51,13 @@ impl ProcessChart {
             .map(|(id, _)| id)
             .collect();
 
-        // Build ownership map: usage → definition
+        let control_nodes: Vec<(&ElementId, ElementKind)> = elements
+            .iter()
+            .filter(|(_, k)| k.is_control_node())
+            .map(|(id, k)| (id, *k))
+            .collect();
+
+        // Build ownership map: usage/node → definition
         let mut owner_of: HashMap<&str, &str> = HashMap::new();
         for rel in relations {
             if let Relation::FeatureMembership { owner, member } = rel {
@@ -66,10 +84,18 @@ impl ProcessChart {
                 first_child.entry(def_id).or_insert(usage_id);
             }
         }
+        // Also consider control nodes owned by a definition
+        for (node, _) in &control_nodes {
+            let node_id = node.0.as_str();
+            if let Some(def_id) = owner_of.get(node_id) {
+                first_child.entry(def_id).or_insert(node_id);
+            }
+        }
 
         let chart_def = Self::build_chart_def(
             &action_defs,
             &action_usages,
+            &control_nodes,
             &owner_of,
             &first_child,
             &successors,
@@ -84,6 +110,7 @@ impl ProcessChart {
     fn build_chart_def(
         defs: &[&ElementId],
         usages: &[&ElementId],
+        control_nodes: &[(&ElementId, ElementKind)],
         owner_of: &HashMap<&str, &str>,
         first_child: &HashMap<&str, &str>,
         successors: &HashMap<&str, Vec<&str>>,
@@ -129,14 +156,84 @@ impl ProcessChart {
             builder = builder.state(sb);
         }
 
+        // Control nodes — Phase 5 mapping (see doc comment on from_sysml)
+        for (node, kind) in control_nodes {
+            let node_id = node.0.as_str();
+            if !seen.insert(node_id) {
+                return Err(Kr0kiError::DuplicateId(node_id.to_string()));
+            }
+            let mut sb = StateBuilder::atomic(node_id);
+            if let Some(def_id) = owner_of.get(node_id) {
+                sb = sb.parent(*def_id);
+            }
+            match kind {
+                // MergeNode: same as ActionUsage — plain "next" transitions
+                ElementKind::MergeNode | ElementKind::JoinNode => {
+                    if let Some(targets) = successors.get(node_id) {
+                        for &tgt in targets {
+                            sb = sb.on("next", tgt);
+                        }
+                    }
+                }
+                // DecisionNode: one distinct event per branch — "choose:<target>".
+                // lazily stores one transition per event key, so guards on the same
+                // "next" event would silently overwrite each other. Named events are
+                // unambiguous and work with the single-transition-per-event constraint.
+                // Phase 6 note: replace with proper SCXML event-less transitions +
+                // condition expressions once lazily gains Vec<Transition> per event.
+                ElementKind::DecisionNode => {
+                    if let Some(targets) = successors.get(node_id) {
+                        match targets.as_slice() {
+                            [only] => sb = sb.on("next", *only),
+                            many => {
+                                for &tgt in many {
+                                    let event = format!("choose:{tgt}");
+                                    sb = sb.on(event, tgt);
+                                }
+                            }
+                        }
+                    }
+                }
+                // ForkNode: one distinct event per branch — "fork:<target>".
+                // Same rationale as DecisionNode above.
+                // Phase 6: replace with StateBuilder::parallel + child regions.
+                ElementKind::ForkNode => {
+                    if let Some(targets) = successors.get(node_id) {
+                        match targets.as_slice() {
+                            [only] => sb = sb.on("next", *only),
+                            many => {
+                                for &tgt in many {
+                                    let event = format!("fork:{tgt}");
+                                    sb = sb.on(event, tgt);
+                                }
+                            }
+                        }
+                    }
+                }
+                _ => unreachable!("control_nodes filtered to is_control_node()"),
+            }
+            builder = builder.state(sb);
+        }
+
         builder.build().map_err(Kr0kiError::ChartBuild)
     }
 
-    /// Send event `"next"` to advance the chart. Returns `true` if a
-    /// transition fired. `guards` maps guard name → bool (pass empty for
-    /// guard-free charts).
-    pub fn advance(&self, guards: &HashMap<String, bool>) -> bool {
-        self.chart.send(&self.ctx, "next", guards)
+    /// Send event `"next"` to advance the chart (unguarded succession).
+    pub fn advance(&self) -> bool {
+        self.chart.send(&self.ctx, "next", &HashMap::new())
+    }
+
+    /// Route a `DecisionNode` to `target_id` by sending `"choose:<target_id>"`.
+    pub fn decide(&self, target_id: &str) -> bool {
+        let event = format!("choose:{target_id}");
+        self.chart.send(&self.ctx, &event, &HashMap::new())
+    }
+
+    /// Route a `ForkNode` to one branch `target_id` by sending `"fork:<target_id>"`.
+    /// Phase 6: replace with parallel entry once lazily supports Vec<Transition>/event.
+    pub fn fork_to(&self, target_id: &str) -> bool {
+        let event = format!("fork:{target_id}");
+        self.chart.send(&self.ctx, &event, &HashMap::new())
     }
 
     /// Current active leaf state ids.
@@ -191,7 +288,7 @@ mod tests {
             .expect("chart build should succeed");
 
         assert!(chart.is_active("step_a"), "initial state must be step_a");
-        let fired = chart.advance(&HashMap::new());
+        let fired = chart.advance();
         assert!(fired, "transition must fire on 'next'");
         assert!(chart.is_active("step_b"), "active leaf must be step_b after advance");
     }
@@ -217,9 +314,9 @@ mod tests {
             .expect("chart build");
 
         assert!(chart.is_active("a"));
-        chart.advance(&HashMap::new());
+        chart.advance();
         assert!(chart.is_active("b"));
-        chart.advance(&HashMap::new());
+        chart.advance();
         assert!(chart.is_active("c"));
     }
 
@@ -232,11 +329,140 @@ mod tests {
         ];
         let relations = vec![];
 
-        // No initial child → root becomes atomic; orphan has no parent.
-        // ChartBuilder should accept this (two disjoint roots treated as parallel-capable).
-        // We just assert it doesn't error.
         let result = ProcessChart::from_sysml(&elements, &relations);
-        // May succeed or fail depending on lazily's root validation; we accept either.
         let _ = result;
     }
+
+    /// MergeNode: two branches converge at a merge, then continue to "done".
+    /// Topology: a --next--> merge; b --next--> merge; merge --next--> done
+    /// Starting in "a" (first child of proc), advancing routes a→merge→done.
+    #[test]
+    fn merge_node_pass_through() {
+        let elements = vec![
+            (eid("proc"), ElementKind::ActionDefinition),
+            (eid("a"), ElementKind::ActionUsage),
+            (eid("b"), ElementKind::ActionUsage),
+            (eid("merge"), ElementKind::MergeNode),
+            (eid("done"), ElementKind::ActionUsage),
+        ];
+        let relations = vec![
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("a") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("b") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("merge") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("done") },
+            Relation::Succession { source: eid("a"), target: eid("merge") },
+            Relation::Succession { source: eid("b"), target: eid("merge") },
+            Relation::Succession { source: eid("merge"), target: eid("done") },
+        ];
+
+        let chart = ProcessChart::from_sysml(&elements, &relations).expect("merge chart builds");
+        // "a" is first child → initial
+        assert!(chart.is_active("a"));
+        chart.advance(); // a → merge
+        assert!(chart.is_active("merge"));
+        chart.advance(); // merge → done
+        assert!(chart.is_active("done"));
+    }
+
+    /// DecisionNode: two branches — route to "branch_yes" via `decide()`.
+    /// Start at "step_a", advance to decision, then call `decide("branch_yes")`.
+    #[test]
+    fn decision_node_routing() {
+        let elements = vec![
+            (eid("proc"), ElementKind::ActionDefinition),
+            (eid("step_a"), ElementKind::ActionUsage),
+            (eid("decision"), ElementKind::DecisionNode),
+            (eid("branch_yes"), ElementKind::ActionUsage),
+            (eid("branch_no"), ElementKind::ActionUsage),
+        ];
+        let relations = vec![
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("step_a") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("decision") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("branch_yes") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("branch_no") },
+            Relation::Succession { source: eid("step_a"), target: eid("decision") },
+            Relation::Succession { source: eid("decision"), target: eid("branch_yes") },
+            Relation::Succession { source: eid("decision"), target: eid("branch_no") },
+        ];
+
+        let chart = ProcessChart::from_sysml(&elements, &relations)
+            .expect("decision chart builds");
+
+        assert!(chart.is_active("step_a"));
+        chart.advance(); // step_a → decision
+        assert!(chart.is_active("decision"));
+
+        let fired = chart.decide("branch_yes");
+        assert!(fired, "decide(branch_yes) must fire");
+        assert!(chart.is_active("branch_yes"), "must be in branch_yes after decide");
+    }
+
+    /// ForkNode: two branches — route to branch "b" via `fork_to()`.
+    /// Advance through fork to one branch, then continue to join and end.
+    #[test]
+    fn fork_node_fanout() {
+        let elements = vec![
+            (eid("proc"), ElementKind::ActionDefinition),
+            (eid("start"), ElementKind::ActionUsage),
+            (eid("fork"), ElementKind::ForkNode),
+            (eid("b"), ElementKind::ActionUsage),
+            (eid("c"), ElementKind::ActionUsage),
+            (eid("join"), ElementKind::JoinNode),
+            (eid("end"), ElementKind::ActionUsage),
+        ];
+        let relations = vec![
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("start") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("fork") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("b") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("c") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("join") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("end") },
+            Relation::Succession { source: eid("start"), target: eid("fork") },
+            Relation::Succession { source: eid("fork"), target: eid("b") },
+            Relation::Succession { source: eid("fork"), target: eid("c") },
+            Relation::Succession { source: eid("b"), target: eid("join") },
+            Relation::Succession { source: eid("c"), target: eid("join") },
+            Relation::Succession { source: eid("join"), target: eid("end") },
+        ];
+
+        let chart = ProcessChart::from_sysml(&elements, &relations)
+            .expect("fork/join chart builds");
+
+        assert!(chart.is_active("start"));
+        chart.advance(); // start → fork
+        assert!(chart.is_active("fork"));
+
+        let fired = chart.fork_to("b");
+        assert!(fired, "fork_to(b) must fire");
+        assert!(chart.is_active("b"), "must be in branch b");
+        chart.advance(); // b → join
+        assert!(chart.is_active("join"));
+        chart.advance(); // join → end
+        assert!(chart.is_active("end"));
+    }
+
+    /// JoinNode with single outgoing Succession (plain "next").
+    #[test]
+    fn join_node_single_successor() {
+        let elements = vec![
+            (eid("proc"), ElementKind::ActionDefinition),
+            (eid("a"), ElementKind::ActionUsage),
+            (eid("join"), ElementKind::JoinNode),
+            (eid("done"), ElementKind::ActionUsage),
+        ];
+        let relations = vec![
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("a") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("join") },
+            Relation::FeatureMembership { owner: eid("proc"), member: eid("done") },
+            Relation::Succession { source: eid("a"), target: eid("join") },
+            Relation::Succession { source: eid("join"), target: eid("done") },
+        ];
+
+        let chart = ProcessChart::from_sysml(&elements, &relations).expect("join chart builds");
+        assert!(chart.is_active("a"));
+        chart.advance();
+        assert!(chart.is_active("join"));
+        chart.advance();
+        assert!(chart.is_active("done"));
+    }
 }

crates/ledgerr-cloud/src/reconcile.rs

@@ -75,17 +75,47 @@ impl ReconcileRunner {
 
     /// Run all four providers. Never panics or short-circuits on a single
     /// provider's error — every provider gets its own [`ProviderStatus`].
+    ///
+    /// Each provider call is wrapped in a [`PROVIDER_TIMEOUT_SECS`]-second
+    /// timeout so a hanging CLI (e.g. `gcloud auth print-access-token` with
+    /// no credentials on a machine that has the SDK installed but no auth
+    /// context) maps to a `fail` entry rather than hanging the whole run.
     pub async fn run(&self) -> ReconcileReport {
         let providers = vec![
-            run_one("aws", &self.aws).await,
-            run_one("gcp", &self.gcp).await,
-            run_one("azure", &self.azure).await,
-            run_one("hf", &self.hf).await,
+            run_timed("aws", &self.aws).await,
+            run_timed("gcp", &self.gcp).await,
+            run_timed("azure", &self.azure).await,
+            run_timed("hf", &self.hf).await,
         ];
         ReconcileReport { providers }
     }
 }
 
+/// Per-provider wall-clock timeout. CLI calls (`gcloud`, `az`) can hang
+/// indefinitely when installed but unconfigured; this bound keeps the run
+/// from blocking the MCP server.
+const PROVIDER_TIMEOUT_SECS: u64 = 8;
+
+async fn run_timed<P: BudgetProvider>(name: &str, provider: &P) -> ProviderStatus {
+    match tokio::time::timeout(
+        std::time::Duration::from_secs(PROVIDER_TIMEOUT_SECS),
+        run_one(name, provider),
+    )
+    .await
+    {
+        Ok(status) => status,
+        Err(_elapsed) => ProviderStatus {
+            provider: name.to_string(),
+            status: ReconcileStatus::Fail,
+            cap_cake: None,
+            authoritative: false,
+            error: Some(format!(
+                "{name} budget check timed out after {PROVIDER_TIMEOUT_SECS}s"
+            )),
+        },
+    }
+}
+
 /// Run a single provider's `check_auth` + `fetch_budget`, catching every
 /// `Result` into a [`ProviderStatus`]. Generic over `P: BudgetProvider`
 /// rather than `dyn` — the trait's `impl Future` return position is not