diff --git a/CHANGELOG.md b/CHANGELOG.md index d686f8351..137768c43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -257,6 +257,39 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). pre-existing DB-error branch already produced — fail-open, matching `ShadowSentinel`'s documented defence-in-depth contract; the primary `PolicyGateExecutor`/`TrajectorySentinel` gates are unaffected and continue to run regardless (#6269). +- **Orchestration**: `PlanVerifier::verify_plan()` (whole-plan completeness verification, run + once after all DAG tasks complete) had no tool-call grounding — only the per-task `verify()` + path gained deterministic grounding against the real tool-execution trace in #6278/PR #6286, + which explicitly scoped whole-plan out with a `TODO(critic)` marker. A hallucinated + aggregated-output claim (e.g. "ran the full test suite across all tasks") could pass whole-plan + verification ungrounded. `verify_plan()` now grounds against the DAG-wide **union** of every + completed task's real `tool_trace`, rebuilt from transcripts at whole-plan-verify time by + reimplementing the same resolution logic `build_tool_trace_for_task` uses for the per-task path + (independent of whether per-task `Verify` ran for a given task — this is deliberate + defense-in-depth against a future dispatch mode that skips it). Trace + availability is all-or-nothing at the DAG level: the aggregate is `Some(union)` only if every + completed task's trace resolves; any one unavailable trace (e.g. a `RunInline` task, whose + in-loop trace is never persisted) degrades the whole aggregate to `None` and grounding fails + open, exactly reproducing prior ungrounded behavior (now with a `DEBUG` log noting the + degradation). Whole-plan grounding is strictly weaker than per-task grounding at catching a + single task's own hallucination (a claim grounds if *any* task in the plan really performed + it) — it is additive defense-in-depth, not a replacement. The trace-path resolution loop is + offloaded to `spawn_blocking` to avoid N synchronous transcript reads blocking the async + finalization path. `specs/009-orchestration/spec.md` updated with the new grounding contract, + Key Invariant, and AC-13..AC-16 (#6287). +- **Orchestration**: `execute_partial_replan_dag` (whole-plan replan execution) silently rejected + every replan attempt against a non-empty graph — i.e. always, in practice, since whole-plan + replan only ever runs after at least one task has completed. `replan_from_plan` assigns gap-task + IDs continuing the parent graph's numbering (so the final merge into `completed_graph.tasks` + stays globally unique), but the standalone partial `TaskGraph` built to execute those gap tasks + is validated by `dag::validate`, which requires 0-based positional IDs (`tasks[i].id == + TaskId(i)`) for any freestanding graph. The mismatch made `DagScheduler::new` reject the partial + graph outright (`"invalid graph: task at index 0 has id 1 (expected 0)"`), fail-opening to no + replan every time. This is a distinct, pre-existing defect independent of the whole-plan + grounding work above — unrelated to the matching/grounding contract, purely a task-ID-numbering + bug in replan execution — surfaced by the new end-to-end test added for #6287's review pass. + Gap-task IDs are now remapped to local 0-based IDs for the partial scheduler run and back to the + original global IDs on the way out. - **Worktree**: `--bare` silently skipped the entire worktree subsystem bootstrap (`WorktreeManager` construction, `probe_capabilities`) with no warning when `worktree.enabled = true` in the active config — the 6th confirmed instance of the `--bare` diff --git a/crates/zeph-core/src/agent/plan.rs b/crates/zeph-core/src/agent/plan.rs index ab8686d77..4dd05b0cf 100644 --- a/crates/zeph-core/src/agent/plan.rs +++ b/crates/zeph-core/src/agent/plan.rs @@ -521,6 +521,21 @@ impl Agent { return None; } + let trace_paths = self.resolve_whole_plan_trace_paths(scheduler.graph()); + let tool_trace = match trace_paths { + Some(paths) => Self::build_whole_plan_tool_trace(paths).await, + None => None, + }; + if tool_trace.is_none() { + tracing::debug!( + "whole-plan verify: tool-trace union unavailable — at least one completed \ + task's trace could not be resolved (e.g. a RunInline task, whose in-loop trace \ + is never persisted, or an unreadable/partial transcript); grounding skipped for \ + this whole-plan verify (fail-open, matches per-task behavior on an unavailable \ + trace)" + ); + } + let verify_provider = self .services .orchestration @@ -537,7 +552,7 @@ impl Agent { &self.services.orchestration.orchestration_config, ); let result = verifier - .verify_plan(&goal, &truncated_output) + .verify_plan(&goal, &truncated_output, tool_trace.as_deref()) .instrument(tracing::info_span!("core.plan.whole_plan_verify")) .await; @@ -577,15 +592,145 @@ impl Agent { self.execute_partial_replan_dag(gap_tasks, &goal).await } + /// Resolve the transcript path for every completed-with-result task in `graph`, as a + /// synchronous prerequisite step for [`Self::build_whole_plan_tool_trace`]. + /// + /// Takes a plain `&TaskGraph` (not `&DagScheduler`) so it is testable without spinning up + /// full scheduler machinery — the caller passes `scheduler.graph()`. + /// + /// Pure in-memory work (`agent_transcript_dir` is a `HashMap` lookup, no I/O) — deliberately + /// kept synchronous and `&self`-borrowing so it never needs to cross an `.await` point, + /// which would otherwise require `Agent: Sync` to keep the caller's future `Send` (spec + /// 009 § Whole-Plan Grounding, issue #6287). Returns `None` the moment any one task's path + /// cannot be resolved (missing `agent_id` — e.g. a `RunInline` task, whose in-loop trace is + /// never persisted — or no `SubAgentManager`/transcript dir), matching the all-or-nothing + /// availability contract. + fn resolve_whole_plan_trace_paths( + &self, + graph: &zeph_orchestration::TaskGraph, + ) -> Option> { + use zeph_orchestration::TaskStatus; + + let mut paths: Vec = Vec::new(); + for task in graph + .tasks + .iter() + .filter(|t| t.status == TaskStatus::Completed && t.result.is_some()) + { + let Some(agent_id) = task.result.as_ref().and_then(|r| r.agent_id.as_deref()) else { + tracing::debug!( + task_id = %task.id, + "whole-plan tool-trace union: task has no agent_id (RunInline dispatch or \ + missing), aggregate unavailable" + ); + return None; + }; + let Some(mgr) = self.services.orchestration.subagent_manager.as_ref() else { + tracing::debug!( + task_id = %task.id, + agent_id = %agent_id, + "whole-plan tool-trace union: no SubAgentManager configured, aggregate \ + unavailable" + ); + return None; + }; + let Some(dir) = mgr.agent_transcript_dir(agent_id) else { + tracing::debug!( + task_id = %task.id, + agent_id = %agent_id, + "whole-plan tool-trace union: transcript dir unresolvable, aggregate \ + unavailable" + ); + return None; + }; + paths.push(dir.join(format!("{agent_id}.jsonl"))); + } + Some(paths) + } + + /// Build the DAG-wide **union** of every completed task's real `tool_trace`, rebuilt from + /// transcripts at whole-plan-verify time (spec 009 § Whole-Plan Grounding, issue #6287). + /// + /// Availability is all-or-nothing: returns `Some(union)` iff **every** completed task with + /// a `result` resolves to a trace (`Some`, including `Some(vec![])` for a task that + /// genuinely ran zero tools); returns `None` the moment any one task's trace cannot be + /// resolved (missing `agent_id` — e.g. a `RunInline` task, whose in-loop trace is never + /// persisted — or an unreadable/partial transcript). An incomplete union is never returned: + /// a union missing part of the real record could false-positive an honest claim, mirroring + /// the per-task `None`-means-unavailable contract lifted to the DAG level. + /// + /// Transcript path resolution ([`Self::resolve_whole_plan_trace_paths`]) happens entirely + /// before this method's only `.await`, so `self` is never held across it — this is a free + /// associated function taking owned `paths` rather than `&self` specifically to keep the + /// generated future `Send` regardless of `Agent`'s `Sync` bound. The actual synchronous + /// file reads + JSON parsing (`TranscriptReader::load_strict`) are offloaded to + /// `spawn_blocking` since this loop can run N reads back-to-back with no yield point on the + /// async finalization path. + async fn build_whole_plan_tool_trace( + paths: Vec, + ) -> Option> { + if paths.is_empty() { + return Some(Vec::new()); + } + + let read_result = tokio::task::spawn_blocking(move || { + let mut union = Vec::new(); + for path in &paths { + match zeph_subagent::TranscriptReader::load_strict(path) { + Ok(messages) => { + union.extend(super::scheduler_loop::tool_trace_from_messages(&messages)); + } + Err(e) => return Err(format!("{}: {e}", path.display())), + } + } + Ok(union) + }) + .await; + + match read_result { + Ok(Ok(union)) => Some(union), + Ok(Err(e)) => { + tracing::debug!( + error = %e, + "whole-plan tool-trace union: transcript read failed, aggregate unavailable" + ); + None + } + Err(join_err) => { + tracing::warn!( + error = %join_err, + "whole-plan tool-trace union: spawn_blocking panicked, aggregate unavailable" + ); + None + } + } + } + pub(super) async fn execute_partial_replan_dag( &mut self, gap_tasks: Vec, goal: &str, ) -> Option> { - use zeph_orchestration::{DagScheduler, RuleBasedRouter, TaskStatus}; - + use zeph_orchestration::{DagScheduler, RuleBasedRouter, TaskId, TaskStatus}; + + // `replan_from_plan` assigns gap-task IDs continuing the parent graph's numbering + // (`next_id..`, so downstream merges into `completed_graph.tasks` stay globally + // unique), but `dag::validate` requires a freshly-constructed standalone `TaskGraph`'s + // task IDs to be 0-based and positional (`tasks[i].id == TaskId(i)`) — otherwise + // `DagScheduler::new` rejects the graph outright. Remap to local 0-based IDs for this + // scheduler run, then remap back to the original global IDs on the way out. Safe + // because whole-plan gap tasks are always independent roots with no `depends_on` + // cross-references to fix up (see `replan_from_plan`'s doc comment). + let base_id = gap_tasks.first().map_or(0, |t| t.id.0); let mut partial_graph = zeph_orchestration::TaskGraph::new(goal); - partial_graph.tasks = gap_tasks; + partial_graph.tasks = gap_tasks + .into_iter() + .enumerate() + .map(|(i, mut task)| { + task.id = TaskId(u32::try_from(i).unwrap_or(u32::MAX)); + task + }) + .collect(); let mut partial_config = self.services.orchestration.orchestration_config.clone(); partial_config.max_replans = 0; @@ -654,6 +799,10 @@ impl Agent { .tasks .into_iter() .filter(|t| t.status == TaskStatus::Completed) + .map(|mut t| { + t.id = TaskId(t.id.0 + base_id); + t + }) .collect(); if completed.is_empty() { @@ -1379,4 +1528,416 @@ mod tests { fn durable_enabled_when_both_flags_true() { assert!(durable_orchestration_enabled(Some(&enabled_cfg()))); } + + // --- #6287: whole-plan verifier grounding — DAG-wide tool-trace union (spec 009 § + // Whole-Plan Grounding) --- + + /// Spawns a "worker" sub-agent via [`crate::agent::Agent`]'s `AgentCommand::Background` + /// path — the same machinery production code and `scheduler_loop`'s + /// `spawn_worker_and_wait_completed` use — but reuses an already-configured + /// `SubAgentManager` across multiple calls so several agent ids stay simultaneously + /// resolvable via `agent_transcript_dir`. Needed to build a multi-task DAG-wide trace union + /// in these tests (the `scheduler_loop.rs` helper recreates the manager on every call, + /// which would evict the previous spawn's id). + async fn spawn_worker_and_wait_completed_shared( + agent: &mut crate::agent::Agent, + tmp: &std::path::Path, + ) -> String { + use zeph_subagent::def::{SkillFilter, SubAgentPermissions, ToolPolicy}; + use zeph_subagent::hooks::SubagentHooks; + use zeph_subagent::{AgentCommand, SubAgentDef, SubAgentManager, SubAgentState}; + + if agent.services.orchestration.subagent_manager.is_none() { + agent.services.orchestration.subagent_config.transcript_dir = Some(tmp.to_path_buf()); + agent + .services + .orchestration + .subagent_config + .transcript_enabled = true; + + let mut mgr = SubAgentManager::new(4); + mgr.definitions_mut().push(SubAgentDef { + name: "worker".into(), + description: "A worker bot".into(), + model: None, + tools: ToolPolicy::InheritAll, + disallowed_tools: vec![], + permissions: SubAgentPermissions { + max_turns: 1, + ..SubAgentPermissions::default() + }, + skills: SkillFilter::default(), + system_prompt: "You are a worker.".into(), + hooks: SubagentHooks::default(), + memory: None, + source: None, + file_path: None, + }); + agent.services.orchestration.subagent_manager = Some(mgr); + } + + let spawn_resp = agent + .handle_agent_command(AgentCommand::Background { + name: "worker".into(), + prompt: "do a task".into(), + }) + .await + .expect("Background spawn must return Some"); + let short_id = spawn_resp + .split("id: ") + .nth(1) + .expect("response must contain 'id: '") + .trim_end_matches(')') + .trim() + .to_string(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let mgr = agent + .services + .orchestration + .subagent_manager + .as_ref() + .unwrap(); + let statuses = mgr.statuses(); + let found = statuses.iter().find(|(id, _)| id.starts_with(&short_id)); + if let Some((id, status)) = found { + match status.state { + SubAgentState::Completed => break id.clone(), + SubAgentState::Failed => { + panic!("sub-agent Failed unexpectedly: {:?}", status.last_message); + } + _ => {} + } + } + assert!( + std::time::Instant::now() <= deadline, + "sub-agent did not complete within timeout" + ); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + } + + /// Appends a real `ToolUse`("bash", `{"command": "cargo test"}`)/`ToolResult` round to the + /// `.jsonl` transcript at `jsonl_path` (mirrors `scheduler_loop.rs`'s private helper of the + /// same shape — duplicated here since test helpers are module-private). + async fn append_tool_round(jsonl_path: &std::path::Path) { + let writer = zeph_subagent::TranscriptWriter::new(jsonl_path).unwrap(); + writer + .append( + 1000, + &zeph_llm::provider::Message::from_parts( + zeph_llm::provider::Role::Assistant, + vec![zeph_llm::provider::MessagePart::ToolUse { + id: "call-1".into(), + name: "bash".into(), + input: serde_json::json!({ "command": "cargo test" }), + }], + ), + ) + .await + .unwrap(); + writer + .append( + 1001, + &zeph_llm::provider::Message::from_parts( + zeph_llm::provider::Role::User, + vec![zeph_llm::provider::MessagePart::ToolResult { + tool_use_id: "call-1".into(), + content: "ok".into(), + is_error: false, + }], + ), + ) + .await + .unwrap(); + } + + fn completed_task_with_agent(id: u32, agent_id: &str) -> zeph_orchestration::TaskNode { + let mut task = zeph_orchestration::TaskNode::new(id, format!("t{id}"), "d"); + task.status = zeph_orchestration::TaskStatus::Completed; + task.result = Some(zeph_orchestration::TaskResult { + output: "done".to_string(), + artifacts: vec![], + duration_ms: 0, + agent_id: Some(agent_id.to_string()), + agent_def: None, + }); + task + } + + /// AC-13/M3 (integration-level): two completed spawn tasks with intact-but-empty + /// transcripts (zero tool calls) resolve to an aggregate `Some(vec![])`, NOT `None` — the + /// pitfall the critic flagged as most likely to be gotten wrong (spec 009 § Whole-Plan + /// Grounding). An empty-but-available union is the tightest detection case, so collapsing it + /// to `None` would silently fail the whole feature open for every pure-LLM-task plan. + #[tokio::test] + async fn whole_plan_trace_union_stays_some_empty_for_zero_tool_completed_tasks() { + use crate::agent::agent_tests::*; + + let tmp = tempfile::tempdir().unwrap(); + let provider = mock_provider(vec![ + "task completed successfully".into(), + "task completed successfully".into(), + ]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + + let id1 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + let id2 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + + let mut graph = zeph_orchestration::TaskGraph::new("goal"); + graph.tasks = vec![ + completed_task_with_agent(0, &id1), + completed_task_with_agent(1, &id2), + ]; + + let paths = agent + .resolve_whole_plan_trace_paths(&graph) + .expect("both tasks have resolvable agent_ids/transcript dirs"); + assert_eq!(paths.len(), 2); + + let union = Agent::::build_whole_plan_tool_trace(paths) + .await + .expect("intact-but-empty transcripts must resolve to Some(union), not None"); + assert!( + union.is_empty(), + "neither worker ran a tool, so the union must be Some(vec![]): {union:?}" + ); + } + + /// AC-13 (integration-level): a real tool call recorded in ONE task's transcript is present + /// in the DAG-wide union alongside the other (tool-free) task's contribution. + #[tokio::test] + async fn whole_plan_trace_union_combines_multiple_tasks() { + use crate::agent::agent_tests::*; + + let tmp = tempfile::tempdir().unwrap(); + let provider = mock_provider(vec![ + "task completed successfully".into(), + "task completed successfully".into(), + ]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + + let id1 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + let id2 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + + let dir = agent + .services + .orchestration + .subagent_manager + .as_ref() + .unwrap() + .agent_transcript_dir(&id1) + .unwrap() + .to_path_buf(); + append_tool_round(&dir.join(format!("{id1}.jsonl"))).await; + + let mut graph = zeph_orchestration::TaskGraph::new("goal"); + graph.tasks = vec![ + completed_task_with_agent(0, &id1), + completed_task_with_agent(1, &id2), + ]; + + let paths = agent.resolve_whole_plan_trace_paths(&graph).unwrap(); + let union = Agent::::build_whole_plan_tool_trace(paths) + .await + .expect("both transcripts are intact"); + assert!( + union + .iter() + .any(|t| t.tool == "bash" && t.args_summary.as_deref() == Some("cargo test")), + "union must include the tool call recorded on task 0's transcript: {union:?}" + ); + } + + /// AC-15 (integration-level): a `RunInline` task anywhere in the DAG (no `agent_id`, its + /// in-loop trace is never persisted) makes the WHOLE aggregate `None` — fail-open, + /// reproducing today's ungrounded behavior exactly — even though another task in the same + /// DAG has a perfectly resolvable transcript. + #[tokio::test] + async fn whole_plan_trace_union_none_when_any_task_is_run_inline() { + use crate::agent::agent_tests::*; + + let tmp = tempfile::tempdir().unwrap(); + let provider = mock_provider(vec!["task completed successfully".into()]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + + let id1 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + + let mut inline_task = zeph_orchestration::TaskNode::new(1, "t1", "d"); + inline_task.status = zeph_orchestration::TaskStatus::Completed; + inline_task.result = Some(zeph_orchestration::TaskResult { + output: "done inline".to_string(), + artifacts: vec![], + duration_ms: 0, + agent_id: None, // RunInline: no agent_id, trace is ephemeral/never persisted. + agent_def: None, + }); + + let mut graph = zeph_orchestration::TaskGraph::new("goal"); + graph.tasks = vec![completed_task_with_agent(0, &id1), inline_task]; + + let paths = agent.resolve_whole_plan_trace_paths(&graph); + assert!( + paths.is_none(), + "any RunInline task in the DAG must degrade the whole aggregate to None (fail-open)" + ); + } + + /// AC-16 (empty/single-task DAG): no completed tasks in the graph means the trace-path + /// resolution loop never runs, and the union is vacuously available (`Some(vec![])`) — the + /// caller (`run_whole_plan_verify`) never actually reaches this path in practice since + /// `truncated_output.is_empty()` already short-circuits first, but the helper itself must + /// not spuriously report unavailability for an empty task set. + #[tokio::test] + async fn whole_plan_trace_union_empty_graph_is_vacuously_available() { + use crate::agent::agent_tests::*; + + let agent = QuickTestAgent::minimal("noop").agent; + let graph = zeph_orchestration::TaskGraph::new("goal"); + + let paths = agent + .resolve_whole_plan_trace_paths(&graph) + .expect("no completed tasks means vacuously available, not unavailable"); + assert!(paths.is_empty()); + + let union = Agent::::build_whole_plan_tool_trace(paths) + .await + .expect("empty path list must resolve to Some(vec![])"); + assert!(union.is_empty()); + } + + /// End-to-end wiring test for `run_whole_plan_verify` itself (code review Important-1): + /// every sub-component (trace-union building, `verify_plan` grounding) is covered in + /// isolation above, but nothing previously called `run_whole_plan_verify` directly — this + /// is the project's documented "wire X into Y" defect class (a piece built and unit-tested + /// in isolation, but never proven reachable from its real call site). A hallucinated + /// whole-plan claim (unmatched against the completed task's real, empty trace) must flow + /// through grounding -> `should_replan` -> `replan_from_plan` -> `execute_partial_replan_dag` + /// and produce a non-`None`, non-empty result — proving the pipeline is genuinely wired, not + /// just each piece in isolation. + #[tokio::test] + async fn run_whole_plan_verify_end_to_end_hallucinated_claim_triggers_replan() { + use crate::agent::agent_tests::*; + use zeph_orchestration::{DagScheduler, GraphStatus, RuleBasedRouter}; + + let tmp = tempfile::tempdir().unwrap(); + let provider = mock_provider(vec![ + "task completed successfully".into(), + r#"{"complete": true, "gaps": [], "confidence": 0.5, + "claimed_executions": ["bash: cargo test"]}"# + .into(), + r#"{"tasks": [{"title": "fix gap", "description": "address the gap", + "agent_hint": null}]}"# + .into(), + "gap task completed successfully".into(), + ]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + agent.services.orchestration.orchestration_config = crate::config::OrchestrationConfig { + enabled: true, + verify_completeness: true, + ..crate::config::OrchestrationConfig::default() + }; + + let id1 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + + let mut graph = zeph_orchestration::TaskGraph::new("goal"); + graph.tasks = vec![completed_task_with_agent(0, &id1)]; + + let available_agents = agent + .services + .orchestration + .subagent_manager + .as_ref() + .map(|m| m.definitions().to_vec()) + .unwrap_or_default(); + let mut scheduler = DagScheduler::resume_from( + graph, + &agent.services.orchestration.orchestration_config, + Box::new(RuleBasedRouter), + available_agents, + None, + ) + .unwrap(); + + let result = agent + .run_whole_plan_verify(&mut scheduler, GraphStatus::Completed) + .await; + + let extra_tasks = result.expect( + "hallucinated whole-plan claim must trigger the full grounding -> replan -> \ + execute pipeline, not silently fail open", + ); + assert!( + !extra_tasks.is_empty(), + "the replan must actually produce and execute at least one gap task" + ); + } + + /// Companion to the hallucinated-claim wiring test above: an honest whole-plan claim + /// (matches nothing because there is nothing to match, and the LLM claims nothing) must + /// pass through `run_whole_plan_verify` end-to-end and correctly return `None` — no + /// spurious replan triggered by the real pipeline. + #[tokio::test] + async fn run_whole_plan_verify_end_to_end_honest_claim_returns_none() { + use crate::agent::agent_tests::*; + use zeph_orchestration::{DagScheduler, GraphStatus, RuleBasedRouter}; + + let tmp = tempfile::tempdir().unwrap(); + let provider = mock_provider(vec![ + "task completed successfully".into(), + r#"{"complete": true, "gaps": [], "confidence": 0.95}"#.into(), + ]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + agent.services.orchestration.orchestration_config = crate::config::OrchestrationConfig { + enabled: true, + verify_completeness: true, + ..crate::config::OrchestrationConfig::default() + }; + + let id1 = spawn_worker_and_wait_completed_shared(&mut agent, tmp.path()).await; + + let mut graph = zeph_orchestration::TaskGraph::new("goal"); + graph.tasks = vec![completed_task_with_agent(0, &id1)]; + + let available_agents = agent + .services + .orchestration + .subagent_manager + .as_ref() + .map(|m| m.definitions().to_vec()) + .unwrap_or_default(); + let mut scheduler = DagScheduler::resume_from( + graph, + &agent.services.orchestration.orchestration_config, + Box::new(RuleBasedRouter), + available_agents, + None, + ) + .unwrap(); + + let result = agent + .run_whole_plan_verify(&mut scheduler, GraphStatus::Completed) + .await; + + assert!( + result.is_none(), + "an honest, grounded whole-plan verdict must not trigger a replan: {result:?}" + ); + } } diff --git a/crates/zeph-core/src/agent/scheduler_loop.rs b/crates/zeph-core/src/agent/scheduler_loop.rs index df8457031..f5929949b 100644 --- a/crates/zeph-core/src/agent/scheduler_loop.rs +++ b/crates/zeph-core/src/agent/scheduler_loop.rs @@ -57,7 +57,7 @@ fn network_denied_for_task(task: Option<&zeph_orchestration::TaskNode>) -> bool /// sub-agent was canceled mid-call) is still included, defaulting `ok` to `true` — grounding's /// matching rule does not consult `ok` (existence, not outcome, is in scope), so this default /// cannot cause a false grounding match/mismatch. -fn tool_trace_from_messages( +pub(super) fn tool_trace_from_messages( messages: &[zeph_llm::provider::Message], ) -> Vec { use std::collections::HashMap; diff --git a/crates/zeph-orchestration/src/verifier.rs b/crates/zeph-orchestration/src/verifier.rs index 7b93f0d66..c92dbba2c 100644 --- a/crates/zeph-orchestration/src/verifier.rs +++ b/crates/zeph-orchestration/src/verifier.rs @@ -11,6 +11,7 @@ //! All LLM call failures are fail-open: `verify()` returns `complete = true` on //! error; `replan()` returns an empty `Vec`. Verification never blocks execution. +use std::fmt::Write as _; use std::time::Duration; use serde::{Deserialize, Serialize}; @@ -34,6 +35,13 @@ const MAX_GAP_DESCRIPTION_LEN: usize = 500; /// "this narration is long enough to plausibly describe a tool invocation." const NARRATIVE_HEAVY_OUTPUT_LEN_THRESHOLD: usize = 200; +/// Maximum number of trace entries rendered into the whole-plan verify prompt's advisory trace +/// section (spec 009 § Whole-Plan Grounding, "Prompt bound vs. grounding input"). The full +/// (uncapped) union is always passed to the deterministic [`ground`] call — this constant only +/// bounds the prompt's token footprint on large DAGs and can never cause a false-positive since +/// grounding itself never sees a truncated slice. +const MAX_WHOLE_PLAN_TRACE_ENTRIES: usize = 200; + /// A single real tool invocation recorded during a task's execution. /// /// Built from `MessagePart::ToolUse`/`ToolResult` pairs — either read from the sub-agent @@ -284,12 +292,29 @@ impl PlanVerifier

{ output: &str, vr: VerifyResponse, tool_trace: Option<&[ToolCallSummary]>, + ) -> VerificationResult { + self.apply_grounding(output, vr, tool_trace, &task.id.to_string()) + } + + /// Shared grounding core used by both the per-task (`verify`) and whole-plan (`verify_plan`) + /// paths: run the deterministic [`ground`] stage over a successful LLM response and project + /// the result into a `VerificationResult`, updating the override-accounting counters/logs + /// along the way (spec 009 § Verifier Tool-Call Grounding, Observability). `log_ctx` is the + /// identifier included in the log fields — the task id for per-task verification, or a + /// fixed marker (e.g. `"whole_plan"`) for whole-plan verification — so the two paths never + /// drift on how override-accounting is logged or counted (a security invariant). + fn apply_grounding( + &mut self, + output: &str, + vr: VerifyResponse, + tool_trace: Option<&[ToolCallSummary]>, + log_ctx: &str, ) -> VerificationResult { if vr.claimed_executions.is_empty() { warn!( - task_id = %task.id, - "no claimed_executions to ground for this task (either the LLM reported no tool \ - executions, or the field was missing/null and defaulted to empty)" + context = %log_ctx, + "no claimed_executions to ground (either the LLM reported no tool executions, \ + or the field was missing/null and defaulted to empty)" ); } if narrative_heavy_empty_claims(output, &vr.claimed_executions) { @@ -304,7 +329,7 @@ impl PlanVerifier

{ if llm_complete && !outcome.complete { self.grounding_overrides_total = self.grounding_overrides_total.saturating_add(1); warn!( - task_id = %task.id, + context = %log_ctx, unmatched_claims = ?outcome.unmatched_claims, matched = vr.claimed_executions.len() - outcome.unmatched_claims.len(), total_claims = vr.claimed_executions.len(), @@ -419,19 +444,38 @@ impl PlanVerifier

{ /// /// The aggregated output is expected to be pre-truncated by the caller to stay /// within the token budget before calling this method. + /// + /// `tool_trace` is the DAG-wide **union** of every completed task's real `tool_trace`, + /// rebuilt by the caller from transcripts (spec 009 § Whole-Plan Grounding, issue #6287): + /// `None` when unavailable (at least one completed task's trace could not be resolved — + /// grounding fails open for the whole plan), `Some(&[])` when every completed task + /// genuinely ran zero tools, `Some(&[…])` otherwise. The LLM's verdict is deterministically + /// cross-checked against it by the crate-internal `ground()` function before being + /// projected into the returned `VerificationResult` — identical mechanism to [`Self::verify`], + /// applied to the DAG-wide union instead of a single task's trace. #[tracing::instrument( name = "orchestration.verifier.verify_plan", - skip(self, goal, aggregated_output) + skip(self, goal, aggregated_output, tool_trace) )] - pub async fn verify_plan(&mut self, goal: &str, aggregated_output: &str) -> VerificationResult { - let messages = build_verify_plan_prompt(goal, aggregated_output, &self.sanitizer); + pub async fn verify_plan( + &mut self, + goal: &str, + aggregated_output: &str, + tool_trace: Option<&[ToolCallSummary]>, + ) -> VerificationResult { + let messages = + build_verify_plan_prompt(goal, aggregated_output, tool_trace, &self.sanitizer); - let result = tokio::time::timeout(self.timeout, self.provider.chat_typed(&messages)).await; + let result = tokio::time::timeout( + self.timeout, + self.provider.chat_typed::(&messages), + ) + .await; match result { Ok(Ok(vr)) => { self.consecutive_failures = 0; - vr + self.apply_grounding(aggregated_output, vr, tool_trace, "whole_plan") } Ok(Err(e)) => { self.consecutive_failures = self.consecutive_failures.saturating_add(1); @@ -798,6 +842,7 @@ fn render_tool_trace(tool_trace: Option<&[ToolCallSummary]>) -> String { fn build_verify_plan_prompt( goal: &str, aggregated_output: &str, + tool_trace: Option<&[ToolCallSummary]>, sanitizer: &Arc, ) -> Vec { let system = "You are a plan completion verifier. Evaluate whether the aggregated output \ @@ -808,17 +853,31 @@ fn build_verify_plan_prompt( \"gaps\": [\n\ {\"description\": \"what was missing\", \"severity\": \"critical|important|minor\"}\n\ ],\n\ - \"confidence\": 0.0-1.0\n\ + \"confidence\": 0.0-1.0,\n\ + \"claimed_executions\": [\": \", ...]\n\ }\n\n\ severity levels:\n\ - critical: essential goal requirement not addressed\n\ - important: partial coverage that affects goal quality\n\ - - minor: nice to have, does not affect core goal" + - minor: nice to have, does not affect core goal\n\n\ + claimed_executions: list every tool or command invocation the aggregated \ + output narrative claims occurred anywhere in the plan, one entry per \ + invocation, in the form \": \" (quote the command verbatim \ + from the narration). Leave empty if the output does not claim any tool \ + executions. This list is cross-checked against the actual tool-execution log \ + below — list every claim so a hallucinated completion (output narrates a \ + command that never really ran anywhere in the plan) can be detected. If the \ + output claims a specific command or tool was executed, cross-check it \ + yourself against that log too: a claim with no matching real execution \ + anywhere in the plan is always at least an important gap." .to_string(); let safe_output = sanitizer.sanitize_task_output(aggregated_output); + let trace_section = render_whole_plan_tool_trace(tool_trace); - let user = format!("Original goal: {goal}\n\nAggregated plan output:\n{safe_output}"); + let user = format!( + "Original goal: {goal}\n\nAggregated plan output:\n{safe_output}\n\n{trace_section}" + ); vec![ Message::from_legacy(Role::System, system), @@ -826,6 +885,46 @@ fn build_verify_plan_prompt( ] } +/// Render the DAG-wide real tool-execution trace union into the labeled prompt section the +/// whole-plan verify LLM is instructed to cross-check `claimed_executions` against. Unlike +/// [`render_tool_trace`], this caps the number of rendered entries at +/// [`MAX_WHOLE_PLAN_TRACE_ENTRIES`] to keep the prompt bounded on large DAGs — the full, +/// uncapped union still feeds the deterministic [`ground`] call separately (spec 009 § +/// Whole-Plan Grounding, "Prompt bound vs. grounding input"). +fn render_whole_plan_tool_trace(tool_trace: Option<&[ToolCallSummary]>) -> String { + match tool_trace { + None => { + "Actual tool executions across the plan: unavailable (could not be read)".to_string() + } + Some([]) => "Actual tool executions across the plan: none recorded".to_string(), + Some(entries) => { + let total = entries.len(); + let lines: Vec = entries + .iter() + .take(MAX_WHOLE_PLAN_TRACE_ENTRIES) + .map(|e| { + let args = e.args_summary.as_deref().unwrap_or("(args not captured)"); + let status = if e.ok { "" } else { " [failed]" }; + format!("- {}: {args}{status}", e.tool) + }) + .collect(); + let mut section = format!( + "Actual tool executions across the plan:\n{}", + lines.join("\n") + ); + if total > MAX_WHOLE_PLAN_TRACE_ENTRIES { + let _ = write!( + section, + "\n... ({} more entries omitted from this prompt; the full record is still \ + checked)", + total - MAX_WHOLE_PLAN_TRACE_ENTRIES + ); + } + section + } + } +} + fn build_replan_from_plan_prompt( goal: &str, gaps: &[&Gap], @@ -1253,7 +1352,7 @@ mod tests { let mut verifier = PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); let result = verifier - .verify_plan("write a web server", "here is the server code") + .verify_plan("write a web server", "here is the server code", None) .await; assert!(result.complete); assert!(result.gaps.is_empty()); @@ -1268,7 +1367,7 @@ mod tests { let mut verifier = PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); let result = verifier - .verify_plan("write a web server", "partial output") + .verify_plan("write a web server", "partial output", None) .await; assert!(!result.complete); assert_eq!(result.gaps.len(), 3); @@ -1282,7 +1381,7 @@ mod tests { }; let mut verifier = PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); - let result = verifier.verify_plan("goal", "output").await; + let result = verifier.verify_plan("goal", "output", None).await; assert!(result.complete); assert!(result.gaps.is_empty()); assert!(result.confidence.abs() < f64::EPSILON); @@ -1374,7 +1473,7 @@ mod tests { }; let mut verifier = PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); - let result = verifier.verify_plan("goal", "output").await; + let result = verifier.verify_plan("goal", "output", None).await; assert!(!result.complete); assert!((result.confidence - 0.6).abs() < 0.01); // The caller is responsible for gating on threshold; verify_plan just returns the result. @@ -1396,7 +1495,7 @@ mod tests { }; let mut verifier = PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); - let result = verifier.verify_plan("goal", "output").await; + let result = verifier.verify_plan("goal", "output", None).await; let threshold = 0.7_f64; let should_replan = !result.complete && result.confidence < threshold && !result.gaps.is_empty(); @@ -1472,7 +1571,7 @@ mod tests { #[tokio::test] async fn verify_plan_timeout_is_fail_open() { let mut verifier = slow_verifier(); - let result = verifier.verify_plan("goal", "output").await; + let result = verifier.verify_plan("goal", "output", None).await; assert!(result.complete, "timeout must be fail-open (complete=true)"); assert!(result.gaps.is_empty()); } @@ -1510,7 +1609,7 @@ mod tests { async fn verify_plan_timeout_increments_counter_and_crosses_threshold() { let mut verifier = slow_verifier(); for _ in 0..3 { - let _ = verifier.verify_plan("goal", "output").await; + let _ = verifier.verify_plan("goal", "output", None).await; } assert_eq!( verifier.consecutive_failures(), @@ -1880,4 +1979,180 @@ mod tests { assert!(result.complete); assert_eq!(verifier.grounding_overrides_total(), 0); } + + // --- #6287: whole-plan verifier grounding (spec 009 § Whole-Plan Grounding) --- + + fn hallucinated_verify_plan_json() -> String { + r#"{ + "complete": true, + "gaps": [], + "confidence": 0.9, + "claimed_executions": ["bash: cargo test"] + }"# + .to_string() + } + + /// AC-14 (pure `ground()` over a DAG-wide union): a union built from two different tasks' + /// traces still grounds a claim correctly — `ground()` has no per-task attribution, so this + /// is really the same `ground()` already covered elsewhere, exercised here specifically + /// over a multi-task union to document the whole-plan usage shape. + #[test] + fn ground_ac14_union_of_multiple_tasks_grounds_claim() { + let union = vec![ + tool_call("bash", Some("cargo build"), true), + tool_call("bash", Some("cargo test --all-features"), true), + ]; + let outcome = ground( + true, + vec![], + &["bash: cargo test".to_string()], + Some(&union), + ); + assert!( + outcome.complete, + "claim matches the second task's real call" + ); + assert!(outcome.unmatched_claims.is_empty()); + } + + /// M3 regression at the `verify_plan()` entry point: an aggregate that is `Some(&[])` + /// (every completed task genuinely ran zero tools — the tightest detection case, NOT the + /// same as an unavailable `None` aggregate) must still ground and catch a hallucinated + /// claim, ending in `complete: false`. + #[tokio::test] + async fn verify_plan_end_to_end_hallucinated_claim_on_empty_union_overrides_complete() { + let provider = MockProvider { + response: Ok(hallucinated_verify_plan_json()), + }; + let mut verifier = + PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); + let union: Vec = vec![]; + let result = verifier + .verify_plan( + "run the test suite", + "I ran cargo test across the whole plan and it passed", + Some(&union), + ) + .await; + + assert!(!result.complete, "hallucinated claim must be caught"); + assert!( + result + .gaps + .iter() + .any(|g| g.severity == GapSeverity::Critical) + ); + assert_eq!(verifier.grounding_overrides_total(), 1); + } + + /// Honest completion: the same claim matched against a real union entry (contributed by + /// some task in the DAG — not necessarily the one that narrated it) stays complete. + #[tokio::test] + async fn verify_plan_end_to_end_honest_claim_stays_complete() { + let provider = MockProvider { + response: Ok(hallucinated_verify_plan_json()), + }; + let mut verifier = + PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); + let union = vec![tool_call("bash", Some("cargo test --all-features"), true)]; + let result = verifier + .verify_plan( + "run the test suite", + "I ran cargo test across the whole plan and it passed", + Some(&union), + ) + .await; + + assert!(result.complete); + assert_eq!(verifier.grounding_overrides_total(), 0); + } + + /// AC-15 (verifier-side half): an unavailable aggregate (`None` — the caller's contract for + /// "at least one completed task's trace could not be resolved") never overrides, even + /// though the narration claims a tool call. The DAG-level trace-resolution behavior itself + /// (`RunInline` task → `None`) is exercised as an integration test in + /// `scheduler_loop.rs`. + #[tokio::test] + async fn verify_plan_end_to_end_none_aggregate_never_overrides() { + let provider = MockProvider { + response: Ok(hallucinated_verify_plan_json()), + }; + let mut verifier = + PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); + let result = verifier + .verify_plan( + "run the test suite", + "I ran cargo test across the whole plan and it passed", + None, + ) + .await; + + assert!(result.complete); + assert_eq!(verifier.grounding_overrides_total(), 0); + } + + /// Override-accounting counters must not drift between the per-task and whole-plan paths — + /// both go through the shared `apply_grounding` helper, so a whole-plan override increments + /// the exact same `grounding_overrides_total` counter as a per-task override. + #[tokio::test] + async fn verify_plan_override_increments_same_counter_as_per_task_path() { + let provider = MockProvider { + response: Ok(hallucinated_verify_json()), + }; + let mut verifier = + PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); + let task = TaskNode::new(0, "run tests", "run cargo test and report result"); + let trace: Vec = vec![]; + let _ = verifier + .verify(&task, "I ran cargo test and it passed", Some(&trace)) + .await; + assert_eq!(verifier.grounding_overrides_total(), 1); + + // A second override via verify_plan() on the SAME verifier instance must accumulate, + // not reset or diverge onto a separate counter. + let union: Vec = vec![]; + let _ = verifier + .verify_plan( + "goal", + "I ran cargo test across the whole plan", + Some(&union), + ) + .await; + assert_eq!(verifier.grounding_overrides_total(), 2); + } + + /// LLM serialization gate (`.claude/rules/branching.md`): a real round-trip against a live + /// Ollama model, exercising the new `build_verify_plan_prompt` schema (the + /// `claimed_executions` field + trace section added for #6287) end-to-end through + /// `chat_typed::`. Ignored by default (requires a local Ollama instance with + /// `qwen2.5:7b` pulled) — run manually with `cargo nextest run -p zeph-orchestration + /// --features llm-planning -- --ignored`. + #[tokio::test] + #[ignore = "requires a local Ollama instance with qwen2.5:7b"] + async fn verify_plan_live_ollama_round_trip_does_not_error() { + let provider = zeph_llm::ollama::OllamaProvider::new( + "http://localhost:11434", + "qwen2.5:7b".into(), + "nomic-embed-text-v2-moe".into(), + ); + let mut verifier = + PlanVerifier::new(provider, test_sanitizer(), &OrchestrationConfig::default()); + let union = vec![tool_call("bash", Some("cargo test --all-features"), true)]; + let result = verifier + .verify_plan( + "run the full test suite and report results", + "I ran cargo test across the whole plan and all tests passed", + Some(&union), + ) + .await; + // The call must complete without a 400/422 surfacing as a hard failure (fail-open on + // any LLM error still returns a well-formed VerificationResult, so a non-panicking + // return with a real HTTP round-trip already proves the schema round-trips). + println!( + "live verify_plan result: complete={} gaps={} confidence={}", + result.complete, + result.gaps.len(), + result.confidence + ); + } } diff --git a/specs/009-orchestration/spec.md b/specs/009-orchestration/spec.md index 74ad908b9..8adf005a0 100644 --- a/specs/009-orchestration/spec.md +++ b/specs/009-orchestration/spec.md @@ -375,6 +375,7 @@ VMAO (Verify-and-Modify Adaptive Orchestration) extends Plan Verification with a - Grounding MUST fail open (skip the override, pass the LLM's verdict through unmodified) when `tool_trace` is unavailable (`None`); an available-but-empty trace is not itself a gap — only an unmatched claim is. This is a narrower, grounding-specific fail-open, distinct from the whole-`verify()` fail-open on LLM error/timeout above - `ground()` MUST be pure — no I/O, no second LLM call, no randomness — and MUST NOT be implemented as a regex/substring scan of the raw narration; the claim set comes only from the LLM's structured `claimed_executions` - Grounding on the ensemble path MUST run as one `ground()` call over the **union** of `claimed_executions` across all responded members, after `merge()` — never inside `merge()`, never majority/intersection +- `verify_plan()` MUST run the same deterministic `ground()` stage over the **DAG-wide union** of every completed task's `tool_trace`; the aggregate MUST be `None` (fail open) if **any** completed-with-result task's trace is unavailable — never `Some(partial_union)`, which could false-positive an honest claim — see [[#Whole-Plan Grounding (issue #6287)]] --- @@ -547,30 +548,92 @@ to `merge()` itself: contract — i.e. today's pre-fix behavior (`complete: true`), no worse than before but not specially rescued by this feature. -### Whole-Plan Verification — Explicitly Out of Scope - -`verify_plan()` / `replan_from_plan()` (aggregated, whole-plan verification) remain ungrounded. -This scope-out is safe only under a precondition that has been **verified true**, not merely -assumed: - -**Precondition (verified):** every completed task's narrated output passes through the grounded -per-task `verify()` (`SchedulerAction::Verify`) before it can reach the whole-plan verifier via -`collect_and_truncate_task_outputs`. `handle_completed_outcome` (`tick/mod.rs:582-583`) emits -`SchedulerAction::Verify` unconditionally for every completed task whenever -`verify_completeness = true` (no per-task -skip, including when `max_replans = 0`); both the spawn and RunInline dispatch paths converge on -this function through the shared `TaskOutcome::Completed` enum. Whole-plan -`run_whole_plan_verify` is gated by the same `verify_completeness` flag. Therefore no dispatch -mode today lets a task reach whole-plan aggregation without a prior grounded per-task `Verify`, -and the whole-plan verifier can only *add* replan tasks — it cannot launder a hallucination into -acceptance. - -If any future dispatch mode changes this convergence, the grounding read for that mode MUST be -re-verified before this scope-out can be relied on again. - - +### Whole-Plan Grounding (issue #6287) + +`verify_plan()` grounds its aggregated-output verdict against the **union** of every completed +task's real `tool_trace`, applying the identical deterministic `ground()` stage the per-task +path uses. This is independent defense-in-depth: per-task `verify()` already grounds each task's +own narration against its own trace, so today's whole-plan grounding is *additive*, not the sole +guard. Its purpose is to close the structural gap that a future dispatch mode which lets a task +reach whole-plan aggregation *without* a prior grounded per-task `Verify` would otherwise open — +laundering a hallucinated claim that only surfaces once outputs are aggregated across the DAG. + +**Aggregation is transcript-derived, not per-task-verify-derived.** At `run_whole_plan_verify` +time the DAG-wide trace is rebuilt from source, per completed-with-result task, by reimplementing +the same resolution logic `build_tool_trace_for_task` uses for the per-task path (`TaskResult.agent_id` +→ `SubAgentManager::agent_transcript_dir()` → `TranscriptReader::load_strict`) — split across +`resolve_whole_plan_trace_paths` (synchronous path resolution) and `build_whole_plan_tool_trace` +(the actual reads, offloaded to `spawn_blocking`). `build_tool_trace_for_task` itself is +module-private to `scheduler_loop.rs` and is not called directly from the whole-plan path (a +sibling module); only its inner `tool_trace_from_messages` conversion (messages → +`Vec`) is actually shared, bumped to `pub(super)` for that purpose. Rebuilding +from the transcript — rather than reusing a value cached during per-task `Verify` — is deliberate: +it makes whole-plan grounding correct *even when per-task `Verify` was skipped* for some task, +which is exactly the future gap this closes. The reads are already-persisted transcripts; no +per-task LLM claim-extraction is re-run. + +**Trace availability is all-or-nothing, lifted to the DAG level.** The aggregate is +`Some(union)` only if **every** completed-with-result task resolves to `Some(trace)`; if **any** +one resolves to `None` (unavailable — unreadable/partial transcript, or a RunInline task whose +in-loop trace is ephemeral and has no transcript file), the whole aggregate degrades to `None` +and grounding is skipped entirely (fail-open), exactly reproducing today's ungrounded behavior. +This mirrors the per-task `None`-means-unavailable contract: a union missing part of the real +execution record could false-positive an honest claim, so an incomplete union must never drive an +override. Consequence and documented limitation: a DAG containing any RunInline task fails open at +whole-plan grounding (RunInline traces are not persisted); persisting them onto `TaskResult` to +cover that case is a deliberate future enhancement, not part of this MVP. Under the default +`RuleBasedRouter`, a DAG is either all-spawn or all-inline (it returns `None` — i.e. RunInline — +iff `available_agents` is empty, otherwise every task routes to a sub-agent), so in practice this +limitation manifests as an all-or-nothing feature toggle per deployment: a no-subagent-defs +deployment gets zero whole-plan grounding, silently, unless the caller logs the degradation (see +`run_whole_plan_verify`'s DEBUG log on aggregate unavailability). A **custom** `AgentRouter` +implementation, however, can freely mix spawn and RunInline dispatch within a single DAG — in that +case a single stray inline task disables whole-plan grounding for the *entire* plan, even though +every other task's transcript is perfectly readable. This is a real, not merely theoretical, +failure mode for any deployment using a router other than the default. + +**Union, not per-task attribution — and materially weaker detection than the per-task path.** +`ground()` checks each aggregated `claimed_executions` entry against *any* union member. Grounding +against the union (rather than the originating task's trace) is strictly more lenient — a claim +grounds if it matches any task's real call — which is the correct fail-open bias; task-boundary +attribution is intentionally not reconstructed from the concatenated output. This leniency has a +real cost: a per-task hallucination (task T's own narration claims a command T never ran) PASSES +whole-plan grounding if **any other task U** in the same DAG genuinely ran that command, even +though per-task `verify()` on T alone would have caught it. Because the sole reason this feature +exists is to defend against a *future dispatch mode that skips per-task `Verify`* — in which +whole-plan grounding becomes the **only** guard for that task — this is exactly the scenario where +whole-plan grounding runs at its *weakest*. Whole-plan grounding is therefore genuine +defense-in-depth only, layered on top of (never a replacement for) per-task grounding; it is not +sized to catch every hallucination on its own, only to close the specific structural gap described +above. + +**Prompt bound vs. grounding input.** The full union is passed to the deterministic `ground()` +call (which alone binds the verdict). The advisory tool-trace section rendered into the +`verify_plan` prompt is capped at a fixed entry count to keep the prompt bounded on large DAGs; +because `ground()` runs in Rust over the complete slice, capping the *prompt* rendering can never +introduce a false-positive (unlike dropping entries from the grounding input, which is forbidden). +`aggregated_output` remains pre-truncated by the caller as today. + +**Scope alignment.** Only `Completed` tasks with a `result` contribute both output and trace, so +partial/failed tasks are excluded symmetrically from both sides. Output truncation only *removes* +claims (never adds), and a union that is a superset of the referenced calls is always safe, so +truncation cannot desync the two sides. Whole-plan verify is single-provider `PlanVerifier` only +(the ensemble path is per-task); no ensemble union is constructed here. + +`replan_from_plan()` needs no change: it consumes the grounded `gaps` that `verify_plan()` now +emits, so a grounding-forced `Critical` gap flows through the existing `should_replan` gate into +whole-plan replan unchanged. As with the per-task path, `TaskNode.status` stays `Completed` — +grounding remains observational, adding replan tasks, never gating acceptance. + +**Note (execution contract, not part of the grounding contract itself):** `execute_partial_replan_dag` +runs `replan_from_plan()`'s gap tasks in a standalone `DagScheduler`/`TaskGraph`, which +`dag::validate` requires to carry 0-based positional task IDs (`tasks[i].id == TaskId(i)`) — but +`replan_from_plan()` assigns gap-task IDs continuing the *parent* graph's numbering so the final +merge into `completed_graph.tasks` stays globally unique. `execute_partial_replan_dag` reconciles +this by remapping gap-task IDs to local 0-based IDs for the partial scheduler run and back to the +original global IDs on the way out (fixed alongside #6287's end-to-end wiring test, which was the +first test to exercise this path with a non-empty parent graph and exposed a pre-existing +rejection here — see CHANGELOG). ### Config @@ -617,6 +680,25 @@ separate opt-in flag, no config/migration/wizard entries required. - **AC-12 (`args_summary: None` on a real same-tool entry):** real trace entry `ToolUse{name:"bash", args_summary:None}`, honest claim `"bash: "` ⇒ `complete:true`, no gap (an entry with no captured args is treated as inconclusive, not a mismatch). +- **AC-13 (whole-plan hallucination caught) — integration-level** (`zeph-core` + `scheduler_loop.rs`/`plan.rs`, extends the `build_tool_trace_for_task_parity_*` fixture family): + two spawn tasks with readable transcripts, DAG-wide union `Some(&[bash: cargo build])`, + `verify_plan` LLM returns `{complete:true, claimed_executions:["bash: cargo test"], gaps:[]}` ⇒ + final `VerificationResult{complete:false}` with a `Critical` gap naming the unmatched claim, + flowing into `replan_from_plan`. +- **AC-14 (whole-plan honest completion) — unit-level** (`zeph-orchestration` `verifier.rs`, pure + `ground()`/mock-provider `verify_plan()` test, no I/O): union `Some(&[bash: cargo test])`, + `verify_plan` claims `["bash: cargo test"]` ⇒ `complete:true`, no grounding gap. +- **AC-15 (whole-plan fail-open on any unavailable task trace) — integration-level** (`zeph-core`, + same fixture family as AC-13): two completed tasks, one resolves to `Some(trace)` and one to + `None` (e.g. a RunInline task, or an unreadable transcript) ⇒ aggregate is `None`, grounding + skipped, `verify_plan`'s LLM verdict passes through unmodified — no spurious whole-plan replan. + Reproduces today's ungrounded behavior exactly. +- **AC-16 (empty/single-task DAG) — integration-level** (`zeph-core`, trace-path resolution over a + hand-built `TaskGraph`): empty aggregated output ⇒ `run_whole_plan_verify` returns early + (unchanged); a graph with no completed tasks resolves to a vacuously-available `Some(vec![])` + aggregate, not `None`; a single completed task's trace forms a one-element union and grounds + normally. ---