diff --git a/crates/protocol/src/fleet.rs b/crates/protocol/src/fleet.rs index 3dba033887..114ccfb277 100644 --- a/crates/protocol/src/fleet.rs +++ b/crates/protocol/src/fleet.rs @@ -1126,6 +1126,12 @@ pub struct FleetReceipt { /// existed) deserializable. #[serde(default, skip_serializing_if = "Option::is_none")] pub resolved_route: Option, + /// Saved exec session id holding the worker's full transcript, when the + /// worker persisted one on completion (the exec stream's + /// `session_capture.saved_session_id`). Callers resolve the final + /// assistant reply via `GET /v1/sessions/{id}`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub saved_session_id: Option, /// Effective worker authority for this task (#3211). #[serde(default, skip_serializing_if = "Option::is_none")] pub effective_permissions: Option, @@ -1493,6 +1499,7 @@ mod tests { notes: None, }), resolved_route: None, + saved_session_id: None, effective_permissions: None, }; let json = serde_json::to_string(&receipt).unwrap(); @@ -1521,6 +1528,7 @@ mod tests { notes: Some("manual verification required".to_string()), }), resolved_route: None, + saved_session_id: None, effective_permissions: None, }; @@ -1689,6 +1697,7 @@ mod tests { model_source: Some("task.model".to_string()), source: "resolver".to_string(), }), + saved_session_id: None, effective_permissions: Some(FleetEffectivePermissions { write: true, network: true, diff --git a/crates/tui/src/exec_agent.rs b/crates/tui/src/exec_agent.rs index a30f5b1c8d..78e8d906c7 100644 --- a/crates/tui/src/exec_agent.rs +++ b/crates/tui/src/exec_agent.rs @@ -937,6 +937,7 @@ pub(crate) async fn run_exec_agent( if let Some(id) = saved_session_id.as_ref() { emit_exec_stream_event(&ExecStreamEvent::SessionCapture { content: exec_stream_session_ref(id), + saved_session_id: id.clone(), })?; } // Resolved output ceiling and its provenance, surfaced so a @@ -954,6 +955,12 @@ pub(crate) async fn run_exec_agent( &latest_model, ) .as_str(); + // The deliverable is the final assistant reply of the + // session, not the cumulative stream output: a + // multi-step turn streams pre-tool commentary first, + // and that commentary is not part of the answer. + let final_answer = exec_stream_final_answer_text(&latest_messages) + .unwrap_or_else(|| summary.output.trim().to_string()); emit_exec_stream_event(&ExecStreamEvent::Metadata { meta: Box::new(ExecStreamMeta { receipt_kind: "terminal", @@ -988,7 +995,10 @@ pub(crate) async fn run_exec_agent( &latest_messages, latest_system_prompt.as_ref(), ), - visible_final_answer_chars: summary.output.chars().count(), + visible_final_answer_chars: final_answer.chars().count(), + visible_final_answer_excerpt: exec_stream_final_answer_excerpt( + &final_answer, + ), resume_command: saved_session_id .as_deref() .map(exec_stream_resume_hint) diff --git a/crates/tui/src/fleet/alerts.rs b/crates/tui/src/fleet/alerts.rs index 220df76eb2..2efb219182 100644 --- a/crates/tui/src/fleet/alerts.rs +++ b/crates/tui/src/fleet/alerts.rs @@ -715,6 +715,7 @@ mod tests { notes: Some("regex scorer could not be compiled".to_string()), }), resolved_route: None, + saved_session_id: None, effective_permissions: None, }; diff --git a/crates/tui/src/fleet/control.rs b/crates/tui/src/fleet/control.rs index 9ebf4b1a6a..56886d3f52 100644 --- a/crates/tui/src/fleet/control.rs +++ b/crates/tui/src/fleet/control.rs @@ -753,6 +753,7 @@ mod tests { model_source: None, source: "resolver".to_string(), }), + saved_session_id: None, effective_permissions: None, } } diff --git a/crates/tui/src/fleet/executor.rs b/crates/tui/src/fleet/executor.rs index a508e8545e..8a5c396cfe 100644 --- a/crates/tui/src/fleet/executor.rs +++ b/crates/tui/src/fleet/executor.rs @@ -26,6 +26,7 @@ use codewhale_protocol::fleet::{FleetHostSpec, FleetTaskSpec, FleetWorkerEventPa use super::host::{FleetHostAdapter, FleetWorkerCommand}; use super::profile::AgentProfile; +use super::task_spec::FleetWorkerFinalAnswer; use super::worker_runtime::{ fleet_task_prompt, fleet_task_prompt_with_profiles, fleet_worker_launch_reasoning_effort, fleet_worker_launch_route, @@ -364,6 +365,12 @@ fn build_worker_exec_command_from_prompt( /// `{"type": "...", ...}` (see `ExecStreamEvent` in `main.rs`). pub fn map_exec_stream_line(line: &str) -> Option { let value: serde_json::Value = serde_json::from_str(line.trim()).ok()?; + map_exec_stream_value(&value) +} + +/// [`map_exec_stream_line`] on an already-parsed line, so the incremental +/// stream reader parses each frame exactly once. +fn map_exec_stream_value(value: &serde_json::Value) -> Option { match value.get("type").and_then(serde_json::Value::as_str)? { "tool_use" => { let tool = value @@ -431,23 +438,50 @@ enum ParsedTerminalRoute { Invalid, } +/// The `meta` object of a terminal exec receipt, or `None` for every other +/// stream line. +fn exec_terminal_meta( + value: &serde_json::Value, +) -> Option<&serde_json::Map> { + if value.get("type").and_then(serde_json::Value::as_str) != Some("metadata") { + return None; + } + let meta = value.get("meta").and_then(serde_json::Value::as_object)?; + (meta.get("receipt_kind").and_then(serde_json::Value::as_str) == Some("terminal")) + .then_some(meta) +} + +/// The worker's visible final answer from a terminal exec receipt: the +/// emitter already bounded and redacted `visible_final_answer_excerpt`, and +/// `visible_final_answer_chars` is the real pre-truncation length. +fn parse_exec_terminal_final_answer(value: &serde_json::Value) -> Option { + let meta = exec_terminal_meta(value)?; + let excerpt = meta + .get("visible_final_answer_excerpt") + .and_then(serde_json::Value::as_str)? + .trim(); + if excerpt.is_empty() { + return None; + } + let chars = meta + .get("visible_final_answer_chars") + .and_then(serde_json::Value::as_u64) + .and_then(|chars| usize::try_from(chars).ok()) + .unwrap_or_else(|| excerpt.chars().count()); + Some(FleetWorkerFinalAnswer { + excerpt: excerpt.to_string(), + chars, + }) +} + /// Parse one allowlisted, secret-free route identity from terminal exec /// metadata. Once a line declares itself as a terminal receipt, malformed /// route fields are distinct from ordinary non-terminal stream noise so a /// prior valid record cannot survive contradictory evidence. -fn parse_exec_terminal_route(line: &str) -> ParsedTerminalRoute { - let Ok(value) = serde_json::from_str::(line.trim()) else { +fn parse_exec_terminal_route(value: &serde_json::Value) -> ParsedTerminalRoute { + let Some(meta) = exec_terminal_meta(value) else { return ParsedTerminalRoute::NotTerminal; }; - if value.get("type").and_then(serde_json::Value::as_str) != Some("metadata") { - return ParsedTerminalRoute::NotTerminal; - } - let Some(meta) = value.get("meta").and_then(serde_json::Value::as_object) else { - return ParsedTerminalRoute::NotTerminal; - }; - if meta.get("receipt_kind").and_then(serde_json::Value::as_str) != Some("terminal") { - return ParsedTerminalRoute::NotTerminal; - } let route = (|| { let provider = meta.get("provider")?.as_str()?.trim(); @@ -481,7 +515,8 @@ fn parse_exec_terminal_route(line: &str) -> ParsedTerminalRoute { #[cfg(test)] fn map_exec_terminal_route(line: &str) -> Option { - match parse_exec_terminal_route(line) { + let value: serde_json::Value = serde_json::from_str(line).ok()?; + match parse_exec_terminal_route(&value) { ParsedTerminalRoute::Valid(route) => Some(route), ParsedTerminalRoute::NotTerminal | ParsedTerminalRoute::Invalid => None, } @@ -547,6 +582,48 @@ struct WorkerStream { terminal_route: TerminalRouteEvidence, /// When this worker process was started, for per-task wall-clock limits (R5). started_at: std::time::Instant, + /// The worker's visible final answer from its terminal exec receipt. This + /// is the task's deliverable for report/summary work that produces no + /// file artifact; surfaced as `Completed.summary` and in the receipt note + /// so receipts stop reporting "no verifiable output" for a worker that + /// wrote a full report. Bounded by the emitter, so nothing accumulates + /// here. + final_answer: Option, + /// Saved exec session id reported by the worker's `session_capture` event. + /// Resolving it via `GET /v1/sessions/{id}` yields the full transcript + /// (the worker's final assistant reply). + saved_session_id: Option, +} + +impl WorkerStream { + /// Observe one raw stream-json frame: record route evidence, the final + /// answer, and the saved-session id, and map it to a ledger payload. The + /// frame is parsed exactly once. + fn observe_line(&mut self, line: &[u8]) -> Option { + let Ok(line) = std::str::from_utf8(line) else { + // stream-json is a UTF-8 contract. Never accept a lossy-decoded route + // receipt: replacement characters could turn corrupt provider/model + // bytes into apparently valid provenance. + self.terminal_route.observe(ParsedTerminalRoute::Invalid); + return None; + }; + let value: serde_json::Value = serde_json::from_str(line.trim()).ok()?; + self.terminal_route + .observe(parse_exec_terminal_route(&value)); + if let Some(answer) = parse_exec_terminal_final_answer(&value) { + self.final_answer = Some(answer); + } + if value.get("type").and_then(serde_json::Value::as_str) == Some("session_capture") + && let Some(id) = value + .get("saved_session_id") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|id| !id.is_empty()) + { + self.saved_session_id = Some(id.to_string()); + } + map_exec_stream_value(&value) + } } #[derive(Debug, Clone, Default)] @@ -583,22 +660,6 @@ impl TerminalRouteEvidence { } } -fn observe_worker_stream_line( - terminal_route: &mut TerminalRouteEvidence, - line: &[u8], -) -> Option { - let Ok(line) = std::str::from_utf8(line) else { - // stream-json is a UTF-8 contract. Never accept a lossy-decoded route - // receipt: replacement characters could turn corrupt provider/model - // bytes into apparently valid provenance. - terminal_route.observe(ParsedTerminalRoute::Invalid); - return None; - }; - let line = line.trim_end(); - terminal_route.observe(parse_exec_terminal_route(line)); - map_exec_stream_line(line) -} - enum WorkerStreamHost { Local, Ssh(String), @@ -618,6 +679,13 @@ pub struct FleetWorkerTerminalEvent { /// Non-terminal payloads discovered by the mandatory post-exit drain. pub tail_payloads: Vec, pub reported_route: Option, + /// The worker's visible final answer from its terminal exec receipt, + /// whatever the outcome: a worker that fails after writing most of a + /// report keeps the text on its receipt. + pub final_answer: Option, + /// Saved exec session id reported by the worker's `session_capture` event, + /// when one was persisted on completion. + pub saved_session_id: Option, /// A real headless exec process must report its actual route. Callers use /// this bit to distinguish a missing/invalid report (fail closed) from /// pre-launch or simulated paths that only have declared route intent. @@ -710,6 +778,8 @@ impl FleetExecutor { terminal: false, terminal_route: TerminalRouteEvidence::default(), started_at: std::time::Instant::now(), + final_answer: None, + saved_session_id: None, }, ); Ok(handle) @@ -802,7 +872,7 @@ impl FleetExecutor { stream.pending.extend_from_slice(&buf); while let Some(idx) = stream.pending.iter().position(|byte| *byte == b'\n') { let line: Vec = stream.pending.drain(..=idx).collect(); - if let Some(event) = observe_worker_stream_line(&mut stream.terminal_route, &line) { + if let Some(event) = stream.observe_line(&line) { events.push(event); } } @@ -834,7 +904,7 @@ impl FleetExecutor { .get_mut(key) .and_then(|adapter| adapter.read_status(worker_id).ok())?, }; - let terminal = match status.state { + let mut terminal = match status.state { super::host::FleetHostWorkerState::Running | super::host::FleetHostWorkerState::Draining | super::host::FleetHostWorkerState::Unknown => return None, @@ -851,26 +921,31 @@ impl FleetExecutor { // between the scheduler's ordinary drain and this status poll cannot // be lost when the worker is forgotten. let mut tail_payloads = self.drain_events(worker_id); - if let Some(stream) = self.streams.get_mut(worker_id) { - let trailing_line = std::mem::take(&mut stream.pending); - if trailing_line.iter().any(|byte| !byte.is_ascii_whitespace()) - && let Some(payload) = - observe_worker_stream_line(&mut stream.terminal_route, &trailing_line) - { - tail_payloads.push(payload); - } + let stream = self.streams.get_mut(worker_id)?; + let trailing_line = std::mem::take(&mut stream.pending); + if trailing_line.iter().any(|byte| !byte.is_ascii_whitespace()) + && let Some(payload) = stream.observe_line(&trailing_line) + { + tail_payloads.push(payload); } - if let Some(stream) = self.streams.get_mut(worker_id) { - stream.terminal = true; + stream.terminal = true; + let final_answer = stream.final_answer.take(); + // Surface the visible final answer on a successful completion so + // report/summary tasks (no scorer, no file artifact) show their + // deliverable instead of "no verifiable output". `Failed` has no + // summary slot; the receipt keeps the text via `final_answer`. + if let (Some(answer), FleetWorkerEventPayload::Completed { summary, .. }) = + (final_answer.as_ref(), &mut terminal) + { + *summary = Some(answer.excerpt.clone()); } Some(FleetWorkerTerminalEvent { payload: terminal, exit_code: status.exit_code, tail_payloads, - reported_route: self - .streams - .get(worker_id) - .and_then(|stream| stream.terminal_route.reported_route().cloned()), + reported_route: stream.terminal_route.reported_route().cloned(), + final_answer, + saved_session_id: stream.saved_session_id.take(), requires_reported_route: true, }) } @@ -992,6 +1067,8 @@ mod tests { terminal: false, terminal_route: TerminalRouteEvidence::default(), started_at: std::time::Instant::now(), + final_answer: None, + saved_session_id: None, }, ); } @@ -1633,7 +1710,8 @@ mod tests { let observe = |lines: &[&str]| { let mut evidence = TerminalRouteEvidence::default(); for line in lines { - evidence.observe(parse_exec_terminal_route(line)); + let value: serde_json::Value = serde_json::from_str(line).unwrap(); + evidence.observe(parse_exec_terminal_route(&value)); } evidence.reported_route().cloned() }; @@ -1722,6 +1800,116 @@ mod tests { assert!(exec.all_terminal()); } + #[cfg(unix)] + fn run_worker_to_terminal(script: &str, worker_id: &str) -> FleetWorkerTerminalEvent { + let tmp = tempfile::TempDir::new().unwrap(); + let mut exec = FleetExecutor::new(tmp.path()); + let command = FleetWorkerCommand::new("sh", vec!["-c".to_string(), script.to_string()]); + exec.start_worker(worker_id, command, None).unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + exec.drain_events(worker_id); + if let Some(term) = exec.poll_terminal_with_status(worker_id) { + break term; + } + assert!( + std::time::Instant::now() < deadline, + "worker did not terminate in time" + ); + std::thread::sleep(std::time::Duration::from_millis(20)); + } + } + + #[cfg(unix)] + #[test] + fn completed_worker_surfaces_terminal_final_answer_as_summary() { + // Report/summary tasks produce their deliverable as the final + // assistant reply, not a file artifact. The exec side emits a bounded + // excerpt plus the real length on its terminal receipt; the executor + // reads that (never the streamed `content` deltas, which are the run + // thinking out loud) and attaches it to `Completed.summary` and the + // terminal event so a receipt can show the actual result. + let script = r#"printf '%s\n' '{"type":"content","content":"let me look first"}' '{"type":"metadata","meta":{"receipt_kind":"terminal","provider":"custom","provider_id":"remote-x","model":"worker-model","visible_final_answer_chars":9000,"visible_final_answer_excerpt":"the report..."}}' '{"type":"done"}'"#; + let terminal = run_worker_to_terminal(script, "w1"); + + match &terminal.payload { + FleetWorkerEventPayload::Completed { summary, .. } => { + assert_eq!(summary.as_deref(), Some("the report...")); + } + other => panic!("expected Completed, got {other:?}"), + } + assert_eq!( + terminal.final_answer, + Some(FleetWorkerFinalAnswer { + excerpt: "the report...".to_string(), + chars: 9000, + }) + ); + } + + #[cfg(unix)] + #[test] + fn failed_worker_keeps_terminal_final_answer_on_terminal_event() { + // A worker that fails after writing most of a report still reports + // its visible answer on the terminal receipt; the executor keeps it + // on the terminal event so the receipt can retain the text. + let script = r#"printf '%s\n' '{"type":"error","error":"boom"}' '{"type":"metadata","meta":{"receipt_kind":"terminal","provider":"custom","provider_id":"remote-x","model":"worker-model","visible_final_answer_chars":12,"visible_final_answer_excerpt":"partial text"}}'; exit 1"#; + let terminal = run_worker_to_terminal(script, "w-failed"); + + assert!( + matches!(terminal.payload, FleetWorkerEventPayload::Failed { .. }), + "{:?}", + terminal.payload + ); + assert_eq!( + terminal + .final_answer + .as_ref() + .map(|answer| answer.excerpt.as_str()), + Some("partial text") + ); + } + + #[cfg(unix)] + #[test] + fn completed_worker_surfaces_session_capture_id_for_full_transcript() { + // The worker persists its full transcript as a saved session and + // reports the recoverable id via `session_capture.saved_session_id`. + // The executor must capture that id on the terminal event so a caller + // can resolve the final assistant reply through `GET /v1/sessions/{id}`. + let script = r#"printf '%s\n' '{"type":"session_capture","content":"","saved_session_id":"session-abc"}' '{"type":"done"}'"#; + let terminal = run_worker_to_terminal(script, "w-session"); + + assert_eq!(terminal.saved_session_id.as_deref(), Some("session-abc")); + assert!(terminal.final_answer.is_none()); + } + + #[test] + fn terminal_final_answer_ignores_empty_and_nonterminal_receipts() { + let parse = + |line: &str| parse_exec_terminal_final_answer(&serde_json::from_str(line).unwrap()); + assert!(parse(r#"{"type":"content","content":"streamed"}"#).is_none()); + assert!( + parse(r#"{"type":"metadata","meta":{"receipt_kind":"turn","visible_final_answer_excerpt":"x"}}"#) + .is_none() + ); + assert!( + parse(r#"{"type":"metadata","meta":{"receipt_kind":"terminal","visible_final_answer_excerpt":" "}}"#) + .is_none() + ); + // A receipt without the count falls back to the excerpt length. + assert_eq!( + parse( + r#"{"type":"metadata","meta":{"receipt_kind":"terminal","visible_final_answer_excerpt":"héllo"}}"# + ), + Some(FleetWorkerFinalAnswer { + excerpt: "héllo".to_string(), + chars: 5, + }) + ); + } + #[cfg(unix)] #[test] fn terminal_poll_final_drains_route_metadata_and_tail_payloads() { diff --git a/crates/tui/src/fleet/ledger.rs b/crates/tui/src/fleet/ledger.rs index 66a785697c..2f54a6bed3 100644 --- a/crates/tui/src/fleet/ledger.rs +++ b/crates/tui/src/fleet/ledger.rs @@ -2763,6 +2763,7 @@ mod tests { notes: Some("verifier note contained super-secret".to_string()), }), resolved_route: None, + saved_session_id: None, effective_permissions: None, }) .unwrap(); @@ -3542,6 +3543,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }; assert!( @@ -3825,6 +3827,7 @@ mod tests { artifacts: vec![], score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }) .unwrap(); @@ -3940,6 +3943,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }, ) @@ -4186,6 +4190,7 @@ mod tests { artifacts: vec![], score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }; ledger.record_receipt(receipt.clone()).unwrap(); diff --git a/crates/tui/src/fleet/manager.rs b/crates/tui/src/fleet/manager.rs index ec3257002e..6445c130cf 100644 --- a/crates/tui/src/fleet/manager.rs +++ b/crates/tui/src/fleet/manager.rs @@ -1499,6 +1499,8 @@ impl FleetManager { exit_code: None, tail_payloads: Vec::new(), reported_route: None, + final_answer: None, + saved_session_id: None, requires_reported_route: false, }; let _ = self.record_task_outcome(&task, terminal)?; @@ -1547,6 +1549,8 @@ impl FleetManager { exit_code: None, tail_payloads: Vec::new(), reported_route: None, + final_answer: None, + saved_session_id: None, requires_reported_route: false, }; let _ = self.record_task_outcome(&task, terminal)?; @@ -1666,6 +1670,8 @@ impl FleetManager { exit_code, tail_payloads, reported_route, + final_answer, + saved_session_id, requires_reported_route, } = terminal; let (receipt_result, failure_kind, exit_code) = task_receipt_outcome(&payload, exit_code); @@ -1722,6 +1728,8 @@ impl FleetManager { attempt: task.entry.attempts, exit_code, artifacts, + final_answer, + saved_session_id, resolved_route, effective_permissions, }; @@ -1740,8 +1748,19 @@ impl FleetManager { result: receipt_result, failure_kind, artifacts: verification_input.artifacts, - score: None, + // No scorer ran, but a worker that failed after writing most + // of a report keeps its visible answer on the receipt rather + // than losing it with the failed attempt. + score: verification_input + .final_answer + .as_ref() + .map(|answer| FleetScore { + value: 0.0, + max: Some(1.0), + notes: Some(answer.receipt_note()), + }), resolved_route: verification_input.resolved_route, + saved_session_id: verification_input.saved_session_id, effective_permissions: verification_input.effective_permissions, } }; @@ -1801,6 +1820,7 @@ impl FleetManager { artifacts, score: None, resolved_route: self.resolve_task_route(&task.task_spec), + saved_session_id: None, effective_permissions: self.resolve_task_effective_permissions(task), }; let payload = FleetWorkerEventPayload::Cancelled { @@ -2295,11 +2315,19 @@ fn receipt_summary(receipt: &FleetReceipt) -> String { .and_then(|score| score.notes.as_deref()) .filter(|notes| !notes.trim().is_empty()) { - summary.push_str(&format!(" notes={notes}")); + // Notes may carry the worker's final-answer excerpt; the inspection + // summary is a one-line status surface. + summary.push_str(&format!( + " notes={}", + crate::utils::truncate_with_ellipsis(notes, RECEIPT_SUMMARY_NOTES_BYTES, "...") + )); } summary } +/// Byte bound on receipt notes inside the one-line inspection summary. +const RECEIPT_SUMMARY_NOTES_BYTES: usize = 240; + fn latest_error_for_worker(state: &FleetLedgerState, worker_id: &str) -> Option { state .latest_events @@ -3405,6 +3433,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }) .unwrap(); diff --git a/crates/tui/src/fleet/task_spec.rs b/crates/tui/src/fleet/task_spec.rs index 40104a93a8..7098ed29d4 100644 --- a/crates/tui/src/fleet/task_spec.rs +++ b/crates/tui/src/fleet/task_spec.rs @@ -77,6 +77,28 @@ impl FleetTaskSpecFile { } } +/// The worker's visible final answer as carried by the terminal exec +/// `metadata` receipt: `excerpt` is already bounded and secret-redacted by the +/// emitter (`visible_final_answer_excerpt`), `chars` is the real +/// pre-truncation length (`visible_final_answer_chars`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FleetWorkerFinalAnswer { + pub excerpt: String, + pub chars: usize, +} + +impl FleetWorkerFinalAnswer { + /// The receipt note for a task whose only deliverable is its answer text. + /// The excerpt is used verbatim: it was bounded and redacted once at the + /// emitter, and the ledger redacts receipt notes again on write. + pub fn receipt_note(&self) -> String { + format!( + "worker produced {} characters of deliverable: {}", + self.chars, self.excerpt + ) + } +} + #[derive(Debug, Clone)] pub struct FleetTaskVerificationInput { pub run_id: FleetRunId, @@ -86,6 +108,13 @@ pub struct FleetTaskVerificationInput { pub attempt: u32, pub exit_code: Option, pub artifacts: Vec, + /// The worker's visible final answer, as reported by its terminal exec + /// receipt. Report/summary tasks with no scorer and no file artifact + /// surface this as their deliverable instead of "no verifiable output". + pub final_answer: Option, + /// Saved exec session id holding the worker's full transcript, when the + /// worker persisted one on completion. + pub saved_session_id: Option, /// Resolved-route snapshot to persist on the receipt (#3154). pub resolved_route: Option, /// Effective worker authority snapshot to persist on the receipt (#3211). @@ -330,10 +359,20 @@ pub fn verify_task_result( "manual scorer configured", "manual verification is required to finalize this receipt", ), - None if !has_verifiable_artifact(input) => partial( - "no scorer configured and no verifiable artifacts recorded", - "worker exited successfully but produced no verifiable output", - ), + None if !has_verifiable_artifact(input) => match input + .final_answer + .as_ref() + .filter(|answer| !answer.excerpt.trim().is_empty()) + { + Some(answer) => partial( + "no scorer configured; worker produced a summary deliverable", + answer.receipt_note(), + ), + None => partial( + "no scorer configured and no verifiable artifacts recorded", + "worker exited successfully but produced no verifiable output", + ), + }, None => partial( "no scorer configured", "task has artifacts but no deterministic scorer", @@ -392,6 +431,7 @@ pub fn prepare_verification_receipt( artifacts, score: Some(verification.score), resolved_route: input.resolved_route.clone(), + saved_session_id: input.saved_session_id.clone(), effective_permissions: input.effective_permissions.clone(), }; Ok(receipt) @@ -978,6 +1018,8 @@ mod tests { attempt: 1, exit_code: Some(0), artifacts: vec![], + final_answer: None, + saved_session_id: None, resolved_route: None, effective_permissions: None, }; @@ -1065,6 +1107,40 @@ mod tests { ); } + #[test] + fn unscored_worker_surfaces_summary_deliverable_instead_of_no_output() { + let tmp = TempDir::new().unwrap(); + let input = FleetTaskVerificationInput { + run_id: FleetRunId::from("run-1"), + task_id: "task-a".to_string(), + worker_id: "worker-1".to_string(), + attempt: 1, + exit_code: Some(0), + artifacts: vec![], + final_answer: Some(FleetWorkerFinalAnswer { + excerpt: "The Changelog review is complete".to_string(), + chars: 32, + }), + saved_session_id: None, + resolved_route: None, + effective_permissions: None, + }; + let verification = verify_task_result(tmp.path(), &task("unscored", None), &input); + assert_eq!(verification.result, FleetTaskResult::Partial); + let notes = verification + .score + .notes + .as_deref() + .unwrap_or_default() + .to_string(); + assert!( + notes.contains("worker produced 32 characters of deliverable"), + "unexpected notes: {notes}" + ); + assert!(notes.contains("Changelog review is complete")); + assert!(!notes.contains("no verifiable output")); + } + #[test] fn fleet_task_spec_receipt_records_artifacts_scores_and_failure_kind() { let tmp = TempDir::new().unwrap(); @@ -1087,6 +1163,8 @@ mod tests { attempt: 3, exit_code: Some(1), artifacts: vec![log], + final_answer: None, + saved_session_id: None, resolved_route: None, effective_permissions: Some(FleetEffectivePermissions { write: false, @@ -1146,6 +1224,8 @@ mod tests { attempt: 1, exit_code: Some(1), artifacts: Vec::new(), + final_answer: None, + saved_session_id: None, resolved_route: None, effective_permissions: None, }; diff --git a/crates/tui/src/lib.rs b/crates/tui/src/lib.rs index 7c0b603fc7..7d51f8057d 100644 --- a/crates/tui/src/lib.rs +++ b/crates/tui/src/lib.rs @@ -11378,7 +11378,13 @@ struct ExecStreamMeta { #[serde(skip_serializing_if = "Option::is_none")] tool_catalog_sha256: Option, input_analysis: ExecStreamInputAnalysis, + /// Real character count of the visible final answer, before any bound. visible_final_answer_chars: usize, + /// Bounded, secret-redacted excerpt of the visible final answer (see + /// [`exec_stream_final_answer_excerpt`]). Omitted when the run produced + /// no visible answer. + #[serde(skip_serializing_if = "String::is_empty")] + visible_final_answer_excerpt: String, session_id: String, resume_command: String, workspace: String, @@ -11479,7 +11485,17 @@ enum ExecStreamEvent { event: serde_json::Value, }, #[serde(rename = "session_capture")] - SessionCapture { content: String }, + SessionCapture { + /// Redacted fingerprint for logs/forensics, the same value the + /// terminal `metadata.session_id` carries; never the recoverable id. + content: String, + /// The real saved-session id a caller can resolve via + /// `GET /v1/sessions/{id}` to read the worker's full transcript. This + /// is the only place the exec stream carries the raw id: `metadata` + /// stays fingerprint-only so a captured terminal receipt is safe to + /// log on its own. + saved_session_id: String, + }, #[serde(rename = "service_released")] #[cfg(unix)] ServiceReleased { @@ -11787,6 +11803,7 @@ async fn run_workflow_tool_command_inner( tool_catalog_sha256: None, input_analysis: ExecStreamInputAnalysis::default(), visible_final_answer_chars: result.content.chars().count(), + visible_final_answer_excerpt: exec_stream_final_answer_excerpt(&result.content), session_id: String::new(), resume_command: String::new(), workspace: workspace.display().to_string(), @@ -12201,11 +12218,61 @@ fn exec_stream_session_ref(session_id: &str) -> String { crate::utils::redacted_identifier_for_log(session_id) } +/// Resume hint for the terminal `metadata` receipt. `metadata` carries only +/// the session fingerprint, so the hint names the `session_capture` field +/// that holds the recoverable id instead of pretending to redact one. fn exec_stream_resume_hint(session_id: &str) -> String { if session_id.trim().is_empty() { String::new() } else { - "codewhale exec --resume ".to_string() + "codewhale exec --resume ".to_string() + } +} + +/// Character bound for `metadata.visible_final_answer_excerpt`. The excerpt +/// is a status surface (fleet receipts, event labels, runtime API payloads), +/// not the transcript: the full answer lives in the saved session and the +/// worker's stream-json log, and `visible_final_answer_chars` carries the real +/// length so a consumer can tell a bounded excerpt from a short answer. +const EXEC_STREAM_FINAL_ANSWER_EXCERPT_CHARS: usize = 4_000; + +/// The final visible assistant reply for the terminal receipt: the text +/// blocks of the last assistant-like message in the persisted session. +/// `ExecSummary::output` accumulates every stream delta of the run, +/// including pre-tool commentary from earlier steps of a multi-step +/// turn, so the cumulative output is only a fallback when the session +/// carries no assistant text at all. +fn exec_stream_final_answer_text(messages: &[Message]) -> Option { + let text = messages + .iter() + .rev() + .find(|message| message.role.is_assistant_like())? + .content + .iter() + .filter_map(|block| match block { + ContentBlock::Text { text, .. } => Some(text.as_str()), + _ => None, + }) + .collect::>() + .join("\n") + .trim() + .to_string(); + (!text.is_empty()).then_some(text) +} + +/// Bound and secret-redact the visible final answer once, at the emitter, so +/// every downstream consumer reads the same excerpt. +fn exec_stream_final_answer_excerpt(output: &str) -> String { + let redacted = codewhale_config::persistence::redact_secrets(output.trim()); + let mut chars = redacted.chars(); + let excerpt: String = chars + .by_ref() + .take(EXEC_STREAM_FINAL_ANSWER_EXCERPT_CHARS) + .collect(); + if chars.next().is_some() { + format!("{excerpt}...") + } else { + excerpt } } @@ -17044,6 +17111,7 @@ api_key = "test-only-key" ( ExecStreamEvent::SessionCapture { content: "x".to_string(), + saved_session_id: "session-x".to_string(), }, "session_capture", ), @@ -17179,6 +17247,7 @@ api_key = "test-only-key" tool_catalog_sha256: Some("sha256:tools".to_string()), input_analysis: ExecStreamInputAnalysis::default(), visible_final_answer_chars: 17, + visible_final_answer_excerpt: "the visible reply".to_string(), session_id: exec_stream_session_ref(raw_session_id), resume_command: exec_stream_resume_hint(raw_session_id), workspace: "/tmp/work".to_string(), @@ -17204,21 +17273,113 @@ api_key = "test-only-key" ); assert_eq!( parsed["meta"]["resume_command"], - "codewhale exec --resume " + "codewhale exec --resume " ); assert_eq!(parsed["meta"]["workspace"], "/tmp/work"); assert_eq!(parsed["meta"]["message_count"], 4); assert_eq!(parsed["meta"]["visible_final_answer_chars"], 17); + assert_eq!( + parsed["meta"]["visible_final_answer_excerpt"], + "the visible reply" + ); + // Contract (#5946): the raw saved-session id is carried by exactly one + // field, `session_capture.saved_session_id`. The `metadata` receipt + // above stays fingerprint-only, and the capture's own `content` keeps + // the same fingerprint so both surfaces can be correlated in a log. let capture = ExecStreamEvent::SessionCapture { content: exec_stream_session_ref(raw_session_id), + saved_session_id: raw_session_id.to_string(), }; let capture_json = serde_json::to_string(&capture).expect("serializes"); - assert!(!capture_json.contains(raw_session_id)); let parsed_capture: serde_json::Value = serde_json::from_str(&capture_json).expect("valid json"); assert_eq!(parsed_capture["type"], "session_capture"); + assert_eq!(parsed_capture["content"], parsed["meta"]["session_id"]); assert_ne!(parsed_capture["content"], raw_session_id); + assert_eq!(parsed_capture["saved_session_id"], raw_session_id); + assert!(parsed_capture.get("session_id").is_none(), "{capture_json}"); + } + + #[test] + fn exec_stream_final_answer_excerpt_is_bounded_and_redacted() { + assert_eq!( + exec_stream_final_answer_excerpt(" short reply \n"), + "short reply" + ); + let long = "x".repeat(EXEC_STREAM_FINAL_ANSWER_EXCERPT_CHARS + 5); + let excerpt = exec_stream_final_answer_excerpt(&long); + assert_eq!( + excerpt.chars().count(), + EXEC_STREAM_FINAL_ANSWER_EXCERPT_CHARS + 3 + ); + assert!(excerpt.ends_with("...")); + let leaked = exec_stream_final_answer_excerpt("token: sk-ant-must-not-leak-1234567890"); + assert!(!leaked.contains("sk-ant-must-not-leak"), "{leaked}"); + } + + #[test] + fn exec_stream_final_answer_text_is_the_last_assistant_reply() { + // Multi-step turn: pre-tool commentary, a tool result, then a + // distinct final answer. The receipt must carry only the final + // reply, not the cumulative stream output. + let messages = vec![ + Message { + role: Role::User, + content: vec![ContentBlock::Text { + text: "write the report".to_string(), + cache_control: None, + }], + }, + Message { + role: Role::Assistant, + content: vec![ContentBlock::Text { + text: "let me check the workspace first".to_string(), + cache_control: None, + }], + }, + Message { + role: Role::User, + content: vec![ContentBlock::ToolResult { + tool_use_id: "call-1".to_string(), + content: "listed files".to_string(), + is_error: Some(false), + content_blocks: None, + }], + }, + Message { + role: Role::Assistant, + content: vec![ + ContentBlock::thinking("final reasoning"), + ContentBlock::Text { + text: "the final report".to_string(), + cache_control: None, + }, + ], + }, + ]; + assert_eq!( + exec_stream_final_answer_text(&messages), + Some("the final report".to_string()) + ); + } + + #[test] + fn exec_stream_final_answer_text_requires_assistant_text() { + assert_eq!(exec_stream_final_answer_text(&[]), None); + let user_only = vec![Message { + role: Role::User, + content: vec![ContentBlock::Text { + text: "prompt".to_string(), + cache_control: None, + }], + }]; + assert_eq!(exec_stream_final_answer_text(&user_only), None); + let textless_assistant = vec![Message { + role: Role::Assistant, + content: vec![ContentBlock::thinking("reasoning only")], + }]; + assert_eq!(exec_stream_final_answer_text(&textless_assistant), None); } #[test] diff --git a/crates/tui/src/runtime_api.rs b/crates/tui/src/runtime_api.rs index aae9155982..15c6fe31a0 100644 --- a/crates/tui/src/runtime_api.rs +++ b/crates/tui/src/runtime_api.rs @@ -2688,6 +2688,7 @@ fn fleet_receipt_json(receipt: &codewhale_protocol::fleet::FleetReceipt) -> Valu "retry_eligible": retry_eligible, "score": score_json, "artifacts": receipt.artifacts.iter().map(fleet_artifact_json).collect::>(), + "saved_session_id": receipt.saved_session_id.clone(), "evidence_available": evidence_available, }) } @@ -2738,6 +2739,9 @@ fn artifact_kind_label(kind: &FleetArtifactKind) -> String { } } +/// Bound on the `Completed.summary` excerpt inside a lifecycle event label. +const FLEET_EVENT_LABEL_SUMMARY_CHARS: usize = 160; + fn fleet_event_label(payload: &FleetWorkerEventPayload) -> String { match payload { FleetWorkerEventPayload::Queued => "queued".to_string(), @@ -2768,7 +2772,15 @@ fn fleet_event_label(payload: &FleetWorkerEventPayload) -> String { FleetWorkerEventPayload::Artifact(artifact) => { format!("artifact kind={}", artifact_kind_label(&artifact.kind)) } - FleetWorkerEventPayload::Completed { exit_code, summary } => match (exit_code, summary) { + // `summary` may carry the worker's bounded final-answer excerpt (up + // to a few thousand chars); the label is a one-line status surface, + // so it gets a short excerpt while `payload` keeps the full text. + FleetWorkerEventPayload::Completed { exit_code, summary } => match ( + exit_code, + summary + .as_deref() + .map(|summary| truncate_text(summary, FLEET_EVENT_LABEL_SUMMARY_CHARS)), + ) { (Some(code), Some(summary)) => format!("completed exit_code={code} {summary}"), (Some(code), None) => format!("completed exit_code={code}"), (None, Some(summary)) => format!("completed {summary}"), diff --git a/crates/tui/src/runtime_api/tests.rs b/crates/tui/src/runtime_api/tests.rs index 9f9b3f26b2..7822d0124d 100644 --- a/crates/tui/src/runtime_api/tests.rs +++ b/crates/tui/src/runtime_api/tests.rs @@ -9567,6 +9567,7 @@ fn fleet_receipt_json_pass_result_has_no_failure_fields() { artifacts: Vec::new(), score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9600,6 +9601,7 @@ fn fleet_receipt_json_verifier_failure_is_not_retry_eligible() { artifacts: Vec::new(), score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9631,6 +9633,7 @@ fn fleet_receipt_json_transport_failure_is_retry_eligible() { artifacts: Vec::new(), score: None, resolved_route: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9666,6 +9669,7 @@ fn fleet_receipt_json_receipt_artifact_sets_evidence_available() { notes: Some("all checks pass".to_string()), }), resolved_route: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9742,6 +9746,8 @@ async fn fleet_receipt_api_list_and_get_round_trip() -> Result<()> { attempt: 1, exit_code: Some(0), artifacts: Vec::new(), + final_answer: None, + saved_session_id: None, resolved_route: None, effective_permissions: None, }; diff --git a/docs/AGENT_RUNTIME.md b/docs/AGENT_RUNTIME.md index 753e9ffc4c..064d5a01e3 100644 --- a/docs/AGENT_RUNTIME.md +++ b/docs/AGENT_RUNTIME.md @@ -153,6 +153,35 @@ run/phase/task/gate receipt while a Workflow is in flight and is retained as a typed `WorkflowEvent` in the Runtime execution ledger; the enclosing Runtime worker still owns the terminal `done` or `error`. One vocabulary, two surfaces. +`session_capture` is emitted once, when the exec run persisted its transcript +as a saved session, and carries the recoverable id in exactly one place: + +```json +{"type": "session_capture", "schema": "codewhale.exec-stream", "schema_version": 1, + "content": "", "saved_session_id": "01J…"} +``` + +- `saved_session_id` is the raw saved-session id. The Runtime executor + captures it onto the task's `FleetReceipt.saved_session_id` (also exposed by + the runtime API's receipt payload), so a client can resolve the worker's full + final reply via `GET /v1/sessions/{id}` instead of re-reading the worker log. +- `content` is the same redacted fingerprint the terminal `metadata.session_id` + carries, so a captured `metadata` receipt stays safe to log on its own and + the two events can still be correlated. `metadata.resume_command` therefore + names this field (`codewhale exec --resume `) + rather than carrying the id itself. + +The terminal `metadata` receipt also carries the worker's visible final answer: +`visible_final_answer_chars` is the real character count of the final +assistant reply, and `visible_final_answer_excerpt` is a bounded (4,000 +characters, `...` when cut), secret-redacted excerpt of it, omitted when the +run produced no visible answer. The Runtime executor reads the excerpt from +this receipt — never from the streamed `content` deltas, which are the run +thinking out loud — and attaches it to `Completed.summary` and, for a task +with no scorer and no file artifact, to the receipt notes as the task's +deliverable. Lifecycle event labels and worker inspection summaries show a +short excerpt; the event `payload` and the receipt keep the full excerpt. + `turn_usage` is the per-model-call usage receipt, emitted once per model request (turn-step) when the provider reported usage for that call: diff --git a/docs/zh_hans/AGENT_RUNTIME.md b/docs/zh_hans/AGENT_RUNTIME.md index 05a24d24c4..7e60dfb6c5 100644 --- a/docs/zh_hans/AGENT_RUNTIME.md +++ b/docs/zh_hans/AGENT_RUNTIME.md @@ -75,6 +75,18 @@ worker 在 `spawn_depth = 0` 运行，并且可以在满足 `spawn_depth + 1 ≤ fleet 账本持久化的是 worker 自身的事件流，而不是另一套模拟的分类法。`codewhale exec --output-format stream-json` 会发出 `{"type": "content" | "tool_use" | "tool_result" | "sandbox_denied" | "workflow_event" | "session_capture" | "turn_usage" | "metadata" | "done" | "error"}` 行，它们映射到 fleet 账本的 `FleetWorkerEventPayload`（`RunningTool`、`WorkflowEvent`、`Running`、`Completed`、`Failed` 等）。`workflow_event` 在 Workflow 飞行期间携带类型化的 run/phase/task/gate 回执，并作为类型化的 `WorkflowEvent` 保留在 Fleet 账本中；外层 worker 仍然拥有终态 `done` 或 `error`。一套词汇，两个表面。 +`session_capture` 在 exec 运行把自己的对话记录持久化为已保存会话时发出一次，并且只在这一个地方携带可恢复的 id： + +```json +{"type": "session_capture", "schema": "codewhale.exec-stream", "schema_version": 1, + "content": "", "saved_session_id": "01J…"} +``` + +- `saved_session_id` 是原始的已保存会话 id。Runtime 执行器会把它记录到任务的 `FleetReceipt.saved_session_id`（runtime API 的回执载荷也会暴露它），这样客户端可以通过 `GET /v1/sessions/{id}` 获取 worker 的完整最终回复，而不必重新读取 worker 日志。 +- `content` 是与终态 `metadata.session_id` 相同的脱敏指纹，因此单独截获的 `metadata` 回执仍然可以安全写入日志，两个事件之间也仍可关联。相应地，`metadata.resume_command` 指向该字段（`codewhale exec --resume `），而不是自己携带 id。 + +终态 `metadata` 回执还携带 worker 可见的最终回答：`visible_final_answer_chars` 是最终助手回复的真实字符数，`visible_final_answer_excerpt` 是它的有界（4,000 字符，截断时以 `...` 结尾）、已脱敏的摘录；运行没有产生可见回答时省略该字段。Runtime 执行器从这个回执读取摘录——绝不从流式 `content` 增量读取，那是运行过程中的"边想边说"——并把它附加到 `Completed.summary`；对于没有评分器也没有文件工件的任务，还会作为任务的交付物写入回执备注。生命周期事件标签和 worker 检视摘要只显示短摘录；事件 `payload` 和回执保留完整摘录。 + `turn_usage` 是每次模型调用的用量回执，当 provider 为该调用报告了用量时，每个模型请求（turn 步骤）发出一次： ```json