test(orchestration): add direct SC-4/SC-9 assertions for Mode-1 recovery

CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
+### Testing
+
+- **zeph-orchestration** / **zeph-core**: added direct test coverage for two spec-075 §7
+  success criteria that were only transitively covered (#6301, deferred from #6243/#6265).
+  `test_build_prompt_includes_mode1_recovered_state_injection` drives Mode-1 recovery
+  end-to-end (`propagate_failure` → `try_recover` → synthetic `TaskResult`) and asserts the
+  injected `state_injection` value actually reaches `build_task_prompt()` for a downstream
+  dependent, without leaking the internal `__recovery__` completion marker (SC-4).
+  `finalize_plan_execution_recovery_derived_task_counted_in_tasks_completed` asserts a
+  `__recovery__`-derived `Completed` task is counted by `finalize_plan_completed`'s generic,
+  status-based `tasks_completed` counter — with a sibling `Failed` task in the same graph
+  confirmed excluded, giving a real recovered-vs-failed contrast (SC-9). No production code
+  changed; both mechanisms already behaved correctly, this closes the direct-assertion gap
+  only.
+
 ### Docs
 
 - **Spec 072 §4**: updated to enumerate all four persistence surfaces that must strip `MessagePart::Image`

crates/zeph-core/src/agent/tests/compaction_e2e.rs

@@ -1528,6 +1528,71 @@ async fn finalize_plan_execution_completed_increments_aggregator_metrics() {
     );
 }
 
+/// SC-9 regression guard on the *absence* of recovery-specific special-casing -- this does
+/// NOT drive `try_recover()` end-to-end: that function is private to `zeph-orchestration` and
+/// not exercisable cross-crate without adding a production-only export purely for this test,
+/// which is out of scope for this issue. Instead it proves that a recovery-*shaped* task --
+/// shaped exactly as `try_recover()` in `zeph-orchestration`'s `dag.rs` synthesizes it
+/// (`status=Completed`, `result.agent_def=Some("__recovery__")`) -- flows through
+/// `finalize_plan_completed`'s (crates/zeph-core/src/agent/plan.rs) generic, purely
+/// status-derived counting path exactly like any other `Completed` task, with no
+/// `agent_def`-based exclusion. A sibling `Failed` task (no recovery) in the same graph is
+/// asserted NOT counted, making the recovered-vs-failed distinction a real contrast rather
+/// than a trivially-true single-task positive.
+#[cfg(feature = "scheduler")]
+#[tokio::test]
+async fn finalize_plan_execution_recovery_derived_task_counted_in_tasks_completed() {
+    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("recovery finalize test");
+    let mut recovered = TaskNode::new(0, "task-recovered", "desc");
+    recovered.status = TaskStatus::Completed;
+    recovered.result = Some(TaskResult {
+        output: "fallback output".into(),
+        artifacts: vec![],
+        duration_ms: 0,
+        agent_id: None,
+        agent_def: Some("__recovery__".to_string()),
+    });
+    graph.tasks.push(recovered);
+
+    let mut failed = TaskNode::new(1, "task-failed", "desc");
+    failed.status = TaskStatus::Failed;
+    failed.result = Some(TaskResult {
+        output: "error: no recovery configured".into(),
+        artifacts: vec![],
+        duration_ms: 0,
+        agent_id: None,
+        agent_def: None,
+    });
+    graph.tasks.push(failed);
+
+    graph.status = GraphStatus::Completed;
+
+    agent
+        .finalize_plan_execution(graph, GraphStatus::Completed)
+        .await
+        .unwrap();
+
+    let snapshot = rx.borrow().clone();
+    assert_eq!(
+        snapshot.orchestration.tasks_completed, 1,
+        "only the __recovery__-derived Completed task must be counted in tasks_completed via \
+         the generic status-based counting path; the sibling Failed task must not be counted; \
+         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-orchestration/src/scheduler/router.rs

@@ -96,8 +96,12 @@ impl DagScheduler {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::graph::{TaskResult, TaskStatus};
+    use crate::dag::propagate_failure;
+    use crate::graph::{
+        FailureStrategy, GraphStatus, RecoveryAction, TaskId, TaskResult, TaskStatus,
+    };
     use crate::scheduler::tests::*;
+    use crate::topology::build_rev_adj;
 
     #[test]
     fn test_build_prompt_no_deps() {
@@ -209,4 +213,48 @@ mod tests {
             "prompt must contain truncation notice. Prompt: {prompt}"
         );
     }
+
+    #[test]
+    fn test_build_prompt_includes_mode1_recovered_state_injection() {
+        // Drives the real Mode-1 recovery path (propagate_failure -> try_recover in dag.rs)
+        // end-to-end, rather than hand-constructing the recovered TaskNode, so this test
+        // breaks if try_recover()'s field mapping or completion marker ever changes.
+        let mut graph = graph_from_nodes(vec![make_node(0, &[]), make_node(1, &[0])]);
+        graph.status = GraphStatus::Running;
+        graph.tasks[0].status = TaskStatus::Failed;
+        graph.tasks[0].failure_strategy = Some(FailureStrategy::Abort);
+        graph.tasks[0].recovery = Some(RecoveryAction {
+            state_injection: Some("fallback output".to_string()),
+        });
+
+        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[0].status, TaskStatus::Completed);
+
+        // DagScheduler::new independently requires a freshly-`Created` graph; recovery leaves
+        // `graph.status` untouched (`Running`), so reset it here purely to satisfy that
+        // unrelated constructor invariant -- it does not affect the recovered task state above.
+        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("fallback output"),
+            "dependent's prompt must include the recovered dependency's synthetic output. Prompt: {prompt}"
+        );
+        assert!(
+            !prompt.contains("__recovery__"),
+            "agent_def marker must not leak into the prompt. Prompt: {prompt}"
+        );
+    }
 }