From cc7fc115f822936c6e9a30a7b39e3daa7fd34769 Mon Sep 17 00:00:00 2001 From: Yidhar <80593214+Yidhar@users.noreply.github.com> Date: Sat, 27 Jun 2026 15:28:32 +0800 Subject: [PATCH 1/3] fix(kohaku): show only LLM messages, not kt runtime noise When kt is the core, the chat flooded with backend runtime detail and the delegation sentinel leaked into the live bubble. Root causes (all inside the kohaku adapter, per the provider-differences-stay-in-the-adapter invariant): - The consumer mirrored every `kt --json` protocol line as terminal output, replaying the whole machine stream into the live-execution card. Only genuine non-JSON stdout is now surfaced; protocol lines are parsed-only. - kt's native vocabulary (text/activity/turn_start/turn_end/error) was unknown to the shared recognizers, so activity/turn_start rendered as generic [activity] markers. classify() now recognizes the full vocabulary; turn.rs drops telemetry + turn_start and translates genuine tool/subagent calls into normalized (collapsed) tool_call cards. - The sentinel streamed token-by-token, leaking fragments into the bubble. A stateful SentinelDisplayFilter withholds any BEGIN..END block (and partial markers split across deltas) from the display channel while the full body still reaches the router for delegation. - Structured turn errors are surfaced as the response reason instead of a raw JSONL dump; no-delta turn_end.text bodies now stream live. Frontend: renderMessageBody strips sentinel blocks defensively (complete + unclosed-trailing) regardless of delegate parsing. The shared strip_sentinel_blocks stays conservative (no cross-provider behavior change). Tests: fake_kt gained a split-sentinel path; a new integration test asserts no protocol mirroring and no sentinel leakage into the chat. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/components/ui/RenderHelpers.tsx | 9 + .../switchyard-provider-api/src/sentinel.rs | 12 +- .../switchyard-provider-kohaku/src/jsonl.rs | 70 +++- crates/switchyard-provider-kohaku/src/turn.rs | 354 +++++++++++++++--- tests/bin/fake_kt.rs | 19 + tests/tests/kohaku_integration.rs | 59 +++ 6 files changed, 455 insertions(+), 68 deletions(-) diff --git a/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx b/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx index 562d71f..f1c24e7 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 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..171d3ad 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, @@ -102,95 +103,265 @@ pub async fn run_kohaku_turn( 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 the shared + // normalized vocabulary so they render as clean, collapsed + // tool cards (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. Only genuine tool/subagent calls are surfaced (as +/// collapsed `tool_call` cards, matching codex/claude); a `*_start` becomes a +/// running card that the frontend merges with its later `*_done`/`*_error` (by +/// name) 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": "tool_call", + "name": 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,79 @@ 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 card (the frontend merges it with the + // later done/error by name into one transitioning card). + 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("tool_call") + ); + assert_eq!(start.get("name").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("name").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..dfc9fb5 100644 --- a/tests/tests/kohaku_integration.rs +++ b/tests/tests/kohaku_integration.rs @@ -99,6 +99,65 @@ 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; + 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, + _ => {} + } + } + + assert_eq!( + terminal_outputs, 0, + "kt --json protocol lines must not be mirrored as terminal output" + ); + 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(); From 04f8e1ec1e58d738d8827ff84fc80f23b6fa6d08 Mon Sep 17 00:00:00 2001 From: Yidhar <80593214+Yidhar@users.noreply.github.com> Date: Sat, 27 Jun 2026 15:41:37 +0800 Subject: [PATCH 2/3] fix(kohaku): count kt tool activity as commands, drop kt.exe driver headline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The live-execution card showed "正在运行 kt.exe" and counted the driver as a command, while the real kt tool activity wasn't counted. Both stemmed from the kt.exe execution_telemetry feeding the frontend's `commandLine` (which drives both the running headline and `commandCount`): - Stop emitting execution_telemetry for kohaku — kt.exe is Switchyard's own driver, not a model action. The resolved invocation is still recorded in the archived raw-output artifact for replay/diagnostics. - Map genuine kt tool/subagent activity to command_execution items (was tool_call) so the live card counts them ("已运行 N 条命令") and shows the running one in its headline ("正在运行 ") — the same surfaces codex/claude drive via their command items. start->done/error merge (by command) into one transitioning card. The integration test now asserts kohaku emits no execution_telemetry and that tool activity surfaces as counted command_execution items. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/switchyard-provider-kohaku/src/turn.rs | 56 ++++++++++--------- tests/tests/kohaku_integration.rs | 19 +++++++ 2 files changed, 49 insertions(+), 26 deletions(-) diff --git a/crates/switchyard-provider-kohaku/src/turn.rs b/crates/switchyard-provider-kohaku/src/turn.rs index 171d3ad..717325e 100644 --- a/crates/switchyard-provider-kohaku/src/turn.rs +++ b/crates/switchyard-provider-kohaku/src/turn.rs @@ -92,14 +92,12 @@ 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 { @@ -185,10 +183,10 @@ pub async fn run_kohaku_turn( activity_type, value, } => { - // Translate genuine tool/subagent calls into the shared - // normalized vocabulary so they render as clean, collapsed - // tool cards (matching codex/claude). Pure KT runtime - // telemetry (processing_*, token_usage, session_info, + // 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) { @@ -326,13 +324,15 @@ fn partial_begin_suffix_len(tail: &str) -> usize { /// 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. Only genuine tool/subagent calls are surfaced (as -/// collapsed `tool_call` cards, matching codex/claude); a `*_start` becomes a -/// running card that the frontend merges with its later `*_done`/`*_error` (by -/// name) 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. +/// 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, @@ -356,8 +356,8 @@ fn normalize_tool_activity( .filter(|s| !s.trim().is_empty()) .unwrap_or("tool"); Some(serde_json::json!({ - "item_type": "tool_call", - "name": name, + "item_type": "command_execution", + "command": name, "status": status, })) } @@ -454,15 +454,16 @@ mod tests { 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 card (the frontend merges it with the - // later done/error by name into one transitioning card). + // 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("tool_call") + Some("command_execution") ); - assert_eq!(start.get("name").and_then(|v| v.as_str()), Some("read")); + 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") @@ -480,7 +481,10 @@ mod tests { &serde_json::json!({"metadata": {"name": "write"}}), ) .unwrap(); - assert_eq!(errored.get("name").and_then(|v| v.as_str()), Some("write")); + 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") diff --git a/tests/tests/kohaku_integration.rs b/tests/tests/kohaku_integration.rs index dfc9fb5..5d1eb52 100644 --- a/tests/tests/kohaku_integration.rs +++ b/tests/tests/kohaku_integration.rs @@ -124,6 +124,8 @@ async fn kohaku_keeps_chat_clean_of_protocol_and_sentinel() { 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") => { @@ -132,6 +134,12 @@ async fn kohaku_keeps_chat_clean_of_protocol_and_sentinel() { } } 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()); + } + } _ => {} } } @@ -140,6 +148,17 @@ async fn kohaku_keeps_chat_clean_of_protocol_and_sentinel() { 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:?}" From ed8caee84d418166eb277ec6f98de346bd415a3d Mon Sep 17 00:00:00 2001 From: Yidhar <80593214+Yidhar@users.noreply.github.com> Date: Sat, 27 Jun 2026 20:01:00 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(gui):=20reflect=20completed=20commands?= =?UTF-8?q?=20on=20the=20live=20card,=20not=20"=E6=9A=82=E6=97=A0=E5=B7=A5?= =?UTF-8?q?=E5=85=B7=E4=BA=8B=E4=BB=B6"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After kohaku stopped emitting the kt.exe execution_telemetry, a turn whose tools had all completed (none currently running, and no launcher commandLine to fall back on) fell through to the "Provider 正在运行，暂无工具事件…" status even while `commandCount` showed commands had run — a confusing contradiction. The live-execution card now falls back to the most recent command that ran ("已运行 ") when none is currently active, keeping the headline consistent with the count. This also improves codex/claude (the last real command shows between batches instead of the launcher invocation). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../frontend/src/components/ui/RenderHelpers.tsx | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx b/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx index f1c24e7..145beca 100644 --- a/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx +++ b/crates/switchyard-gui/frontend/src/components/ui/RenderHelpers.tsx @@ -2606,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