diff --git a/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx b/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx index 562d71f..145beca 100644 --- a/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx +++ b/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx @@ -552,6 +552,15 @@ export function renderMessageBody( } } + // Defense-in-depth: a sentinel is a routing/control artifact and must never + // appear in the bubble, even when it isn't a recognized delegate or arrives + // in fragments mid-stream. Strip every complete block, then drop an unclosed + // trailing block (a dangling BEGIN with no END yet). Mirrors the Rust + // strip_sentinel_blocks semantics. + remainingText = remainingText + .replace(/<<>>[\s\S]*?<<>>/g, '') + .replace(/<<>>[\s\S]*$/, ''); + if (!remainingText.trim() && !delegateCard) return null; // Split by code blocks first @@ -2597,12 +2606,19 @@ export function renderTurnActivitySummary( const phaseIsActive = runtimePhase === 'running' || runtimePhase === 'output_completed' || runtimePhase === 'finalizing'; const showSpinner = phaseIsActive || turnIsActive || hasRunningTool || hasLiveTerminal; const detailToolCalls = displayToolCalls.filter((tool) => !actionableTools.includes(tool)); - const commandActivity = runningCommand - ? activeToolInputLabel(runningCommand) + // Prefer a currently-running command; otherwise fall back to the most recent + // command that already ran. Without this, a provider whose tools have all + // completed (e.g. kohaku between tool batches, with no launcher commandLine) + // would fall through to the "暂无工具事件" status even though `commandCount` + // shows commands ran — a confusing contradiction. + const latestCommandTool = runningCommand + ?? [...displayToolCalls].reverse().find((tool) => isCommandTool(tool)); + const commandActivity = latestCommandTool + ? activeToolInputLabel(latestCommandTool) : commandLine ? truncateActivityText(`${commandLine}${commandArgsSuffix(state.commandArgs)}`, 180) : null; - const commandVerb = runningCommand || hasRunningTool || hasLiveTerminal || runtimePhase === 'running' + const commandVerb = runningCommand || hasRunningTool || hasLiveTerminal || (!latestCommandTool && runtimePhase === 'running') ? '正在运行' : '已运行'; const editActivity = latestEdit diff --git a/crates/switchyard-provider-api/src/sentinel.rs b/crates/switchyard-provider-api/src/sentinel.rs index b0db9db..531e570 100644 --- a/crates/switchyard-provider-api/src/sentinel.rs +++ b/crates/switchyard-provider-api/src/sentinel.rs @@ -1,7 +1,10 @@ use crate::ProviderError; -const SENTINEL_BEGIN: &str = "<<>>"; -const SENTINEL_END: &str = "<<>>"; +/// The opening marker of a Switchyard control-channel (e.g. delegate) block. +/// Public so streaming adapters can gate it out of the live display channel. +pub const SENTINEL_BEGIN: &str = "<<>>"; +/// The closing marker of a Switchyard control-channel block. +pub const SENTINEL_END: &str = "<<>>"; /// Extract all JSON blocks delimited by sentinel markers from provider text output. pub fn extract_sentinel_blocks(text: &str) -> Vec<&str> { @@ -37,7 +40,10 @@ pub fn strip_sentinel_blocks(text: &str) -> String { if let Some(end) = text[json_start..].find(SENTINEL_END) { search_from = json_start + end + SENTINEL_END.len(); } else { - // Unclosed block — keep the rest as-is + // Unclosed block — keep the rest as-is. This shared strip stays + // conservative (it runs on every provider's final response); a + // streaming adapter that needs to withhold a pending block does so + // in its own display channel (see the kohaku SentinelDisplayFilter). search_from = abs_begin + SENTINEL_BEGIN.len(); } } diff --git a/crates/switchyard-provider-kohaku/src/jsonl.rs b/crates/switchyard-provider-kohaku/src/jsonl.rs index fca25a0..00dc74a 100644 --- a/crates/switchyard-provider-kohaku/src/jsonl.rs +++ b/crates/switchyard-provider-kohaku/src/jsonl.rs @@ -2,14 +2,18 @@ //! //! The `switchyard-headless` fork emits one JSON object per line on stdout: //! `turn_start`, `text`, `activity` (with `activity_type` + nested -//! `metadata`), `turn_end` (`status`/`text`/`error`/`usage`), and `error`. -//! Every line is also passed through verbatim as an `ItemUpdated` event for -//! the diagnostics drawer; this module only classifies the lines that affect -//! the assistant text and the turn outcome. +//! `metadata`), `turn_end` (`status`/`text`/`error`/`usage`), and a top-level +//! `error`. This module classifies each line into a [`KohakuEvent`]; the +//! adapter (`turn.rs`) then maps only the user-facing signal onto events: +//! assistant `text` deltas (sentinel-gated), genuine tool/subagent `activity` +//! (normalized to collapsed tool cards), and the turn outcome/error. Pure +//! runtime telemetry and `turn_start` are dropped so the chat shows only the +//! model's message; the raw protocol is preserved solely in the archived +//! stdout artifact, not as live `ItemUpdated` events. use serde_json::Value; -#[derive(Debug, PartialEq, Eq)] +#[derive(Debug, PartialEq)] pub enum KohakuEvent { /// An assistant text delta (`{"type":"text","content":...}`). Text(String), @@ -19,7 +23,16 @@ pub enum KohakuEvent { text: String, error: Option, }, - /// activity / turn_start / error / anything else — passed through as-is. + /// Runtime activity (`{"type":"activity","activity_type":...,"metadata":...}`). + /// `value` is the whole activity object so the adapter can read + /// `activity_type` / `detail` / `metadata` when deciding what (if anything) + /// to surface. + Activity { activity_type: String, value: Value }, + /// The turn-start marker (`{"type":"turn_start",...}`). + TurnStart, + /// A top-level fatal error (`{"type":"error","content":...}`). + Error(String), + /// Anything else — carries no user-facing signal. Other, } @@ -47,6 +60,23 @@ pub fn classify(json: &Value) -> KohakuEvent { .and_then(Value::as_str) .map(str::to_string), }, + Some("activity") => KohakuEvent::Activity { + activity_type: json + .get("activity_type") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(), + value: json.clone(), + }, + Some("turn_start") => KohakuEvent::TurnStart, + Some("error") => KohakuEvent::Error( + json.get("content") + .and_then(Value::as_str) + .or_else(|| json.get("error").and_then(Value::as_str)) + .or_else(|| json.get("message").and_then(Value::as_str)) + .unwrap_or("") + .to_string(), + ), _ => KohakuEvent::Other, } } @@ -100,14 +130,34 @@ mod tests { } #[test] - fn activity_and_turn_start_are_other() { + fn classifies_activity_with_type() { assert_eq!( - classify(&json!({"type": "activity", "activity_type": "tool_start"})), - KohakuEvent::Other + classify(&json!({"type": "activity", "activity_type": "tool_start", "detail": "read"})), + KohakuEvent::Activity { + activity_type: "tool_start".to_string(), + value: json!({"type": "activity", "activity_type": "tool_start", "detail": "read"}), + } ); + } + + #[test] + fn classifies_turn_start() { assert_eq!( classify(&json!({"type": "turn_start", "agent": "x"})), - KohakuEvent::Other + KohakuEvent::TurnStart + ); + } + + #[test] + fn classifies_top_level_error() { + assert_eq!( + classify(&json!({"type": "error", "content": "fatal boom"})), + KohakuEvent::Error("fatal boom".to_string()) ); } + + #[test] + fn unknown_type_is_other() { + assert_eq!(classify(&json!({"type": "mystery"})), KohakuEvent::Other); + } } diff --git a/crates/switchyard-provider-kohaku/src/turn.rs b/crates/switchyard-provider-kohaku/src/turn.rs index 77bb8d6..717325e 100644 --- a/crates/switchyard-provider-kohaku/src/turn.rs +++ b/crates/switchyard-provider-kohaku/src/turn.rs @@ -9,6 +9,7 @@ use tokio::sync::mpsc; use uuid::Uuid; +use switchyard_provider_api::sentinel::{SENTINEL_BEGIN, SENTINEL_END}; use switchyard_provider_api::*; use switchyard_provider_subprocess::{ StreamingOutputLine, SubprocessConfig, build_subprocess_invocation_plan, build_turn_result, @@ -91,106 +92,276 @@ pub async fn run_kohaku_turn( let (line_tx, mut line_rx) = mpsc::channel::(256); - event_tx - .send(ProviderEvent::execution_telemetry( - turn_id, - "kohaku", - &plan.execution, - )) - .await - .ok(); + // NOTE: kohaku deliberately does NOT emit execution_telemetry. `kt.exe` is + // Switchyard's own driver, not a model action — surfacing it would make the + // live-execution card headline read "正在运行 kt.exe" and count the driver + // as a command. The real per-tool activity (below) drives the headline and + // the command count instead; the resolved invocation is still recorded in + // the archived raw-output artifact for replay/diagnostics. let event_tx_clone = event_tx.clone(); let consumer = tokio::spawn(async move { - // `assistant_message` accumulates streamed `text` deltas (the primary - // response body). `final_text` is the consolidated `turn_end.text`, - // used only when no deltas arrived (e.g. a tool-only turn). + // `assistant_message` accumulates the FULL streamed body — including any + // routing sentinel block, plus a consolidated `turn_end.text` on the + // no-delta path — and is what the router parses for delegation (it + // strips the block for the final display). `display` gates what reaches + // the chat bubble so a sentinel never flickers into view while + // streaming. `turn_error` captures a structured failure so we surface + // the reason instead of dumping the raw JSONL as the response body. let mut assistant_message = String::new(); - let mut final_text: Option = None; + let mut turn_error: Option = None; + let mut display = SentinelDisplayFilter::new(); while let Some(output_line) = line_rx.recv().await { let line = output_line.text; let protocol_line = line.trim_end_matches(['\r', '\n']); - event_tx_clone - .send(ProviderEvent::terminal_output( - turn_id, - "kohaku", - &line, - Some("merged"), - Some(output_line.transport.as_str()), - )) - .await - .ok(); if protocol_line.is_empty() { continue; } - if let Ok(json) = serde_json::from_str::(protocol_line) { - match classify(&json) { - KohakuEvent::Text(text) if !text.is_empty() => { - assistant_message.push_str(&text); - event_tx_clone - .send(ProviderEvent::text_message(turn_id, "kohaku", &text)) - .await - .ok(); - } - KohakuEvent::TurnEnd { text, .. } => { - if final_text.is_none() && !text.is_empty() { - final_text = Some(text); - } - } - _ => {} - } - // Pass the raw JSON through for the diagnostics drawer and any - // downstream observability (tool/subagent activity, usage). + // `kt --json` stdout is pure machine protocol, one JSON object per + // line. Mirroring it to the terminal channel would replay the whole + // protocol into the live-execution card (runtime-detail flood), so + // only genuine NON-JSON stdout is surfaced as terminal output. + let Ok(json) = serde_json::from_str::(protocol_line) else { event_tx_clone - .send(ProviderEvent::new( + .send(ProviderEvent::terminal_output( turn_id, - EventType::ItemUpdated, "kohaku", - json, - )) - .await - .ok(); - } else if !protocol_line.trim().is_empty() { - // Non-JSON line — surface as text so unexpected output is - // visible when debugging. - event_tx_clone - .send(ProviderEvent::text_message( - turn_id, - "kohaku", - protocol_line, + &line, + Some("merged"), + Some(output_line.transport.as_str()), )) .await .ok(); + continue; + }; + + match classify(&json) { + KohakuEvent::Text(text) if !text.is_empty() => { + assistant_message.push_str(&text); + let revealed = display.push(&text); + if !revealed.is_empty() { + event_tx_clone + .send(ProviderEvent::text_message(turn_id, "kohaku", &revealed)) + .await + .ok(); + } + } + KohakuEvent::TurnEnd { + status, + text, + error, + } => { + if status == "error" { + if turn_error.is_none() { + turn_error = error + .filter(|e| !e.trim().is_empty()) + .or_else(|| (!text.trim().is_empty()).then(|| text.clone())) + .or_else(|| Some("kohaku: turn ended with error".to_string())); + } + } else if assistant_message.is_empty() && !text.is_empty() { + // No `text` deltas arrived — the body was delivered + // consolidated in `turn_end.text`. Stream it now (gated) + // so the bubble fills live instead of only appearing + // after finalize + DB refresh. + assistant_message.push_str(&text); + let revealed = display.push(&text); + if !revealed.is_empty() { + event_tx_clone + .send(ProviderEvent::text_message(turn_id, "kohaku", &revealed)) + .await + .ok(); + } + } + } + KohakuEvent::Error(message) => { + if turn_error.is_none() && !message.is_empty() { + turn_error = Some(message); + } + } + KohakuEvent::Activity { + activity_type, + value, + } => { + // Translate genuine tool/subagent calls into normalized + // command_execution items so they're counted and surfaced in + // the live-execution card (matching codex/claude). Pure KT + // runtime telemetry (processing_*, token_usage, session_info, + // compact_*, …) is backend noise and is dropped so the chat + // shows only the model's message. + if let Some(item) = normalize_tool_activity(&activity_type, &value) { + event_tx_clone + .send(ProviderEvent::new( + turn_id, + EventType::ItemUpdated, + "kohaku", + item, + )) + .await + .ok(); + } + } + // Empty text, turn_start, and unrecognized lines carry no + // user-facing signal — drop them from the chat stream. + KohakuEvent::Text(_) | KohakuEvent::TurnStart | KohakuEvent::Other => {} } } - if assistant_message.is_empty() { - final_text.unwrap_or_default() - } else { - assistant_message + ConsumerResult { + assistant_message, + turn_error, } }); let result = run_subprocess_streaming(&config, &line_tx, cancel).await; drop(line_tx); - let streamed_text = consumer.await.unwrap_or_default(); + let consumed = consumer.await.unwrap_or_default(); let output = match result { Ok(o) => o, Err(e) => return Err(handle_subprocess_error(e, turn_id, "kohaku", event_tx).await), }; + // `emit_completion_event` derives the lifecycle from the exit code — kt + // exits non-zero on a turn error, so a failure already surfaces a + // `turn_failed` event (don't double-emit one here). emit_completion_event(&output, turn_id, "kohaku", event_tx).await; - let response_text = if streamed_text.is_empty() { - output.stdout.trim().to_string() - } else { - streamed_text + // Use the full streamed assistant body (the router strips any sentinel for + // display). On failure, surface the structured error message so the reason + // isn't lost — appended after any partial prose, never replacing the raw + // JSONL protocol dump. + let response_text = match consumed.turn_error { + Some(error) if consumed.assistant_message.trim().is_empty() => error, + Some(error) => format!( + "{}\n\n[KohakuTerrarium 错误] {error}", + consumed.assistant_message.trim_end() + ), + None => consumed.assistant_message, }; Ok(build_turn_result(response_text, &output, "kohaku")) } +/// Accumulated outcome of the JSONL consumer task. +#[derive(Default)] +struct ConsumerResult { + /// Full streamed assistant body (including any routing sentinel block, and + /// any consolidated `turn_end.text` for the no-delta path). + assistant_message: String, + /// Structured failure message (`turn_end` error or a top-level `error`). + turn_error: Option, +} + +/// Gates the chat display channel so a routing sentinel block never streams into +/// the bubble. Deltas are accumulated; each `push` returns only the +/// sentinel-safe text that has not yet been shown. The caller keeps the full +/// (ungated) text separately for routing. +#[derive(Default)] +struct SentinelDisplayFilter { + buf: String, + shown: usize, +} + +impl SentinelDisplayFilter { + fn new() -> Self { + Self::default() + } + + /// Append `delta`; return the newly-revealed sentinel-safe text, if any. + fn push(&mut self, delta: &str) -> String { + self.buf.push_str(delta); + let safe = sentinel_safe_display(&self.buf); + if safe.len() > self.shown { + let revealed = safe[self.shown..].to_string(); + self.shown = safe.len(); + revealed + } else { + String::new() + } + } +} + +/// The portion of `text` that is safe to display now: prose outside any +/// complete sentinel block, withholding a pending (unclosed) block and any +/// trailing fragment that could be the start of a `BEGIN` marker. The returned +/// prefix grows monotonically as more deltas arrive. +fn sentinel_safe_display(text: &str) -> String { + let mut out = String::with_capacity(text.len()); + let mut cursor = 0; + loop { + match text[cursor..].find(SENTINEL_BEGIN) { + Some(rel) => { + let begin = cursor + rel; + out.push_str(&text[cursor..begin]); + let after = begin + SENTINEL_BEGIN.len(); + match text[after..].find(SENTINEL_END) { + Some(rel_end) => cursor = after + rel_end + SENTINEL_END.len(), + // Pending block — withhold everything from the marker on. + None => return out, + } + } + None => { + let tail = &text[cursor..]; + let keep = tail.len() - partial_begin_suffix_len(tail); + out.push_str(&tail[..keep]); + return out; + } + } + } +} + +/// Length of the longest suffix of `tail` that is a proper prefix of the +/// `BEGIN` marker, so a marker split across deltas is withheld until complete. +fn partial_begin_suffix_len(tail: &str) -> usize { + let begin = SENTINEL_BEGIN.as_bytes(); + let bytes = tail.as_bytes(); + let max = (begin.len() - 1).min(bytes.len()); + (1..=max) + .rev() + .find(|&k| bytes[bytes.len() - k..] == begin[..k]) + .unwrap_or(0) +} + +/// Map a KohakuTerrarium `activity` event onto Switchyard's normalized item +/// vocabulary, or `None` if it is backend runtime *telemetry* that should not +/// reach the chat. Genuine tool/subagent calls become `command_execution` +/// items so the live-execution card counts them ("已运行 N 条命令") and shows +/// the running one in its headline ("正在运行 ") — the same surfaces +/// codex/claude drive via their command items. A `*_start` becomes a running +/// item that the frontend merges (by command) with its later `*_done`/`*_error` +/// into one transitioning card. Pure telemetry (processing_*, token_usage, +/// *_token_update, session_info, compact_*, tool_promoted, job_cancelled, +/// interrupt, …) and unknown types are dropped so a vocabulary drift in the KT +/// fork never reintroduces flooding. +fn normalize_tool_activity( + activity_type: &str, + value: &serde_json::Value, +) -> Option { + let status = match activity_type { + "tool_start" | "subagent_tool_start" | "subagent_start" => "running", + "tool_done" | "subagent_tool_done" | "subagent_done" => "completed", + "tool_error" | "subagent_tool_error" | "subagent_error" => "failed", + _ => return None, + }; + let name = value + .get("detail") + .and_then(|v| v.as_str()) + .filter(|s| !s.trim().is_empty()) + .or_else(|| { + value + .get("metadata") + .and_then(|m| m.get("name")) + .and_then(|v| v.as_str()) + }) + .filter(|s| !s.trim().is_empty()) + .unwrap_or("tool"); + Some(serde_json::json!({ + "item_type": "command_execution", + "command": name, + "status": status, + })) +} + /// Map Switchyard's sandbox mode + workspace onto `kt` headless flags. fn kohaku_policy_args(policy: &ExecutionPolicy) -> Vec { let preset = match policy.effective_sandbox_mode() { @@ -243,6 +414,83 @@ mod tests { assert_eq!(kohaku_runtime_args(None, Some("low")), Vec::::new()); } + #[test] + fn sentinel_filter_withholds_block_split_across_deltas() { + let mut filter = SentinelDisplayFilter::new(); + assert_eq!(filter.push("Hello "), "Hello "); + // The BEGIN marker is split across two deltas — nothing must leak. + assert_eq!(filter.push("<<>>{\"type\":\"deleg"), ""); + assert_eq!(filter.push("ate\"}<<>>"), ""); + // Prose after the closed block resumes streaming. + assert_eq!(filter.push(" world"), " world"); + } + + #[test] + fn sentinel_filter_passes_plain_text_incrementally() { + let mut filter = SentinelDisplayFilter::new(); + assert_eq!(filter.push("a"), "a"); + assert_eq!(filter.push("bc"), "bc"); + assert_eq!(filter.push(""), ""); + } + + #[test] + fn sentinel_safe_display_strips_a_complete_block() { + let text = "before <<>>X<<>> after"; + assert_eq!(sentinel_safe_display(text), "before after"); + } + + #[test] + fn sentinel_safe_display_withholds_pending_block() { + let text = "keep <<>>{\"partial\":true"; + assert_eq!(sentinel_safe_display(text), "keep "); + } + + #[test] + fn normalize_tool_activity_surfaces_tools_and_drops_telemetry() { + // Pure telemetry and unknown types are dropped. + assert!(normalize_tool_activity("token_usage", &serde_json::json!({})).is_none()); + assert!(normalize_tool_activity("processing_start", &serde_json::json!({})).is_none()); + assert!(normalize_tool_activity("session_info", &serde_json::json!({})).is_none()); + assert!(normalize_tool_activity("brand_new_event", &serde_json::json!({})).is_none()); + + // A tool start becomes a running command_execution (the frontend merges + // it with the later done/error by command into one transitioning card), + // so the live card counts it and shows it in the headline. + let start = + normalize_tool_activity("tool_start", &serde_json::json!({"detail": "read"})).unwrap(); + assert_eq!( + start.get("item_type").and_then(|v| v.as_str()), + Some("command_execution") + ); + assert_eq!(start.get("command").and_then(|v| v.as_str()), Some("read")); + assert_eq!( + start.get("status").and_then(|v| v.as_str()), + Some("running") + ); + + let done = + normalize_tool_activity("tool_done", &serde_json::json!({"detail": "read"})).unwrap(); + assert_eq!( + done.get("status").and_then(|v| v.as_str()), + Some("completed") + ); + + let errored = normalize_tool_activity( + "tool_error", + &serde_json::json!({"metadata": {"name": "write"}}), + ) + .unwrap(); + assert_eq!( + errored.get("command").and_then(|v| v.as_str()), + Some("write") + ); + assert_eq!( + errored.get("status").and_then(|v| v.as_str()), + Some("failed") + ); + } + #[test] fn policy_args_map_sandbox_modes() { let ro = kohaku_policy_args(&ExecutionPolicy::read_only("/repo")); diff --git a/tests/bin/fake_kt.rs b/tests/bin/fake_kt.rs index cbcf3fd..b716ec6 100644 --- a/tests/bin/fake_kt.rs +++ b/tests/bin/fake_kt.rs @@ -7,6 +7,9 @@ //! //! Behaviour switches: //! - `FAKE_KT_FAIL=1` → emit a failed `turn_end` and exit 1. +//! - `FAKE_KT_SENTINEL=1` → emit a SWITCHYARD sentinel block split across two +//! `text` deltas (marker broken mid-string) to exercise the adapter's +//! display gating. use std::io::Write; @@ -54,6 +57,22 @@ fn main() { let echo = format!("echo: {prompt}"); emit(&mut out, serde_json::json!({"type":"text","content": echo})); + + // Optionally stream a routing sentinel block fragmented across two `text` + // deltas (the BEGIN marker is split mid-string). The adapter must withhold + // the whole block from the chat display while still accumulating it for the + // router. + if std::env::var("FAKE_KT_SENTINEL").is_ok() { + emit( + &mut out, + serde_json::json!({"type":"text","content":" Plan: <<>>{\"type\":\"note\"}<<>> done"}), + ); + } + emit( &mut out, serde_json::json!({"type":"activity","activity_type":"tool_start","detail":"read","metadata":{"name":"read","args":{"path":"x"}}}), diff --git a/tests/tests/kohaku_integration.rs b/tests/tests/kohaku_integration.rs index 58a19cd..5d1eb52 100644 --- a/tests/tests/kohaku_integration.rs +++ b/tests/tests/kohaku_integration.rs @@ -99,6 +99,84 @@ async fn kohaku_headless_turn_maps_jsonl_to_events() { assert!(!bundle.artifacts.is_empty(), "should archive raw output"); } +#[tokio::test] +async fn kohaku_keeps_chat_clean_of_protocol_and_sentinel() { + // The fake emits an activity stream, a fragmented SWITCHYARD sentinel block, + // and the usual text — the chat path must show only the model's prose: no + // terminal mirror of the JSONL protocol, and no sentinel leakage. + let mut env = HashMap::new(); + env.insert("FAKE_KT_SENTINEL".to_string(), "1".to_string()); + let provider = KohakuProvider::new(fake_kt(), vec!["@fake/creature".to_string()], env, 30); + let turn_id = uuid::Uuid::now_v7(); + let (tx, mut rx) = tokio::sync::mpsc::channel(256); + + provider + .start_turn( + turn_id, + input("ping"), + policy(), + context(), + tx, + CancellationToken::new(), + ) + .await + .expect("start_turn ok"); + + let mut display = String::new(); + let mut terminal_outputs = 0; + let mut execution_telemetry = 0; + let mut command_executions: Vec = vec![]; + while let Some(e) = rx.recv().await { + match e.payload.get("item_type").and_then(|v| v.as_str()) { + Some("agent_message") => { + if let Some(t) = e.payload.get("text").and_then(|v| v.as_str()) { + display.push_str(t); + } + } + Some("terminal_output") => terminal_outputs += 1, + Some("execution_telemetry") => execution_telemetry += 1, + Some("command_execution") => { + if let Some(c) = e.payload.get("command").and_then(|v| v.as_str()) { + command_executions.push(c.to_string()); + } + } + _ => {} + } + } + + assert_eq!( + terminal_outputs, 0, + "kt --json protocol lines must not be mirrored as terminal output" + ); + assert_eq!( + execution_telemetry, 0, + "the kt.exe driver must not be surfaced as execution_telemetry (it would \ + headline the live card as '正在运行 kt.exe' and inflate the command count)" + ); + // The fake runs a `read` tool — it must reach the live card as a counted + // command_execution (start->done), not vanish. + assert!( + command_executions.iter().any(|c| c == "read"), + "kt tool activity should surface as command_execution items, got: {command_executions:?}" + ); + assert!( + !display.contains("SWITCHYARD_JSON"), + "sentinel must never leak into the chat display, got: {display:?}" + ); + // The surrounding prose still streams (block is withheld, not the text). + assert!(display.contains("echo: ping"), "got: {display:?}"); + assert!(display.contains("Plan:"), "got: {display:?}"); + assert!(display.contains("done"), "got: {display:?}"); + + // The router still sees the full body (block included) for delegation. + let (result, _) = provider.finalize_turn(turn_id).await.expect("finalize"); + assert!( + result.response_text.contains("SWITCHYARD_JSON"), + "response_text (routing input) keeps the sentinel: {:?}", + result.response_text + ); +} + #[tokio::test] async fn kohaku_headless_turn_failure_propagates() { let mut env = HashMap::new();