eidos_kernel/workflow/
context.rs

1use std::collections::BTreeMap;
2
3use super::run::{deps_passed, ensure_run_matches_dag};
4use super::{
5    DagValidationError, TaskContextItem, TaskDag, TaskEventError, TaskOperator, TaskRisk, TaskRun,
6    TaskState, validate_dag,
7};
8
9/// A bounded, task-local read pack derived from a DAG node.
10#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
11pub struct TaskContextSlice {
12    pub dag_id: String,
13    pub task_id: String,
14    pub title: String,
15    pub operator: TaskOperator,
16    pub risk: TaskRisk,
17    #[serde(default, skip_serializing_if = "Option::is_none")]
18    pub run_state: Option<TaskState>,
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    pub ready: Option<bool>,
21    #[serde(default, skip_serializing_if = "Vec::is_empty")]
22    pub blocked_by: Vec<String>,
23    pub depends_on: Vec<String>,
24    pub inputs: Vec<String>,
25    pub artifact_inputs: Vec<String>,
26    pub context: Vec<TaskContextItem>,
27    pub outputs: Vec<String>,
28    pub required_checks: Vec<String>,
29    pub tools: Vec<String>,
30    pub receipts: Vec<String>,
31}
32
33/// Invalid task-context slice requests.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub enum TaskContextError {
36    InvalidDag(DagValidationError),
37    InvalidRun(TaskEventError),
38    UnknownTask(String),
39}
40
41/// Return the smallest useful context slice for one DAG task.
42///
43/// `available_context` is usually the focus pack that produced the DAG. Inputs that resolve to a
44/// context item are returned as read targets; other inputs are preserved as artifact inputs
45/// (`context_pack`, `work_items`, `change_set`, etc.).
46pub fn task_context_slice(
47    dag: &TaskDag,
48    task_id: &str,
49    available_context: &[TaskContextItem],
50) -> Result<TaskContextSlice, TaskContextError> {
51    validate_dag(dag).map_err(TaskContextError::InvalidDag)?;
52    let Some(node) = dag.nodes.iter().find(|node| node.id == task_id) else {
53        return Err(TaskContextError::UnknownTask(task_id.to_string()));
54    };
55
56    let context_by_id = available_context
57        .iter()
58        .map(|item| (item.id.as_str(), item))
59        .collect::<BTreeMap<_, _>>();
60    let mut context = Vec::new();
61    let mut artifact_inputs = Vec::new();
62    for input in &node.inputs {
63        if let Some(item) = context_by_id.get(input.as_str()) {
64            context.push((*item).clone());
65        } else {
66            artifact_inputs.push(input.clone());
67        }
68    }
69    if node.inputs.is_empty() {
70        artifact_inputs.extend(dependency_outputs(dag, node));
71    }
72    artifact_inputs.sort();
73    artifact_inputs.dedup();
74    context.sort_by(|a, b| a.id.cmp(&b.id));
75
76    let mut receipts = node.receipts.clone();
77    receipts.extend(
78        context
79            .iter()
80            .filter_map(|item| (!item.source.trim().is_empty()).then_some(item.source.clone())),
81    );
82    receipts.sort();
83    receipts.dedup();
84
85    Ok(TaskContextSlice {
86        dag_id: dag.id.clone(),
87        task_id: node.id.clone(),
88        title: node.title.clone(),
89        operator: node.operator,
90        risk: node.risk,
91        run_state: None,
92        ready: None,
93        blocked_by: Vec::new(),
94        depends_on: node.depends_on.clone(),
95        inputs: node.inputs.clone(),
96        artifact_inputs,
97        context,
98        outputs: node.outputs.clone(),
99        required_checks: node.required_checks.clone(),
100        tools: node.tools.clone(),
101        receipts,
102    })
103}
104
105fn dependency_outputs(dag: &TaskDag, node: &super::TaskNode) -> Vec<String> {
106    let nodes = dag
107        .nodes
108        .iter()
109        .map(|node| (node.id.as_str(), node))
110        .collect::<BTreeMap<_, _>>();
111    node.depends_on
112        .iter()
113        .filter_map(|dependency| nodes.get(dependency.as_str()))
114        .flat_map(|dependency| dependency.outputs.iter().cloned())
115        .filter(|output| !output.trim().is_empty())
116        .collect()
117}
118
119/// Return a task context slice annotated with the current run state.
120pub fn task_context_slice_for_run(
121    dag: &TaskDag,
122    run: &TaskRun,
123    task_id: &str,
124    available_context: &[TaskContextItem],
125) -> Result<TaskContextSlice, TaskContextError> {
126    let mut slice = task_context_slice(dag, task_id, available_context)?;
127    ensure_run_matches_dag(dag, run).map_err(TaskContextError::InvalidRun)?;
128    let Some(state) = run.states.get(task_id).copied() else {
129        return Err(TaskContextError::UnknownTask(task_id.to_string()));
130    };
131    let blocked_by = deps_passed(dag, run, task_id);
132    let ready = state == TaskState::Ready && blocked_by.is_empty();
133    slice.run_state = Some(state);
134    slice.ready = Some(ready);
135    slice.blocked_by = blocked_by;
136    Ok(slice)
137}
eidos_kernel/workflow/context.rs

eidos_kernel/workflow/
context.rs
