diff --git a/CHANGELOG.md b/CHANGELOG.md index cc84fb7f7..8a2f74499 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,28 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). wall-clock cap). `--init` wizard text, `config.toml` comments, and the `TimeoutPolicy`/ `default_idle_timeout_secs` doc comments now describe the enforced semantics and warn that the value must be set above the longest expected single-turn duration. +- **zeph-orchestration**: implemented Mode-2 `route_to` reroute-to-alternate-node recovery + (spec `075-orchestration-node-control-parity` FR-D-01, #6244), deferred from the original + Mode-1 (`state_injection`) recovery feature after the naive design was found unsafe + (N5/N1/N3). A node's `RecoveryAction.route_to: Option` names a fallback node that + starts in a new non-dispatchable `TaskStatus::Dormant` state and is activated + (`Dormant → Ready`) only by the source node's terminal `Abort`-default or + retry-exhausted-`Retry` failure, carrying a `routed_from` marker that seeds the + fallback's prompt with the failed source's sanitized output via a `` + block. If the source never fails, a completion-time sweep resolves the untriggered + fallback (and its transitive downstream subtree) to `Skipped` instead of falsely + reporting a scheduler deadlock. `route_to` is mutually exclusive with `state_injection`, + rejects chained reroutes and N:1 shared targets, and is validated at graph construction + time. `DispatchStrategy::LevelBarrier` treats a `Dormant` fallback as parked (never + blocking barrier advancement) and lets an activated fallback dispatch out-of-level, + closing a silent livelock where a route_to source deeper than its depth-0 fallback would + never get a chance to run. `/plan retry` re-arms a previously-activated fallback branch + back to `Dormant` so retry semantics compose correctly with a prior reroute. A rerouted + source stays terminal `Failed` even inside a `Completed` graph (by design, to keep the + aggregator/grounding `Completed`-only filters from pulling the failure's error output + into synthesis) and is now tallied into `tasks_failed` on the `finalize_plan_completed` + path. No config/CLI/wizard surface: `route_to`, like `state_injection`, is authored + programmatically rather than exposed to the LLM planner schema. - `zeph-mcp`/`zeph-core`: an MCP server can now request a larger per-call truncation limit for a single tool result via `_meta["zeph/maxResultSizeChars"]`, bounded by a new diff --git a/crates/zeph-core/src/agent/plan.rs b/crates/zeph-core/src/agent/plan.rs index 3a417b52f..bb45915ed 100644 --- a/crates/zeph-core/src/agent/plan.rs +++ b/crates/zeph-core/src/agent/plan.rs @@ -910,9 +910,22 @@ impl Agent { .iter() .filter(|t| t.status == zeph_orchestration::TaskStatus::Skipped) .count() as u64; + // D3 (spec-075 FR-D-01): on a Completed graph, a Failed task can only be a + // rerouted Mode-2 source — Mode-1 relabels Failed -> Completed, Skip relabels + // Failed -> Skipped, and Abort/retry-exhausted-without-reroute sets the graph + // Failed (not Completed), so this branch is never reached with an unrecovered + // Failed task. If a future recovery mechanism ever leaves a Failed task inside + // a Completed graph without being a rerouted source, this tally would silently + // misclassify it as "rerouted" — re-verify the invariant before adding one. + let rerouted_failed_count = completed_graph + .tasks + .iter() + .filter(|t| t.status == zeph_orchestration::TaskStatus::Failed) + .count() as u64; self.update_metrics(|m| { m.orchestration.tasks_completed += completed_count; m.orchestration.tasks_skipped += skipped_count; + m.orchestration.tasks_failed += rerouted_failed_count; }); let aggregator_provider = self @@ -978,6 +991,14 @@ impl Agent { ) -> Result<&'static str, error::AgentError> { use std::fmt::Write; + // M2 (spec-075 FR-D-01, cosmetic): a Mode-2 route_to fallback can persist in + // TaskStatus::Dormant into a Failed graph (the Abort/retry-exhausted path sets + // graph.status = Failed and returns before the completion sweep runs — see + // `check_graph_completion`'s `resolve_dormant_after_terminal` doc). A Dormant + // task is counted in none of the buckets below, so failed+cancelled+completed+ + // skipped may not sum to tasks.len() on this path. Benign: `/plan retry` + // re-arms it if its source is reset, or the completion sweep resolves it once + // a retried graph heads to Completed. Not treated as an error here. let failed_tasks: Vec<_> = completed_graph .tasks .iter() diff --git a/crates/zeph-core/src/agent/tests/compaction_e2e.rs b/crates/zeph-core/src/agent/tests/compaction_e2e.rs index ffec0021f..78af96e4f 100644 --- a/crates/zeph-core/src/agent/tests/compaction_e2e.rs +++ b/crates/zeph-core/src/agent/tests/compaction_e2e.rs @@ -1593,6 +1593,74 @@ async fn finalize_plan_execution_recovery_derived_task_counted_in_tasks_complete ); } +/// D3 (spec-075 FR-D-01): on a `Completed` graph, a terminal-`Failed` task can only be a +/// Mode-2 rerouted source (Mode-1 relabels Failed -> Completed, Skip relabels Failed -> +/// Skipped, Abort/retry-exhausted-without-reroute sets the graph `Failed`, not `Completed`). +/// `finalize_plan_completed` must tally it into `tasks_failed` so it is metric-visible +/// without touching its status semantics (grounding/aggregator must keep ignoring it -- +/// see the Completed-only filters this deliberately does not disturb). +#[cfg(feature = "scheduler")] +#[tokio::test] +async fn finalize_plan_execution_completed_graph_tallies_rerouted_failed_source() { + use zeph_subagent::SubAgentManager; + + let provider = mock_provider(vec!["synthesis".into()]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let (tx, rx) = watch::channel(MetricsSnapshot::default()); + let mut agent = Agent::new(provider, channel, registry, None, 5, executor).with_metrics(tx); + agent.services.orchestration.orchestration_config.enabled = true; + agent.services.orchestration.subagent_manager = Some(SubAgentManager::new(4)); + + let mut graph = TaskGraph::new("route_to finalize test"); + + // The rerouted source: stays terminal Failed even though the graph completes. + let mut source = TaskNode::new(0, "task-source", "desc"); + source.status = TaskStatus::Failed; + source.result = Some(TaskResult { + output: "boom: connection refused".into(), + artifacts: vec![], + duration_ms: 0, + agent_id: None, + agent_def: None, + }); + graph.tasks.push(source); + + // The activated fallback: ran to completion. + let mut fallback = TaskNode::new(1, "task-fallback", "desc"); + fallback.status = TaskStatus::Completed; + fallback.routed_from = Some(zeph_orchestration::TaskId(0)); + fallback.result = Some(TaskResult { + output: "fallback output".into(), + artifacts: vec![], + duration_ms: 5, + agent_id: None, + agent_def: None, + }); + graph.tasks.push(fallback); + + graph.status = GraphStatus::Completed; + + agent + .finalize_plan_execution(graph, GraphStatus::Completed) + .await + .unwrap(); + + let snapshot = rx.borrow().clone(); + assert_eq!( + snapshot.orchestration.tasks_failed, 1, + "the rerouted source must be tallied into tasks_failed even on a Completed graph; \ + got: {}", + snapshot.orchestration.tasks_failed + ); + assert_eq!( + snapshot.orchestration.tasks_completed, 1, + "the activated fallback must still be tallied into tasks_completed; got: {}", + snapshot.orchestration.tasks_completed + ); +} + /// Regression for #1879: mixed failure — some tasks failed, some canceled. /// Message must say "Plan failed. X/M tasks failed, Y canceled:" (not misleading). #[cfg(feature = "scheduler")] diff --git a/crates/zeph-core/src/agent/tests/inline_tool_loop_tests.rs b/crates/zeph-core/src/agent/tests/inline_tool_loop_tests.rs index bb30eb58d..1d0015558 100644 --- a/crates/zeph-core/src/agent/tests/inline_tool_loop_tests.rs +++ b/crates/zeph-core/src/agent/tests/inline_tool_loop_tests.rs @@ -624,6 +624,7 @@ mod run_inline_timeout { }); node0.recovery = Some(zeph_orchestration::RecoveryAction { state_injection: Some("recovered output".to_string()), + route_to: None, }); let mut node1 = TaskNode::new(1, "dependent task", "consume the recovered output"); node1.depends_on = vec![zeph_orchestration::TaskId(0)]; diff --git a/crates/zeph-orchestration/src/dag.rs b/crates/zeph-orchestration/src/dag.rs index 84eb59f53..7c3a68a47 100644 --- a/crates/zeph-orchestration/src/dag.rs +++ b/crates/zeph-orchestration/src/dag.rs @@ -33,10 +33,25 @@ use super::graph::{ /// - No task sets both `recovery` and `verify_predicate` (a predicate-gated task must /// not be recovery-eligible — recovery bypasses the completion-event handler where /// predicate verification runs). +/// - No task sets both `recovery.state_injection` and `recovery.route_to` (Mode 1 and +/// Mode 2 are mutually exclusive recovery modes on the same node). +/// - Every `recovery.route_to` target references a valid index and is not a +/// self-reroute. +/// - A `recovery.route_to` target has an empty `depends_on` — it may only ever become +/// `Ready` via Mode-2 activation (the crate-internal `try_reroute`), never via the +/// `Pending` arm of [`ready_tasks`]. This is what makes `TaskStatus::Dormant` sound. +/// - A `recovery.route_to` target does not itself set `recovery.route_to` — chained +/// reroutes are unsupported in v1 (fail closed rather than silently mishandle the +/// transitive re-arm semantics on retry). +/// - Every `recovery.route_to` target has exactly one source (rejects `count > 1`, +/// keeping the crate-internal `resolve_dormant_after_terminal`'s source lookup and +/// the retry re-arm pass single-source; N:1 shared fallback fan-in is deferred). /// -/// Also warns (does not reject) when a task sets `recovery` but its effective failure -/// strategy (`task.failure_strategy.unwrap_or(default_failure_strategy)`) is `Skip` or -/// `Ask` — those arms never consult recovery, so it would be configured but inert. +/// Also rejects (upgraded from Mode-1's warn) a task that sets `recovery.route_to` when +/// its effective failure strategy (`task.failure_strategy.unwrap_or(default_failure_strategy)`) +/// is `Skip` or `Ask` — those arms never consult recovery, and `route_to`'s on-failure +/// edge must never coincide with the Skip-BFS arm. Still only warns for +/// `recovery.state_injection` under `Skip`/`Ask` (Mode-1, unchanged behavior). /// /// # Errors /// @@ -61,6 +76,10 @@ pub fn validate( )); } + // route_to target -> count of sources pointing at it (invariant: exactly one). + let mut route_to_target_counts: std::collections::HashMap = + std::collections::HashMap::new(); + for (i, task) in tasks.iter().enumerate() { // Invariant: tasks[i].id == TaskId(i) let expected = u32::try_from(i).map_err(|_| { @@ -95,12 +114,30 @@ pub fn validate( ))); } - if task.recovery.is_some() { + if let Some(recovery) = &task.recovery { + if recovery.state_injection.is_some() && recovery.route_to.is_some() { + return Err(OrchestrationError::InvalidGraph(format!( + "task {i} sets both recovery.state_injection and recovery.route_to — \ + Mode 1 and Mode 2 recovery are mutually exclusive" + ))); + } + + validate_route_to( + i, + task, + recovery, + tasks, + default_failure_strategy, + &mut route_to_target_counts, + )?; + let effective_strategy = task.failure_strategy.unwrap_or(default_failure_strategy); - if matches!( - effective_strategy, - FailureStrategy::Skip | FailureStrategy::Ask - ) { + if recovery.state_injection.is_some() + && matches!( + effective_strategy, + FailureStrategy::Skip | FailureStrategy::Ask + ) + { tracing::warn!( task_index = i, strategy = ?effective_strategy, @@ -111,7 +148,19 @@ pub fn validate( } } - // Cycle detection + root check via toposort + for (target, count) in route_to_target_counts { + if count > 1 { + return Err(OrchestrationError::InvalidGraph(format!( + "task {target} is the recovery.route_to target of {count} sources — \ + exactly one source per target is required (N:1 shared fallback is deferred)" + ))); + } + } + + // Cycle detection + root check via toposort. route_to edges are excluded from + // toposort by construction (they are read from `depends_on`, which route_to never + // touches) — they are on-failure edges, not dependency edges, so no cycle-detection + // change is needed here. let sorted = toposort(tasks)?; // After a successful toposort every task was visited; verify at least one root @@ -125,6 +174,71 @@ pub fn validate( Ok(()) } +/// Validate a single task's Mode-2 `recovery.route_to` configuration. Extracted from +/// [`validate`] (which would otherwise exceed clippy's line-count threshold): checks the +/// target index is in range and not a self-reroute, that the target has an empty +/// `depends_on` (invariant 4 — see [`TaskStatus::Dormant`]), that the target does not +/// itself set `route_to` (no chained reroutes in v1), tallies the target into +/// `route_to_target_counts` for the caller's N:1 post-loop check, and rejects an +/// effective `Skip`/`Ask` failure strategy on the source. No-op when `recovery.route_to` +/// is `None`. +fn validate_route_to( + i: usize, + task: &TaskNode, + recovery: &crate::graph::RecoveryAction, + tasks: &[TaskNode], + default_failure_strategy: FailureStrategy, + route_to_target_counts: &mut std::collections::HashMap, +) -> Result<(), OrchestrationError> { + let Some(target) = recovery.route_to else { + return Ok(()); + }; + + if target == task.id { + return Err(OrchestrationError::InvalidGraph(format!( + "task {i} sets recovery.route_to to itself — self-reroute is not allowed" + ))); + } + if target.index() >= tasks.len() { + return Err(OrchestrationError::InvalidGraph(format!( + "task {i} sets recovery.route_to to non-existent task {target}" + ))); + } + + let target_task = &tasks[target.index()]; + if !target_task.depends_on.is_empty() { + return Err(OrchestrationError::InvalidGraph(format!( + "task {i} routes to task {target}, but {target} has a non-empty depends_on \ + — a route_to target must only become ready via on-failure activation" + ))); + } + if target_task + .recovery + .as_ref() + .is_some_and(|r| r.route_to.is_some()) + { + return Err(OrchestrationError::InvalidGraph(format!( + "task {i} routes to task {target}, but {target} itself sets recovery.route_to \ + — chained `route_to` is not supported in v1" + ))); + } + *route_to_target_counts.entry(target).or_insert(0) += 1; + + let effective_strategy = task.failure_strategy.unwrap_or(default_failure_strategy); + if matches!( + effective_strategy, + FailureStrategy::Skip | FailureStrategy::Ask + ) { + return Err(OrchestrationError::InvalidGraph(format!( + "task {i} sets recovery.route_to but its effective failure strategy is \ + {effective_strategy:?} — route_to must never coincide with the Skip-BFS \ + or Ask-pause arms" + ))); + } + + Ok(()) +} + /// Topological sort using Kahn's algorithm. /// /// Returns tasks in dependency order (roots first). @@ -289,15 +403,156 @@ fn try_recover(graph: &mut TaskGraph, failed_id: TaskId) -> bool { true } +/// Mark every Mode-2 `route_to` fallback target [`TaskStatus::Dormant`], parking it +/// until its source's terminal failure activates it via [`try_reroute`]. +/// +/// Iterates all tasks; for each with `recovery.route_to == Some(target)`, sets +/// `graph.tasks[target].status = Dormant` **only if `target.status == Pending`**. The +/// `== Pending` guard makes this idempotent and restart-safe: a reloaded graph whose +/// target was already activated (`Ready`/`Running`/`Completed`/`Dormant`) or resolved +/// (`Skipped`) is never re-dormanted. +/// +/// Call site: top of `DagScheduler::init_common`, ordered **before** the root-activation +/// loop (a `route_to` target's `depends_on` is empty by `validate` invariant, so the +/// root-activation loop would otherwise flip it straight to `Ready` on a fresh graph). +pub(crate) fn mark_dormant_route_to_targets(graph: &mut TaskGraph) { + let targets: Vec = graph + .tasks + .iter() + .filter_map(|t| t.recovery.as_ref().and_then(|r| r.route_to)) + .collect(); + for target in targets { + let node = &mut graph.tasks[target.index()]; + if node.status == TaskStatus::Pending { + node.status = TaskStatus::Dormant; + } + } +} + +/// Attempt Mode-2 reroute for a failed task. +/// +/// If `graph.tasks[failed_id].recovery.route_to == Some(target)` **and +/// `target.status == Dormant`**: activates the target (`Dormant → Ready`), sets +/// `target.routed_from = Some(failed_id)`, leaves the source `failed_id` terminal +/// `Failed`, and returns `true` — the failure is absorbed exactly like Mode-1 +/// (`graph.status` untouched, independent branches continue). Returns `false` (no +/// mutation) when no `route_to` is configured, or when the target is not currently +/// `Dormant` (already activated via another path, or mid-retry re-arm race) — this +/// runtime status guard is load-bearing (spec-075 FR-D-01): it is what stops +/// `try_reroute` from clobbering a live node. +/// +/// Mutates synchronously with no `.await` — preserves the same-tick snapshot atomicity +/// durability guarantee (FR-016), same as [`try_recover`]. +fn try_reroute(graph: &mut TaskGraph, failed_id: TaskId) -> bool { + let Some(target) = graph.tasks[failed_id.index()] + .recovery + .as_ref() + .and_then(|r| r.route_to) + else { + return false; + }; + if graph.tasks[target.index()].status != TaskStatus::Dormant { + tracing::warn!( + task_id = %failed_id, + target = %target, + target_status = %graph.tasks[target.index()].status, + "orchestration.dag.try_reroute: route_to target is not Dormant, skipping activation" + ); + return false; + } + let node = &mut graph.tasks[target.index()]; + node.status = TaskStatus::Ready; + node.routed_from = Some(failed_id); + tracing::info!( + task_id = %failed_id, + target = %target, + "orchestration.dag.try_reroute: Mode-2 reroute activated" + ); + true +} + +/// Mark `seed` and all its transitive non-terminal dependents [`TaskStatus::Skipped`]. +/// +/// Shared BFS core for the `Skip` failure-strategy arm and +/// [`resolve_dormant_after_terminal`]'s un-triggered-fallback resolution. Returns the +/// `Running` dependents found along the way — the caller must cancel them, because +/// marking a task `Skipped` in the data structure does not stop execution. +/// +/// `rev_adj[i]` must contain the IDs of all tasks that depend on task `i`. +fn skip_subtree(graph: &mut TaskGraph, seed: TaskId, rev_adj: &[Vec]) -> Vec { + let mut to_cancel = Vec::new(); + let mut queue: VecDeque = VecDeque::new(); + queue.push_back(seed); + + while let Some(current) = queue.pop_front() { + let dependents = rev_adj.get(current.index()).map_or(&[] as &[TaskId], |v| v); + for &dep_id in dependents { + if !graph.tasks[dep_id.index()].status.is_terminal() { + if graph.tasks[dep_id.index()].status == TaskStatus::Running { + to_cancel.push(dep_id); + } + graph.tasks[dep_id.index()].status = TaskStatus::Skipped; + queue.push_back(dep_id); + } + } + } + + to_cancel +} + +/// Resolve every still-[`TaskStatus::Dormant`] `route_to` fallback whose source has +/// terminalized without rerouting. +/// +/// For each `Dormant` task `F`, finds its unique source `S` (`validate` guarantees +/// exactly one). If `S.status.is_terminal()` — it succeeded, or was itself terminalized +/// by an unrelated cascade/skip without ever calling [`try_reroute`] (which would have +/// flipped `F` to `Ready`) — marks `F` [`TaskStatus::Skipped`] and skips its transitive +/// subtree via [`skip_subtree`] (any task depending on `F` would otherwise strand in +/// `Pending` forever, since `F` never reaches `Completed`). +/// +/// Call site: top of `check_graph_completion`, **before** the `all_terminal` / +/// deadlock-detection logic — a still-`Dormant` node is non-terminal and excluded from +/// `ready_tasks()`, so without this sweep a successful plan with an untriggered fallback +/// would be misreported as a scheduler deadlock. Returns the list of resolved (skipped) +/// task IDs, purely for caller-side logging; an empty return means no mutation occurred. +pub(crate) fn resolve_dormant_after_terminal( + graph: &mut TaskGraph, + rev_adj: &[Vec], +) -> Vec { + let dormant_sources: Vec<(TaskId, TaskId)> = graph + .tasks + .iter() + .filter(|t| t.status == TaskStatus::Dormant) + .filter_map(|target| { + graph + .tasks + .iter() + .find(|t| t.recovery.as_ref().and_then(|r| r.route_to) == Some(target.id)) + .map(|source| (target.id, source.id)) + }) + .collect(); + + let mut resolved = Vec::new(); + for (target, source) in dormant_sources { + if graph.tasks[source.index()].status.is_terminal() { + graph.tasks[target.index()].status = TaskStatus::Skipped; + resolved.push(target); + skip_subtree(graph, target, rev_adj); + } + } + resolved +} + /// Handle a task failure. Applies the effective failure strategy and mutates the graph. /// /// Returns the list of `Running` task IDs that the caller should cancel (for `Abort` strategy). /// -/// - `Abort`: sets `graph.status = Failed`, returns all currently `Running` task IDs. +/// - `Abort`: tries Mode-1 recovery, then Mode-2 reroute; if neither applies, sets +/// `graph.status = Failed` and returns all currently `Running` task IDs. /// - `Skip`: marks the failed task `Skipped` and transitively skips all non-terminal dependents /// using BFS over a reverse adjacency list. /// - `Retry`: if `retry_count < max_retries`, increments counter and resets task to `Ready`. -/// Otherwise falls through to `Abort`. +/// Otherwise tries Mode-1 recovery, then Mode-2 reroute, then falls through to `Abort`. /// - `Ask`: sets `graph.status = Paused`. /// /// `rev_adj[i]` must contain the IDs of all tasks that depend on task `i` (pre-built by the @@ -326,6 +581,9 @@ pub fn propagate_failure( if try_recover(graph, failed_id) { return Vec::new(); } + if try_reroute(graph, failed_id) { + return Vec::new(); + } graph.status = GraphStatus::Failed; // Return IDs of all currently Running tasks for the caller to cancel graph @@ -337,30 +595,13 @@ pub fn propagate_failure( } FailureStrategy::Skip => { - // Mark the failed task as Skipped + // Mark the failed task as Skipped, then transitively skip all non-terminal + // dependents. route_to targets are never reached here: they are not + // `depends_on`-dependents of the failed task (validate invariant), and + // route_to is rejected under an effective Skip strategy at graph + // construction time. graph.tasks[failed_id.index()].status = TaskStatus::Skipped; - - // BFS to transitively skip all non-terminal dependents. - // Collect Running tasks that are being skipped — the caller must cancel them, - // because marking a task Skipped in the data structure does not stop execution. - let mut to_cancel = Vec::new(); - let mut queue: VecDeque = VecDeque::new(); - queue.push_back(failed_id); - - while let Some(current) = queue.pop_front() { - let dependents = rev_adj.get(current.index()).map_or(&[] as &[TaskId], |v| v); - for &dep_id in dependents { - if !graph.tasks[dep_id.index()].status.is_terminal() { - if graph.tasks[dep_id.index()].status == TaskStatus::Running { - to_cancel.push(dep_id); - } - graph.tasks[dep_id.index()].status = TaskStatus::Skipped; - queue.push_back(dep_id); - } - } - } - - to_cancel + skip_subtree(graph, failed_id, rev_adj) } FailureStrategy::Retry => { @@ -370,10 +611,14 @@ pub fn propagate_failure( graph.tasks[failed_id.index()].status = TaskStatus::Ready; Vec::new() } else { - // Retry exhausted — try Mode-1 recovery before falling through to Abort + // Retry exhausted — try Mode-1 recovery, then Mode-2 reroute, before + // falling through to Abort. if try_recover(graph, failed_id) { return Vec::new(); } + if try_reroute(graph, failed_id) { + return Vec::new(); + } graph.status = GraphStatus::Failed; graph .tasks @@ -463,6 +708,10 @@ pub fn reset_for_retry( return Ok(()); } + // `seeds` is moved into the Skipped-BFS queue below; the route_to re-arm pass + // (D2, spec-075 FR-D-01) needs its own copy of the just-reset Failed source IDs. + let seeds_for_reroute = seeds.clone(); + // BFS from seeds: reset Skipped dependents back to Pending. let mut queue: std::collections::VecDeque = seeds.into_iter().collect(); while let Some(current) = queue.pop_front() { @@ -475,6 +724,51 @@ pub fn reset_for_retry( } } + // route_to re-arm pass (D2, spec-075 FR-D-01): a rerouted source that is reset to + // Ready must re-arm its entire fallback branch back to the parked/quiescent state, + // else a source that now succeeds finds its fallback already Ready/beyond and + // dispatches anyway (defeating Mode 2), or a source that fails again finds its + // fallback not Dormant and `try_reroute`'s runtime guard refuses to re-activate it + // (permanently disabling Mode 2 for that source/target pair). This is a SEPARATE + // pass from the Skipped-BFS above, keyed on `route_to` rather than `depends_on`: + // a route_to target is never a `depends_on`-dependent of its source (validate + // invariant forces the target's `depends_on` empty), so the Skipped-BFS provably + // never reaches it — walking `rev_adj` from the target's own subtree is required. + for s_id in seeds_for_reroute { + let Some(target) = graph.tasks[s_id.index()] + .recovery + .as_ref() + .and_then(|r| r.route_to) + else { + continue; + }; + + let target_node = &mut graph.tasks[target.index()]; + target_node.status = TaskStatus::Dormant; + target_node.routed_from = None; + target_node.retry_count = 0; + target_node.result = None; + + // BFS the target's own transitive dependents, resetting any non-Pending status + // (Completed/Failed/Skipped/Canceled/Ready/Running left by a prior fallback + // run) back to Pending for a clean re-run. Idempotent: a target that never ran + // has dependents already Pending/Dormant, so this is a no-op. + let mut re_arm_queue: VecDeque = VecDeque::new(); + re_arm_queue.push_back(target); + while let Some(current) = re_arm_queue.pop_front() { + let dependents = rev_adj.get(current.index()).map_or(&[] as &[TaskId], |v| v); + for &dep_id in dependents { + let dep = &mut graph.tasks[dep_id.index()]; + if dep.status != TaskStatus::Pending { + dep.status = TaskStatus::Pending; + dep.retry_count = 0; + dep.result = None; + re_arm_queue.push_back(dep_id); + } + } + } + } + graph.status = GraphStatus::Running; Ok(()) } @@ -790,6 +1084,7 @@ mod tests { let mut tasks = vec![make_node(0, &[])]; tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); tasks[0].verify_predicate = Some(crate::graph::VerifyPredicate::Natural( "criterion".to_string(), @@ -803,6 +1098,7 @@ mod tests { let mut tasks = vec![make_node(0, &[])]; tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); assert!(validate(&tasks, 20, FailureStrategy::Abort).is_ok()); } @@ -812,6 +1108,7 @@ mod tests { let mut tasks = vec![make_node(0, &[])]; tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); tasks[0].failure_strategy = Some(FailureStrategy::Skip); assert!(validate(&tasks, 20, FailureStrategy::Abort).is_ok()); @@ -822,6 +1119,7 @@ mod tests { let mut tasks = vec![make_node(0, &[])]; tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); tasks[0].failure_strategy = Some(FailureStrategy::Ask); assert!(validate(&tasks, 20, FailureStrategy::Abort).is_ok()); @@ -832,6 +1130,7 @@ mod tests { let mut tasks = vec![make_node(0, &[])]; tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); tasks[0].failure_strategy = Some(FailureStrategy::Retry); assert!(validate(&tasks, 20, FailureStrategy::Abort).is_ok()); @@ -840,10 +1139,501 @@ mod tests { let mut tasks2 = vec![make_node(0, &[])]; tasks2[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); assert!(validate(&tasks2, 20, FailureStrategy::Abort).is_ok()); } + // --- route_to (Mode 2) validate tests (spec-075 FR-D-01) --- + + fn make_route_to_pair() -> Vec { + // B(1) routes to F(0). Both have empty depends_on: F per invariant (4), and B + // because route_to is an on-failure edge, not a dependency edge -- B must NOT + // depend on F (that would be the rejected N5 topology). + let mut tasks = vec![make_node(0, &[]), make_node(1, &[])]; + tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + tasks + } + + #[test] + fn test_validate_route_to_valid_pair_ok() { + let tasks = make_route_to_pair(); + assert!(validate(&tasks, 20, FailureStrategy::Abort).is_ok()); + } + + #[test] + fn test_validate_route_to_self_reroute_rejected() { + let mut tasks = vec![make_node(0, &[])]; + tasks[0].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_out_of_range_rejected() { + let mut tasks = vec![make_node(0, &[])]; + tasks[0].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(99)), + }); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_and_state_injection_mutually_exclusive() { + let mut tasks = make_route_to_pair(); + tasks[1].recovery.as_mut().unwrap().state_injection = Some("fallback".to_string()); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_target_with_deps_rejected() { + // F(0) must have empty depends_on; give it one. + let mut tasks = vec![ + make_node(0, &[]), + make_node(1, &[0]), + make_node(2, &[1]), // F=2, but depends on 1 -- invalid target + ]; + tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(2)), + }); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_chained_rejected() { + // M3: F itself must not set route_to (chained reroute unsupported in v1). + let mut tasks = vec![ + make_node(0, &[]), // F2 (final target) + make_node(1, &[]), // F (chains to F2) + make_node(2, &[]), // B (routes to F) + ]; + tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + tasks[2].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(1)), + }); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_n_to_one_rejected() { + // Two sources routing to the same target F. + let mut tasks = vec![ + make_node(0, &[]), // F + make_node(1, &[]), // source 1 + make_node(2, &[]), // source 2 + ]; + tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + tasks[2].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_under_skip_strategy_rejected() { + // Upgraded from Mode-1's warn to a hard error for route_to. + let mut tasks = make_route_to_pair(); + tasks[1].failure_strategy = Some(FailureStrategy::Skip); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_under_ask_strategy_rejected() { + let mut tasks = make_route_to_pair(); + tasks[1].failure_strategy = Some(FailureStrategy::Ask); + let err = validate(&tasks, 20, FailureStrategy::Abort).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + #[test] + fn test_validate_route_to_under_default_skip_strategy_rejected() { + // Effective strategy via graph default (no per-task override) must also reject. + let tasks = make_route_to_pair(); + let err = validate(&tasks, 20, FailureStrategy::Skip).unwrap_err(); + assert_matches!(err, OrchestrationError::InvalidGraph(_)); + } + + // --- mark_dormant_route_to_targets tests --- + + #[test] + fn test_mark_dormant_route_to_targets_marks_pending_target() { + let mut graph = graph_from_nodes(make_route_to_pair()); + mark_dormant_route_to_targets(&mut graph); + assert_eq!(graph.tasks[0].status, TaskStatus::Dormant); + } + + #[test] + fn test_mark_dormant_route_to_targets_guard_skips_non_pending() { + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Completed; + mark_dormant_route_to_targets(&mut graph); + assert_eq!( + graph.tasks[0].status, + TaskStatus::Completed, + "guard must not re-dormant an already-terminal target" + ); + } + + #[test] + fn test_mark_dormant_route_to_targets_no_route_to_is_noop() { + let mut graph = graph_from_nodes(vec![make_node(0, &[])]); + mark_dormant_route_to_targets(&mut graph); + assert_eq!(graph.tasks[0].status, TaskStatus::Pending); + } + + // --- try_reroute / propagate_failure Mode-2 tests --- + + #[test] + fn test_propagate_failure_abort_reroutes_to_dormant_target() { + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.status = GraphStatus::Running; + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Failed; + graph.tasks[1].failure_strategy = Some(FailureStrategy::Abort); + + let __ra = make_rev_adj(&graph); + let to_cancel = propagate_failure(&mut graph, TaskId(1), &__ra); + + assert!(to_cancel.is_empty()); + assert_eq!( + graph.tasks[1].status, + TaskStatus::Failed, + "source stays terminal Failed" + ); + assert_eq!(graph.tasks[0].status, TaskStatus::Ready, "target activated"); + assert_eq!(graph.tasks[0].routed_from, Some(TaskId(1))); + assert_eq!( + graph.status, + GraphStatus::Running, + "graph.status must be left untouched by reroute" + ); + } + + #[test] + fn test_propagate_failure_retry_exhausted_reroutes_to_dormant_target() { + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.status = GraphStatus::Running; + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Failed; + graph.tasks[1].failure_strategy = Some(FailureStrategy::Retry); + graph.tasks[1].max_retries = Some(3); + graph.tasks[1].retry_count = 3; + + let __ra = make_rev_adj(&graph); + propagate_failure(&mut graph, TaskId(1), &__ra); + + assert_eq!(graph.tasks[0].status, TaskStatus::Ready); + assert_eq!(graph.tasks[0].routed_from, Some(TaskId(1))); + assert_eq!(graph.status, GraphStatus::Running); + } + + #[test] + fn test_propagate_failure_reroute_runtime_guard_refuses_non_dormant_target() { + // Target already Ready (e.g. re-arm race / already activated) — try_reroute + // must refuse to clobber it and fall through to Abort instead. + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.status = GraphStatus::Running; + graph.tasks[0].status = TaskStatus::Ready; // NOT Dormant + graph.tasks[1].status = TaskStatus::Failed; + graph.tasks[1].failure_strategy = Some(FailureStrategy::Abort); + + let __ra = make_rev_adj(&graph); + propagate_failure(&mut graph, TaskId(1), &__ra); + + assert_eq!( + graph.tasks[0].status, + TaskStatus::Ready, + "runtime guard must not mutate a non-Dormant target" + ); + assert_eq!(graph.tasks[0].routed_from, None); + assert_eq!( + graph.status, + GraphStatus::Failed, + "must fall through to Abort when reroute is refused" + ); + } + + #[test] + fn test_route_to_target_dependent_becomes_ready_after_reroute() { + // F(0) <- routed by B(1); G(2) depends_on F(0). + let mut graph = graph_from_nodes(vec![ + make_node(0, &[]), + make_node(1, &[]), + make_node(2, &[0]), + ]); + graph.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + graph.status = GraphStatus::Running; + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Failed; + graph.tasks[1].failure_strategy = Some(FailureStrategy::Abort); + + let __ra = make_rev_adj(&graph); + propagate_failure(&mut graph, TaskId(1), &__ra); + assert_eq!(graph.tasks[0].status, TaskStatus::Ready); + + // F completes -> G must unblock via the normal Pending arm. + graph.tasks[0].status = TaskStatus::Completed; + let ready = ready_tasks(&graph); + assert!(ready.contains(&TaskId(2))); + } + + // --- resolve_dormant_after_terminal tests --- + + #[test] + fn test_resolve_dormant_after_terminal_skips_untriggered_fallback_on_source_success() { + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Completed; // source succeeded, never rerouted + + let __ra = make_rev_adj(&graph); + let resolved = resolve_dormant_after_terminal(&mut graph, &__ra); + + assert_eq!(resolved, vec![TaskId(0)]); + assert_eq!(graph.tasks[0].status, TaskStatus::Skipped); + } + + #[test] + fn test_resolve_dormant_after_terminal_skips_subtree() { + // F(0) <- routed by B(1); G(2) depends_on F(0). Source succeeds without + // rerouting: F must resolve Skipped and drag G down with it. + let mut graph = graph_from_nodes(vec![ + make_node(0, &[]), + make_node(1, &[]), + make_node(2, &[0]), + ]); + graph.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Completed; + graph.tasks[2].status = TaskStatus::Pending; + + let __ra = make_rev_adj(&graph); + resolve_dormant_after_terminal(&mut graph, &__ra); + + assert_eq!(graph.tasks[0].status, TaskStatus::Skipped); + assert_eq!( + graph.tasks[2].status, + TaskStatus::Skipped, + "F's downstream subtree must be skipped when the fallback is never triggered" + ); + } + + #[test] + fn test_resolve_dormant_after_terminal_noop_while_source_running() { + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Running; // not terminal yet + + let __ra = make_rev_adj(&graph); + let resolved = resolve_dormant_after_terminal(&mut graph, &__ra); + + assert!(resolved.is_empty()); + assert_eq!(graph.tasks[0].status, TaskStatus::Dormant); + } + + #[test] + fn test_resolve_dormant_after_terminal_ignores_activated_target() { + // Target already Ready (activated by a prior reroute) must never be touched + // by the sweep, even if its source is terminal-Failed. + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Ready; + graph.tasks[0].routed_from = Some(TaskId(1)); + graph.tasks[1].status = TaskStatus::Failed; + + let __ra = make_rev_adj(&graph); + let resolved = resolve_dormant_after_terminal(&mut graph, &__ra); + + assert!(resolved.is_empty()); + assert_eq!(graph.tasks[0].status, TaskStatus::Ready); + } + + // --- reset_for_retry route_to re-arm tests (D2, spec-075 FR-D-01) --- + + #[test] + fn test_reset_for_retry_rearms_dormant_fallback_that_never_ran() { + // Source failed (never rerouted, target still Dormant); graph failed for an + // unrelated reason. Retry must leave the (already-Dormant) target untouched. + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Failed; + graph.status = GraphStatus::Failed; + + let __ra = make_rev_adj(&graph); + reset_for_retry(&mut graph, &__ra).unwrap(); + + assert_eq!( + graph.tasks[1].status, + TaskStatus::Ready, + "source reset for retry" + ); + assert_eq!(graph.tasks[0].status, TaskStatus::Dormant); + assert_eq!(graph.tasks[0].routed_from, None); + } + + #[test] + fn test_reset_for_retry_rearms_activated_fallback_case_a_source_now_succeeds() { + // D2 case (a): F already ran (activated + Completed) from a prior reroute; the + // graph later failed for an unrelated reason. On retry, S is reset to Ready and + // F must be re-armed back to Dormant -- if S now succeeds, F must NOT dispatch. + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Completed; // F already ran + graph.tasks[0].routed_from = Some(TaskId(1)); + graph.tasks[0].result = Some(TaskResult { + output: "stale fallback output".to_string(), + artifacts: Vec::new(), + duration_ms: 5, + agent_id: None, + agent_def: None, + }); + graph.tasks[1].status = TaskStatus::Failed; // S: the route_to source + graph.status = GraphStatus::Failed; + + let __ra = make_rev_adj(&graph); + reset_for_retry(&mut graph, &__ra).unwrap(); + + assert_eq!(graph.tasks[1].status, TaskStatus::Ready); + assert_eq!( + graph.tasks[0].status, + TaskStatus::Dormant, + "activated fallback must be re-armed to Dormant on retry" + ); + assert_eq!( + graph.tasks[0].routed_from, None, + "stale routed_from must be cleared" + ); + assert!( + graph.tasks[0].result.is_none(), + "stale result must be cleared" + ); + + // S now succeeds -- F must stay parked, not dispatch. + graph.tasks[1].status = TaskStatus::Completed; + let ready = ready_tasks(&graph); + assert!( + !ready.contains(&TaskId(0)), + "re-armed Dormant fallback must not dispatch when the source now succeeds" + ); + } + + #[test] + fn test_reset_for_retry_rearms_activated_fallback_case_b_source_fails_again() { + // D2 case (b): same setup, but after retry S fails again -- try_reroute must + // fire again (F was correctly re-Dormant, not stuck Ready/beyond). + let mut graph = graph_from_nodes(make_route_to_pair()); + graph.tasks[0].status = TaskStatus::Completed; + graph.tasks[0].routed_from = Some(TaskId(1)); + graph.tasks[1].status = TaskStatus::Failed; + graph.status = GraphStatus::Failed; + + let __ra = make_rev_adj(&graph); + reset_for_retry(&mut graph, &__ra).unwrap(); + assert_eq!(graph.tasks[0].status, TaskStatus::Dormant); + + // S fails again. + graph.tasks[1].status = TaskStatus::Failed; + graph.tasks[1].failure_strategy = Some(FailureStrategy::Abort); + let to_cancel = propagate_failure(&mut graph, TaskId(1), &__ra); + + assert!(to_cancel.is_empty()); + assert_eq!( + graph.tasks[0].status, + TaskStatus::Ready, + "reroute must fire again after the re-arm" + ); + assert_eq!(graph.tasks[0].routed_from, Some(TaskId(1))); + } + + #[test] + fn test_reset_for_retry_rearm_resets_fallback_subtree() { + // F(0) <- routed by S(1); G(2) depends_on F(0). A prior reroute ran F to + // Completed and G to Completed too. Retry must walk F's subtree and reset G + // back to Pending for a clean re-run. + let mut graph = graph_from_nodes(vec![ + make_node(0, &[]), + make_node(1, &[]), + make_node(2, &[0]), + ]); + graph.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + graph.tasks[0].status = TaskStatus::Completed; + graph.tasks[0].routed_from = Some(TaskId(1)); + graph.tasks[1].status = TaskStatus::Failed; + graph.tasks[2].status = TaskStatus::Completed; + graph.status = GraphStatus::Failed; + + let __ra = make_rev_adj(&graph); + reset_for_retry(&mut graph, &__ra).unwrap(); + + assert_eq!(graph.tasks[0].status, TaskStatus::Dormant); + assert_eq!( + graph.tasks[2].status, + TaskStatus::Pending, + "F's downstream subtree must reset to Pending alongside the re-arm" + ); + } + + #[test] + fn test_reset_for_retry_does_not_rearm_untouched_route_to_source() { + // S succeeded (not in `seeds`); the graph failed for an unrelated reason. The + // untouched source's fallback branch must not be re-armed. + let mut graph = graph_from_nodes(vec![ + make_node(0, &[]), // F (unrelated route_to target) + make_node(1, &[]), // S (succeeded) + make_node(2, &[]), // unrelated failed task causing graph Failed + ]); + graph.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + graph.tasks[0].status = TaskStatus::Dormant; + graph.tasks[1].status = TaskStatus::Completed; // S succeeded, never rerouted + graph.tasks[2].status = TaskStatus::Failed; + graph.status = GraphStatus::Failed; + + let __ra = make_rev_adj(&graph); + reset_for_retry(&mut graph, &__ra).unwrap(); + + assert_eq!( + graph.tasks[1].status, + TaskStatus::Completed, + "S was not reset (not Failed)" + ); + assert_eq!( + graph.tasks[0].status, + TaskStatus::Dormant, + "F must be untouched since its source was never reset" + ); + } + // --- toposort tests --- #[test] @@ -1184,6 +1974,7 @@ mod tests { graph.tasks[0].failure_strategy = Some(FailureStrategy::Abort); graph.tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback output".to_string()), + route_to: None, }); let __ra = make_rev_adj(&graph); @@ -1216,6 +2007,7 @@ mod tests { graph.tasks[0].retry_count = 3; // at max — exhausted graph.tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback output".to_string()), + route_to: None, }); let __ra = make_rev_adj(&graph); @@ -1267,6 +2059,7 @@ mod tests { graph.tasks[0].failure_strategy = Some(FailureStrategy::Abort); graph.tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback output".to_string()), + route_to: None, }); graph.tasks[1].status = TaskStatus::Pending; @@ -1291,6 +2084,7 @@ mod tests { graph.tasks[0].failure_strategy = Some(FailureStrategy::Skip); graph.tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback output".to_string()), + route_to: None, }); let __ra = make_rev_adj(&graph); @@ -1306,6 +2100,7 @@ mod tests { graph.tasks[0].failure_strategy = Some(FailureStrategy::Ask); graph.tasks[0].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("fallback output".to_string()), + route_to: None, }); let __ra = make_rev_adj(&graph); diff --git a/crates/zeph-orchestration/src/graph.rs b/crates/zeph-orchestration/src/graph.rs index 65e532cfd..6b4e34850 100644 --- a/crates/zeph-orchestration/src/graph.rs +++ b/crates/zeph-orchestration/src/graph.rs @@ -158,10 +158,23 @@ impl FromStr for GraphId { /// → Failed (error; then failure strategy applies) /// → Skipped (upstream failed with Skip strategy) /// → Canceled (graph aborted while task was running) +/// +/// Dormant → Ready (on-failure activation: this node's `route_to` source failed +/// terminally, see `dag::try_reroute`) +/// → Skipped (source terminalized without rerouting, see +/// `dag::resolve_dormant_after_terminal`) /// ``` /// +/// `Dormant` is the Mode-2 `route_to` fallback-node marker (spec-075 FR-D-01): a task +/// with `recovery.route_to == Some(F)` set on another node starts `Dormant` instead of +/// `Pending` (see `dag::mark_dormant_route_to_targets`) and is excluded from +/// [`ready_tasks`](crate::dag::ready_tasks) dispatch until explicitly activated by its +/// source's terminal failure. It never survives to graph termination — see +/// `dag::resolve_dormant_after_terminal`. +/// /// Only `Completed`, `Failed`, `Skipped`, and `Canceled` are terminal — see -/// [`TaskStatus::is_terminal`]. +/// [`TaskStatus::is_terminal`]. `Dormant` is intentionally **not** terminal: it is a +/// parked pre-dispatch state, not an end state. /// /// # Examples /// @@ -170,6 +183,7 @@ impl FromStr for GraphId { /// /// assert!(TaskStatus::Completed.is_terminal()); /// assert!(!TaskStatus::Running.is_terminal()); +/// assert!(!TaskStatus::Dormant.is_terminal()); /// assert_eq!(TaskStatus::Pending.to_string(), "pending"); /// ``` #[non_exhaustive] @@ -190,6 +204,12 @@ pub enum TaskStatus { Skipped, /// Task was running when the graph was aborted ([`FailureStrategy::Abort`]). Canceled, + /// Mode-2 `route_to` fallback node parked before activation. Never dispatched by + /// [`ready_tasks`](crate::dag::ready_tasks); only reachable via + /// `dag::try_reroute` (on-failure activation to `Ready`) or + /// `dag::resolve_dormant_after_terminal` (terminal resolution to `Skipped`). Not + /// terminal — see [`TaskStatus::is_terminal`]. + Dormant, } impl TaskStatus { @@ -213,6 +233,7 @@ impl fmt::Display for TaskStatus { TaskStatus::Failed => write!(f, "failed"), TaskStatus::Skipped => write!(f, "skipped"), TaskStatus::Canceled => write!(f, "canceled"), + TaskStatus::Dormant => write!(f, "dormant"), } } } @@ -396,25 +417,40 @@ pub struct TimeoutPolicy { /// Declarative recovery action applied on a node's terminal failure. /// -/// v1 supports Mode 1 (`state_injection`) only: on `Abort`-default or retry-exhausted -/// `Retry` failure, the node is marked [`TaskStatus::Completed`] with the given output -/// substituted as its [`TaskResult`], letting the graph continue past the failure. Mode 2 -/// (reroute to an alternate node) is deferred — see -/// `specs/075-orchestration-node-control-parity/spec.md` §7. +/// Two mutually exclusive modes (`dag::validate` rejects a node that sets both): +/// +/// - **Mode 1** (`state_injection`): on `Abort`-default or retry-exhausted `Retry` +/// failure, the node is marked [`TaskStatus::Completed`] with the given output +/// substituted as its [`TaskResult`], letting the graph continue past the failure. +/// - **Mode 2** (`route_to`): on the same failure conditions, an alternate fallback +/// node is activated instead (`TaskStatus::Dormant → Ready`) and the failed node's +/// output is injected into the fallback's prompt. See `dag::try_reroute`, +/// `dag::mark_dormant_route_to_targets`, and the [`TaskStatus::Dormant`] state-machine +/// doc for the full mechanism. /// /// # Examples /// /// ```rust /// use zeph_orchestration::graph::RecoveryAction; /// -/// let recovery = RecoveryAction { state_injection: Some("fallback output".to_string()) }; +/// let recovery = RecoveryAction { +/// state_injection: Some("fallback output".to_string()), +/// route_to: None, +/// }; /// assert_eq!(recovery.state_injection.as_deref(), Some("fallback output")); /// ``` #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RecoveryAction { /// Substitute output injected as this node's [`TaskResult::output`] on recovery. - /// `None` disables recovery (equivalent to omitting the field entirely). + /// `None` disables Mode-1 recovery. Mutually exclusive with `route_to`. pub state_injection: Option, + /// Mode-2 fallback target: on this node's terminal failure, activate the task at + /// this ID (`Dormant → Ready`) instead of aborting. `None` disables Mode-2 + /// recovery. Mutually exclusive with `state_injection`. The target must have an + /// empty `depends_on` and must not itself set `route_to` (`dag::validate` enforces + /// both, v1 does not support chained reroutes). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub route_to: Option, } /// A single node in the task DAG. @@ -437,6 +473,7 @@ pub struct RecoveryAction { /// assert!(node.asset_sensitivity.is_none()); /// assert!(node.timeout.is_none()); /// assert!(node.recovery.is_none()); +/// assert!(node.routed_from.is_none()); /// ``` #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TaskNode { @@ -518,11 +555,21 @@ pub struct TaskNode { #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout: Option, - /// Declarative Mode-1 recovery action applied on terminal failure. `None` = no - /// recovery, existing `Abort`/retry-exhausted-`Retry` behavior is unchanged. See + /// Declarative recovery action applied on terminal failure. `None` = no recovery, + /// existing `Abort`/retry-exhausted-`Retry` behavior is unchanged. See /// [`RecoveryAction`]. #[serde(default, skip_serializing_if = "Option::is_none")] pub recovery: Option, + + /// Set at Mode-2 activation time to the ID of the task whose terminal failure + /// activated this node (`dag::try_reroute`). `None` for every task that was never + /// a `route_to` fallback target, and cleared back to `None` when a fallback branch + /// is re-armed to `Dormant` on `/plan retry` (`dag::reset_for_retry`). Read by + /// `build_task_prompt` to inject the failed source's sanitized output and by the + /// `LevelBarrier` dispatch gate to let an activated fallback bypass the level + /// check. Persisted so a mid-fallback restart does not lose the injection source. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub routed_from: Option, } impl TaskNode { @@ -551,6 +598,7 @@ impl TaskNode { asset_sensitivity: None, timeout: None, recovery: None, + routed_from: None, } } } @@ -1093,6 +1141,7 @@ mod tests { }); node.recovery = Some(RecoveryAction { state_injection: Some("fallback".to_string()), + route_to: None, }); let json = serde_json::to_string(&node).unwrap(); let restored: TaskNode = serde_json::from_str(&json).unwrap(); diff --git a/crates/zeph-orchestration/src/router.rs b/crates/zeph-orchestration/src/router.rs index 40ec35ca5..1c05a08db 100644 --- a/crates/zeph-orchestration/src/router.rs +++ b/crates/zeph-orchestration/src/router.rs @@ -165,6 +165,7 @@ mod tests { asset_sensitivity: None, timeout: None, recovery: None, + routed_from: None, } } diff --git a/crates/zeph-orchestration/src/scheduler/mod.rs b/crates/zeph-orchestration/src/scheduler/mod.rs index ffae80044..d6c713a5d 100644 --- a/crates/zeph-orchestration/src/scheduler/mod.rs +++ b/crates/zeph-orchestration/src/scheduler/mod.rs @@ -381,11 +381,11 @@ impl DagScheduler { graph.status = GraphStatus::Running; - for task in &mut graph.tasks { - if task.depends_on.is_empty() && task.status == TaskStatus::Pending { - task.status = TaskStatus::Ready; - } - } + // Root-activation moved to `init_common` (D1, spec-075 FR-D-01): a Mode-2 + // `route_to` target has empty `depends_on` by validate invariant and starts + // `Pending`, so activating roots here — before dormancy marking runs — would + // flip it straight to `Ready`, firing the fallback on a fresh graph. See + // `init_common` for the dormancy-first ordering. // Validate cascade_routing dependency on topology_selection. if config.cascade_routing && !config.topology_selection { @@ -529,18 +529,41 @@ impl DagScheduler { /// Build a fully-initialized `DagScheduler` from the supplied pre-configured state. /// - /// Called by both [`DagScheduler::new`] and [`DagScheduler::resume_from`] after they - /// perform their graph-state-specific setup (status transitions, root-task marking, - /// running-map reconstruction). Centralizes all field construction that is identical - /// between the two constructors. + /// Called by both [`DagScheduler::new`] and [`DagScheduler::resume_from`] (via + /// [`DagScheduler::resume_from_durable`]) after they perform their + /// graph-state-specific setup (status transitions, running-map reconstruction). + /// Centralizes all field construction that is identical between the constructors — + /// this is the single chokepoint every graph passes through, which is why the + /// Mode-2 dormancy marking and root-activation loop live here (D1, spec-075 + /// FR-D-01) rather than duplicated per constructor. fn init_common( - graph: TaskGraph, + mut graph: TaskGraph, running: HashMap, config: &OrchestrationConfig, router: Box, available_agents: Vec, admission_gate: Option, ) -> Self { + // D1 (spec-075 FR-D-01): dormancy marking MUST run before root-activation. + // A Mode-2 `route_to` target has empty `depends_on` (validate invariant) and + // starts `Pending`; the root-activation loop below unconditionally flips an + // empty-`depends_on` `Pending` task to `Ready`. Marking dormancy first turns + // that target `Dormant`, so the (guard: `== Pending`) root-activation loop + // skips it — the fallback stays parked instead of firing on a fresh graph. + // + // This loop was moved here from `new()` verbatim: `ready_tasks()`'s `Pending` + // arm already treats an empty-`depends_on` `Pending` task as ready (its + // `all_deps_done` check is vacuously true), so eagerly materializing `Ready` + // here changes no dispatch decision on `resume_from`/`resume_from_durable` — + // and neither path ever carries a `Pending` route_to target anyway, since a + // `Created` graph always enters through `new()` first. + dag::mark_dormant_route_to_targets(&mut graph); + for task in &mut graph.tasks { + if task.depends_on.is_empty() && task.status == TaskStatus::Pending { + task.status = TaskStatus::Ready; + } + } + let agent_provider_map: HashMap = available_agents .iter() .filter_map(|def| { @@ -929,6 +952,60 @@ mod tests { assert_eq!(scheduler.graph().status, GraphStatus::Running); } + #[test] + fn test_new_marks_route_to_target_dormant_not_ready() { + // D1 (spec-075 FR-D-01): dormancy marking must run before root-activation in + // `init_common`. A route_to target has empty depends_on (validate invariant) + // and starts Pending -- without the D1 ordering fix, the root-activation loop + // would flip it straight to Ready on a fresh graph, firing the fallback + // immediately instead of parking it. + let mut graph = graph_from_nodes(vec![make_node(0, &[]), make_node(1, &[])]); + graph.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + let scheduler = make_scheduler(graph); + assert_eq!( + scheduler.graph().tasks[0].status, + TaskStatus::Dormant, + "route_to target must start Dormant, not be swept into Ready by root-activation" + ); + assert_eq!( + scheduler.graph().tasks[1].status, + TaskStatus::Ready, + "the route_to source itself is a normal root and must still activate" + ); + } + + #[test] + fn test_resume_from_leaves_dormant_route_to_target_dormant() { + // Tester/reviewer-flagged gap: D1's claim that "the resume-path root-activation + // loop is provably inert for a Dormant target" was only argued analytically in the + // design handoffs, never test-enforced. This drives the real `resume_from` entry + // point (as a persisted-checkpoint reload would) with a graph where the route_to + // target F is already `Dormant` -- simulating a graph paused/persisted after `new()` + // already ran dormancy marking once -- and asserts `init_common`'s dormancy pass and + // root-activation loop leave it parked rather than re-marking it `Ready`. + let mut graph = graph_from_nodes(vec![make_node(0, &[]), make_node(1, &[])]); + graph.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(TaskId(0)), + }); + graph.tasks[0].status = TaskStatus::Dormant; + graph.status = GraphStatus::Failed; + + let config = make_config(); + let scheduler = + DagScheduler::resume_from(graph, &config, Box::new(FirstRouter), vec![], None).unwrap(); + + assert_eq!( + scheduler.graph().tasks[0].status, + TaskStatus::Dormant, + "an already-Dormant route_to target must stay parked across resume_from, not be \ + reactivated to Ready by the root-activation loop" + ); + } + #[test] fn test_new_validates_empty_graph() { let graph = graph_from_nodes(vec![]); diff --git a/crates/zeph-orchestration/src/scheduler/planner.rs b/crates/zeph-orchestration/src/scheduler/planner.rs index b0a76dc23..c2b984b64 100644 --- a/crates/zeph-orchestration/src/scheduler/planner.rs +++ b/crates/zeph-orchestration/src/scheduler/planner.rs @@ -54,11 +54,15 @@ impl DagScheduler { self.max_parallel = self.topology.max_parallel; self.topology_dirty = false; if self.topology.strategy == DispatchStrategy::LevelBarrier { + // D4 (spec-075 FR-D-01): a Dormant route_to fallback is parked, not + // blocking — exclude it from the min-active-depth floor, else a Dormant + // node at a shallow depth pulls `current_level` back down after every + // `inject_tasks` and re-serializes levels the barrier already passed. let min_active = self .graph .tasks .iter() - .filter(|t| !t.status.is_terminal()) + .filter(|t| !t.status.is_terminal() && t.status != TaskStatus::Dormant) .filter_map(|t| self.topology.depths.get(&t.id).copied()) .min(); if let Some(min_depth) = min_active { @@ -68,6 +72,17 @@ impl DagScheduler { } /// Advance the `LevelBarrier` level when all tasks at the current level are terminal. + /// + /// A [`TaskStatus::Dormant`] task is treated as parked/non-blocking here (D4, + /// spec-075 FR-D-01): `validate` forces a `route_to` target to depth 0 + /// (`depends_on.is_empty()`), and without this the barrier would never advance past + /// a still-Dormant fallback sitting at level 0 while its (deeper) source is still + /// running — a silent livelock invisible to the deadlock detector, since the + /// gated-but-ready source never shows `ready_tasks()` as empty. + /// [`super::DagScheduler::check_graph_completion`]'s `resolve_dormant_after_terminal` + /// sweep is what eventually resolves a Dormant node still parked at graph + /// completion time — this predicate only keeps the barrier itself from stalling on + /// one before that sweep runs. pub(super) fn advance_level_barrier_if_needed(&mut self) { if self.topology.strategy != DispatchStrategy::LevelBarrier { return; @@ -79,7 +94,9 @@ impl DagScheduler { .get(&t.id) .copied() .unwrap_or(usize::MAX); - task_depth != self.current_level || t.status.is_terminal() + task_depth != self.current_level + || t.status.is_terminal() + || t.status == TaskStatus::Dormant }); if all_current_level_terminal { let max_depth = self.topology.depth; @@ -91,7 +108,9 @@ impl DagScheduler { .get(&t.id) .copied() .unwrap_or(usize::MAX); - d == self.current_level && !t.status.is_terminal() + d == self.current_level + && !t.status.is_terminal() + && t.status != TaskStatus::Dormant }); if has_non_terminal { break; @@ -112,6 +131,24 @@ impl DagScheduler { if running_in_graph_now != 0 || !self.running.is_empty() { return vec![]; } + + // Mode-2 completion-time resolution sweep (spec-075 FR-D-01): must run BEFORE + // the `all_terminal`/deadlock checks below. A still-Dormant route_to fallback + // is non-terminal and excluded from `ready_tasks()`, so without this sweep a + // successful plan carrying an untriggered fallback would be misreported as a + // scheduler deadlock. This is the quiescent-tick chokepoint: it runs whenever + // `check_graph_completion` is reached with no Running tasks, covering every way + // a route_to source can terminalize without rerouting (success, upstream-skip, + // cancel) in one place. NOTE: this sweep does NOT run on the Abort/retry- + // exhausted `graph.status = Failed` path — `tick()` returns before reaching + // `check_graph_completion` on that path, so a Dormant fallback can persist into + // a Failed graph. That is acceptable: `/plan retry` (`dag::reset_for_retry`) + // re-arms it if its source is reset, or this sweep resolves it once the + // retried graph heads to Completed. + if !dag::resolve_dormant_after_terminal(&mut self.graph, &self.topology.rev_adj).is_empty() + { + self.graph_dirty = true; + } let all_terminal = self.graph.tasks.iter().all(|t| t.status.is_terminal()); if all_terminal { self.graph.status = GraphStatus::Completed; @@ -418,6 +455,219 @@ mod tests { assert_eq!(scheduler.current_level, 1); } + // --- Mode-2 route_to LevelBarrier tests (D4, spec-075 FR-D-01) --- + // + // A(0, depth0) -> B(1, depth1, route_to=F(2)). F(2, depth0, fallback, depends_on=[]). + // `validate` forces a route_to target's `depends_on` empty, so F is always a graph + // root — this graph naturally classifies as `Mixed` (two roots), not `Hierarchical`. + // The LevelBarrier strategy and per-task depths are forced manually below (same + // override pattern as `current_level` elsewhere in this file) to exercise the D4 + // barrier-parking predicates against a route_to source sitting deeper than its + // depth-0 fallback — the exact shape the critic confirmed hangs without the fix. + + fn make_route_to_level_barrier_graph() -> crate::graph::TaskGraph { + let mut g = graph_from_nodes(vec![ + make_node(0, &[]), + make_node(1, &[0]), + make_node(2, &[]), + ]); + g.tasks[1].recovery = Some(crate::graph::RecoveryAction { + state_injection: None, + route_to: Some(crate::graph::TaskId(2)), + }); + g + } + + fn force_level_barrier_with_route_to_depths(scheduler: &mut DagScheduler) { + use crate::graph::TaskId; + use crate::topology::{DispatchStrategy, build_rev_adj}; + scheduler.topology.strategy = DispatchStrategy::LevelBarrier; + scheduler.topology.depth = 1; + scheduler.topology.depths = [(TaskId(0), 0), (TaskId(1), 1), (TaskId(2), 0)] + .into_iter() + .collect(); + scheduler.topology.rev_adj = build_rev_adj(&scheduler.graph.tasks); + scheduler.current_level = 0; + } + + #[test] + fn test_level_barrier_route_to_source_succeeds_dormant_fallback_resolves_without_hang() { + use crate::graph::TaskId; + + let graph = make_route_to_level_barrier_graph(); + let config = zeph_config::OrchestrationConfig { + topology_selection: true, + max_parallel: 4, + ..make_config() + }; + let mut scheduler = DagScheduler::new( + graph, + &config, + Box::new(FirstRouter), + vec![make_def("worker")], + None, + ) + .unwrap(); + assert_eq!( + scheduler.graph.tasks[2].status, + TaskStatus::Dormant, + "F must start Dormant" + ); + + force_level_barrier_with_route_to_depths(&mut scheduler); + + // Tick 1: only A (depth 0) dispatches. Dormant F sits at depth 0 too but must + // not block dispatch or the barrier's advancement predicate. + let actions = scheduler.tick(); + let spawned: Vec<_> = actions + .iter() + .filter_map(|a| { + if let SchedulerAction::Spawn { task_id, .. } = a { + Some(*task_id) + } else { + None + } + }) + .collect(); + assert_eq!(spawned, vec![TaskId(0)]); + + scheduler.graph.tasks[0].status = TaskStatus::Completed; + scheduler.running.clear(); + + // Tick 2: before the D4 fix, the still-Dormant F at level 0 would prevent the + // barrier from ever advancing, so B (depth 1) would never dispatch, never fail, + // and route_to would never fire -- a silent livelock invisible to the deadlock + // detector (ready_tasks() is non-empty: B is ready but level-gated). + let actions2 = scheduler.tick(); + assert_eq!( + scheduler.current_level, 1, + "barrier must advance past level 0 despite the Dormant F sitting there" + ); + let spawned2: Vec<_> = actions2 + .iter() + .filter_map(|a| { + if let SchedulerAction::Spawn { task_id, .. } = a { + Some(*task_id) + } else { + None + } + }) + .collect(); + assert_eq!( + spawned2, + vec![TaskId(1)], + "B must dispatch once the barrier advances" + ); + + // B completes without ever failing -> route_to never fires; F must resolve via + // the completion-time sweep rather than strand the graph. + scheduler.graph.tasks[1].status = TaskStatus::Completed; + scheduler.running.clear(); + + let actions3 = scheduler.tick(); + assert_eq!( + scheduler.graph.tasks[2].status, + TaskStatus::Skipped, + "untriggered fallback must resolve Skipped via the completion sweep" + ); + assert!( + actions3.iter().any(|a| matches!( + a, + SchedulerAction::Done { + status: crate::graph::GraphStatus::Completed + } + )), + "graph must complete, not deadlock, once the fallback resolves: {actions3:?}" + ); + } + + #[test] + fn test_level_barrier_route_to_source_fails_fallback_activates_out_of_level() { + use crate::graph::TaskId; + use crate::scheduler::{RunningTask, TaskEvent, TaskOutcome}; + + let graph = make_route_to_level_barrier_graph(); + let config = zeph_config::OrchestrationConfig { + topology_selection: true, + max_parallel: 4, + ..make_config() + }; + let mut scheduler = DagScheduler::new( + graph, + &config, + Box::new(FirstRouter), + vec![make_def("worker")], + None, + ) + .unwrap(); + + force_level_barrier_with_route_to_depths(&mut scheduler); + + // Advance to level 1: A dispatches and completes, B dispatches. + scheduler.tick(); + scheduler.graph.tasks[0].status = TaskStatus::Completed; + scheduler.running.clear(); + scheduler.tick(); + assert_eq!(scheduler.current_level, 1); + assert_eq!(scheduler.graph.tasks[1].status, TaskStatus::Running); + + // B fails terminally. + scheduler.running.insert( + TaskId(1), + RunningTask { + agent_handle_id: "h1".to_string(), + agent_def_name: "worker".to_string(), + started_at: std::time::Instant::now(), + admission_permit: None, + last_progress_at: None, + }, + ); + scheduler.buffered_events.push_back(TaskEvent { + task_id: TaskId(1), + agent_handle_id: "h1".to_string(), + outcome: TaskOutcome::Failed { + error: "simulated failure".to_string(), + }, + }); + + // Tick: try_reroute activates F (Dormant -> Ready, routed_from = Some(B)); F + // must dispatch on this same tick, bypassing the depth-0-vs-current_level(1+) + // gate rather than waiting for the barrier to wind back down to level 0. + let actions = scheduler.tick(); + assert_eq!(scheduler.graph.tasks[1].status, TaskStatus::Failed); + assert_eq!(scheduler.graph.tasks[2].status, TaskStatus::Running); + assert_eq!( + scheduler.graph.tasks[2].routed_from, + Some(TaskId(1)), + "activated fallback must record its source" + ); + assert!( + actions.iter().any( + |a| matches!(a, SchedulerAction::Spawn { task_id, .. } if *task_id == TaskId(2)) + ), + "F must dispatch out-of-level on the same tick it is activated: {actions:?}" + ); + assert_eq!( + scheduler.graph.status, + crate::graph::GraphStatus::Running, + "graph must stay Running -- the failure was absorbed by the reroute" + ); + + // F completes -> graph reaches Completed with B terminal-Failed alongside it. + scheduler.graph.tasks[2].status = TaskStatus::Completed; + scheduler.running.clear(); + let actions2 = scheduler.tick(); + assert!( + actions2.iter().any(|a| matches!( + a, + SchedulerAction::Done { + status: crate::graph::GraphStatus::Completed + } + )), + "graph must complete once the activated fallback finishes: {actions2:?}" + ); + } + #[test] fn resume_from_preserves_topology_classification() { use crate::graph::GraphStatus; diff --git a/crates/zeph-orchestration/src/scheduler/router.rs b/crates/zeph-orchestration/src/scheduler/router.rs index 872054134..7548fa42d 100644 --- a/crates/zeph-orchestration/src/scheduler/router.rs +++ b/crates/zeph-orchestration/src/scheduler/router.rs @@ -15,9 +15,32 @@ impl DagScheduler { /// Uses char-boundary-safe truncation (S1 fix) to avoid panics on multi-byte UTF-8. /// Dependency output is sanitized (SEC-ORCH-01) and titles are XML-escaped to prevent /// prompt injection via crafted task outputs. + /// + /// Mode-2 `route_to` injection (spec-075 FR-D-01): when `task.routed_from` is set, + /// prepends a `` block with the failed source's sanitized output. + /// This runs **before** the empty-`depends_on` early return below, because a + /// `route_to` target's `depends_on` is always empty (`validate` invariant) — it has + /// no `Completed`-dependency channel to the source, so the `routed_from` marker is + /// the sole path for the source's output to reach this prompt. pub(super) fn build_task_prompt(&self, task: &TaskNode) -> String { + let recovery_block = task.routed_from.map(|src_id| { + let src = &self.graph.tasks[src_id.index()]; + let escaped_id = xml_escape(&src.id.to_string()); + let escaped_title = xml_escape(&src.title); + let safe_output = src + .result + .as_ref() + .map_or_else(String::new, |r| self.sanitizer.sanitize_task_output(&r.output)); + format!( + "\n## Task \"{escaped_id}\": \"{escaped_title}\" (failed; this task is the recovery fallback)\n{safe_output}\n\n\n" + ) + }); + if task.depends_on.is_empty() { - return task.description.clone(); + return match recovery_block { + Some(block) => format!("{block}Your task: {}", task.description), + None => task.description.clone(), + }; } let completed_deps: Vec<&TaskNode> = task @@ -101,6 +124,7 @@ mod tests { FailureStrategy, GraphStatus, RecoveryAction, TaskId, TaskResult, TaskStatus, }; use crate::scheduler::tests::*; + use crate::scheduler::{RunningTask, SchedulerAction, TaskEvent, TaskOutcome}; use crate::topology::build_rev_adj; #[test] @@ -225,6 +249,7 @@ mod tests { graph.tasks[0].failure_strategy = Some(FailureStrategy::Abort); graph.tasks[0].recovery = Some(RecoveryAction { state_injection: Some("fallback output".to_string()), + route_to: None, }); let rev_adj = build_rev_adj(&graph.tasks); @@ -257,4 +282,136 @@ mod tests { "agent_def marker must not leak into the prompt. Prompt: {prompt}" ); } + + #[test] + fn test_build_prompt_includes_mode2_routed_from_injection() { + // Drives the real Mode-2 reroute path (propagate_failure -> try_reroute in + // dag.rs) end-to-end: F(1) is a route_to target for B(0), which fails. + use crate::graph::TaskStatus as TS; + + // B=0 (source), F=1 (target, empty depends_on). + let mut graph = graph_from_nodes(vec![make_node(0, &[]), make_node(1, &[])]); + graph.tasks[0].recovery = Some(RecoveryAction { + state_injection: None, + route_to: Some(TaskId(1)), + }); + graph.status = GraphStatus::Running; + graph.tasks[1].status = TS::Dormant; + graph.tasks[0].status = TS::Failed; + graph.tasks[0].failure_strategy = Some(FailureStrategy::Abort); + graph.tasks[0].result = Some(TaskResult { + output: "boom: connection refused".to_string(), + artifacts: vec![], + duration_ms: 5, + agent_id: None, + agent_def: None, + }); + + let rev_adj = build_rev_adj(&graph.tasks); + let to_cancel = propagate_failure(&mut graph, TaskId(0), &rev_adj); + assert!(to_cancel.is_empty()); + assert_eq!(graph.tasks[1].status, TS::Ready); + assert_eq!(graph.tasks[1].routed_from, Some(TaskId(0))); + + graph.status = GraphStatus::Created; + let config = make_config(); + let scheduler = DagScheduler::new( + graph, + &config, + Box::new(FirstRouter), + vec![make_def("worker")], + None, + ) + .unwrap(); + + let prompt = scheduler.build_task_prompt(&scheduler.graph.tasks[1]); + assert!( + prompt.contains(""), + "prompt must include the recovery-source block. Prompt: {prompt}" + ); + assert!( + prompt.contains("boom: connection refused"), + "prompt must include the failed source's sanitized output. Prompt: {prompt}" + ); + assert!( + prompt.contains("Your task:"), + "prompt must still include the target's own task description. Prompt: {prompt}" + ); + } + + #[test] + fn test_build_prompt_mode2_routed_from_injection_via_real_failure_path() { + // Regression for the reviewer's Critical finding: the test above manually + // pre-sets `graph.tasks[0].result` before calling `propagate_failure`, a + // precondition that never arises via the real dispatch pipeline -- production + // Failed-transition sites (`tick/mod.rs`'s spawn-failure, `handle_failed_outcome`, + // and timeout paths) never populated `.result`, so this test masked the gap where + // Mechanism 4's injection was silently inert. This test instead drives the real + // `handle_failed_outcome` path end-to-end via `tick()`, exactly as a live agent + // failure event would, and asserts the fallback's prompt actually contains the + // failed source's error content. + let mut graph = graph_from_nodes(vec![make_node(0, &[]), make_node(1, &[])]); + graph.tasks[0].recovery = Some(RecoveryAction { + state_injection: None, + route_to: Some(TaskId(1)), + }); + graph.tasks[0].failure_strategy = Some(FailureStrategy::Abort); + + let mut scheduler = make_scheduler(graph); + assert_eq!( + scheduler.graph.tasks[1].status, + TaskStatus::Dormant, + "route_to target must start Dormant" + ); + + // Simulate task 0 having been dispatched and now failing for real, via the same + // event pipeline a live sub-agent failure uses. + scheduler.graph.tasks[0].status = TaskStatus::Running; + scheduler.running.insert( + TaskId(0), + RunningTask { + agent_handle_id: "h0".to_string(), + agent_def_name: "worker".to_string(), + started_at: std::time::Instant::now(), + admission_permit: None, + last_progress_at: None, + }, + ); + scheduler.buffered_events.push_back(TaskEvent { + task_id: TaskId(0), + agent_handle_id: "h0".to_string(), + outcome: TaskOutcome::Failed { + error: "boom: connection refused".to_string(), + }, + }); + // A single `tick()` both processes the failure event (activating the Dormant + // fallback via `try_reroute`) and dispatches it (it is now Ready with no + // dependencies), so the returned `Spawn` action carries the exact prompt a live + // sub-agent would receive -- built by the same `build_task_prompt` call this test + // is regression-testing. + let actions = scheduler.tick(); + + assert_eq!(scheduler.graph.tasks[0].status, TaskStatus::Failed); + assert_eq!(scheduler.graph.tasks[1].routed_from, Some(TaskId(0))); + + let prompt = actions + .iter() + .find_map(|a| match a { + SchedulerAction::Spawn { + task_id, prompt, .. + } if *task_id == TaskId(1) => Some(prompt), + _ => None, + }) + .expect("fallback task must have been dispatched with a Spawn action this tick"); + assert!( + prompt.contains(""), + "prompt must include the recovery-source block. Prompt: {prompt}" + ); + assert!( + prompt.contains("boom: connection refused"), + "prompt must include the failed source's real error output, populated by the \ + production Failed-transition path in tick/mod.rs -- not a manually-set result. \ + Prompt: {prompt}" + ); + } } diff --git a/crates/zeph-orchestration/src/scheduler/tick/mod.rs b/crates/zeph-orchestration/src/scheduler/tick/mod.rs index 0d5283770..bd2846c73 100644 --- a/crates/zeph-orchestration/src/scheduler/tick/mod.rs +++ b/crates/zeph-orchestration/src/scheduler/tick/mod.rs @@ -139,8 +139,19 @@ impl DagScheduler { break; } - // LevelBarrier: only dispatch tasks at the current level. - if self.topology.strategy == DispatchStrategy::LevelBarrier { + let task = &self.graph.tasks[task_id.index()]; + + // LevelBarrier: only dispatch tasks at the current level. Exception (D4, + // spec-075 FR-D-01): a just-activated Mode-2 fallback (`Ready` with + // `routed_from.is_some()`, set only by `dag::try_reroute`) bypasses the + // level gate. It must dispatch immediately regardless of `current_level` — + // waiting for the barrier to reach its depth-0 level again could stall + // indefinitely on unrelated deeper levels. This is safe because `validate` + // forces every route_to target to have an empty `depends_on`: it has no + // prerequisites, so dispatching it out-of-level can never run ahead of + // anything it depends on. + let is_activated_fallback = task.routed_from.is_some(); + if self.topology.strategy == DispatchStrategy::LevelBarrier && !is_activated_fallback { let task_depth = self .topology .depths @@ -152,8 +163,6 @@ impl DagScheduler { } } - let task = &self.graph.tasks[task_id.index()]; - // Sequential tasks: only one may run at a time within the scheduler. // Independent sequential tasks in separate DAG branches are still // serialized here (they share exclusive-resource intent by annotation). @@ -387,6 +396,18 @@ impl DagScheduler { ); self.graph_dirty = true; self.graph.tasks[task_id.index()].status = TaskStatus::Failed; + // Populate `.result` with the spawn error so Mode-2 `routed_from` prompt injection + // (router.rs's `build_task_prompt`) has real content to surface, and so + // `finalize_plan_failed`'s error-message formatting doesn't fall back to + // "unknown error". No agent was ever spawned, so `agent_id`/`agent_def`/`duration_ms` + // stay at their zero values. + self.graph.tasks[task_id.index()].result = Some(TaskResult { + output: error_excerpt, + artifacts: Vec::new(), + duration_ms: 0, + agent_id: None, + agent_def: None, + }); let cancel_ids = dag::propagate_failure(&mut self.graph, task_id, &self.topology.rev_adj); let mut actions = Vec::new(); for cancel_task_id in cancel_ids { @@ -632,6 +653,18 @@ impl DagScheduler { "task failed" ); self.graph.tasks[task_id.index()].status = TaskStatus::Failed; + // Populate `.result` with the failure error so Mode-2 `routed_from` prompt injection + // (router.rs's `build_task_prompt`) has real content to surface, and so + // `finalize_plan_failed`'s error-message formatting doesn't fall back to + // "unknown error". `agent_id`/`agent_def`/`duration_ms` are left at their zero + // values -- this is failure diagnostics, not a completed-task provenance record. + self.graph.tasks[task_id.index()].result = Some(TaskResult { + output: error_excerpt, + artifacts: Vec::new(), + duration_ms: 0, + agent_id: None, + agent_def: None, + }); if let Some(ref mut detector) = self.cascade_detector { detector.record_outcome(task_id, false, &self.graph); @@ -837,6 +870,10 @@ impl DagScheduler { let removed = self.running.remove(&task_id); self.graph.tasks[task_id.index()].status = TaskStatus::Failed; + // `.result` is populated below (cause-aware) so Mode-2 `routed_from` prompt + // injection (router.rs's `build_task_prompt`) has real content to surface, and so + // `finalize_plan_failed`'s error-message formatting doesn't fall back to + // "unknown error". let duration_ms = removed.as_ref().map_or(0, |r| { u64::try_from(r.started_at.elapsed().as_millis()).unwrap_or(u64::MAX) diff --git a/crates/zeph-orchestration/src/scheduler/tick/tests.rs b/crates/zeph-orchestration/src/scheduler/tick/tests.rs index 86c27f2c5..79811baf1 100644 --- a/crates/zeph-orchestration/src/scheduler/tick/tests.rs +++ b/crates/zeph-orchestration/src/scheduler/tick/tests.rs @@ -440,6 +440,7 @@ fn test_cascade_chain_threshold_preempts_recovery() { // output — the assertion below proves it never gets the chance to. scheduler.graph.tasks[2].recovery = Some(crate::graph::RecoveryAction { state_injection: Some("should never be applied".to_string()), + route_to: None, }); for (id, handle) in [(TaskId(0), "h0"), (TaskId(1), "h1"), (TaskId(2), "h2")] { @@ -480,9 +481,23 @@ fn test_cascade_chain_threshold_preempts_recovery() { "the recovery-configured node must NOT be recovered — cascade-abort preempts \ propagate_failure() (and thus try_recover()) entirely" ); - assert!( - scheduler.graph.tasks[2].result.is_none(), - "no synthetic recovery TaskResult should ever have been set" + assert_eq!( + scheduler.graph.tasks[2] + .result + .as_ref() + .map(|r| r.output.as_str()), + Some("boom"), + "result must hold the plain failure error, not a Mode-1 recovery substitution \ + (agent_id/agent_def would also be set to the recovery marker if try_recover had run)" + ); + assert_eq!( + scheduler.graph.tasks[2] + .result + .as_ref() + .and_then(|r| r.agent_def.as_deref()), + None, + "no synthetic recovery TaskResult (with its recovery marker agent_def) should ever \ + have been set" ); }