feat(orchestration): implement Mode-2 route_to reroute-to-alternate recovery

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<TaskId>` 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 `<recovery-source>`
+  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

crates/zeph-core/src/agent/plan.rs

@@ -910,9 +910,22 @@ impl<C: crate::channel::Channel> Agent<C> {
             .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<C: crate::channel::Channel> Agent<C> {
     ) -> 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()

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")]

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)];