mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-08 22:21:45 +00:00
Remove core-engine feature flag and old execution loop
Fix parity gaps (handler errors → fail outcomes, panic.txt, goal gate message, fail-with-no-edge message, visit limit source, terminal completion normalization) then delete ~1,250 lines of old-path code (LoopState, run_failed_hook, mirror_graph_attributes, execute_with_retry, run_internal) and all cfg gating. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
bf63ba8b70
commit
f022298728
8 changed files with 76 additions and 1376 deletions
|
|
@ -2,6 +2,21 @@ use std::fmt;
|
|||
|
||||
use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeMeta, StageStatus};
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum VisitLimitSource {
|
||||
Node,
|
||||
Graph,
|
||||
}
|
||||
|
||||
impl fmt::Display for VisitLimitSource {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::Node => write!(f, "node"),
|
||||
Self::Graph => write!(f, "graph"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Structured failure data on handler errors. Maps to FabroError's
|
||||
/// is_retryable(), failure_class(), failure_signature_hint(), to_fail_outcome().
|
||||
#[derive(Debug, Clone)]
|
||||
|
|
@ -28,11 +43,12 @@ pub enum CoreError {
|
|||
Cancelled,
|
||||
#[error("blocked: {message}")]
|
||||
Blocked { message: String },
|
||||
#[error("node \"{node_id}\" visited {visits} times (limit {limit})")]
|
||||
#[error("node \"{node_id}\" visited {visits} times ({limit_source} limit {limit}); run is stuck in a cycle")]
|
||||
VisitLimitExceeded {
|
||||
node_id: String,
|
||||
visits: usize,
|
||||
limit: usize,
|
||||
limit_source: VisitLimitSource,
|
||||
},
|
||||
#[error("stall timeout on node \"{node_id}\"")]
|
||||
StallTimeout { node_id: String },
|
||||
|
|
@ -101,10 +117,11 @@ mod tests {
|
|||
CoreError::VisitLimitExceeded {
|
||||
node_id: "n1".into(),
|
||||
visits: 5,
|
||||
limit: 3
|
||||
limit: 3,
|
||||
limit_source: VisitLimitSource::Node,
|
||||
}
|
||||
.to_string(),
|
||||
"node \"n1\" visited 5 times (limit 3)"
|
||||
"node \"n1\" visited 5 times (node limit 3); run is stuck in a cycle"
|
||||
);
|
||||
assert_eq!(
|
||||
CoreError::StallTimeout {
|
||||
|
|
|
|||
|
|
@ -131,8 +131,7 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
continue;
|
||||
}
|
||||
let outcome = Outcome::fail(&format!(
|
||||
"goal gate failed for node \"{}\"",
|
||||
failed_node_id
|
||||
"goal gate unsatisfied for node {failed_node_id} and no retry target"
|
||||
));
|
||||
self.lifecycle.on_run_end(&outcome, &state).await;
|
||||
return Ok((outcome, state));
|
||||
|
|
@ -148,6 +147,7 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
node_id: node.id().to_string(),
|
||||
visits,
|
||||
limit: max,
|
||||
limit_source: crate::error::VisitLimitSource::Node,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -157,6 +157,7 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
node_id: node.id().to_string(),
|
||||
visits,
|
||||
limit: global_max,
|
||||
limit_source: crate::error::VisitLimitSource::Graph,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -225,7 +226,13 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
self.lifecycle.on_run_start(graph, &state).await?;
|
||||
}
|
||||
NextStep::End => {
|
||||
let outcome = last_outcome.clone();
|
||||
let mut outcome = last_outcome.clone();
|
||||
if outcome.status == StageStatus::Fail {
|
||||
outcome = Outcome::fail(&format!(
|
||||
"stage {} failed with no outgoing fail edge",
|
||||
node.id()
|
||||
));
|
||||
}
|
||||
self.lifecycle.on_run_end(&outcome, &state).await;
|
||||
return Ok((outcome, state));
|
||||
}
|
||||
|
|
@ -317,17 +324,19 @@ impl<G: Graph + 'static> Executor<G> {
|
|||
tokio::time::sleep(delay).await;
|
||||
}
|
||||
Err(e) => {
|
||||
let fail_result =
|
||||
NodeResult::from_error(&e, start.elapsed(), attempt, policy.max_attempts);
|
||||
// Convert handler error to fail outcome so routing continues
|
||||
let outcome = e.to_fail_outcome();
|
||||
let result =
|
||||
NodeResult::new(outcome, start.elapsed(), attempt, policy.max_attempts);
|
||||
let ctx = AttemptResultContext {
|
||||
node,
|
||||
result: &fail_result,
|
||||
result: &result,
|
||||
attempt,
|
||||
will_retry: false,
|
||||
backoff_delay: None,
|
||||
};
|
||||
self.lifecycle.after_attempt(&ctx, state).await?;
|
||||
return Err(e);
|
||||
return Ok(result);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -978,7 +987,8 @@ mod tests {
|
|||
handler.clone() as Arc<dyn NodeHandler<TestGraph>>,
|
||||
)
|
||||
.await;
|
||||
assert!(result.is_err());
|
||||
// Non-retryable errors become fail outcomes, routing continues through the linear graph
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(handler.calls(), 1);
|
||||
}
|
||||
|
||||
|
|
@ -998,7 +1008,8 @@ mod tests {
|
|||
handler.clone() as Arc<dyn NodeHandler<TestGraph>>,
|
||||
)
|
||||
.await;
|
||||
assert!(result.is_err());
|
||||
// Errors become fail outcomes, routing continues through the linear graph
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(handler.calls(), 1);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ pub mod state;
|
|||
pub mod test_fixtures;
|
||||
|
||||
pub use context::Context;
|
||||
pub use error::{CoreError, HandlerErrorDetail, Result};
|
||||
pub use error::{CoreError, HandlerErrorDetail, Result, VisitLimitSource};
|
||||
pub use executor::{Executor, ExecutorBuilder, ExecutorSettings};
|
||||
pub use graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec};
|
||||
pub use handler::NodeHandler;
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ doctest = false
|
|||
|
||||
[features]
|
||||
default = []
|
||||
core-engine = []
|
||||
exedev = ["fabro-sandbox/exe", "fabro-config/exedev"]
|
||||
|
||||
[dependencies]
|
||||
|
|
|
|||
|
|
@ -97,6 +97,10 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
|
|||
}
|
||||
Err(panic_payload) => {
|
||||
let msg = format_panic_message(panic_payload);
|
||||
let visit = context.node_visit_count().max(1);
|
||||
let panic_dir = crate::engine::node_dir(&self.run_dir, &gv_node.id, visit);
|
||||
let _ = std::fs::create_dir_all(&panic_dir);
|
||||
let _ = std::fs::write(panic_dir.join("panic.txt"), &msg);
|
||||
Err(CoreError::handler(HandlerErrorDetail {
|
||||
message: msg,
|
||||
retryable: false,
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ pub struct WorkflowLifecycle {
|
|||
/// Shared git checkpoint result (written by git, read by event)
|
||||
checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
|
||||
/// True when constructed with a checkpoint; cleared after first on_run_start.
|
||||
/// Gates mirror_graph_attributes on initial resume.
|
||||
/// Gates context seeding on initial resume.
|
||||
is_initial_resume: AtomicBool,
|
||||
// Config needed for context seeding
|
||||
graph: Arc<fabro_graphviz::graph::types::Graph>,
|
||||
|
|
@ -194,9 +194,9 @@ impl WorkflowLifecycle {
|
|||
impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
|
||||
async fn on_run_start(&self, graph: &WorkflowGraph, state: &WfRunState) -> CoreResult<()> {
|
||||
// Re-seed context keys (fires on initial start AND after every loop restart).
|
||||
// mirror_graph_attributes: skip on initial checkpoint resume (context already has them)
|
||||
// Skip on initial checkpoint resume (context already has them).
|
||||
if self.is_initial_resume.swap(false, Ordering::Relaxed) {
|
||||
// First on_run_start after checkpoint resume — skip mirror_graph_attributes
|
||||
// First on_run_start after checkpoint resume — skip context seeding
|
||||
} else {
|
||||
// Mirror graph-level attributes into the core context
|
||||
if !self.graph.goal().is_empty() {
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -33,7 +33,7 @@ pub struct EngineServices {
|
|||
pub emitter: Arc<EventEmitter>,
|
||||
pub sandbox: Arc<dyn Sandbox>,
|
||||
/// Git state for the current run. Set via `set_git_state` at the start of
|
||||
/// `run_internal` and read by parallel/fan-in handlers.
|
||||
/// `run_via_core` and read by parallel/fan-in handlers.
|
||||
pub(crate) git_state: std::sync::RwLock<Option<Arc<GitState>>>,
|
||||
/// Hook runner for user-defined lifecycle hooks.
|
||||
pub hook_runner: Option<Arc<HookRunner>>,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue