mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-15 23:32:46 +00:00
Rename fabro_core::RunState to ExecutionState
Resolves the name collision with fabro_store::RunState. The core type represents live in-memory execution state (current node, visits, context), while the store type is an event-sourced projection of a full run record. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
22115a180b
commit
a667fb7472
15 changed files with 170 additions and 154 deletions
|
|
@ -13,7 +13,7 @@ use crate::lifecycle::{
|
|||
RunLifecycle,
|
||||
};
|
||||
use crate::outcome::{NodeResult, NodeResultExt, Outcome, StageStatus};
|
||||
use crate::state::RunState;
|
||||
use crate::state::ExecutionState;
|
||||
use tokio::time::sleep;
|
||||
|
||||
#[derive(Default)]
|
||||
|
|
@ -88,8 +88,8 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
pub async fn run(
|
||||
&self,
|
||||
graph: &G,
|
||||
mut state: RunState<G::Meta>,
|
||||
) -> Result<(Outcome<G::Meta>, RunState<G::Meta>)> {
|
||||
mut state: ExecutionState<G::Meta>,
|
||||
) -> Result<(Outcome<G::Meta>, ExecutionState<G::Meta>)> {
|
||||
self.lifecycle.on_run_start(graph, &state).await?;
|
||||
|
||||
loop {
|
||||
|
|
@ -248,7 +248,7 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
async fn execute_with_retry(
|
||||
&self,
|
||||
node: &G::Node,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
graph: &G,
|
||||
) -> Result<NodeResult<G::Meta>> {
|
||||
let policy = self.handler.retry_policy(node, graph);
|
||||
|
|
@ -352,7 +352,7 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
&self,
|
||||
node: &G::Node,
|
||||
outcome: &Outcome<G::Meta>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
graph: &G,
|
||||
) -> Result<NextStep> {
|
||||
// Jump takes priority
|
||||
|
|
@ -434,7 +434,7 @@ mod tests {
|
|||
handler: Arc<dyn NodeHandler<TestGraph>>,
|
||||
) -> Result<Outcome> {
|
||||
let g = linear_graph(node_ids);
|
||||
let state = RunState::new(&g)?;
|
||||
let state = ExecutionState::new(&g)?;
|
||||
let executor = ExecutorBuilder::new(handler).build();
|
||||
executor
|
||||
.run(&g, state)
|
||||
|
|
@ -458,13 +458,13 @@ mod tests {
|
|||
struct LogLifecycle(Arc<Mutex<Vec<String>>>);
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for LogLifecycle {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &ExecutionState) -> Result<()> {
|
||||
self.0.lock().unwrap().push("start".into());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(LogLifecycle(log.clone())))
|
||||
|
|
@ -477,7 +477,7 @@ mod tests {
|
|||
async fn executor_builder_sets_cancel_token() {
|
||||
let token = Arc::new(AtomicBool::new(true)); // already cancelled
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.cancel_token(token)
|
||||
|
|
@ -498,7 +498,7 @@ mod tests {
|
|||
vec![TestEdge::new("work", "end")],
|
||||
"work",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.build();
|
||||
|
|
@ -525,7 +525,7 @@ mod tests {
|
|||
Ok(Outcome::fail("first attempt")),
|
||||
Ok(Outcome::success()),
|
||||
]));
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>).build();
|
||||
let (result, _) = executor.run(&g, state).await.unwrap();
|
||||
|
|
@ -544,7 +544,7 @@ mod tests {
|
|||
"work",
|
||||
);
|
||||
// No retry target, and handler fails
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(AlwaysFailHandler::new("nope")) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -559,7 +559,11 @@ mod tests {
|
|||
struct TrackingLifecycle(Arc<Mutex<Vec<String>>>);
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for TrackingLifecycle {
|
||||
async fn before_node(&self, node: &TestNode, _s: &RunState) -> Result<NodeDecision> {
|
||||
async fn before_node(
|
||||
&self,
|
||||
node: &TestNode,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
self.0
|
||||
.lock()
|
||||
.unwrap()
|
||||
|
|
@ -570,7 +574,7 @@ mod tests {
|
|||
&self,
|
||||
node: &TestNode,
|
||||
_r: &mut NodeResult,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.0
|
||||
.lock()
|
||||
|
|
@ -582,7 +586,7 @@ mod tests {
|
|||
&self,
|
||||
node: &TestNode,
|
||||
_goal_gates_passed: bool,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) {
|
||||
self.0
|
||||
.lock()
|
||||
|
|
@ -591,7 +595,7 @@ mod tests {
|
|||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(TrackingLifecycle(log.clone())))
|
||||
|
|
@ -617,7 +621,7 @@ mod tests {
|
|||
&self,
|
||||
node: &TestNode,
|
||||
_goal_gates_passed: bool,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) {
|
||||
self.0
|
||||
.lock()
|
||||
|
|
@ -626,7 +630,7 @@ mod tests {
|
|||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(TerminalTracker(log.clone())))
|
||||
|
|
@ -650,7 +654,7 @@ mod tests {
|
|||
],
|
||||
"loop_node",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.build();
|
||||
|
|
@ -669,7 +673,7 @@ mod tests {
|
|||
vec![TestEdge::new("a", "b"), TestEdge::new("b", "a")],
|
||||
"a",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.max_node_visits(3)
|
||||
|
|
@ -694,7 +698,7 @@ mod tests {
|
|||
],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(AlwaysFailHandler::new("oops")) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -718,7 +722,7 @@ mod tests {
|
|||
],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.build();
|
||||
|
|
@ -752,7 +756,7 @@ mod tests {
|
|||
vec![TestEdge::new("start", "end")],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(JumpHandler) as Arc<dyn NodeHandler<TestGraph>>).build();
|
||||
let (result, _) = executor.run(&g, state).await.unwrap();
|
||||
|
|
@ -787,7 +791,7 @@ mod tests {
|
|||
],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.max_node_visits(5)
|
||||
.build();
|
||||
|
|
@ -802,7 +806,7 @@ mod tests {
|
|||
struct StartTracker(Arc<Mutex<Vec<String>>>);
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for StartTracker {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &ExecutionState) -> Result<()> {
|
||||
self.0.lock().unwrap().push("on_run_start".into());
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -832,7 +836,7 @@ mod tests {
|
|||
],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(handler as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(StartTracker(log.clone())))
|
||||
.max_node_visits(5)
|
||||
|
|
@ -850,7 +854,7 @@ mod tests {
|
|||
vec![TestEdge::new("start", "end").with_label("success")],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(AlwaysFailHandler::new("boom")) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -863,7 +867,7 @@ mod tests {
|
|||
async fn executor_no_edge_after_success_returns_success() {
|
||||
// Node succeeds with no outgoing edges → run ends with success
|
||||
let g = TestGraph::new(vec![TestNode::new("only")], vec![], "only");
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.build();
|
||||
|
|
@ -894,7 +898,7 @@ mod tests {
|
|||
}
|
||||
|
||||
let g = linear_graph(&["start", "work", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(CancellingHandler(token_clone)) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -1057,7 +1061,7 @@ mod tests {
|
|||
}
|
||||
// No outgoing edges from "start" so PartialSuccess becomes the run result
|
||||
let g = TestGraph::new(vec![TestNode::new("start")], vec![], "start");
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(ExhaustedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.build();
|
||||
|
|
@ -1074,7 +1078,7 @@ mod tests {
|
|||
async fn before_attempt(
|
||||
&self,
|
||||
ctx: &AttemptContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
self.0.lock().unwrap().push(ctx.attempt);
|
||||
Ok(NodeDecision::Continue)
|
||||
|
|
@ -1101,7 +1105,7 @@ mod tests {
|
|||
}),
|
||||
);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(handler as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(AttemptTracker(attempt_log.clone())))
|
||||
.build();
|
||||
|
|
@ -1118,7 +1122,7 @@ mod tests {
|
|||
async fn after_attempt(
|
||||
&self,
|
||||
ctx: &AttemptResultContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.0.lock().unwrap().push((ctx.attempt, ctx.will_retry));
|
||||
Ok(())
|
||||
|
|
@ -1145,7 +1149,7 @@ mod tests {
|
|||
}),
|
||||
);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(handler as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(RetryTracker(retry_log.clone())))
|
||||
.build();
|
||||
|
|
@ -1164,7 +1168,7 @@ mod tests {
|
|||
async fn before_attempt(
|
||||
&self,
|
||||
ctx: &AttemptContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
if ctx.attempt >= 2 {
|
||||
|
|
@ -1195,7 +1199,7 @@ mod tests {
|
|||
}),
|
||||
);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(SkipOnSecondAttempt(call_count_clone)))
|
||||
.build();
|
||||
|
|
@ -1245,7 +1249,11 @@ mod tests {
|
|||
struct SkipFirst(Mutex<bool>);
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for SkipFirst {
|
||||
async fn before_node(&self, node: &TestNode, _s: &RunState) -> Result<NodeDecision> {
|
||||
async fn before_node(
|
||||
&self,
|
||||
node: &TestNode,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
if node.id() == "start" {
|
||||
let mut skipped = self.0.lock().unwrap();
|
||||
if !*skipped {
|
||||
|
|
@ -1257,7 +1265,7 @@ mod tests {
|
|||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(SkipFirst(Mutex::new(false))))
|
||||
|
|
@ -1271,12 +1279,16 @@ mod tests {
|
|||
struct Blocker;
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for Blocker {
|
||||
async fn before_node(&self, _n: &TestNode, _s: &RunState) -> Result<NodeDecision> {
|
||||
async fn before_node(
|
||||
&self,
|
||||
_n: &TestNode,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
Ok(NodeDecision::Block("blocked".into()))
|
||||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(Blocker))
|
||||
|
|
@ -1294,14 +1306,14 @@ mod tests {
|
|||
&self,
|
||||
_n: &TestNode,
|
||||
result: &mut NodeResult,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
result.outcome.notes = Some("mutated".into());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(Mutator))
|
||||
|
|
@ -1318,7 +1330,7 @@ mod tests {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
_ctx: &EdgeContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<EdgeDecision> {
|
||||
Ok(EdgeDecision::Override("alt".into()))
|
||||
}
|
||||
|
|
@ -1332,7 +1344,7 @@ mod tests {
|
|||
vec![TestEdge::new("start", "end")],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(Redirector))
|
||||
|
|
@ -1349,13 +1361,13 @@ mod tests {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
_ctx: &EdgeContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<EdgeDecision> {
|
||||
Ok(EdgeDecision::Block("edge blocked".into()))
|
||||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(EdgeBlocker))
|
||||
|
|
@ -1375,14 +1387,14 @@ mod tests {
|
|||
node: &TestNode,
|
||||
_r: &NodeResult,
|
||||
_next_node_id: Option<&str>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.0.lock().unwrap().push(node.id().to_string());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "work", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(CheckpointTracker(log.clone())))
|
||||
|
|
@ -1397,16 +1409,16 @@ mod tests {
|
|||
struct RunTracker(Arc<Mutex<Vec<String>>>);
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for RunTracker {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &ExecutionState) -> Result<()> {
|
||||
self.0.lock().unwrap().push("start".into());
|
||||
Ok(())
|
||||
}
|
||||
async fn on_run_end(&self, _o: &Outcome, _s: &RunState) {
|
||||
async fn on_run_end(&self, _o: &Outcome, _s: &ExecutionState) {
|
||||
self.0.lock().unwrap().push("end".into());
|
||||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(RunTracker(log.clone())))
|
||||
|
|
@ -1424,7 +1436,7 @@ mod tests {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
ctx: &EdgeContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<EdgeDecision> {
|
||||
self.0
|
||||
.lock()
|
||||
|
|
@ -1456,7 +1468,7 @@ mod tests {
|
|||
vec![TestEdge::new("start", "end")],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(JumpHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(JumpTracker(log.clone())))
|
||||
|
|
@ -1500,7 +1512,7 @@ mod tests {
|
|||
&self,
|
||||
node: &TestNode,
|
||||
result: &mut NodeResult,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
if node.id() == "work" {
|
||||
if let Some(ref notes) = result.outcome.notes {
|
||||
|
|
@ -1512,7 +1524,7 @@ mod tests {
|
|||
}
|
||||
|
||||
let g = linear_graph(&["start", "work", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(ContextWriter) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(NoteCapture(log.clone())))
|
||||
|
|
@ -1533,7 +1545,7 @@ mod tests {
|
|||
node: &TestNode,
|
||||
_r: &NodeResult,
|
||||
next_node_id: Option<&str>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.0
|
||||
.lock()
|
||||
|
|
@ -1543,7 +1555,7 @@ mod tests {
|
|||
}
|
||||
}
|
||||
let g = linear_graph(&["start", "work", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(NextNodeTracker(log.clone())))
|
||||
|
|
@ -1591,7 +1603,7 @@ mod tests {
|
|||
&self,
|
||||
node: &TestNode,
|
||||
_result: &NodeResult,
|
||||
state: &RunState,
|
||||
state: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
let shared = state.context.get_string("shared", "missing");
|
||||
let completed = state.completed_nodes.join(",");
|
||||
|
|
@ -1607,7 +1619,7 @@ mod tests {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
ctx: &EdgeContext<'_, TestGraph>,
|
||||
state: &RunState,
|
||||
state: &ExecutionState,
|
||||
) -> Result<EdgeDecision> {
|
||||
let shared = state.context.get_string("shared", "missing");
|
||||
self.0
|
||||
|
|
@ -1619,7 +1631,7 @@ mod tests {
|
|||
}
|
||||
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(ContextWriter) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(RecordTracker(log.clone())))
|
||||
|
|
@ -1646,7 +1658,7 @@ mod tests {
|
|||
&self,
|
||||
node: &TestNode,
|
||||
goal_gates_passed: bool,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) {
|
||||
self.0
|
||||
.lock()
|
||||
|
|
@ -1657,7 +1669,7 @@ mod tests {
|
|||
|
||||
// Test 1: goal gates pass
|
||||
let g = linear_graph(&["work", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(GateTracker(log.clone())))
|
||||
|
|
@ -1675,7 +1687,7 @@ mod tests {
|
|||
vec![TestEdge::new("work", "end")],
|
||||
"work",
|
||||
);
|
||||
let state2 = RunState::new(&g2).unwrap();
|
||||
let state2 = ExecutionState::new(&g2).unwrap();
|
||||
let executor2 = ExecutorBuilder::new(
|
||||
Arc::new(AlwaysFailHandler::new("nope")) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -1734,7 +1746,7 @@ mod tests {
|
|||
],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(LogHandler(log_clone)) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -1794,7 +1806,7 @@ mod tests {
|
|||
],
|
||||
"start",
|
||||
);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(ContextChecker { log: log.clone() }) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -1826,7 +1838,7 @@ mod tests {
|
|||
)
|
||||
.with_retry_target("work", "work");
|
||||
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>).build();
|
||||
let (result, _) = executor.run(&g, state).await.unwrap();
|
||||
|
|
@ -1856,7 +1868,7 @@ mod tests {
|
|||
)
|
||||
.with_retry_target("work", "recovery");
|
||||
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(handler.clone() as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.max_node_visits(5)
|
||||
.build();
|
||||
|
|
@ -1889,7 +1901,7 @@ mod tests {
|
|||
}
|
||||
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(
|
||||
Arc::new(SlowHandler(stall_clone)) as Arc<dyn NodeHandler<TestGraph>>
|
||||
)
|
||||
|
|
@ -1950,7 +1962,7 @@ mod tests {
|
|||
}
|
||||
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor = ExecutorBuilder::new(Arc::new(FailOnceHandler {
|
||||
stall: stall_clone,
|
||||
calls: AtomicU32::new(0),
|
||||
|
|
@ -1976,7 +1988,7 @@ mod tests {
|
|||
async fn before_attempt(
|
||||
&self,
|
||||
_ctx: &AttemptContext<'_, TestGraph>,
|
||||
_s: &RunState,
|
||||
_s: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
self.0.cancel();
|
||||
sleep(Duration::from_secs(10)).await;
|
||||
|
|
@ -1985,7 +1997,7 @@ mod tests {
|
|||
}
|
||||
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let executor =
|
||||
ExecutorBuilder::new(Arc::new(AlwaysSucceedHandler) as Arc<dyn NodeHandler<TestGraph>>)
|
||||
.lifecycle(Box::new(SlowBeforeAttempt(stall_clone)))
|
||||
|
|
|
|||
|
|
@ -26,4 +26,4 @@ pub use outcome::{
|
|||
};
|
||||
pub use retry::{BackoffPolicy, RetryPolicy};
|
||||
pub use stall::{ActivityMonitor, StallGuard, StallWatchdog};
|
||||
pub use state::RunState;
|
||||
pub use state::ExecutionState;
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ use async_trait::async_trait;
|
|||
use crate::error::Result;
|
||||
use crate::graph::Graph;
|
||||
use crate::outcome::{NodeResult, Outcome, OutcomeMeta};
|
||||
use crate::state::RunState;
|
||||
use crate::state::ExecutionState;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum NodeDecision<M: OutcomeMeta = ()> {
|
||||
|
|
@ -46,7 +46,7 @@ pub struct EdgeContext<'a, G: Graph> {
|
|||
|
||||
#[async_trait]
|
||||
pub trait RunLifecycle<G: Graph>: Send + Sync {
|
||||
async fn on_run_start(&self, _graph: &G, _state: &RunState<G::Meta>) -> Result<()> {
|
||||
async fn on_run_start(&self, _graph: &G, _state: &ExecutionState<G::Meta>) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -54,14 +54,14 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
&self,
|
||||
_node: &G::Node,
|
||||
_goal_gates_passed: bool,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) {
|
||||
}
|
||||
|
||||
async fn before_node(
|
||||
&self,
|
||||
_node: &G::Node,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<NodeDecision<G::Meta>> {
|
||||
Ok(NodeDecision::Continue)
|
||||
}
|
||||
|
|
@ -69,7 +69,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
async fn before_attempt(
|
||||
&self,
|
||||
_ctx: &AttemptContext<'_, G>,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<NodeDecision<G::Meta>> {
|
||||
Ok(NodeDecision::Continue)
|
||||
}
|
||||
|
|
@ -77,7 +77,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
async fn after_attempt(
|
||||
&self,
|
||||
_ctx: &AttemptResultContext<'_, G>,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -86,7 +86,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
&self,
|
||||
_node: &G::Node,
|
||||
_result: &mut NodeResult<G::Meta>,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -95,7 +95,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
&self,
|
||||
_node: &G::Node,
|
||||
_result: &NodeResult<G::Meta>,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -103,7 +103,7 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
_ctx: &EdgeContext<'_, G>,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<EdgeDecision> {
|
||||
Ok(EdgeDecision::Continue)
|
||||
}
|
||||
|
|
@ -113,12 +113,12 @@ pub trait RunLifecycle<G: Graph>: Send + Sync {
|
|||
_node: &G::Node,
|
||||
_result: &NodeResult<G::Meta>,
|
||||
_next_node_id: Option<&str>,
|
||||
_state: &RunState<G::Meta>,
|
||||
_state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_run_end(&self, _outcome: &Outcome<G::Meta>, _state: &RunState<G::Meta>) {}
|
||||
async fn on_run_end(&self, _outcome: &Outcome<G::Meta>, _state: &ExecutionState<G::Meta>) {}
|
||||
}
|
||||
|
||||
/// No-op lifecycle that passes through everything.
|
||||
|
|
@ -141,7 +141,7 @@ impl<G: Graph> CompositeLifecycle<G> {
|
|||
|
||||
#[async_trait]
|
||||
impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
||||
async fn on_run_start(&self, graph: &G, state: &RunState<G::Meta>) -> Result<()> {
|
||||
async fn on_run_start(&self, graph: &G, state: &ExecutionState<G::Meta>) -> Result<()> {
|
||||
for child in &self.children {
|
||||
child.on_run_start(graph, state).await?;
|
||||
}
|
||||
|
|
@ -152,7 +152,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
&self,
|
||||
node: &G::Node,
|
||||
goal_gates_passed: bool,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) {
|
||||
for child in &self.children {
|
||||
child
|
||||
|
|
@ -164,7 +164,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
async fn before_node(
|
||||
&self,
|
||||
node: &G::Node,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<NodeDecision<G::Meta>> {
|
||||
for child in &self.children {
|
||||
match child.before_node(node, state).await? {
|
||||
|
|
@ -178,7 +178,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
async fn before_attempt(
|
||||
&self,
|
||||
ctx: &AttemptContext<'_, G>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<NodeDecision<G::Meta>> {
|
||||
for child in &self.children {
|
||||
match child.before_attempt(ctx, state).await? {
|
||||
|
|
@ -192,7 +192,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
async fn after_attempt(
|
||||
&self,
|
||||
ctx: &AttemptResultContext<'_, G>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
for child in &self.children {
|
||||
child.after_attempt(ctx, state).await?;
|
||||
|
|
@ -204,7 +204,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
&self,
|
||||
node: &G::Node,
|
||||
result: &mut NodeResult<G::Meta>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
for child in &self.children {
|
||||
child.after_node(node, result, state).await?;
|
||||
|
|
@ -216,7 +216,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
&self,
|
||||
node: &G::Node,
|
||||
result: &NodeResult<G::Meta>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
for child in &self.children {
|
||||
child.after_record(node, result, state).await?;
|
||||
|
|
@ -227,7 +227,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
ctx: &EdgeContext<'_, G>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<EdgeDecision> {
|
||||
for child in &self.children {
|
||||
match child.on_edge_selected(ctx, state).await? {
|
||||
|
|
@ -243,7 +243,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
node: &G::Node,
|
||||
result: &NodeResult<G::Meta>,
|
||||
next_node_id: Option<&str>,
|
||||
state: &RunState<G::Meta>,
|
||||
state: &ExecutionState<G::Meta>,
|
||||
) -> Result<()> {
|
||||
for child in &self.children {
|
||||
child
|
||||
|
|
@ -253,7 +253,7 @@ impl<G: Graph + 'static> RunLifecycle<G> for CompositeLifecycle<G> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_run_end(&self, outcome: &Outcome<G::Meta>, state: &RunState<G::Meta>) {
|
||||
async fn on_run_end(&self, outcome: &Outcome<G::Meta>, state: &ExecutionState<G::Meta>) {
|
||||
for child in &self.children {
|
||||
child.on_run_end(outcome, state).await;
|
||||
}
|
||||
|
|
@ -308,7 +308,7 @@ mod tests {
|
|||
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for RecordingLifecycle {
|
||||
async fn on_run_start(&self, _graph: &TestGraph, _state: &RunState) -> Result<()> {
|
||||
async fn on_run_start(&self, _graph: &TestGraph, _state: &ExecutionState) -> Result<()> {
|
||||
self.log
|
||||
.lock()
|
||||
.unwrap()
|
||||
|
|
@ -320,7 +320,7 @@ mod tests {
|
|||
&self,
|
||||
_node: &TestNode,
|
||||
_goal_gates_passed: bool,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -328,7 +328,11 @@ mod tests {
|
|||
.push(format!("{}:on_terminal_reached", self.name));
|
||||
}
|
||||
|
||||
async fn before_node(&self, _node: &TestNode, _state: &RunState) -> Result<NodeDecision> {
|
||||
async fn before_node(
|
||||
&self,
|
||||
_node: &TestNode,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
self.log
|
||||
.lock()
|
||||
.unwrap()
|
||||
|
|
@ -344,7 +348,7 @@ mod tests {
|
|||
async fn before_attempt(
|
||||
&self,
|
||||
_ctx: &AttemptContext<'_, TestGraph>,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<NodeDecision> {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -361,7 +365,7 @@ mod tests {
|
|||
async fn after_attempt(
|
||||
&self,
|
||||
_ctx: &AttemptResultContext<'_, TestGraph>,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -374,7 +378,7 @@ mod tests {
|
|||
&self,
|
||||
_node: &TestNode,
|
||||
_result: &mut NodeResult,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -387,7 +391,7 @@ mod tests {
|
|||
&self,
|
||||
_node: &TestNode,
|
||||
_result: &NodeResult,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -399,7 +403,7 @@ mod tests {
|
|||
async fn on_edge_selected(
|
||||
&self,
|
||||
_ctx: &EdgeContext<'_, TestGraph>,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<EdgeDecision> {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -418,7 +422,7 @@ mod tests {
|
|||
_node: &TestNode,
|
||||
_result: &NodeResult,
|
||||
_next_node_id: Option<&str>,
|
||||
_state: &RunState,
|
||||
_state: &ExecutionState,
|
||||
) -> Result<()> {
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -427,7 +431,7 @@ mod tests {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_run_end(&self, _outcome: &Outcome, _state: &RunState) {
|
||||
async fn on_run_end(&self, _outcome: &Outcome, _state: &ExecutionState) {
|
||||
self.log
|
||||
.lock()
|
||||
.unwrap()
|
||||
|
|
@ -439,7 +443,7 @@ mod tests {
|
|||
async fn default_lifecycle_is_noop() {
|
||||
let lc = NoopLifecycle;
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
assert!(
|
||||
<NoopLifecycle as RunLifecycle<TestGraph>>::on_run_start(&lc, &g, &state)
|
||||
.await
|
||||
|
|
@ -462,7 +466,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
lc.on_run_start(&g, &state).await.unwrap();
|
||||
let calls = log.lock().unwrap().clone();
|
||||
assert_eq!(calls, vec!["a:on_run_start", "b:on_run_start"]);
|
||||
|
|
@ -479,7 +483,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let decision = lc.before_node(&node, &state).await.unwrap();
|
||||
assert!(matches!(decision, NodeDecision::Skip(_)));
|
||||
|
|
@ -499,7 +503,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let decision = lc.before_node(&node, &state).await.unwrap();
|
||||
assert!(matches!(decision, NodeDecision::Block(_)));
|
||||
|
|
@ -518,7 +522,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let ctx = AttemptContext {
|
||||
node: &node,
|
||||
|
|
@ -542,7 +546,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let ctx = AttemptContext {
|
||||
node: &node,
|
||||
|
|
@ -561,7 +565,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1);
|
||||
let ctx = AttemptResultContext {
|
||||
|
|
@ -587,7 +591,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let outcome = Outcome::success();
|
||||
let edge = g.outgoing_edges("start").into_iter().next().unwrap();
|
||||
let ctx = EdgeContext {
|
||||
|
|
@ -615,7 +619,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let outcome = Outcome::success();
|
||||
let ctx = EdgeContext {
|
||||
from: "start",
|
||||
|
|
@ -634,7 +638,7 @@ mod tests {
|
|||
let log = Arc::new(Mutex::new(Vec::new()));
|
||||
let lc = CompositeLifecycle::new(vec![Box::new(RecordingLifecycle::new("a", log.clone()))]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let outcome = Outcome::success();
|
||||
let ctx = EdgeContext::<TestGraph> {
|
||||
from: "start",
|
||||
|
|
@ -658,7 +662,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let mut result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1);
|
||||
lc.after_node(&node, &mut result, &state).await.unwrap();
|
||||
|
|
@ -674,7 +678,7 @@ mod tests {
|
|||
Box::new(RecordingLifecycle::new("b", log.clone())),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
let node = g.get_node("start").unwrap();
|
||||
let result = NodeResult::new(Outcome::success(), Duration::ZERO, 1, 1);
|
||||
lc.after_record(&node, &result, &state).await.unwrap();
|
||||
|
|
@ -695,7 +699,7 @@ mod tests {
|
|||
|
||||
#[async_trait]
|
||||
impl RunLifecycle<TestGraph> for OrderedLifecycle {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &RunState) -> Result<()> {
|
||||
async fn on_run_start(&self, _g: &TestGraph, _s: &ExecutionState) -> Result<()> {
|
||||
let order = self.counter.fetch_add(1, Ordering::SeqCst);
|
||||
self.log
|
||||
.lock()
|
||||
|
|
@ -723,7 +727,7 @@ mod tests {
|
|||
}),
|
||||
]);
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::new(&g).unwrap();
|
||||
let state = ExecutionState::new(&g).unwrap();
|
||||
lc.on_run_start(&g, &state).await.unwrap();
|
||||
let calls = log.lock().unwrap().clone();
|
||||
assert_eq!(calls, vec!["first:0", "second:1", "third:2"]);
|
||||
|
|
|
|||
|
|
@ -5,9 +5,9 @@ use crate::error::Result;
|
|||
use crate::graph::{Graph, NodeSpec};
|
||||
use crate::outcome::{NodeResult, Outcome, OutcomeMeta};
|
||||
|
||||
impl<M: OutcomeMeta> std::fmt::Debug for RunState<M> {
|
||||
impl<M: OutcomeMeta> std::fmt::Debug for ExecutionState<M> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("RunState")
|
||||
f.debug_struct("ExecutionState")
|
||||
.field("current_node_id", &self.current_node_id)
|
||||
.field("completed_nodes", &self.completed_nodes)
|
||||
.field("stage_index", &self.stage_index)
|
||||
|
|
@ -16,7 +16,7 @@ impl<M: OutcomeMeta> std::fmt::Debug for RunState<M> {
|
|||
}
|
||||
}
|
||||
|
||||
pub struct RunState<M: OutcomeMeta = ()> {
|
||||
pub struct ExecutionState<M: OutcomeMeta = ()> {
|
||||
pub context: Context,
|
||||
pub current_node_id: String,
|
||||
pub completed_nodes: Vec<String>,
|
||||
|
|
@ -28,7 +28,7 @@ pub struct RunState<M: OutcomeMeta = ()> {
|
|||
pub cancelled: bool,
|
||||
}
|
||||
|
||||
impl<M: OutcomeMeta> RunState<M> {
|
||||
impl<M: OutcomeMeta> ExecutionState<M> {
|
||||
pub fn new<G: Graph>(graph: &G) -> Result<Self> {
|
||||
let start = graph.find_start_node()?;
|
||||
Ok(Self {
|
||||
|
|
@ -98,7 +98,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_new_from_graph() {
|
||||
let g = linear_graph(&["start", "work", "end"]);
|
||||
let state = RunState::<()>::new(&g).unwrap();
|
||||
let state = ExecutionState::<()>::new(&g).unwrap();
|
||||
assert_eq!(state.current_node_id, "start");
|
||||
assert!(state.completed_nodes.is_empty());
|
||||
assert!(state.node_outcomes.is_empty());
|
||||
|
|
@ -109,7 +109,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_record_updates_all_fields() {
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
let result = NodeResult::new(Outcome::success(), Duration::from_millis(50), 2, 3);
|
||||
state.record("start", &result);
|
||||
|
||||
|
|
@ -122,7 +122,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_record_applies_context_updates() {
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
let mut outcome = Outcome::success();
|
||||
outcome.context_updates.insert("key".into(), json!("value"));
|
||||
let result = NodeResult::new(outcome, Duration::ZERO, 1, 1);
|
||||
|
|
@ -133,7 +133,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_advance_updates_current_and_previous() {
|
||||
let g = linear_graph(&["start", "mid", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
assert_eq!(state.current_node_id, "start");
|
||||
assert!(state.previous_node_id.is_none());
|
||||
|
||||
|
|
@ -149,7 +149,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_restart_clears_progress_keeps_visits() {
|
||||
let g = linear_graph(&["start", "work", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
state.increment_visits("start");
|
||||
state.increment_visits("work");
|
||||
state.record(
|
||||
|
|
@ -174,7 +174,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_current_node_from_graph() {
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let state = RunState::<()>::new(&g).unwrap();
|
||||
let state = ExecutionState::<()>::new(&g).unwrap();
|
||||
let node = state.current_node(&g).unwrap();
|
||||
assert_eq!(node.id(), "start");
|
||||
}
|
||||
|
|
@ -182,7 +182,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_increment_visits() {
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
assert_eq!(state.increment_visits("start"), 1);
|
||||
assert_eq!(state.increment_visits("start"), 2);
|
||||
assert_eq!(state.increment_visits("other"), 1);
|
||||
|
|
@ -191,7 +191,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_restart_with_new_context() {
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
state.context.set("key", json!("old_value"));
|
||||
state.increment_visits("start");
|
||||
|
||||
|
|
@ -210,7 +210,7 @@ mod tests {
|
|||
#[test]
|
||||
fn run_state_restart_without_context_preserves() {
|
||||
let g = linear_graph(&["start", "end"]);
|
||||
let mut state = RunState::<()>::new(&g).unwrap();
|
||||
let mut state = ExecutionState::<()>::new(&g).unwrap();
|
||||
state.context.set("key", json!("value"));
|
||||
|
||||
state.restart("start", None);
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ use async_trait::async_trait;
|
|||
use fabro_core::graph::NodeSpec;
|
||||
use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, RunLifecycle};
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use crate::artifact::{ArtifactStore, offload_large_values, sync_artifacts_to_env};
|
||||
use crate::asset_snapshot::collect_assets;
|
||||
|
|
@ -17,7 +17,7 @@ use crate::outcome::StageUsage;
|
|||
use fabro_core::error::Result as CoreResult;
|
||||
use fabro_core::lifecycle::NodeDecision;
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
|
||||
|
||||
|
|
|
|||
|
|
@ -3,13 +3,13 @@ use async_trait::async_trait;
|
|||
use fabro_core::error::Result as CoreResult;
|
||||
use fabro_core::lifecycle::RunLifecycle;
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use crate::graph::WorkflowGraph;
|
||||
use crate::graph::WorkflowNode;
|
||||
use crate::outcome::{StageStatus, StageUsage};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
|
||||
/// Sub-lifecycle responsible for auto-status override on nodes with `auto_status=true`.
|
||||
|
|
|
|||
|
|
@ -6,14 +6,14 @@ use async_trait::async_trait;
|
|||
use fabro_core::error::{CoreError, Result as CoreResult};
|
||||
use fabro_core::lifecycle::{EdgeContext, EdgeDecision, RunLifecycle};
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use crate::error::{FailureCategory, FailureSignature, FailureSignatureExt};
|
||||
use crate::graph::WorkflowGraph;
|
||||
use crate::graph::WorkflowNode;
|
||||
use crate::outcome::{OutcomeExt, StageStatus, StageUsage};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
|
||||
/// Sub-lifecycle responsible for tracking failure signatures and tripping the
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ use fabro_core::error::Result as CoreResult;
|
|||
use fabro_core::graph::NodeSpec;
|
||||
use fabro_core::lifecycle::RunLifecycle;
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use super::circuit_breaker::CircuitBreakerLifecycle;
|
||||
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent, append_workflow_event};
|
||||
|
|
@ -19,7 +19,7 @@ use crate::outcome::StageUsage;
|
|||
use crate::run_options::RunOptions;
|
||||
use fabro_graphviz::graph::types::Graph as GvGraph;
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
|
||||
/// Sub-lifecycle responsible for emitting store-backed run lifecycle events.
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ use fabro_core::lifecycle::{
|
|||
AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle,
|
||||
};
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use super::circuit_breaker::CircuitBreakerLifecycle;
|
||||
use super::git::GitCheckpointResult;
|
||||
|
|
@ -25,7 +25,7 @@ use crate::outcome::{
|
|||
};
|
||||
use fabro_types::{RunId, StatusReason};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
type FailureSignatureSnapshot = (
|
||||
Option<BTreeMap<String, usize>>,
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ use async_trait::async_trait;
|
|||
use fabro_core::error::Result as CoreResult;
|
||||
use fabro_core::graph::NodeSpec;
|
||||
use fabro_core::lifecycle::{EdgeContext, EdgeDecision, NodeDecision, RunLifecycle};
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode};
|
||||
|
||||
use crate::context::keys;
|
||||
|
|
@ -14,7 +14,7 @@ use crate::graph::WorkflowNode;
|
|||
use crate::handler::llm::preamble::build_preamble;
|
||||
use crate::outcome::StageUsage;
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
|
||||
|
||||
/// Graphviz edge captured from edge selection, passed to the next node's before_node
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ use fabro_core::error::{CoreError, Result as CoreResult};
|
|||
use fabro_core::graph::NodeSpec;
|
||||
use fabro_core::lifecycle::RunLifecycle;
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use crate::artifact::ArtifactStore;
|
||||
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
|
||||
|
|
@ -21,7 +21,7 @@ use crate::run_dump::RunDump;
|
|||
use crate::run_options::RunOptions;
|
||||
use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
|
||||
/// Result of a git checkpoint operation, shared with EventLifecycle.
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ use fabro_core::lifecycle::{
|
|||
AttemptContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle,
|
||||
};
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use crate::graph::WorkflowGraph;
|
||||
use crate::graph::WorkflowNode;
|
||||
|
|
@ -18,7 +18,7 @@ use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
|
|||
use fabro_sandbox::Sandbox;
|
||||
use fabro_types::RunId;
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
|
||||
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ use fabro_core::lifecycle::{
|
|||
AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle,
|
||||
};
|
||||
use fabro_core::outcome::NodeResult;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
|
||||
use crate::artifact::ArtifactStore;
|
||||
use crate::context;
|
||||
|
|
@ -48,7 +48,7 @@ use self::git::{GitCheckpointResult, GitLifecycle};
|
|||
use self::hook::HookLifecycle;
|
||||
use crate::outcome::OutcomeExt;
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfRunState = ExecutionState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
type WfNodeDecision = NodeDecision<Option<StageUsage>>;
|
||||
|
||||
|
|
|
|||
|
|
@ -137,7 +137,7 @@ mod tests {
|
|||
use fabro_core::executor::ExecutorBuilder;
|
||||
use fabro_core::lifecycle::NoopLifecycle;
|
||||
use fabro_core::outcome::StageStatus;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
use fabro_graphviz::graph::AttrValue;
|
||||
use fabro_graphviz::graph::types::{Edge, Graph, Node};
|
||||
|
||||
|
|
@ -183,7 +183,7 @@ mod tests {
|
|||
|
||||
let wf_graph = WorkflowGraph(Arc::new(graph));
|
||||
let handler: Arc<dyn NodeHandler<WorkflowGraph>> = Arc::new(SpikeHandler);
|
||||
let state = RunState::new(&wf_graph).unwrap();
|
||||
let state = ExecutionState::new(&wf_graph).unwrap();
|
||||
|
||||
let executor = ExecutorBuilder::new(handler)
|
||||
.lifecycle(Box::new(NoopLifecycle))
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ use std::time::Instant;
|
|||
|
||||
use fabro_core::executor::ExecutorBuilder;
|
||||
use fabro_core::handler::NodeHandler;
|
||||
use fabro_core::state::RunState;
|
||||
use fabro_core::state::ExecutionState;
|
||||
use tokio::time::sleep;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
|
|
@ -114,7 +114,7 @@ pub async fn execute(init: Initialized) -> Executed {
|
|||
}
|
||||
|
||||
let state = if let Some(ref cp) = checkpoint {
|
||||
match RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) {
|
||||
match ExecutionState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) {
|
||||
Ok(mut s) => {
|
||||
for (k, v) in &cp.context_values {
|
||||
s.context.set(k.clone(), v.clone());
|
||||
|
|
@ -162,7 +162,7 @@ pub async fn execute(init: Initialized) -> Executed {
|
|||
}
|
||||
}
|
||||
} else if let Some(seed) = seed_context {
|
||||
match RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) {
|
||||
match ExecutionState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) {
|
||||
Ok(s) => {
|
||||
for (k, v) in seed.snapshot() {
|
||||
s.context.set(k, v);
|
||||
|
|
@ -187,7 +187,7 @@ pub async fn execute(init: Initialized) -> Executed {
|
|||
}
|
||||
}
|
||||
} else {
|
||||
match RunState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) {
|
||||
match ExecutionState::new(&wf_graph).map_err(|e| FabroError::engine(e.to_string())) {
|
||||
Ok(s) => s,
|
||||
Err(err) => {
|
||||
return Executed {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue