From d85055a58548f64b90d4a1c80060600869478a58 Mon Sep 17 00:00:00 2001 From: Ben Gao Date: Sun, 6 Sep 2026 13:26:33 +0800 Subject: [PATCH 1/4] feat(fleet): surface worker deliverables via summary and saved-session reply Accumulate streamed content into Completed.summary so receipt notes show a bounded deliverable excerpt instead of 'no verifiable output'. Emit the real saved-session id in the session_capture stream event, persist it on FleetReceipt, and expose it via the runtime API so a client can resolve the worker's final assistant reply through GET /v1/sessions/{id}. --- crates/protocol/src/fleet.rs | 8 ++ crates/tui/src/exec_agent.rs | 1 + crates/tui/src/fleet/alerts.rs | 1 + crates/tui/src/fleet/control.rs | 1 + crates/tui/src/fleet/executor.rs | 161 ++++++++++++++++++++++++++-- crates/tui/src/fleet/ledger.rs | 5 + crates/tui/src/fleet/manager.rs | 12 +++ crates/tui/src/fleet/task_spec.rs | 90 +++++++++++++++- crates/tui/src/lib.rs | 14 ++- crates/tui/src/runtime_api.rs | 1 + crates/tui/src/runtime_api/tests.rs | 6 ++ 11 files changed, 288 insertions(+), 12 deletions(-) diff --git a/crates/protocol/src/fleet.rs b/crates/protocol/src/fleet.rs index 3dba033887..c532f2ac37 100644 --- a/crates/protocol/src/fleet.rs +++ b/crates/protocol/src/fleet.rs @@ -1126,6 +1126,11 @@ 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. Callers resolve the final assistant + /// reply via `GET /v1/sessions/{id}`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_id: Option, /// Effective worker authority for this task (#3211). #[serde(default, skip_serializing_if = "Option::is_none")] pub effective_permissions: Option, @@ -1493,6 +1498,7 @@ mod tests { notes: None, }), resolved_route: None, + session_id: None, effective_permissions: None, }; let json = serde_json::to_string(&receipt).unwrap(); @@ -1521,6 +1527,7 @@ mod tests { notes: Some("manual verification required".to_string()), }), resolved_route: None, + session_id: None, effective_permissions: None, }; @@ -1689,6 +1696,7 @@ mod tests { model_source: Some("task.model".to_string()), source: "resolver".to_string(), }), + 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..de1a64d979 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), + session_id: id.clone(), })?; } // Resolved output ceiling and its provenance, surfaced so a diff --git a/crates/tui/src/fleet/alerts.rs b/crates/tui/src/fleet/alerts.rs index 220df76eb2..6cc68c2667 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, + session_id: None, effective_permissions: None, }; diff --git a/crates/tui/src/fleet/control.rs b/crates/tui/src/fleet/control.rs index 9ebf4b1a6a..f3e19fd1d9 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(), }), + session_id: None, effective_permissions: None, } } diff --git a/crates/tui/src/fleet/executor.rs b/crates/tui/src/fleet/executor.rs index a508e8545e..ab39392cc9 100644 --- a/crates/tui/src/fleet/executor.rs +++ b/crates/tui/src/fleet/executor.rs @@ -547,6 +547,15 @@ struct WorkerStream { terminal_route: TerminalRouteEvidence, /// When this worker process was started, for per-task wall-clock limits (R5). started_at: std::time::Instant, + /// Accumulated assistant text from `content` stream events. This is the + /// task's visible deliverable for report/summary work that produces no file + /// artifact; surfaced as `Completed.summary` so receipts stop reporting + /// "no verifiable output" for a worker that wrote a full report. + answer: String, + /// 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). + session_id: Option, } #[derive(Debug, Clone, Default)] @@ -585,6 +594,8 @@ impl TerminalRouteEvidence { fn observe_worker_stream_line( terminal_route: &mut TerminalRouteEvidence, + answer: &mut String, + session_id: &mut Option, line: &[u8], ) -> Option { let Ok(line) = std::str::from_utf8(line) else { @@ -596,9 +607,50 @@ fn observe_worker_stream_line( }; let line = line.trim_end(); terminal_route.observe(parse_exec_terminal_route(line)); + // Accumulate the worker's visible assistant text so report/summary tasks + // (no scorer, no file artifact) still surface their deliverable as + // `Completed.summary` instead of "no verifiable output". Also capture the + // saved exec session id so a caller can resolve the full transcript. + if let Ok(value) = serde_json::from_str::(line) { + match value.get("type").and_then(serde_json::Value::as_str) { + Some("content") => { + if let Some(content) = value.get("content").and_then(serde_json::Value::as_str) { + answer.push_str(content); + } + } + Some("session_capture") => { + if let Some(id) = value.get("session_id").and_then(serde_json::Value::as_str) + && !id.trim().is_empty() + { + *session_id = Some(id.to_string()); + } + } + _ => {} + } + } map_exec_stream_line(line) } +const MAX_WORKER_SUMMARY_CHARS: usize = 4_000; + +/// Bound and redact the worker's accumulated answer before surfacing it as +/// `Completed.summary`. The summary is a status surface (receipt notes, event +/// labels, runtime API payloads), not the forensic worker log — the full text +/// already lives in the worker's stream-json file. +fn bounded_worker_summary(answer: &str) -> String { + let redacted = codewhale_config::persistence::redact_secrets(answer); + let mut chars = redacted.chars(); + let preview = chars + .by_ref() + .take(MAX_WORKER_SUMMARY_CHARS) + .collect::(); + if chars.next().is_some() { + format!("{preview}...") + } else { + preview + } +} + enum WorkerStreamHost { Local, Ssh(String), @@ -618,6 +670,9 @@ pub struct FleetWorkerTerminalEvent { /// Non-terminal payloads discovered by the mandatory post-exit drain. pub tail_payloads: Vec, pub reported_route: Option, + /// Saved exec session id reported by the worker's `session_capture` event, + /// when one was persisted on completion. + pub 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 +765,8 @@ impl FleetExecutor { terminal: false, terminal_route: TerminalRouteEvidence::default(), started_at: std::time::Instant::now(), + answer: String::new(), + session_id: None, }, ); Ok(handle) @@ -802,7 +859,12 @@ 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) = observe_worker_stream_line( + &mut stream.terminal_route, + &mut stream.answer, + &mut stream.session_id, + &line, + ) { events.push(event); } } @@ -834,7 +896,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, @@ -854,14 +916,35 @@ impl FleetExecutor { 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) + && let Some(payload) = observe_worker_stream_line( + &mut stream.terminal_route, + &mut stream.answer, + &mut stream.session_id, + &trailing_line, + ) { tail_payloads.push(payload); } } - if let Some(stream) = self.streams.get_mut(worker_id) { - stream.terminal = true; + let answer = self + .streams + .get_mut(worker_id) + .map(|stream| { + stream.terminal = true; + std::mem::take(&mut stream.answer) + }) + .unwrap_or_default(); + let session_id = self + .streams + .get_mut(worker_id) + .and_then(|stream| stream.session_id.take()); + // Attach the accumulated visible answer to a successful completion so + // report/summary tasks (no scorer, no file artifact) surface their + // deliverable instead of "no verifiable output". + if !answer.trim().is_empty() + && let FleetWorkerEventPayload::Completed { summary, .. } = &mut terminal + { + *summary = Some(bounded_worker_summary(&answer)); } Some(FleetWorkerTerminalEvent { payload: terminal, @@ -871,6 +954,7 @@ impl FleetExecutor { .streams .get(worker_id) .and_then(|stream| stream.terminal_route.reported_route().cloned()), + session_id, requires_reported_route: true, }) } @@ -992,6 +1076,8 @@ mod tests { terminal: false, terminal_route: TerminalRouteEvidence::default(), started_at: std::time::Instant::now(), + answer: String::new(), + session_id: None, }, ); } @@ -1722,6 +1808,69 @@ mod tests { assert!(exec.all_terminal()); } + #[cfg(unix)] + #[test] + fn completed_worker_surfaces_accumulated_content_as_summary() { + // Report/summary tasks produce their deliverable as streamed text, not + // a file artifact. The executor must accumulate `content` events and + // attach them to the terminal `Completed.summary` so a receipt can show + // the actual result instead of "no verifiable output". + let tmp = tempfile::TempDir::new().unwrap(); + let mut exec = FleetExecutor::new(tmp.path()); + let script = r#"printf '%s\n' '{"type":"content","content":"part one "}' '{"type":"content","content":"part two"}' '{"type":"done"}'"#; + let command = FleetWorkerCommand::new("sh", vec!["-c".to_string(), script.to_string()]); + exec.start_worker("w1", command, None).unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let terminal = loop { + exec.drain_events("w1"); + if let Some(term) = exec.poll_terminal("w1") { + break term; + } + assert!( + std::time::Instant::now() < deadline, + "worker did not terminate in time" + ); + std::thread::sleep(std::time::Duration::from_millis(20)); + }; + + match terminal { + FleetWorkerEventPayload::Completed { summary, .. } => { + assert_eq!(summary.as_deref(), Some("part one part two")); + } + other => panic!("expected Completed, got {other:?}"), + } + } + + #[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`. 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 tmp = tempfile::TempDir::new().unwrap(); + let mut exec = FleetExecutor::new(tmp.path()); + let script = r#"printf '%s\n' '{"type":"session_capture","content":"","session_id":"session-abc"}' '{"type":"done"}'"#; + let command = FleetWorkerCommand::new("sh", vec!["-c".to_string(), script.to_string()]); + exec.start_worker("w-session", command, None).unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let terminal = loop { + exec.drain_events("w-session"); + if let Some(term) = exec.poll_terminal_with_status("w-session") { + break term; + } + assert!( + std::time::Instant::now() < deadline, + "worker did not terminate in time" + ); + std::thread::sleep(std::time::Duration::from_millis(20)); + }; + + assert_eq!(terminal.session_id.as_deref(), Some("session-abc")); + } + #[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..ca2809770b 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, + session_id: None, effective_permissions: None, }) .unwrap(); @@ -3542,6 +3543,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, + session_id: None, effective_permissions: None, }; assert!( @@ -3825,6 +3827,7 @@ mod tests { artifacts: vec![], score: None, resolved_route: None, + session_id: None, effective_permissions: None, }) .unwrap(); @@ -3940,6 +3943,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, + session_id: None, effective_permissions: None, }, ) @@ -4186,6 +4190,7 @@ mod tests { artifacts: vec![], score: None, resolved_route: None, + 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..efeed383a3 100644 --- a/crates/tui/src/fleet/manager.rs +++ b/crates/tui/src/fleet/manager.rs @@ -1499,6 +1499,7 @@ impl FleetManager { exit_code: None, tail_payloads: Vec::new(), reported_route: None, + session_id: None, requires_reported_route: false, }; let _ = self.record_task_outcome(&task, terminal)?; @@ -1547,6 +1548,7 @@ impl FleetManager { exit_code: None, tail_payloads: Vec::new(), reported_route: None, + session_id: None, requires_reported_route: false, }; let _ = self.record_task_outcome(&task, terminal)?; @@ -1666,6 +1668,7 @@ impl FleetManager { exit_code, tail_payloads, reported_route, + session_id, requires_reported_route, } = terminal; let (receipt_result, failure_kind, exit_code) = task_receipt_outcome(&payload, exit_code); @@ -1715,6 +1718,10 @@ impl FleetManager { (None, false) => self.resolve_task_route(&task.task_spec), }; let effective_permissions = self.resolve_task_effective_permissions(task); + let summary = match &payload { + FleetWorkerEventPayload::Completed { summary, .. } => summary.clone(), + _ => None, + }; let verification_input = FleetTaskVerificationInput { run_id: task.entry.run_id.clone(), task_id: task.entry.task_id.clone(), @@ -1722,6 +1729,8 @@ impl FleetManager { attempt: task.entry.attempts, exit_code, artifacts, + summary, + session_id, resolved_route, effective_permissions, }; @@ -1742,6 +1751,7 @@ impl FleetManager { artifacts: verification_input.artifacts, score: None, resolved_route: verification_input.resolved_route, + session_id: verification_input.session_id, effective_permissions: verification_input.effective_permissions, } }; @@ -1801,6 +1811,7 @@ impl FleetManager { artifacts, score: None, resolved_route: self.resolve_task_route(&task.task_spec), + session_id: None, effective_permissions: self.resolve_task_effective_permissions(task), }; let payload = FleetWorkerEventPayload::Cancelled { @@ -3405,6 +3416,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, + 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..d848bf2c2d 100644 --- a/crates/tui/src/fleet/task_spec.rs +++ b/crates/tui/src/fleet/task_spec.rs @@ -86,6 +86,13 @@ pub struct FleetTaskVerificationInput { pub attempt: u32, pub exit_code: Option, pub artifacts: Vec, + /// Accumulated visible assistant text from the worker stream. Report/summary + /// tasks with no scorer and no file artifact surface this as their + /// deliverable instead of "no verifiable output". + pub summary: Option, + /// Saved exec session id holding the worker's full transcript, when the + /// worker persisted one on completion. + pub 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 +337,27 @@ 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 + .summary + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + { + Some(summary) => partial( + "no scorer configured; worker produced a summary deliverable", + format!( + "worker produced {} characters of deliverable: {}", + summary.chars().count(), + bounded_receipt_excerpt(summary), + ), + ), + 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 +416,7 @@ pub fn prepare_verification_receipt( artifacts, score: Some(verification.score), resolved_route: input.resolved_route.clone(), + session_id: input.session_id.clone(), effective_permissions: input.effective_permissions.clone(), }; Ok(receipt) @@ -612,6 +637,26 @@ fn has_verifiable_artifact(input: &FleetTaskVerificationInput) -> bool { }) } +/// Bound and redact the worker's visible deliverable for a receipt note. +/// Receipts are status surfaces, not the forensic worker log, so a bounded, +/// whitespace-normalized, secret-redacted excerpt is enough to show the user +/// what a report/summary task actually produced. +fn bounded_receipt_excerpt(value: &str) -> String { + const MAX_RECEIPT_EXCERPT_CHARS: usize = 600; + let redacted = codewhale_config::persistence::redact_secrets(value); + let normalized = redacted.split_whitespace().collect::>().join(" "); + let mut chars = normalized.chars(); + let preview = chars + .by_ref() + .take(MAX_RECEIPT_EXCERPT_CHARS) + .collect::(); + if chars.next().is_some() { + format!("{preview}...") + } else { + preview + } +} + #[derive(Debug)] struct EvidenceReadError { failure_kind: FleetTaskFailureKind, @@ -978,6 +1023,8 @@ mod tests { attempt: 1, exit_code: Some(0), artifacts: vec![], + summary: None, + session_id: None, resolved_route: None, effective_permissions: None, }; @@ -1065,6 +1112,37 @@ 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![], + summary: Some("The Changelog review is complete".to_string()), + 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 +1165,8 @@ mod tests { attempt: 3, exit_code: Some(1), artifacts: vec![log], + summary: None, + session_id: None, resolved_route: None, effective_permissions: Some(FleetEffectivePermissions { write: false, @@ -1146,6 +1226,8 @@ mod tests { attempt: 1, exit_code: Some(1), artifacts: Vec::new(), + summary: None, + 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..dd443fdb35 100644 --- a/crates/tui/src/lib.rs +++ b/crates/tui/src/lib.rs @@ -11479,7 +11479,13 @@ enum ExecStreamEvent { event: serde_json::Value, }, #[serde(rename = "session_capture")] - SessionCapture { content: String }, + SessionCapture { + /// Redacted fingerprint for logs/forensics; 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. + session_id: String, + }, #[serde(rename = "service_released")] #[cfg(unix)] ServiceReleased { @@ -17044,6 +17050,7 @@ api_key = "test-only-key" ( ExecStreamEvent::SessionCapture { content: "x".to_string(), + session_id: "session-x".to_string(), }, "session_capture", ), @@ -17212,13 +17219,16 @@ api_key = "test-only-key" let capture = ExecStreamEvent::SessionCapture { content: exec_stream_session_ref(raw_session_id), + 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"); + // The log fingerprint stays redacted; the recoverable id is a distinct + // field so a caller can resolve the saved session without the log path. assert_ne!(parsed_capture["content"], raw_session_id); + assert_eq!(parsed_capture["session_id"], raw_session_id); } #[test] diff --git a/crates/tui/src/runtime_api.rs b/crates/tui/src/runtime_api.rs index aae9155982..96bfcc8ec8 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::>(), + "session_id": receipt.session_id.clone(), "evidence_available": evidence_available, }) } diff --git a/crates/tui/src/runtime_api/tests.rs b/crates/tui/src/runtime_api/tests.rs index 9f9b3f26b2..ef8dc89501 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, + 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, + 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, + 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, + 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(), + summary: None, + session_id: None, resolved_route: None, effective_permissions: None, }; From 4de6dc17f6a5b2b047b453bccc98bd5237eb21d8 Mon Sep 17 00:00:00 2001 From: CodeWhale Bot Date: Sun, 6 Sep 2026 17:55:05 -0700 Subject: [PATCH 2/4] =?UTF-8?q?fix(fleet):=20worker=20deliverable=20rework?= =?UTF-8?q?=20=E2=80=94=20excerpt=20at=20the=20emitter,=20parse=20once,=20?= =?UTF-8?q?saved=5Fsession=5Fid?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Maintainer follow-up on #5946 (original by @gaord, preserved below as c58c74912). Keeps the saved-session half; reworks the summary half per review: 1. The excerpt now travels in the exec terminal event: the terminal `metadata` receipt carries `visible_final_answer_excerpt` (bounded, secret-redacted) next to the REAL pre-bound count `visible_final_answer_chars` of `summary.output` — the final reply, not the opening of the run. 2. The fleet executor's frame accumulator is deleted: nothing streams assistant text into per-worker memory anymore; `WorkerStream` only records the terminal receipt's answer. 3. Exactly one bound-and-redact helper, `exec_stream_final_answer_excerpt` (crates/tui/src/lib.rs, 4,000 chars); the executor-side `bounded_worker_summary` and task_spec-side `bounded_receipt_excerpt` duplicates are deleted. 4. The terminal frame is parsed exactly once: `parse_exec_terminal_*` take `&serde_json::Value`, and `WorkerStream::observe_line` parses each line once for route evidence, final answer, session capture, and payload mapping (`map_exec_stream_value`). 5. `session_capture.session_id` renamed to `saved_session_id` everywhere, and protocol `FleetReceipt.session_id` to `saved_session_id`; all consumers updated (runtime_api receipt JSON, manager, task_spec, ledger/alerts/control tests); `metadata` stays fingerprint-only and `metadata.resume_command` now names the field instead of pretending to redact one. 6. Docs updated in docs/AGENT_RUNTIME.md and docs/zh_hans/AGENT_RUNTIME.md. 7. The excerpt also surfaces on FAILED outcomes: it stays on `FleetWorkerTerminalEvent.final_answer` whatever the outcome, and a no-scorer failed/cancelled receipt keeps the text in its score notes; lifecycle event labels show a 160-char excerpt and worker inspection summaries bound notes to 240 bytes while payloads/receipts keep the full excerpt. Gates (RUST_MIN_STACK=16777216, shared target dir): - cargo fmt --all: clean, no changes - cargo clippy --workspace --all-targets --all-features --locked -D warnings -A clippy::uninlined_format_args -A clippy::too_many_arguments -A clippy::unnecessary_map_or: pass, 0 warnings - cargo test -p codewhale-tui --lib --locked -- fleet::executor fleet::task_spec fleet::manager terminal_mode_tests::exec_stream runtime_api::tests::fleet_receipt: 100 passed, 0 failed - cargo test -p codewhale-protocol --locked: 85 passed, 0 failed - cargo test -p codewhale-tui --lib --locked: 11856 passed, 0 failed, 13 ignored (one earlier run had 1 unrelated tmux clipboard flake that passes in isolation; a mass-failure run in between was shared-target-dir cross-worktree contamination, not this change) Signed-off-by: CodeWhale Bot --- CHANGELOG.md | 14 ++ crates/protocol/src/fleet.rs | 13 +- crates/tui/CHANGELOG.md | 14 ++ crates/tui/src/exec_agent.rs | 5 +- crates/tui/src/fleet/alerts.rs | 2 +- crates/tui/src/fleet/control.rs | 2 +- crates/tui/src/fleet/executor.rs | 357 +++++++++++++++------------- crates/tui/src/fleet/ledger.rs | 10 +- crates/tui/src/fleet/manager.rs | 45 ++-- crates/tui/src/fleet/task_spec.rs | 108 +++++---- crates/tui/src/lib.rs | 83 ++++++- crates/tui/src/runtime_api.rs | 15 +- crates/tui/src/runtime_api/tests.rs | 12 +- docs/AGENT_RUNTIME.md | 29 +++ docs/zh_hans/AGENT_RUNTIME.md | 12 + 15 files changed, 461 insertions(+), 260 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cd3b2cb3e..b2c2451d53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Fleet workers now surface their deliverable. The terminal `codewhale exec` + `metadata` receipt carries `visible_final_answer_excerpt`, a bounded, + secret-redacted excerpt of the final assistant reply next to the real + `visible_final_answer_chars` count; the Runtime executor attaches it to + `Completed.summary` and, for a task with no scorer and no file artifact, + to the receipt notes instead of "no verifiable output" — a worker that + fails after writing most of a report keeps the text too. `session_capture` + now carries the raw `saved_session_id` (the `metadata` receipt stays + fingerprint-only), `FleetReceipt.saved_session_id` persists it, and the + runtime API exposes it so a client can resolve the worker's full final + reply via `GET /v1/sessions/{id}` (#5946, thanks @gaord). + ### Changed - `/statusline` drives the bottom chrome again. Since the 0.9.12 shell diff --git a/crates/protocol/src/fleet.rs b/crates/protocol/src/fleet.rs index c532f2ac37..114ccfb277 100644 --- a/crates/protocol/src/fleet.rs +++ b/crates/protocol/src/fleet.rs @@ -1127,10 +1127,11 @@ pub struct FleetReceipt { #[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. Callers resolve the final assistant - /// reply via `GET /v1/sessions/{id}`. + /// 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 session_id: Option, + pub saved_session_id: Option, /// Effective worker authority for this task (#3211). #[serde(default, skip_serializing_if = "Option::is_none")] pub effective_permissions: Option, @@ -1498,7 +1499,7 @@ mod tests { notes: None, }), resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; let json = serde_json::to_string(&receipt).unwrap(); @@ -1527,7 +1528,7 @@ mod tests { notes: Some("manual verification required".to_string()), }), resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; @@ -1696,7 +1697,7 @@ mod tests { model_source: Some("task.model".to_string()), source: "resolver".to_string(), }), - session_id: None, + saved_session_id: None, effective_permissions: Some(FleetEffectivePermissions { write: true, network: true, diff --git a/crates/tui/CHANGELOG.md b/crates/tui/CHANGELOG.md index 47025c6680..6691095972 100644 --- a/crates/tui/CHANGELOG.md +++ b/crates/tui/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Fleet workers now surface their deliverable. The terminal `codewhale exec` + `metadata` receipt carries `visible_final_answer_excerpt`, a bounded, + secret-redacted excerpt of the final assistant reply next to the real + `visible_final_answer_chars` count; the Runtime executor attaches it to + `Completed.summary` and, for a task with no scorer and no file artifact, + to the receipt notes instead of "no verifiable output" — a worker that + fails after writing most of a report keeps the text too. `session_capture` + now carries the raw `saved_session_id` (the `metadata` receipt stays + fingerprint-only), `FleetReceipt.saved_session_id` persists it, and the + runtime API exposes it so a client can resolve the worker's full final + reply via `GET /v1/sessions/{id}` (#5946, thanks @gaord). + ### Changed - `/statusline` drives the bottom chrome again. Since the 0.9.12 shell diff --git a/crates/tui/src/exec_agent.rs b/crates/tui/src/exec_agent.rs index de1a64d979..a5945cc052 100644 --- a/crates/tui/src/exec_agent.rs +++ b/crates/tui/src/exec_agent.rs @@ -937,7 +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), - session_id: id.clone(), + saved_session_id: id.clone(), })?; } // Resolved output ceiling and its provenance, surfaced so a @@ -990,6 +990,9 @@ pub(crate) async fn run_exec_agent( latest_system_prompt.as_ref(), ), visible_final_answer_chars: summary.output.chars().count(), + visible_final_answer_excerpt: exec_stream_final_answer_excerpt( + &summary.output, + ), 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 6cc68c2667..2efb219182 100644 --- a/crates/tui/src/fleet/alerts.rs +++ b/crates/tui/src/fleet/alerts.rs @@ -715,7 +715,7 @@ mod tests { notes: Some("regex scorer could not be compiled".to_string()), }), resolved_route: None, - session_id: 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 f3e19fd1d9..56886d3f52 100644 --- a/crates/tui/src/fleet/control.rs +++ b/crates/tui/src/fleet/control.rs @@ -753,7 +753,7 @@ mod tests { model_source: None, source: "resolver".to_string(), }), - session_id: None, + saved_session_id: None, effective_permissions: None, } } diff --git a/crates/tui/src/fleet/executor.rs b/crates/tui/src/fleet/executor.rs index ab39392cc9..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 { - 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 { +fn parse_exec_terminal_route(value: &serde_json::Value) -> ParsedTerminalRoute { + let Some(meta) = exec_terminal_meta(value) 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,15 +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, - /// Accumulated assistant text from `content` stream events. This is the - /// task's visible deliverable for report/summary work that produces no file - /// artifact; surfaced as `Completed.summary` so receipts stop reporting - /// "no verifiable output" for a worker that wrote a full report. - answer: String, + /// 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). - session_id: Option, + 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)] @@ -592,65 +660,6 @@ impl TerminalRouteEvidence { } } -fn observe_worker_stream_line( - terminal_route: &mut TerminalRouteEvidence, - answer: &mut String, - session_id: &mut Option, - 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)); - // Accumulate the worker's visible assistant text so report/summary tasks - // (no scorer, no file artifact) still surface their deliverable as - // `Completed.summary` instead of "no verifiable output". Also capture the - // saved exec session id so a caller can resolve the full transcript. - if let Ok(value) = serde_json::from_str::(line) { - match value.get("type").and_then(serde_json::Value::as_str) { - Some("content") => { - if let Some(content) = value.get("content").and_then(serde_json::Value::as_str) { - answer.push_str(content); - } - } - Some("session_capture") => { - if let Some(id) = value.get("session_id").and_then(serde_json::Value::as_str) - && !id.trim().is_empty() - { - *session_id = Some(id.to_string()); - } - } - _ => {} - } - } - map_exec_stream_line(line) -} - -const MAX_WORKER_SUMMARY_CHARS: usize = 4_000; - -/// Bound and redact the worker's accumulated answer before surfacing it as -/// `Completed.summary`. The summary is a status surface (receipt notes, event -/// labels, runtime API payloads), not the forensic worker log — the full text -/// already lives in the worker's stream-json file. -fn bounded_worker_summary(answer: &str) -> String { - let redacted = codewhale_config::persistence::redact_secrets(answer); - let mut chars = redacted.chars(); - let preview = chars - .by_ref() - .take(MAX_WORKER_SUMMARY_CHARS) - .collect::(); - if chars.next().is_some() { - format!("{preview}...") - } else { - preview - } -} - enum WorkerStreamHost { Local, Ssh(String), @@ -670,9 +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 session_id: Option, + 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. @@ -765,8 +778,8 @@ impl FleetExecutor { terminal: false, terminal_route: TerminalRouteEvidence::default(), started_at: std::time::Instant::now(), - answer: String::new(), - session_id: None, + final_answer: None, + saved_session_id: None, }, ); Ok(handle) @@ -859,12 +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, - &mut stream.answer, - &mut stream.session_id, - &line, - ) { + if let Some(event) = stream.observe_line(&line) { events.push(event); } } @@ -913,48 +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, - &mut stream.answer, - &mut stream.session_id, - &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); } - let answer = self - .streams - .get_mut(worker_id) - .map(|stream| { - stream.terminal = true; - std::mem::take(&mut stream.answer) - }) - .unwrap_or_default(); - let session_id = self - .streams - .get_mut(worker_id) - .and_then(|stream| stream.session_id.take()); - // Attach the accumulated visible answer to a successful completion so - // report/summary tasks (no scorer, no file artifact) surface their - // deliverable instead of "no verifiable output". - if !answer.trim().is_empty() - && let FleetWorkerEventPayload::Completed { summary, .. } = &mut terminal + 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(bounded_worker_summary(&answer)); + *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()), - session_id, + reported_route: stream.terminal_route.reported_route().cloned(), + final_answer, + saved_session_id: stream.saved_session_id.take(), requires_reported_route: true, }) } @@ -1076,8 +1067,8 @@ mod tests { terminal: false, terminal_route: TerminalRouteEvidence::default(), started_at: std::time::Instant::now(), - answer: String::new(), - session_id: None, + final_answer: None, + saved_session_id: None, }, ); } @@ -1719,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() }; @@ -1809,22 +1801,16 @@ mod tests { } #[cfg(unix)] - #[test] - fn completed_worker_surfaces_accumulated_content_as_summary() { - // Report/summary tasks produce their deliverable as streamed text, not - // a file artifact. The executor must accumulate `content` events and - // attach them to the terminal `Completed.summary` so a receipt can show - // the actual result instead of "no verifiable output". + 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 script = r#"printf '%s\n' '{"type":"content","content":"part one "}' '{"type":"content","content":"part two"}' '{"type":"done"}'"#; let command = FleetWorkerCommand::new("sh", vec!["-c".to_string(), script.to_string()]); - exec.start_worker("w1", command, None).unwrap(); + exec.start_worker(worker_id, command, None).unwrap(); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - let terminal = loop { - exec.drain_events("w1"); - if let Some(term) = exec.poll_terminal("w1") { + loop { + exec.drain_events(worker_id); + if let Some(term) = exec.poll_terminal_with_status(worker_id) { break term; } assert!( @@ -1832,43 +1818,96 @@ mod tests { "worker did not terminate in time" ); std::thread::sleep(std::time::Duration::from_millis(20)); - }; + } + } - match terminal { + #[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("part one part two")); + 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`. 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 tmp = tempfile::TempDir::new().unwrap(); - let mut exec = FleetExecutor::new(tmp.path()); - let script = r#"printf '%s\n' '{"type":"session_capture","content":"","session_id":"session-abc"}' '{"type":"done"}'"#; - let command = FleetWorkerCommand::new("sh", vec!["-c".to_string(), script.to_string()]); - exec.start_worker("w-session", command, None).unwrap(); + // 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"); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - let terminal = loop { - exec.drain_events("w-session"); - if let Some(term) = exec.poll_terminal_with_status("w-session") { - break term; - } - assert!( - std::time::Instant::now() < deadline, - "worker did not terminate in time" - ); - std::thread::sleep(std::time::Duration::from_millis(20)); - }; + assert_eq!(terminal.saved_session_id.as_deref(), Some("session-abc")); + assert!(terminal.final_answer.is_none()); + } - assert_eq!(terminal.session_id.as_deref(), Some("session-abc")); + #[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)] diff --git a/crates/tui/src/fleet/ledger.rs b/crates/tui/src/fleet/ledger.rs index ca2809770b..2f54a6bed3 100644 --- a/crates/tui/src/fleet/ledger.rs +++ b/crates/tui/src/fleet/ledger.rs @@ -2763,7 +2763,7 @@ mod tests { notes: Some("verifier note contained super-secret".to_string()), }), resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }) .unwrap(); @@ -3543,7 +3543,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; assert!( @@ -3827,7 +3827,7 @@ mod tests { artifacts: vec![], score: None, resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }) .unwrap(); @@ -3943,7 +3943,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }, ) @@ -4190,7 +4190,7 @@ mod tests { artifacts: vec![], score: None, resolved_route: None, - session_id: 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 efeed383a3..6445c130cf 100644 --- a/crates/tui/src/fleet/manager.rs +++ b/crates/tui/src/fleet/manager.rs @@ -1499,7 +1499,8 @@ impl FleetManager { exit_code: None, tail_payloads: Vec::new(), reported_route: None, - session_id: None, + final_answer: None, + saved_session_id: None, requires_reported_route: false, }; let _ = self.record_task_outcome(&task, terminal)?; @@ -1548,7 +1549,8 @@ impl FleetManager { exit_code: None, tail_payloads: Vec::new(), reported_route: None, - session_id: None, + final_answer: None, + saved_session_id: None, requires_reported_route: false, }; let _ = self.record_task_outcome(&task, terminal)?; @@ -1668,7 +1670,8 @@ impl FleetManager { exit_code, tail_payloads, reported_route, - session_id, + final_answer, + saved_session_id, requires_reported_route, } = terminal; let (receipt_result, failure_kind, exit_code) = task_receipt_outcome(&payload, exit_code); @@ -1718,10 +1721,6 @@ impl FleetManager { (None, false) => self.resolve_task_route(&task.task_spec), }; let effective_permissions = self.resolve_task_effective_permissions(task); - let summary = match &payload { - FleetWorkerEventPayload::Completed { summary, .. } => summary.clone(), - _ => None, - }; let verification_input = FleetTaskVerificationInput { run_id: task.entry.run_id.clone(), task_id: task.entry.task_id.clone(), @@ -1729,8 +1728,8 @@ impl FleetManager { attempt: task.entry.attempts, exit_code, artifacts, - summary, - session_id, + final_answer, + saved_session_id, resolved_route, effective_permissions, }; @@ -1749,9 +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, - session_id: verification_input.session_id, + saved_session_id: verification_input.saved_session_id, effective_permissions: verification_input.effective_permissions, } }; @@ -1811,7 +1820,7 @@ impl FleetManager { artifacts, score: None, resolved_route: self.resolve_task_route(&task.task_spec), - session_id: None, + saved_session_id: None, effective_permissions: self.resolve_task_effective_permissions(task), }; let payload = FleetWorkerEventPayload::Cancelled { @@ -2306,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 @@ -3416,7 +3433,7 @@ mod tests { artifacts: Vec::new(), score: None, resolved_route: None, - session_id: 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 d848bf2c2d..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,13 +108,13 @@ pub struct FleetTaskVerificationInput { pub attempt: u32, pub exit_code: Option, pub artifacts: Vec, - /// Accumulated visible assistant text from the worker stream. Report/summary - /// tasks with no scorer and no file artifact surface this as their - /// deliverable instead of "no verifiable output". - pub summary: Option, + /// 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 session_id: Option, + 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). @@ -337,27 +359,20 @@ pub fn verify_task_result( "manual scorer configured", "manual verification is required to finalize this receipt", ), - None if !has_verifiable_artifact(input) => { - match input - .summary - .as_deref() - .map(str::trim) - .filter(|s| !s.is_empty()) - { - Some(summary) => partial( - "no scorer configured; worker produced a summary deliverable", - format!( - "worker produced {} characters of deliverable: {}", - summary.chars().count(), - bounded_receipt_excerpt(summary), - ), - ), - None => 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", @@ -416,7 +431,7 @@ pub fn prepare_verification_receipt( artifacts, score: Some(verification.score), resolved_route: input.resolved_route.clone(), - session_id: input.session_id.clone(), + saved_session_id: input.saved_session_id.clone(), effective_permissions: input.effective_permissions.clone(), }; Ok(receipt) @@ -637,26 +652,6 @@ fn has_verifiable_artifact(input: &FleetTaskVerificationInput) -> bool { }) } -/// Bound and redact the worker's visible deliverable for a receipt note. -/// Receipts are status surfaces, not the forensic worker log, so a bounded, -/// whitespace-normalized, secret-redacted excerpt is enough to show the user -/// what a report/summary task actually produced. -fn bounded_receipt_excerpt(value: &str) -> String { - const MAX_RECEIPT_EXCERPT_CHARS: usize = 600; - let redacted = codewhale_config::persistence::redact_secrets(value); - let normalized = redacted.split_whitespace().collect::>().join(" "); - let mut chars = normalized.chars(); - let preview = chars - .by_ref() - .take(MAX_RECEIPT_EXCERPT_CHARS) - .collect::(); - if chars.next().is_some() { - format!("{preview}...") - } else { - preview - } -} - #[derive(Debug)] struct EvidenceReadError { failure_kind: FleetTaskFailureKind, @@ -1023,8 +1018,8 @@ mod tests { attempt: 1, exit_code: Some(0), artifacts: vec![], - summary: None, - session_id: None, + final_answer: None, + saved_session_id: None, resolved_route: None, effective_permissions: None, }; @@ -1122,8 +1117,11 @@ mod tests { attempt: 1, exit_code: Some(0), artifacts: vec![], - summary: Some("The Changelog review is complete".to_string()), - session_id: None, + final_answer: Some(FleetWorkerFinalAnswer { + excerpt: "The Changelog review is complete".to_string(), + chars: 32, + }), + saved_session_id: None, resolved_route: None, effective_permissions: None, }; @@ -1165,8 +1163,8 @@ mod tests { attempt: 3, exit_code: Some(1), artifacts: vec![log], - summary: None, - session_id: None, + final_answer: None, + saved_session_id: None, resolved_route: None, effective_permissions: Some(FleetEffectivePermissions { write: false, @@ -1226,8 +1224,8 @@ mod tests { attempt: 1, exit_code: Some(1), artifacts: Vec::new(), - summary: None, - session_id: None, + 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 dd443fdb35..c60293e737 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, @@ -11480,11 +11486,15 @@ enum ExecStreamEvent { }, #[serde(rename = "session_capture")] SessionCapture { - /// Redacted fingerprint for logs/forensics; never the recoverable id. + /// 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. - session_id: String, + /// `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)] @@ -11793,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(), @@ -12207,11 +12218,37 @@ 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; + +/// 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 } } @@ -17050,7 +17087,7 @@ api_key = "test-only-key" ( ExecStreamEvent::SessionCapture { content: "x".to_string(), - session_id: "session-x".to_string(), + saved_session_id: "session-x".to_string(), }, "session_capture", ), @@ -17186,6 +17223,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(), @@ -17211,24 +17249,49 @@ 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), - session_id: raw_session_id.to_string(), + saved_session_id: raw_session_id.to_string(), }; let capture_json = serde_json::to_string(&capture).expect("serializes"); let parsed_capture: serde_json::Value = serde_json::from_str(&capture_json).expect("valid json"); assert_eq!(parsed_capture["type"], "session_capture"); - // The log fingerprint stays redacted; the recoverable id is a distinct - // field so a caller can resolve the saved session without the log path. + assert_eq!(parsed_capture["content"], parsed["meta"]["session_id"]); assert_ne!(parsed_capture["content"], raw_session_id); - assert_eq!(parsed_capture["session_id"], 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] diff --git a/crates/tui/src/runtime_api.rs b/crates/tui/src/runtime_api.rs index 96bfcc8ec8..15c6fe31a0 100644 --- a/crates/tui/src/runtime_api.rs +++ b/crates/tui/src/runtime_api.rs @@ -2688,7 +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::>(), - "session_id": receipt.session_id.clone(), + "saved_session_id": receipt.saved_session_id.clone(), "evidence_available": evidence_available, }) } @@ -2739,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(), @@ -2769,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 ef8dc89501..7822d0124d 100644 --- a/crates/tui/src/runtime_api/tests.rs +++ b/crates/tui/src/runtime_api/tests.rs @@ -9567,7 +9567,7 @@ fn fleet_receipt_json_pass_result_has_no_failure_fields() { artifacts: Vec::new(), score: None, resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9601,7 +9601,7 @@ fn fleet_receipt_json_verifier_failure_is_not_retry_eligible() { artifacts: Vec::new(), score: None, resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9633,7 +9633,7 @@ fn fleet_receipt_json_transport_failure_is_retry_eligible() { artifacts: Vec::new(), score: None, resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9669,7 +9669,7 @@ fn fleet_receipt_json_receipt_artifact_sets_evidence_available() { notes: Some("all checks pass".to_string()), }), resolved_route: None, - session_id: None, + saved_session_id: None, effective_permissions: None, }; let value = fleet_receipt_json(&receipt); @@ -9746,8 +9746,8 @@ async fn fleet_receipt_api_list_and_get_round_trip() -> Result<()> { attempt: 1, exit_code: Some(0), artifacts: Vec::new(), - summary: None, - session_id: None, + 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 From d3b333b06f960f3898b4c2050655492cb57eefab Mon Sep 17 00:00:00 2001 From: CodeWhale Bot Date: Sun, 6 Sep 2026 20:40:13 -0700 Subject: [PATCH 3/4] chore(ci): retrigger buildkite (suspected base flake, see #5929) Signed-off-by: CodeWhale Bot From f94684cd601f353f75894b6a9c345c34415f4669 Mon Sep 17 00:00:00 2001 From: Ben Gao Date: Mon, 7 Sep 2026 18:45:31 +0800 Subject: [PATCH 4/4] fix(fleet): the receipt excerpt is the final assistant reply, not the cumulative stream Devin follow-up on #5946: in a multi-step turn ExecSummary::output accumulates every streamed delta, including pre-tool commentary from earlier steps, so the terminal metadata could present progress text as the deliverable. Derive visible_final_answer_chars/excerpt from the last assistant-like message of the persisted session; the cumulative output stays only as the fallback when the session carries no assistant text. Signed-off-by: Ben Gao --- crates/tui/src/exec_agent.rs | 10 +++- crates/tui/src/lib.rs | 88 ++++++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 2 deletions(-) diff --git a/crates/tui/src/exec_agent.rs b/crates/tui/src/exec_agent.rs index a5945cc052..78e8d906c7 100644 --- a/crates/tui/src/exec_agent.rs +++ b/crates/tui/src/exec_agent.rs @@ -955,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", @@ -989,9 +995,9 @@ 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( - &summary.output, + &final_answer, ), resume_command: saved_session_id .as_deref() diff --git a/crates/tui/src/lib.rs b/crates/tui/src/lib.rs index c60293e737..7d51f8057d 100644 --- a/crates/tui/src/lib.rs +++ b/crates/tui/src/lib.rs @@ -12236,6 +12236,30 @@ fn exec_stream_resume_hint(session_id: &str) -> String { /// 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 { @@ -17294,6 +17318,70 @@ api_key = "test-only-key" 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] fn exec_stream_input_analysis_reports_prompt_composition() { let system = SystemPrompt::Text("system rules".to_string());