scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-05 08:10:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Align core adapter lifecycle with final plan

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-24 15:29:16 -04:00
parent d285965596
commit 668b70842e
8 changed files with 124 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
lib/crates/fabro-workflows/src/core_adapter/lifecycle/artifact.rs
View file

@ -11,7 +11,6 @@ use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::artifact::{offload_large_values, sync_artifacts_to_env, ArtifactStore};
use crate::engine;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::outcome::StageUsage;
use fabro_core::lifecycle::NodeDecision;
@ -64,6 +63,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
// Swap in a fresh artifact store on restart (don't call clear() — preserves files on disk)
let mut store = self.artifact_store.lock().unwrap();
*store = ArtifactStore::new(self.artifact_base_dir.clone());
*self.attempt_start_epoch.lock().unwrap() = None;
Ok(())
}
@ -92,7 +92,17 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
let epoch = self.attempt_start_epoch.lock().unwrap().unwrap_or(0.0);
let node_id = ctx.node.id();
let visit = state.node_visits.get(node_id).copied().unwrap_or(1);
let stage_dir = engine::node_dir(&self.run_dir, node_id, visit);
let node_slug = if visit <= 1 {
node_id.to_string()
} else {
format!("{node_id}-visit_{visit}")
};
let stage_dir = self
.run_dir
.join("artifacts")
.join("assets")
.join(node_slug)
.join(format!("retry_{}", ctx.attempt));
let _ = std::fs::create_dir_all(&stage_dir);
match crate::asset_snapshot::collect_assets(

9
lib/crates/fabro-workflows/src/core_adapter/lifecycle/circuit_breaker.rs
View file

@ -10,6 +10,7 @@ use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::engine;
use crate::error::{FailureCategory, FailureSignature};
use crate::outcome::{OutcomeExt, StageStatus, StageUsage};
@ -68,7 +69,7 @@ impl RunLifecycle<WorkflowGraph> for CircuitBreakerLifecycle {
let outcome = &result.outcome;
let outcome_failure_category = if outcome.status == StageStatus::Fail {
outcome.failure.as_ref().map(|f| f.category)
engine::classify_outcome(outcome)
} else {
None
};
@ -116,10 +117,10 @@ impl RunLifecycle<WorkflowGraph> for CircuitBreakerLifecycle {
let outcome = ctx.outcome;
// Guard: only TransientInfra failures may trigger loop_restart
let failure_class = outcome.failure_category();
let failure_class = engine::classify_outcome(outcome);
if let Some(fc) = failure_class {
if fc != FailureCategory::TransientInfra {
return Err(CoreError::blocked(format!(
return Ok(EdgeDecision::Block(format!(
"loop_restart blocked: failure_class={fc} (requires transient_infra), failure_reason={}",
outcome.failure_reason().unwrap_or("none"),
)));
@ -140,7 +141,7 @@ impl RunLifecycle<WorkflowGraph> for CircuitBreakerLifecycle {
*count += 1;
let limit = self.loop_restart_signature_limit;
if *count >= limit {
return Err(CoreError::blocked(format!(
return Ok(EdgeDecision::Block(format!(
"loop_restart circuit breaker: signature {sig} repeated {count} times (limit {limit})"
)));
}

30
lib/crates/fabro-workflows/src/core_adapter/lifecycle/event.rs
View file

@ -12,6 +12,7 @@ use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use super::git::GitCheckpointResult;
use crate::artifact::ArtifactStore;
use crate::engine;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage};
@ -86,7 +87,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().map(String::from),
script: None,
script: engine::node_script(gv),
attempt: 1,
max_attempts: 1,
});
@ -118,7 +119,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
name: gv.label().to_string(),
index: state.stage_index,
handler_type: gv.handler_type().map(String::from),
script: None,
script: engine::node_script(gv),
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
});
@ -192,7 +193,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
preferred_label: outcome.preferred_label.clone(),
suggested_next_ids: outcome.suggested_next_ids.clone(),
usage: outcome.usage.clone(),
failure: None,
failure: outcome.failure.clone(),
notes: outcome.notes.clone(),
files_touched: outcome.files_touched.clone(),
attempt: result.attempts as usize,
@ -240,7 +241,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let status = result.outcome.status.to_string();
// Read git checkpoint result (set by GitLifecycle)
let git_result = self.checkpoint_git_result.lock().unwrap().take();
let git_result = self.checkpoint_git_result.lock().unwrap().clone();
let git_sha = git_result.as_ref().and_then(|r| r.commit_sha.clone());
@ -276,15 +277,32 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let duration_ms = self.run_start.lock().unwrap().elapsed().as_millis() as u64;
let artifact_count = self.artifact_store.lock().unwrap().list().len();
let last_sha = self.last_git_sha.lock().unwrap().clone();
let total_cost = {
let sum: f64 = state
.node_outcomes
.values()
.filter_map(|o| o.usage.as_ref()?.cost)
.sum();
if sum > 0.0 {
Some(sum)
} else {
None
}
};
let run_usage = state
.node_outcomes
.values()
.filter_map(|o| o.usage.as_ref().map(fabro_llm::types::Usage::from))
.reduce(|a, b| a + b);
if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess {
self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count,
status: outcome.status.to_string(),
total_cost: None,
total_cost,
final_git_commit_sha: last_sha,
usage: None,
usage: run_usage,
});
} else {
let error_msg = outcome

4
lib/crates/fabro-workflows/src/core_adapter/lifecycle/fidelity.rs
View file

@ -111,7 +111,7 @@ impl RunLifecycle<WorkflowGraph> for FidelityLifecycle {
// 6. Set thread.{tid}.current_node
if let Some(ref tid) = thread_id {
let key = format!("thread.{tid}.current_node");
let key = keys::thread_current_node_key(tid);
state.context.set(key, serde_json::json!(node.id()));
}
@ -130,7 +130,7 @@ impl RunLifecycle<WorkflowGraph> for FidelityLifecycle {
}
// 8. Set INTERNAL_NODE_VISIT_COUNT and CURRENT_NODE
let visits = state.node_visits.get(node.id()).copied().unwrap_or(0);
let visits = state.node_visits.get(node.id()).copied().unwrap_or(1);
state
.context
.set(keys::CURRENT_NODE, serde_json::json!(node.id()));

2
lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs
View file

@ -49,6 +49,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
) -> fabro_core::error::Result<()> {
// Reset last_git_sha (diff base parity)
*self.last_git_sha.lock().unwrap() = None;
*self.checkpoint_git_result.lock().unwrap() = None;
// Init metadata branch (best-effort)
if let (Some(_), Some(ref repo_path)) =
@ -91,6 +92,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
// Skip git checkpoint for the start node (always empty) or if git disabled
if self.start_node_id.as_deref() == Some(node_id) || !self.config.git_checkpoint_enabled {
*self.checkpoint_git_result.lock().unwrap() = None;
return Ok(());
}

24
lib/crates/fabro-workflows/src/core_adapter/lifecycle/hook.rs
View file

@ -13,7 +13,7 @@ use fabro_core::state::RunState;
use super::super::graph::WorkflowGraph;
use super::super::WorkflowNode;
use crate::engine::set_hook_node;
use crate::outcome::{Outcome, StageStatus, StageUsage};
use crate::outcome::{Outcome, OutcomeExt, StageStatus, StageUsage};
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
use fabro_sandbox::Sandbox;
@ -25,7 +25,7 @@ type WfNodeDecision = NodeDecision<Option<StageUsage>>;
pub struct HookLifecycle {
pub hook_runner: Option<Arc<HookRunner>>,
pub sandbox: Arc<dyn Sandbox>,
pub run_dir: PathBuf,
pub hook_work_dir: Option<PathBuf>,
pub run_id: String,
pub graph_name: String,
}
@ -36,7 +36,11 @@ impl HookLifecycle {
return HookDecision::Proceed;
};
runner
.run(hook_ctx, self.sandbox.clone(), Some(&self.run_dir))
.run(
hook_ctx,
self.sandbox.clone(),
self.hook_work_dir.as_deref(),
)
.await
}
}
@ -68,13 +72,17 @@ impl RunLifecycle<WorkflowGraph> for HookLifecycle {
self.run_id.clone(),
self.graph_name.clone(),
);
hook_ctx.cwd = self
.hook_work_dir
.as_ref()
.map(|path| path.display().to_string());
set_hook_node(&mut hook_ctx, gv);
hook_ctx.attempt = Some(ctx.attempt as usize);
hook_ctx.max_attempts = Some(ctx.max_attempts as usize);
let decision = self.run_hook(&hook_ctx).await;
match decision {
HookDecision::Skip { reason } => {
let msg = reason.unwrap_or_else(|| "skipped by hook".into());
let msg = reason.unwrap_or_else(|| "skipped by StageStart hook".into());
Ok(NodeDecision::Skip(Box::new(Outcome::skipped(&msg))))
}
HookDecision::Block { reason } => {
@ -87,7 +95,7 @@ impl RunLifecycle<WorkflowGraph> for HookLifecycle {
async fn after_node(
&self,
_node: &WorkflowNode,
node: &WorkflowNode,
result: &mut WfNodeResult,
_state: &WfRunState,
) -> CoreResult<()> {
@ -103,7 +111,9 @@ impl RunLifecycle<WorkflowGraph> for HookLifecycle {
};
let mut hook_ctx =
HookContext::new(hook_event, self.run_id.clone(), self.graph_name.clone());
set_hook_node(&mut hook_ctx, node.inner());
hook_ctx.status = Some(outcome.status.to_string());
hook_ctx.failure_reason = outcome.failure_reason().map(String::from);
let _ = self.run_hook(&hook_ctx).await;
Ok(())
}
@ -120,6 +130,10 @@ impl RunLifecycle<WorkflowGraph> for HookLifecycle {
);
hook_ctx.edge_from = Some(ctx.from.to_string());
hook_ctx.edge_to = Some(ctx.to.to_string());
hook_ctx.edge_label = ctx
.edge
.as_ref()
.and_then(|edge| edge.inner().label().map(String::from));
let decision = self.run_hook(&hook_ctx).await;
match decision {
HookDecision::Override { edge_to } => Ok(EdgeDecision::Override(edge_to)),

2
lib/crates/fabro-workflows/src/core_adapter/lifecycle/mod.rs
View file

@ -115,7 +115,7 @@ impl WorkflowLifecycle {
let hook = HookLifecycle {
hook_runner,
sandbox: Arc::clone(&sandbox),
run_dir: run_dir.clone(),
hook_work_dir: working_directory.clone().map(PathBuf::from),
run_id: config.run_id.clone(),
graph_name: graph.name.clone(),
};

83
lib/crates/fabro-workflows/src/engine.rs
View file

@ -46,7 +46,7 @@ pub(crate) fn set_hook_node(ctx: &mut HookContext, node: &Node) {
/// 2. String heuristics on `failure_reason`
/// 3. Default to `Deterministic`
#[must_use]
fn classify_outcome(outcome: &Outcome) -> Option<FailureCategory> {
pub(crate) fn classify_outcome(outcome: &Outcome) -> Option<FailureCategory> {
match outcome.status {
StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None,
StageStatus::Fail | StageStatus::Retry => outcome
@ -57,6 +57,7 @@ fn classify_outcome(outcome: &Outcome) -> Option<FailureCategory> {
/// Mutable state carried across loop restarts and recursive `run_internal` calls.
#[derive(Default)]
#[cfg_attr(feature = "core-engine", allow(dead_code))]
struct LoopState {
node_visits: HashMap<String, usize>,
/// Tracks deterministic/structural failure signatures across main-loop stages.
@ -542,7 +543,7 @@ pub(crate) fn is_terminal(node: &Node) -> bool {
node.shape() == "Msquare" || node.handler_type() == Some("exit")
}
fn node_script(node: &Node) -> Option<String> {
pub(crate) fn node_script(node: &Node) -> Option<String> {
node.attrs
.get("script")
.or_else(|| node.attrs.get("tool_command"))
@ -923,6 +924,7 @@ impl WorkflowRunEngine {
}
/// Fire a non-blocking RunFailed hook.
#[cfg_attr(feature = "core-engine", allow(dead_code))]
async fn run_failed_hook(
&self,
run_id: &str,
@ -940,6 +942,7 @@ impl WorkflowRunEngine {
}
/// Mirror graph-level attributes into the context.
#[cfg_attr(feature = "core-engine", allow(dead_code))]
pub(crate) fn mirror_graph_attributes(graph: &Graph, context: &Context) {
if !graph.goal().is_empty() {
context.set(context::keys::GRAPH_GOAL, serde_json::json!(graph.goal()));
@ -955,6 +958,7 @@ impl WorkflowRunEngine {
/// Execute a node handler with retry policy.
/// Returns `(outcome, attempts_used)` where `attempts_used` is the 1-indexed count.
#[allow(clippy::too_many_arguments)]
#[cfg_attr(feature = "core-engine", allow(dead_code))]
async fn execute_with_retry(
&self,
node: &Node,
@ -1181,10 +1185,19 @@ impl WorkflowRunEngine {
/// Returns an error if no start node is found, a node is missing, or a goal gate fails
/// without a retry target.
pub async fn run(&self, graph: &Graph, config: &RunConfig) -> Result<Outcome> {
let (outcome, _context) = self
.run_internal(graph, config, None, None, None, LoopState::default())
.await?;
Ok(outcome)
#[cfg(feature = "core-engine")]
{
let (outcome, _context) = self.run_via_core(graph, config, None, None).await?;
return Ok(outcome);
}
#[cfg(not(feature = "core-engine"))]
{
let (outcome, _context) = self
.run_internal(graph, config, None, None, None, LoopState::default())
.await?;
Ok(outcome)
}
}
/// Run a workflow with full sandbox lifecycle management.
@ -1389,15 +1402,25 @@ impl WorkflowRunEngine {
config: &RunConfig,
seed_context: Context,
) -> Result<(Outcome, Context)> {
self.run_internal(
graph,
config,
None,
None,
Some(seed_context),
LoopState::default(),
)
.await
#[cfg(feature = "core-engine")]
{
return self
.run_via_core(graph, config, None, Some(seed_context))
.await;
}
#[cfg(not(feature = "core-engine"))]
{
self.run_internal(
graph,
config,
None,
None,
Some(seed_context),
LoopState::default(),
)
.await
}
}
/// Resume from a checkpoint. Restores context, completed nodes, and continues
@ -1412,15 +1435,26 @@ impl WorkflowRunEngine {
config: &RunConfig,
checkpoint: &Checkpoint,
) -> Result<Outcome> {
let loop_state = LoopState {
node_visits: HashMap::new(),
loop_failure_signatures: checkpoint.loop_failure_signatures.clone(),
restart_failure_signatures: checkpoint.restart_failure_signatures.clone(),
};
let (outcome, _context) = self
.run_internal(graph, config, Some(checkpoint), None, None, loop_state)
.await?;
Ok(outcome)
#[cfg(feature = "core-engine")]
{
let (outcome, _context) = self
.run_via_core(graph, config, Some(checkpoint), None)
.await?;
return Ok(outcome);
}
#[cfg(not(feature = "core-engine"))]
{
let loop_state = LoopState {
node_visits: HashMap::new(),
loop_failure_signatures: checkpoint.loop_failure_signatures.clone(),
restart_failure_signatures: checkpoint.restart_failure_signatures.clone(),
};
let (outcome, _context) = self
.run_internal(graph, config, Some(checkpoint), None, None, loop_state)
.await?;
Ok(outcome)
}
}
/// Run the workflow through the fabro-core executor with full lifecycle management.
@ -1660,6 +1694,7 @@ impl WorkflowRunEngine {
}
/// Internal run implementation supporting optional checkpoint resume and `start_at` override.
#[cfg_attr(feature = "core-engine", allow(dead_code))]
async fn run_internal(
&self,
graph: &Graph,