lean_ctx_workgraph/
work_graph.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Bounded Work Graph for multi-agent orchestration (P11 / DIM 4).
4//!
5//! Manages parent/child agent delegation with:
6//! - Budget inheritance (child cannot exceed parent)
7//! - Fan-out limits (max concurrent children)
8//! - Stop conditions (stale, over-budget, redundant)
9//! - Provenance tracking for attribution
10
11use std::collections::{BTreeMap, BTreeSet};
12
13use crate::budget_cascade::{BudgetAllocation, CascadeError, cascade_budget, validate_cascade};
14use serde::{Deserialize, Serialize};
15
16pub const WORK_GRAPH_SCHEMA_VERSION: u16 = 1;
17/// Upper bound of persisted delivery attempts within one fence. New starts use
18/// attempt 1; stores written by earlier releases may still carry attempt 2.
19pub const MAX_DELIVERY_ATTEMPTS: u8 = 2;
20const MAX_GRAPH_NODES: usize = 256;
21const MAX_FAN_OUT: usize = 16;
22const MAX_DEPTH: u16 = 8;
23const MAX_LOCAL_CONCURRENCY: usize = 32;
24const MAX_IDENTIFIER_BYTES: usize = 128;
25const MAX_RESULT_CLAIMS: usize = 64;
26const MAX_PATH_CLAIMS: usize = 64;
27
28fn default_local_concurrency() -> usize {
29    MAX_FAN_OUT
30}
31
32#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34pub enum ChildOutcome {
35    Accepted,
36    Rejected,
37    Partial,
38}
39
40#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
41pub struct ResultClaim {
42    pub key: String,
43    pub value: String,
44}
45
46#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
47pub struct ChildExecutionReceipt {
48    pub receipt_id: String,
49    pub node_id: String,
50    pub outcome_ref: String,
51    pub outcome: ChildOutcome,
52    pub tokens_consumed: u64,
53    pub cost_micros_consumed: u64,
54    /// Opaque claim fence returned when queued work becomes active.
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub execution_fence: Option<String>,
57    #[serde(default)]
58    pub claims: Vec<ResultClaim>,
59    /// Who established `tokens_consumed`/`cost_micros_consumed`.
60    #[serde(default, skip_serializing_if = "UsageSource::is_measured")]
61    pub usage_source: UsageSource,
62}
63
64/// Provenance of a receipt's consumption figures. Receipts from earlier
65/// releases carry connector-measured usage; host-run nodes report their own,
66/// which stays bounded by the node budget either way.
67#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
68#[serde(rename_all = "snake_case")]
69pub enum UsageSource {
70    #[default]
71    Measured,
72    Reported,
73}
74
75impl UsageSource {
76    fn is_measured(&self) -> bool {
77        *self == Self::Measured
78    }
79}
80
81/// Canonically verified provider spend that exceeded its authorization.
82///
83/// This is deliberately separate from [`WorkNodeBudget`]: authorization stays
84/// bounded while attribution preserves the provider-reported fact in full.
85#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
86pub struct MeasuredExecutionSpend {
87    pub spend_id: String,
88    pub node_id: String,
89    pub outcome_refs: Vec<String>,
90    pub tokens_consumed: u64,
91    pub cost_micros_consumed: u64,
92    pub execution_fence: String,
93}
94
95impl MeasuredExecutionSpend {
96    pub fn new(
97        node_id: String,
98        outcome_refs: Vec<String>,
99        tokens_consumed: u64,
100        cost_micros_consumed: u64,
101        execution_fence: String,
102    ) -> Self {
103        let spend_id = measured_spend_id(&outcome_refs);
104        Self {
105            spend_id,
106            node_id,
107            outcome_refs,
108            tokens_consumed,
109            cost_micros_consumed,
110            execution_fence,
111        }
112    }
113}
114
115#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
116pub struct ResultConflict {
117    pub key: String,
118    pub values: Vec<String>,
119    pub node_ids: Vec<String>,
120}
121
122#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
123pub struct FusedResults {
124    pub accepted: BTreeMap<String, String>,
125    pub conflicts: Vec<ResultConflict>,
126}
127
128#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
129pub struct WorkAttribution {
130    pub accepted_tokens: u64,
131    pub accepted_cost_micros: u64,
132    pub waste_tokens: u64,
133    pub waste_cost_micros: u64,
134}
135
136#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
137pub struct TeamValueReport {
138    pub accepted_tokens: u64,
139    pub total_tokens: u64,
140    pub accepted_cost_micros: u64,
141    pub total_cost_micros: u64,
142    pub waste_tokens: u64,
143    pub waste_cost_micros: u64,
144    pub accepted_cost_basis_points: u16,
145}
146
147#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
148#[serde(rename_all = "snake_case")]
149pub enum NodeStatus {
150    Pending,
151    Active,
152    Completed,
153    Stopped,
154    Failed,
155}
156
157#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
158#[serde(rename_all = "snake_case")]
159pub enum StopReason {
160    BudgetExhausted,
161    Stale,
162    Redundant,
163    ParentStopped,
164    ManualStop,
165    DepthExceeded,
166    FanOutExceeded,
167    PolicyDenied,
168    LeaseLost,
169    Duplicate,
170    LowValue,
171    ExecutionFailed,
172}
173
174#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
175pub struct WorkNodeBudget {
176    pub tokens_allocated: u64,
177    pub tokens_consumed: u64,
178    pub cost_micros_allocated: u64,
179    pub cost_micros_consumed: u64,
180}
181
182impl WorkNodeBudget {
183    pub fn tokens_remaining(&self) -> u64 {
184        self.tokens_allocated.saturating_sub(self.tokens_consumed)
185    }
186
187    pub fn cost_remaining(&self) -> u64 {
188        self.cost_micros_allocated
189            .saturating_sub(self.cost_micros_consumed)
190    }
191
192    pub fn is_exhausted(&self) -> bool {
193        self.tokens_remaining() == 0 || self.cost_remaining() == 0
194    }
195}
196
197/// Tracks the total budget across an entire delegation chain.
198#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
199pub struct ChainBudget {
200    pub chain_id: String,
201    pub root_budget_tokens: u64,
202    pub total_consumed_tokens: u64,
203    pub total_allocated_tokens: u64,
204    #[serde(default)]
205    pub root_budget_cost_micros: u64,
206    #[serde(default)]
207    pub total_consumed_cost_micros: u64,
208    #[serde(default)]
209    pub total_allocated_cost_micros: u64,
210    /// Verified provider spend above node authorization; never reusable.
211    #[serde(default)]
212    pub unauthorized_tokens: u64,
213    #[serde(default)]
214    pub unauthorized_cost_micros: u64,
215    pub depth: u16,
216}
217
218impl ChainBudget {
219    pub fn remaining(&self) -> u64 {
220        self.root_budget_tokens
221            .saturating_sub(self.total_consumed_tokens)
222            .saturating_sub(self.unauthorized_tokens)
223    }
224
225    pub fn utilization_pct(&self) -> f64 {
226        if self.root_budget_tokens == 0 {
227            return 0.0;
228        }
229
230        (self
231            .total_consumed_tokens
232            .saturating_add(self.unauthorized_tokens) as f64
233            / self.root_budget_tokens as f64)
234            * 100.0
235    }
236}
237
238#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
239pub struct WorkNode {
240    pub node_id: String,
241    pub agent_id: String,
242    pub parent_node_id: Option<String>,
243    pub capsule_ref: String,
244    pub status: NodeStatus,
245    pub budget: WorkNodeBudget,
246    pub depth: u16,
247    pub stop_reason: Option<StopReason>,
248    pub outcome_ref: Option<String>,
249    #[serde(default)]
250    pub claim_attempt: u32,
251    /// Current connector attempt within this execution fence, persisted before launch.
252    #[serde(default, skip_serializing_if = "Option::is_none")]
253    pub delivery_attempt: Option<u8>,
254    #[serde(default, skip_serializing_if = "Option::is_none")]
255    pub execution_fence: Option<String>,
256    #[serde(default, skip_serializing_if = "Option::is_none")]
257    pub lease_expires_epoch_ms: Option<u64>,
258    #[serde(default)]
259    pub path_claims: Vec<String>,
260    #[serde(default)]
261    pub task_ref: String,
262    #[serde(default, skip_serializing_if = "Option::is_none")]
263    pub parent_task_ref: Option<String>,
264    #[serde(default)]
265    pub policy_ref: String,
266    #[serde(default)]
267    pub expected_outcome_ref: String,
268    #[serde(default)]
269    pub execution_started: bool,
270}
271
272/// Bounded, acyclic work graph with enforced fan-out and budget constraints.
273#[derive(Clone, Debug, Serialize, Deserialize)]
274pub struct BoundedWorkGraph {
275    #[serde(default)]
276    graph_id: String,
277    nodes: BTreeMap<String, WorkNode>,
278    children: BTreeMap<String, Vec<String>>,
279    #[serde(default)]
280    chain_budgets: BTreeMap<String, ChainBudget>,
281    #[serde(default)]
282    pending_child_budgets: BTreeMap<String, WorkNodeBudget>,
283    #[serde(default)]
284    pending_child_parents: BTreeMap<String, String>,
285    #[serde(default)]
286    receipts: BTreeMap<String, ChildExecutionReceipt>,
287    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
288    failed_spend: BTreeMap<String, MeasuredExecutionSpend>,
289    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
290    reaped_executions: BTreeMap<String, String>,
291    #[serde(default)]
292    accepted_path: BTreeSet<String>,
293    #[serde(default = "default_local_concurrency")]
294    max_local_concurrency: usize,
295    max_fan_out: usize,
296    max_depth: u16,
297}
298
299impl Default for BoundedWorkGraph {
300    fn default() -> Self {
301        Self::new(MAX_FAN_OUT, MAX_DEPTH)
302    }
303}
304
305mod control_ops;
306mod graph_ops;
307
308fn execution_fence(graph_id: &str, node: &WorkNode) -> String {
309    let material = format!(
310        "{}\0{}\0{}\0{}\0{}",
311        graph_id, node.node_id, node.agent_id, node.capsule_ref, node.claim_attempt
312    );
313    format!("fence:{}", blake3::hash(material.as_bytes()).to_hex())
314}
315
316pub fn path_claims_overlap(left: &str, right: &str) -> bool {
317    left == right
318        || left
319            .strip_prefix(right)
320            .is_some_and(|suffix| suffix.starts_with('/'))
321        || right
322            .strip_prefix(left)
323            .is_some_and(|suffix| suffix.starts_with('/'))
324}
325
326fn validate_identifier(value: &str, field: &'static str) -> Result<(), WorkGraphError> {
327    if value.is_empty() || value.len() > MAX_IDENTIFIER_BYTES || value.chars().any(char::is_control)
328    {
329        return Err(WorkGraphError::InvalidIdentifier(field));
330    }
331    Ok(())
332}
333
334fn validate_budget(budget: &WorkNodeBudget) -> Result<(), WorkGraphError> {
335    if budget.tokens_consumed > budget.tokens_allocated
336        || budget.cost_micros_consumed > budget.cost_micros_allocated
337    {
338        return Err(WorkGraphError::InvalidBudget);
339    }
340    Ok(())
341}
342
343fn validate_receipt(receipt: &ChildExecutionReceipt) -> Result<(), WorkGraphError> {
344    validate_identifier(&receipt.receipt_id, "receipt_id")?;
345    validate_identifier(&receipt.node_id, "node_id")?;
346    validate_identifier(&receipt.outcome_ref, "outcome_ref")?;
347    if let Some(fence) = receipt.execution_fence.as_deref()
348        && (fence.len() != 70
349            || !fence.starts_with("fence:")
350            || !fence[6..]
351                .bytes()
352                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)))
353    {
354        return Err(WorkGraphError::InvalidIdentifier("execution_fence"));
355    }
356    if receipt.claims.len() > MAX_RESULT_CLAIMS {
357        return Err(WorkGraphError::TooManyClaims);
358    }
359    let mut keys = BTreeSet::new();
360    for claim in &receipt.claims {
361        validate_identifier(&claim.key, "claim_key")?;
362        validate_identifier(&claim.value, "claim_value")?;
363        if !keys.insert(&claim.key) {
364            return Err(WorkGraphError::DuplicateClaim(claim.key.clone()));
365        }
366    }
367    Ok(())
368}
369
370fn validate_measured_spend(spend: &MeasuredExecutionSpend) -> Result<(), WorkGraphError> {
371    validate_identifier(&spend.spend_id, "spend_id")?;
372    validate_identifier(&spend.node_id, "node_id")?;
373    if spend.outcome_refs.is_empty() || spend.outcome_refs.len() > 2 {
374        return Err(WorkGraphError::InvalidGraph(
375            "failed spend must reference one or two receipts".into(),
376        ));
377    }
378    let mut outcome_refs = BTreeSet::new();
379    for outcome_ref in &spend.outcome_refs {
380        validate_identifier(outcome_ref, "outcome_ref")?;
381        let digest = outcome_ref
382            .strip_prefix("id:sha256:")
383            .filter(|digest| {
384                digest.len() == 64
385                    && digest
386                        .bytes()
387                        .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
388            })
389            .ok_or(WorkGraphError::InvalidIdentifier("outcome_ref"))?;
390        if !outcome_refs.insert(digest) {
391            return Err(WorkGraphError::InvalidGraph(
392                "duplicate failed spend receipt".into(),
393            ));
394        }
395    }
396    if spend.spend_id != measured_spend_id(&spend.outcome_refs) {
397        return Err(WorkGraphError::InvalidIdentifier("spend_id"));
398    }
399    if spend.execution_fence.len() != 70
400        || !spend.execution_fence.starts_with("fence:")
401        || !spend.execution_fence[6..]
402            .bytes()
403            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
404    {
405        return Err(WorkGraphError::InvalidIdentifier("execution_fence"));
406    }
407    Ok(())
408}
409
410fn measured_spend_id(outcome_refs: &[String]) -> String {
411    format!(
412        "spend:{}",
413        blake3::hash(outcome_refs.join("\0").as_bytes()).to_hex()
414    )
415}
416
417fn validate_path_claim(value: &str) -> Result<(), WorkGraphError> {
418    let relative = value
419        .strip_prefix("path:")
420        .ok_or(WorkGraphError::InvalidPathClaim)?;
421    if relative.is_empty()
422        || relative.len() > 1_024
423        || relative.starts_with('/')
424        || relative.starts_with('\\')
425        || relative
426            .split(['/', '\\'])
427            .any(|part| part.is_empty() || part == "." || part == "..")
428        || relative.chars().any(char::is_control)
429    {
430        return Err(WorkGraphError::InvalidPathClaim);
431    }
432    Ok(())
433}
434
435// ─── Errors ──────────────────────────────────────────────────────────────────
436
437#[derive(Debug, thiserror::Error)]
438pub enum WorkGraphError {
439    #[error("graph at capacity ({MAX_GRAPH_NODES} nodes)")]
440    CapacityExceeded,
441    #[error("duplicate node: {0}")]
442    DuplicateNode(String),
443    #[error("node not found: {0}")]
444    NodeNotFound(String),
445    #[error("parent not active: {0}")]
446    ParentNotActive(String),
447    #[error("depth exceeds max {0}")]
448    DepthExceeded(u16),
449    #[error("fan-out exceeds max {0}")]
450    FanOutExceeded(usize),
451    #[error("child budget ({child_requested}) exceeds parent remaining ({parent_remaining})")]
452    BudgetExceedsParent {
453        child_requested: u64,
454        parent_remaining: u64,
455    },
456    #[error(
457        "invalid status transition for node: {0} (only a pending child of an active parent can \
458         be claimed; completed, failed and cancelled nodes are final — inspect with action=get)"
459    )]
460    InvalidTransition(String),
461    #[error(
462        "local concurrency exceeds max {0} (finish an active node with complete or receipt, \
463         or cancel one, first)"
464    )]
465    LocalConcurrencyExceeded(usize),
466    #[error("invalid or oversized {0}")]
467    InvalidIdentifier(&'static str),
468    #[error("consumed budget exceeds allocation")]
469    InvalidBudget,
470    #[error("duplicate execution receipt: {0}")]
471    DuplicateReceipt(String),
472    #[error("execution receipt exceeds node or chain budget: {0}")]
473    ReceiptExceedsBudget(String),
474    #[error("missing execution receipt for accepted node: {0}")]
475    MissingReceipt(String),
476    #[error("unaccepted node cannot establish an accepted path: {0}")]
477    RejectedAcceptedPath(String),
478    #[error("execution receipt has too many result claims")]
479    TooManyClaims,
480    #[error("node has too many path claims")]
481    TooManyPathClaims,
482    #[error("invalid project-relative path claim")]
483    InvalidPathClaim,
484    #[error("duplicate result claim key: {0}")]
485    DuplicateClaim(String),
486    #[error("work graph accounting overflow")]
487    Overflow,
488    #[error("stale or missing execution fence for node: {0}")]
489    StaleExecutionFence(String),
490    #[error("invalid work graph: {0}")]
491    InvalidGraph(String),
492    #[error("budget cascade error: {0}")]
493    Cascade(#[from] CascadeError),
494}
495
496// ─── Tests ───────────────────────────────────────────────────────────────────
497
498#[cfg(test)]
499pub mod tests;
lean_ctx_workgraph/work_graph.rs

lean_ctx_workgraph/
work_graph.rs
