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

Add failure signatures and circuit breakers for deterministic failure cycle detection

Browse source 
Introduces reason normalization (strip variable data like line numbers,
hex hashes), composite failure signatures (node_id|class|normalized_reason),
and circuit breakers that track signature counts to abort when the same
deterministic failure repeats beyond a configurable limit (default 3).

Key additions:
- normalize_failure_reason() strips hex, digits, whitespace for stable grouping
- FailureSignature type with handler-provided hint priority
- FailureClass::is_signature_tracked() (deterministic + structural only)
- Graph-level loop_restart_signature_limit attribute
- LoopState struct bundling node_visits + signature maps through run_internal
- Loop failure circuit breaker (same node repeating)
- Restart failure circuit breaker (across loop_restart edges)
- Checkpoint persistence for both signature maps with backward compat
- 18 e2e integration tests covering all circuit breaker scenarios

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-01 18:03:09 -05:00
parent f53cc746dc
commit cab25a3e7e
9 changed files with 1830 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -268,6 +268,7 @@ dependencies = [
"nom",
"predicates",
"rand 0.8.5",
"regex",
"scopeguard",
"serde",
"serde_json",

1
crates/arc-workflows/Cargo.toml
View file

@ -37,6 +37,7 @@ dialoguer.workspace = true
daytona-sdk.workspace = true
daytona-api-client.workspace = true
base64.workspace = true
regex.workspace = true
scopeguard = "1"
git2.workspace = true
tokio-util.workspace = true

92
crates/arc-workflows/src/checkpoint.rs
View file

@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::context::Context;
use crate::error::{ArcError, Result};
use crate::error::{ArcError, FailureSignature, Result};
use crate::outcome::Outcome;
/// Serializable snapshot of execution state for crash recovery and resume.
@ -27,10 +27,17 @@ pub struct Checkpoint {
/// SHA of the git commit created at this checkpoint (when running in a worktree).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub git_commit_sha: Option<String>,
/// Failure signature counts within the main loop (deterministic/structural failures).
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub loop_failure_signatures: HashMap<FailureSignature, usize>,
/// Failure signature counts across loop_restart edges.
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub restart_failure_signatures: HashMap<FailureSignature, usize>,
}
impl Checkpoint {
/// Create a checkpoint from the current execution state.
#[allow(clippy::too_many_arguments)]
pub fn from_context(
context: &Context,
current_node: impl Into<String>,
@ -38,6 +45,8 @@ impl Checkpoint {
node_retries: HashMap<String, u32>,
node_outcomes: HashMap<String, Outcome>,
next_node_id: Option<String>,
loop_failure_signatures: HashMap<FailureSignature, usize>,
restart_failure_signatures: HashMap<FailureSignature, usize>,
) -> Self {
Self {
timestamp: Utc::now(),
@ -49,6 +58,8 @@ impl Checkpoint {
node_outcomes,
next_node_id,
git_commit_sha: None,
loop_failure_signatures,
restart_failure_signatures,
}
}
@ -94,6 +105,8 @@ mod tests {
HashMap::new(),
HashMap::new(),
None,
HashMap::new(),
HashMap::new(),
);
assert_eq!(cp.current_node, "node_a");
@ -131,6 +144,8 @@ mod tests {
retries,
outcomes,
Some("next_step".to_string()),
HashMap::new(),
HashMap::new(),
);
cp.save(&path).unwrap();
@ -170,10 +185,83 @@ mod tests {
#[test]
fn serialization_roundtrip() {
let ctx = Context::new();
let cp = Checkpoint::from_context(&ctx, "n1", vec![], HashMap::new(), HashMap::new(), None);
let cp = Checkpoint::from_context(
&ctx,
"n1",
vec![],
HashMap::new(),
HashMap::new(),
None,
HashMap::new(),
HashMap::new(),
);
let json = serde_json::to_string(&cp).unwrap();
let deserialized: Checkpoint = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized.current_node, "n1");
}
#[test]
fn signature_maps_roundtrip() {
use crate::error::{FailureClass, FailureSignature};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("checkpoint.json");
let ctx = Context::new();
let mut loop_sigs = HashMap::new();
loop_sigs.insert(
FailureSignature::new(
"verify",
FailureClass::Deterministic,
None,
Some("test failed"),
),
2,
);
let mut restart_sigs = HashMap::new();
restart_sigs.insert(
FailureSignature::new("build", FailureClass::Structural, None, Some("scope error")),
1,
);
let cp = Checkpoint::from_context(
&ctx,
"verify",
vec![],
HashMap::new(),
HashMap::new(),
None,
loop_sigs,
restart_sigs,
);
cp.save(&path).unwrap();
let loaded = Checkpoint::load(&path).unwrap();
assert_eq!(loaded.loop_failure_signatures.len(), 1);
assert_eq!(loaded.restart_failure_signatures.len(), 1);
let sig = FailureSignature::new(
"verify",
FailureClass::Deterministic,
None,
Some("test failed"),
);
assert_eq!(loaded.loop_failure_signatures.get(&sig), Some(&2));
}
#[test]
fn backward_compat_missing_signature_fields() {
// A checkpoint saved before signatures were added should deserialize with empty maps
let json = r#"{
"timestamp": "2025-01-01T00:00:00Z",
"current_node": "work",
"completed_nodes": ["start"],
"node_retries": {},
"context_values": {},
"logs": []
}"#;
let cp: Checkpoint = serde_json::from_str(json).unwrap();
assert!(cp.loop_failure_signatures.is_empty());
assert!(cp.restart_failure_signatures.is_empty());
}
}

440
crates/arc-workflows/src/engine.rs
View file

@ -16,7 +16,7 @@ use crate::artifact::{offload_large_values, sync_artifacts_to_env, ArtifactStore
use crate::checkpoint::Checkpoint;
use crate::condition::evaluate_condition;
use crate::context::Context;
use crate::error::{classify_failure_reason, ArcError, FailureClass, Result};
use crate::error::{classify_failure_reason, ArcError, FailureClass, FailureSignature, Result};
use crate::event::{EventEmitter, PipelineEvent};
use crate::graph::{Edge, Graph, Node};
use crate::handler::{EngineServices, HandlerRegistry};
@ -59,6 +59,17 @@ fn classify_outcome(outcome: &Outcome) -> Option<FailureClass> {
}
}
/// Mutable state carried across loop restarts and recursive `run_internal` calls.
#[derive(Default)]
struct LoopState {
node_visits: HashMap<String, usize>,
/// Tracks deterministic/structural failure signatures across main-loop stages.
/// Never reset on success — prevents impl-succeeds/verify-fails cycles.
loop_failure_signatures: HashMap<FailureSignature, usize>,
/// Tracks failure signatures across loop_restart edges.
restart_failure_signatures: HashMap<FailureSignature, usize>,
}
// --- Retry policy types ---
/// Configuration for exponential backoff between retry attempts.
@ -666,10 +677,7 @@ pub async fn git_add_worktree_remote(
}
/// Remove a git worktree inside a remote execution environment.
pub async fn git_remove_worktree_remote(
exec_env: &dyn ExecutionEnvironment,
path: &str,
) -> bool {
pub async fn git_remove_worktree_remote(exec_env: &dyn ExecutionEnvironment, path: &str) -> bool {
let cmd = format!("{GIT_REMOTE} worktree remove --force {path}");
matches!(
exec_env.exec_command(&cmd, 30_000, None, None, None).await,
@ -678,10 +686,7 @@ pub async fn git_remove_worktree_remote(
}
/// Fast-forward merge to a given SHA inside a remote execution environment.
pub async fn git_merge_ff_only_remote(
exec_env: &dyn ExecutionEnvironment,
sha: &str,
) -> bool {
pub async fn git_merge_ff_only_remote(exec_env: &dyn ExecutionEnvironment, sha: &str) -> bool {
let cmd = format!("{GIT_REMOTE} merge --ff-only {sha}");
matches!(
exec_env.exec_command(&cmd, 30_000, None, None, None).await,
@ -692,10 +697,7 @@ pub async fn git_merge_ff_only_remote(
/// Get the current HEAD SHA from a remote execution environment.
pub async fn git_head_sha_remote(exec_env: &dyn ExecutionEnvironment) -> Option<String> {
let cmd = format!("{GIT_REMOTE} rev-parse HEAD");
match exec_env
.exec_command(&cmd, 10_000, None, None, None)
.await
{
match exec_env.exec_command(&cmd, 10_000, None, None, None).await {
Ok(r) if r.exit_code == 0 => Some(r.stdout.trim().to_string()),
_ => None,
}
@ -929,7 +931,7 @@ impl PipelineEngine {
/// 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> {
self.run_internal(graph, config, None, None, HashMap::new())
self.run_internal(graph, config, None, None, LoopState::default())
.await
}
@ -945,7 +947,12 @@ impl PipelineEngine {
config: &RunConfig,
checkpoint: &Checkpoint,
) -> Result<Outcome> {
self.run_internal(graph, config, Some(checkpoint), None, HashMap::new())
let loop_state = LoopState {
node_visits: HashMap::new(),
loop_failure_signatures: checkpoint.loop_failure_signatures.clone(),
restart_failure_signatures: checkpoint.restart_failure_signatures.clone(),
};
self.run_internal(graph, config, Some(checkpoint), None, loop_state)
.await
}
@ -956,7 +963,7 @@ impl PipelineEngine {
config: &RunConfig,
resume_checkpoint: Option<&Checkpoint>,
start_at: Option<&str>,
mut node_visits: HashMap<String, usize>,
mut loop_state: LoopState,
) -> Result<Outcome> {
let run_start = Instant::now();
let run_id = config.run_id.clone();
@ -1045,7 +1052,7 @@ impl PipelineEngine {
completed_nodes = cp.completed_nodes.clone();
// Rebuild visit counts from completed_nodes (which records every visit)
for id in &completed_nodes {
*node_visits.entry(id.clone()).or_insert(0) += 1;
*loop_state.node_visits.entry(id.clone()).or_insert(0) += 1;
}
// Gap #5: Restore retry counters from checkpoint
node_retries = cp.node_retries.clone();
@ -1108,7 +1115,10 @@ impl PipelineEngine {
.ok_or_else(|| ArcError::Engine(format!("node not found: {current_node_id}")))?;
// Always track visit count (used for stage directory naming)
let count = node_visits.entry(current_node_id.clone()).or_insert(0);
let count = loop_state
.node_visits
.entry(current_node_id.clone())
.or_insert(0);
*count += 1;
if max_node_visits > 0 && *count > max_node_visits {
return Err(ArcError::Engine(format!(
@ -1173,7 +1183,7 @@ impl PipelineEngine {
}
// Step 2: Execute node handler with retry policy
let visit = *node_visits.get(&current_node_id).unwrap_or(&1);
let visit = *loop_state.node_visits.get(&current_node_id).unwrap_or(&1);
context.set("internal.node_visit_count", serde_json::json!(visit));
context.set("current_node", serde_json::json!(&node.id));
let retry_policy = build_retry_policy(node, graph);
@ -1224,6 +1234,36 @@ impl PipelineEngine {
let outcome_failure_class = classify_outcome(&outcome);
// Circuit breaker: track deterministic/structural failure signatures
let failure_sig = if let Some(fc) = outcome_failure_class {
let sig_hint = outcome
.context_updates
.get("failure_signature")
.and_then(|v| v.as_str());
let sig = FailureSignature::new(
&node.id,
fc,
sig_hint,
outcome.failure_reason.as_deref(),
);
if fc.is_signature_tracked() {
let count = loop_state
.loop_failure_signatures
.entry(sig.clone())
.or_insert(0);
*count += 1;
let limit = graph.loop_restart_signature_limit();
if *count >= limit {
return Err(ArcError::Engine(format!(
"deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})"
)));
}
}
Some(sig)
} else {
None
};
if outcome.status == StageStatus::Fail {
self.services.emitter.emit(&PipelineEvent::StageFailed {
name: node.label().to_string(),
@ -1285,6 +1325,12 @@ impl PipelineEngine {
"failure_class",
serde_json::json!(outcome_failure_class.map_or(String::new(), |fc| fc.to_string())),
);
context.set(
"failure_signature",
serde_json::json!(failure_sig
.as_ref()
.map_or(String::new(), |s| s.to_string())),
);
if let Some(ref pref) = outcome.preferred_label {
context.set("preferred_label", serde_json::json!(pref));
}
@ -1309,6 +1355,8 @@ impl PipelineEngine {
node_retries.clone(),
node_outcomes.clone(),
next_node_id_for_checkpoint,
loop_state.loop_failure_signatures.clone(),
loop_state.restart_failure_signatures.clone(),
);
let checkpoint_path = config.logs_root.join("checkpoint.json");
if let Err(e) = checkpoint.save(&checkpoint_path) {
@ -1457,6 +1505,20 @@ impl PipelineEngine {
incoming_edge = Some(edge);
// Gap #6: Handle loop_restart by recursively running from the target
if edge.loop_restart() {
// Circuit breaker: check restart failure signatures
if let Some(ref sig) = failure_sig {
let count = loop_state
.restart_failure_signatures
.entry(sig.clone())
.or_insert(0);
*count += 1;
let limit = graph.loop_restart_signature_limit();
if *count >= limit {
return Err(ArcError::Engine(format!(
"loop_restart circuit breaker: signature {sig} repeated {count} times (limit {limit})"
)));
}
}
self.services.emitter.emit(&PipelineEvent::LoopRestart {
from_node: node.id.clone(),
to_node: edge.to.clone(),
@ -1466,7 +1528,7 @@ impl PipelineEngine {
config,
None,
Some(&edge.to),
node_visits,
loop_state,
))
.await;
}
@ -3483,4 +3545,342 @@ mod tests {
Some(FailureClass::TransientInfra)
);
}
// --- Circuit breaker tests ---
/// Build a graph where `work` always fails deterministically,
/// and a fail edge loops back to `work`.
fn looping_fail_graph() -> Graph {
let mut g = Graph::new("loop_fail");
g.attrs
.insert("goal".to_string(), AttrValue::String("test".to_string()));
g.attrs
.insert("default_max_retry".to_string(), AttrValue::Integer(0));
let mut start = Node::new("start");
start.attrs.insert(
"shape".to_string(),
AttrValue::String("Mdiamond".to_string()),
);
g.nodes.insert("start".to_string(), start);
let mut work = Node::new("work");
work.attrs.insert(
"type".to_string(),
AttrValue::String("always_fail".to_string()),
);
work.attrs
.insert("max_retries".to_string(), AttrValue::Integer(0));
g.nodes.insert("work".to_string(), work);
let mut exit = Node::new("exit");
exit.attrs.insert(
"shape".to_string(),
AttrValue::String("Msquare".to_string()),
);
g.nodes.insert("exit".to_string(), exit);
g.edges.push(Edge::new("start", "work"));
// Fail loops back
let mut fail_edge = Edge::new("work", "work");
fail_edge.attrs.insert(
"condition".to_string(),
AttrValue::String("outcome=fail".to_string()),
);
g.edges.push(fail_edge);
// Success goes to exit (never taken)
let mut ok_edge = Edge::new("work", "exit");
ok_edge.attrs.insert(
"condition".to_string(),
AttrValue::String("outcome=success".to_string()),
);
g.edges.push(ok_edge);
g
}
/// Handler that always returns transient_infra failure.
struct TransientFailHandler;
#[async_trait]
impl HandlerTrait for TransientFailHandler {
async fn execute(
&self,
_node: &Node,
_context: &Context,
_graph: &Graph,
_logs_root: &Path,
_services: &crate::handler::EngineServices,
) -> std::result::Result<Outcome, ArcError> {
let mut outcome = Outcome::fail("connection refused");
outcome.context_updates.insert(
"failure_class".to_string(),
serde_json::json!("transient_infra"),
);
Ok(outcome)
}
}
/// Handler that fails with a semantically different message each time.
/// Uses words instead of numbers to avoid normalization collapsing them.
struct VaryingFailHandler {
counter: std::sync::atomic::AtomicUsize,
}
static VARYING_REASONS: &[&str] = &[
"syntax error in module alpha",
"type mismatch in module beta",
"missing field in module gamma",
"undefined reference in module delta",
"assertion failed in module epsilon",
];
#[async_trait]
impl HandlerTrait for VaryingFailHandler {
async fn execute(
&self,
_node: &Node,
_context: &Context,
_graph: &Graph,
_logs_root: &Path,
_services: &crate::handler::EngineServices,
) -> std::result::Result<Outcome, ArcError> {
let n = self.counter.fetch_add(1, Ordering::Relaxed);
let reason = VARYING_REASONS[n % VARYING_REASONS.len()];
Ok(Outcome::fail(reason))
}
}
#[tokio::test]
async fn loop_circuit_breaker_aborts_on_repeated_deterministic_failure() {
let dir = tempfile::tempdir().unwrap();
let g = looping_fail_graph();
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
dry_run: false,
run_id: "test-run".into(),
git_checkpoint: None,
base_sha: None,
run_branch: None,
meta_branch: None,
};
let result = engine.run(&g, &config).await;
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(
err.contains("deterministic failure cycle detected"),
"expected circuit breaker error, got: {err}"
);
}
#[tokio::test]
async fn loop_circuit_breaker_ignores_transient_failures() {
let dir = tempfile::tempdir().unwrap();
let mut g = looping_fail_graph();
// Set a high visit limit so we don't trip it; we want to hit the visit limit, not circuit breaker
g.attrs
.insert("max_node_visits".to_string(), AttrValue::Integer(5));
let mut registry = make_registry();
registry.register("always_fail", Box::new(TransientFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
dry_run: false,
run_id: "test-run".into(),
git_checkpoint: None,
base_sha: None,
run_branch: None,
meta_branch: None,
};
let result = engine.run(&g, &config).await;
assert!(result.is_err());
let err = result.unwrap_err().to_string();
// Should hit visit limit, NOT circuit breaker
assert!(
err.contains("exceeded max visit limit"),
"expected visit limit error (transient shouldn't trigger circuit breaker), got: {err}"
);
}
#[tokio::test]
async fn loop_circuit_breaker_different_reasons_get_separate_counters() {
let dir = tempfile::tempdir().unwrap();
let mut g = looping_fail_graph();
// Each failure has a different message, so no signature repeats.
// Should hit max_node_visits instead of circuit breaker.
g.attrs
.insert("max_node_visits".to_string(), AttrValue::Integer(5));
let mut registry = make_registry();
registry.register(
"always_fail",
Box::new(VaryingFailHandler {
counter: std::sync::atomic::AtomicUsize::new(0),
}),
);
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
dry_run: false,
run_id: "test-run".into(),
git_checkpoint: None,
base_sha: None,
run_branch: None,
meta_branch: None,
};
let result = engine.run(&g, &config).await;
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(
err.contains("exceeded max visit limit"),
"expected visit limit (each failure unique), got: {err}"
);
}
#[tokio::test]
async fn restart_circuit_breaker_aborts_on_repeated_failure() {
// In a pipeline with loop_restart edges, a repeating deterministic failure
// triggers a circuit breaker (either loop or restart, depending on topology).
let dir = tempfile::tempdir().unwrap();
let mut g = Graph::new("restart_test");
g.attrs
.insert("goal".to_string(), AttrValue::String("test".to_string()));
g.attrs
.insert("default_max_retry".to_string(), AttrValue::Integer(0));
g.attrs
.insert("max_node_visits".to_string(), AttrValue::Integer(100));
let mut start = Node::new("start");
start.attrs.insert(
"shape".to_string(),
AttrValue::String("Mdiamond".to_string()),
);
g.nodes.insert("start".to_string(), start);
let mut work = Node::new("work");
work.attrs.insert(
"type".to_string(),
AttrValue::String("always_fail".to_string()),
);
work.attrs
.insert("max_retries".to_string(), AttrValue::Integer(0));
g.nodes.insert("work".to_string(), work);
let mut exit = Node::new("exit");
exit.attrs.insert(
"shape".to_string(),
AttrValue::String("Msquare".to_string()),
);
g.nodes.insert("exit".to_string(), exit);
g.edges.push(Edge::new("start", "work"));
// loop_restart edge on failure
let mut restart_edge = Edge::new("work", "start");
restart_edge.attrs.insert(
"condition".to_string(),
AttrValue::String("outcome=fail".to_string()),
);
restart_edge
.attrs
.insert("loop_restart".to_string(), AttrValue::Boolean(true));
g.edges.push(restart_edge);
// Success goes to exit
let mut ok_edge = Edge::new("work", "exit");
ok_edge.attrs.insert(
"condition".to_string(),
AttrValue::String("outcome=success".to_string()),
);
g.edges.push(ok_edge);
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
dry_run: false,
run_id: "test-run".into(),
git_checkpoint: None,
base_sha: None,
run_branch: None,
meta_branch: None,
};
let result = engine.run(&g, &config).await;
assert!(result.is_err());
let err = result.unwrap_err().to_string();
// A circuit breaker fires (loop or restart) rather than looping indefinitely
assert!(
err.contains("failure cycle detected") || err.contains("circuit breaker"),
"expected circuit breaker error, got: {err}"
);
}
#[tokio::test]
async fn failure_signature_stored_in_context() {
let dir = tempfile::tempdir().unwrap();
// Simple pipeline: start -> work (fails) -> exit (via fail edge)
let mut g = Graph::new("sig_context_test");
g.attrs
.insert("goal".to_string(), AttrValue::String("test".to_string()));
g.attrs
.insert("default_max_retry".to_string(), AttrValue::Integer(0));
let mut start = Node::new("start");
start.attrs.insert(
"shape".to_string(),
AttrValue::String("Mdiamond".to_string()),
);
g.nodes.insert("start".to_string(), start);
let mut work = Node::new("work");
work.attrs.insert(
"type".to_string(),
AttrValue::String("always_fail".to_string()),
);
work.attrs
.insert("max_retries".to_string(), AttrValue::Integer(0));
g.nodes.insert("work".to_string(), work);
let mut exit = Node::new("exit");
exit.attrs.insert(
"shape".to_string(),
AttrValue::String("Msquare".to_string()),
);
g.nodes.insert("exit".to_string(), exit);
g.edges.push(Edge::new("start", "work"));
g.edges.push(Edge::new("work", "exit"));
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
dry_run: false,
run_id: "test-run".into(),
git_checkpoint: None,
base_sha: None,
run_branch: None,
meta_branch: None,
};
let _outcome = engine.run(&g, &config).await.unwrap();
// Check the checkpoint for the failure_signature context value
let checkpoint_path = dir.path().join("checkpoint.json");
let cp = Checkpoint::load(&checkpoint_path).unwrap();
let sig_value = cp.context_values.get("failure_signature").unwrap();
let sig_str = sig_value.as_str().unwrap();
assert!(
sig_str.contains("work|deterministic|"),
"expected failure signature in context, got: {sig_str}"
);
}
}

215
crates/arc-workflows/src/error.rs
View file

@ -55,9 +55,15 @@ impl FromStr for FailureClass {
"structural" => Self::Structural,
// Aliases: transient_infra
"transient" | "transient-infra" | "infra_transient" | "transient infra"
| "infrastructure_transient" | "retryable" | "toolchain_workspace_io"
| "toolchain-workspace-io" | "toolchain_or_dependency_registry_unavailable"
"transient"
| "transient-infra"
| "infra_transient"
| "transient infra"
| "infrastructure_transient"
| "retryable"
| "toolchain_workspace_io"
| "toolchain-workspace-io"
| "toolchain_or_dependency_registry_unavailable"
| "toolchain-dependency-registry-unavailable" => Self::TransientInfra,
// Aliases: deterministic
@ -208,6 +214,80 @@ pub fn classify_failure_reason(reason: &str) -> FailureClass {
FailureClass::Deterministic
}
/// Normalize a failure reason for stable signature grouping.
///
/// Replaces variable data (hex strings, digits) with placeholders so that
/// semantically identical errors produce the same signature regardless of
/// line numbers, commit hashes, or timestamps.
pub fn normalize_failure_reason(reason: &str) -> String {
use regex::Regex;
use std::sync::LazyLock;
static HEX_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"\b[0-9a-f]{7,64}\b").unwrap());
static DIGITS_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"\b\d+\b").unwrap());
static COMMA_SPACE_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r",\s+").unwrap());
static WHITESPACE_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"\s+").unwrap());
let s = reason.trim().to_lowercase();
if s.is_empty() {
return String::new();
}
let s = HEX_RE.replace_all(&s, "<hex>");
let s = DIGITS_RE.replace_all(&s, "<n>");
let s = COMMA_SPACE_RE.replace_all(&s, ",");
let s = WHITESPACE_RE.replace_all(&s, " ");
let s = s.trim();
if s.len() > 240 {
s[..240].to_string()
} else {
s.to_string()
}
}
/// Composite key that uniquely identifies a specific recurring failure.
///
/// Format: `node_id|failure_class|normalized_reason`
///
/// Used by circuit breakers to detect when the same failure keeps repeating,
/// e.g. "verify|deterministic|assertion failed in foo_test".
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct FailureSignature(String);
impl FailureSignature {
/// Build a signature from failure context.
///
/// The signature hint from `outcome.context_updates["failure_signature"]` takes
/// priority over the raw `failure_reason`, allowing handlers to provide explicit
/// grouping keys.
pub fn new(
node_id: &str,
failure_class: FailureClass,
signature_hint: Option<&str>,
failure_reason: Option<&str>,
) -> Self {
let reason = signature_hint
.map(normalize_failure_reason)
.filter(|s| !s.is_empty())
.or_else(|| failure_reason.map(normalize_failure_reason))
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "unknown".to_string());
Self(format!("{}|{}|{}", node_id.trim(), failure_class, reason))
}
}
impl fmt::Display for FailureSignature {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl FailureClass {
/// Whether this failure class should be tracked by the cycle breaker.
pub fn is_signature_tracked(self) -> bool {
matches!(self, Self::Deterministic | Self::Structural)
}
}
#[derive(Error, Debug, Clone)]
pub enum ArcError {
#[error("Parse error: {0}")]
@ -1189,4 +1269,133 @@ mod tests {
FailureClass::Deterministic
);
}
// --- normalize_failure_reason tests ---
#[test]
fn normalize_empty_and_whitespace_returns_empty() {
assert_eq!(normalize_failure_reason(""), "");
assert_eq!(normalize_failure_reason(" "), "");
assert_eq!(normalize_failure_reason("\n\t"), "");
}
#[test]
fn normalize_lowercases_and_trims() {
assert_eq!(normalize_failure_reason(" Hello World "), "hello world");
}
#[test]
fn normalize_replaces_hex_strings() {
assert_eq!(
normalize_failure_reason("commit abc123def0"),
"commit <hex>"
);
// Short hex (< 7 chars) not replaced
assert_eq!(normalize_failure_reason("value abcdef"), "value abcdef");
}
#[test]
fn normalize_replaces_digit_sequences() {
assert_eq!(normalize_failure_reason("line 42"), "line <n>");
assert_eq!(normalize_failure_reason("error 0"), "error <n>");
}
#[test]
fn normalize_collapses_comma_space_and_whitespace() {
assert_eq!(normalize_failure_reason("a, b, c"), "a,b,c");
assert_eq!(normalize_failure_reason("a b"), "a b");
}
#[test]
fn normalize_truncates_to_240_chars() {
let long = "a".repeat(300);
let result = normalize_failure_reason(&long);
assert_eq!(result.len(), 240);
}
#[test]
fn normalize_combined_example() {
assert_eq!(
normalize_failure_reason("Error at line 42 in abc123def"),
"error at line <n> in <hex>"
);
}
// --- FailureSignature tests ---
#[test]
fn failure_signature_format() {
let sig = FailureSignature::new(
"verify",
FailureClass::Deterministic,
None,
Some("test failed"),
);
assert_eq!(sig.to_string(), "verify|deterministic|test failed");
}
#[test]
fn failure_signature_display() {
let sig = FailureSignature::new(
"build",
FailureClass::Structural,
None,
Some("scope violation"),
);
assert_eq!(format!("{sig}"), "build|structural|scope violation");
}
#[test]
fn failure_signature_hint_takes_priority() {
let sig = FailureSignature::new(
"verify",
FailureClass::Deterministic,
Some("custom hint"),
Some("raw reason"),
);
assert_eq!(sig.to_string(), "verify|deterministic|custom hint");
}
#[test]
fn failure_signature_missing_reason_falls_back_to_unknown() {
let sig = FailureSignature::new("node", FailureClass::Deterministic, None, None);
assert_eq!(sig.to_string(), "node|deterministic|unknown");
}
#[test]
fn failure_signature_equality_and_hash() {
let sig1 = FailureSignature::new(
"verify",
FailureClass::Deterministic,
None,
Some("test failed"),
);
let sig2 = FailureSignature::new(
"verify",
FailureClass::Deterministic,
None,
Some("test failed"),
);
assert_eq!(sig1, sig2);
let mut map = std::collections::HashMap::new();
map.insert(sig1.clone(), 1);
assert_eq!(map.get(&sig2), Some(&1));
}
// --- is_signature_tracked tests ---
#[test]
fn is_signature_tracked_deterministic_and_structural() {
assert!(FailureClass::Deterministic.is_signature_tracked());
assert!(FailureClass::Structural.is_signature_tracked());
}
#[test]
fn is_signature_tracked_false_for_others() {
assert!(!FailureClass::TransientInfra.is_signature_tracked());
assert!(!FailureClass::BudgetExhausted.is_signature_tracked());
assert!(!FailureClass::Canceled.is_signature_tracked());
assert!(!FailureClass::CompilationLoop.is_signature_tracked());
}
}

11
crates/arc-workflows/src/git.rs
View file

@ -737,6 +737,8 @@ mod tests {
std::collections::HashMap::new(),
std::collections::HashMap::new(),
Some("node_b".to_string()),
std::collections::HashMap::new(),
std::collections::HashMap::new(),
);
let cp_json = serde_json::to_vec_pretty(&cp).unwrap();
store.write_checkpoint("RUN2", &cp_json, &[]).unwrap();
@ -769,6 +771,8 @@ mod tests {
std::collections::HashMap::new(),
std::collections::HashMap::new(),
None,
std::collections::HashMap::new(),
std::collections::HashMap::new(),
);
let cp1_json = serde_json::to_vec_pretty(&cp1).unwrap();
store.write_checkpoint("RUN3", &cp1_json, &[]).unwrap();
@ -780,6 +784,8 @@ mod tests {
std::collections::HashMap::new(),
std::collections::HashMap::new(),
Some("node_c".to_string()),
std::collections::HashMap::new(),
std::collections::HashMap::new(),
);
let cp2_json = serde_json::to_vec_pretty(&cp2).unwrap();
store.write_checkpoint("RUN3", &cp2_json, &[]).unwrap();
@ -846,10 +852,7 @@ mod tests {
#[test]
fn sanitize_ref_component_mixed() {
assert_eq!(
sanitize_ref_component("My Node!@#123"),
"my-node-123"
);
assert_eq!(sanitize_ref_component("My Node!@#123"), "my-node-123");
}
#[test]

42
crates/arc-workflows/src/graph/types.rs
View file

@ -419,6 +419,16 @@ impl Graph {
self.attrs.get("default_thread").and_then(AttrValue::as_str)
}
/// Graph-level `loop_restart_signature_limit` (default 3).
/// When the same failure signature repeats this many times, the pipeline aborts.
pub fn loop_restart_signature_limit(&self) -> usize {
self.attrs
.get("loop_restart_signature_limit")
.and_then(AttrValue::as_i64)
.filter(|&v| v >= 1)
.map_or(3, |v| v as usize)
}
/// Graph-level `max_node_visits` (default 0 = disabled).
pub fn max_node_visits(&self) -> u64 {
self.attrs
@ -746,4 +756,36 @@ mod tests {
);
assert!(node.codergen_mode().is_err());
}
#[test]
fn graph_loop_restart_signature_limit_default() {
let g = Graph::new("empty");
assert_eq!(g.loop_restart_signature_limit(), 3);
}
#[test]
fn graph_loop_restart_signature_limit_set() {
let mut g = Graph::new("test");
g.attrs.insert(
"loop_restart_signature_limit".to_string(),
AttrValue::Integer(5),
);
assert_eq!(g.loop_restart_signature_limit(), 5);
}
#[test]
fn graph_loop_restart_signature_limit_invalid_falls_back() {
let mut g = Graph::new("test");
g.attrs.insert(
"loop_restart_signature_limit".to_string(),
AttrValue::Integer(0),
);
assert_eq!(g.loop_restart_signature_limit(), 3);
g.attrs.insert(
"loop_restart_signature_limit".to_string(),
AttrValue::Integer(-1),
);
assert_eq!(g.loop_restart_signature_limit(), 3);
}
}

45
crates/arc-workflows/src/retro.rs
View file

@ -245,9 +245,7 @@ pub fn derive_retro(
*total_cost.get_or_insert(0.0) += c;
}
let files = outcome
.map(|o| o.files_touched.clone())
.unwrap_or_default();
let files = outcome.map(|o| o.files_touched.clone()).unwrap_or_default();
all_files.extend(files.iter().cloned());
stages.push(StageRetro {
@ -344,6 +342,8 @@ mod tests {
node_outcomes,
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
}
}
@ -355,7 +355,16 @@ mod tests {
.into_iter()
.collect();
let retro = derive_retro("run-1", "my_pipeline", "Fix the bug", &cp, false, None, 20000, &durations);
let retro = derive_retro(
"run-1",
"my_pipeline",
"Fix the bug",
&cp,
false,
None,
20000,
&durations,
);
assert_eq!(retro.run_id, "run-1");
assert_eq!(retro.pipeline_name, "my_pipeline");
@ -393,11 +402,19 @@ mod tests {
},
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
};
let retro = derive_retro(
"run-2", "pipe", "goal", &cp, true,
Some("boom"), 5000, &HashMap::new(),
"run-2",
"pipe",
"goal",
&cp,
true,
Some("boom"),
5000,
&HashMap::new(),
);
assert_eq!(retro.stats.stages_failed, 1);
@ -428,7 +445,10 @@ mod tests {
assert_eq!(retro.smoothness, Some(SmoothnessRating::Smooth));
assert_eq!(retro.intent.as_deref(), Some("Fix authentication bug"));
assert_eq!(retro.outcome.as_deref(), Some("Successfully fixed the login flow"));
assert_eq!(
retro.outcome.as_deref(),
Some("Successfully fixed the login flow")
);
assert_eq!(retro.learnings.as_ref().unwrap().len(), 1);
assert!(retro.friction_points.is_none()); // empty vec -> None
assert_eq!(retro.open_items.as_ref().unwrap().len(), 1);
@ -438,7 +458,16 @@ mod tests {
fn save_and_load_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let cp = make_checkpoint_with_stages();
let mut retro = derive_retro("r1", "pipe", "goal", &cp, false, None, 1000, &HashMap::new());
let mut retro = derive_retro(
"r1",
"pipe",
"goal",
&cp,
false,
None,
1000,
&HashMap::new(),
);
retro.smoothness = Some(SmoothnessRating::Bumpy);
retro.intent = Some("Test intent".to_string());

1033
crates/arc-workflows/tests/integration.rs
View file

File diff suppressed because it is too large Load diff