diff --git a/Cargo.lock b/Cargo.lock index 2c105351f0..672855cd2a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -910,7 +910,10 @@ dependencies = [ "reqwest 0.13.4", "serde", "serde_json", + "tempfile", "tokio", + "tracing", + "wiremock", ] [[package]] @@ -1052,6 +1055,7 @@ dependencies = [ "codewhale-config", "codewhale-core", "codewhale-execpolicy", + "codewhale-hooks", "codewhale-lane", "codewhale-paths", "codewhale-protocol", diff --git a/config.example.toml b/config.example.toml index 0446dcdf49..71fcc40a5f 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1236,6 +1236,19 @@ default_text_model = "deepseek-ai/deepseek-v4-pro" # completion_sound = "beep" # sound_file = "E:\\google\\downloads\\notify.wav" +# Lifecycle event outbox: opt-in JSONL stream of session/turn/subagent +# lifecycle events for supervisors and automation harnesses. One JSON line +# per event (RuntimeEventEnvelope schema), appended and flushed on every +# emit; seq is monotonic per file and recovers from the last line on open. +# UNCOMMENT `path` TO ENABLE — unset/empty = OFF = behavior unchanged. +# Fires for interactive TUI sessions AND headless `codewhale exec` runs. +# See docs/CONFIGURATION.md → Lifecycle Outbox for the file contract. +# [lifecycle_outbox] +# path = "~/.codewhale/notifications/outbox.jsonl" +# webhook_url = "https://example.com/hooks/codewhale" # optional: POST {"at", "event"} JSON per event +# webhook_token = "" # optional: sent as `Authorization: Bearer ` +# # delivery is best-effort: failures are logged and dropped + # Opt-in per-event sound cues (#4817): deterministic, terminal-bell level # (BEL bytes only — functional signals, platform-safe no-op when the terminal # ignores BEL). Off by default. When completion_sound is active, turn-complete diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index 460e0f0669..6679eb6409 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -853,6 +853,10 @@ pub struct ConfigToml { /// lifecycle `[hooks]` table so config rewrites preserve existing hooks. #[serde(default)] pub hook_sinks: Option, + /// Lifecycle event outbox (`[lifecycle_outbox]`). Opt-in: an unset or + /// empty `path` disables the feature and leaves behavior unchanged. + #[serde(default)] + pub lifecycle_outbox: Option, /// Agent Fleet trust and security policy (#3165). When absent, fleet /// workers inherit conservative Sandbox defaults. #[serde(default)] @@ -1553,6 +1557,31 @@ pub struct HookSinksToml { pub unix_socket_path: Option, } +/// On-disk schema for the `[lifecycle_outbox]` table. +/// +/// Opt-in lifecycle event outbox: every emitted event is appended as one +/// JSONL line to `path` in the `RuntimeEventEnvelope` shape +/// (`schema_version, seq, event, kind, thread_id, turn_id, item_id, +/// timestamp, payload`), and optionally POSTed to `webhook_url`. An unset or +/// empty `path` disables the feature entirely — behavior is unchanged from a +/// release without the table. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct LifecycleOutboxToml { + /// Path to the JSONL outbox file. Parent directories are created lazily + /// on the first event. Unset or empty = feature OFF. + #[serde(default)] + pub path: Option, + /// Optional webhook URL. Events are POSTed as `{"at", "event"}` JSON + /// only when this is set (in addition to, never instead of, `path`). + /// Delivery is best-effort: failures are logged and dropped. + #[serde(default)] + pub webhook_url: Option, + /// Optional bearer token sent as `Authorization: Bearer ` on + /// webhook POSTs. Ignored when `webhook_url` is unset. + #[serde(default)] + pub webhook_token: Option, +} + /// On-disk schema for the `[skills]` table (#140). See `config.example.toml` /// for documentation. #[derive(Debug, Clone, Serialize, Deserialize, Default)] diff --git a/crates/config/src/tests.rs b/crates/config/src/tests.rs index a3ce497ac7..71913c6701 100644 --- a/crates/config/src/tests.rs +++ b/crates/config/src/tests.rs @@ -59,6 +59,78 @@ fn verifier_config_rejects_unknown_verdict_policy() { ); } +#[test] +fn lifecycle_outbox_toml_is_off_by_default_and_parses_when_configured() { + // Unset = feature OFF: the table is absent and the field is None. + let absent: ConfigToml = toml::from_str("model = \"demo\"\n").expect("minimal config"); + assert!( + absent.lifecycle_outbox.is_none(), + "unset [lifecycle_outbox] must leave the feature off" + ); + + // An empty table is also off: no path means no outbox file. + let empty: ConfigToml = + toml::from_str("[lifecycle_outbox]\n").expect("empty lifecycle_outbox table"); + let outbox = empty.lifecycle_outbox.expect("table should parse"); + assert!(outbox.path.is_none()); + assert!(outbox.webhook_url.is_none()); + assert!(outbox.webhook_token.is_none()); + + // Full configuration: path plus optional webhook url and token. + let full: ConfigToml = toml::from_str( + r#" + [lifecycle_outbox] + path = "~/.codewhale/notifications/outbox.jsonl" + webhook_url = "https://example.com/hooks/codewhale" + webhook_token = "secret-token" + "#, + ) + .expect("full lifecycle_outbox table"); + let outbox = full.lifecycle_outbox.expect("table should parse"); + assert_eq!( + outbox.path, + Some(PathBuf::from("~/.codewhale/notifications/outbox.jsonl")) + ); + assert_eq!( + outbox.webhook_url.as_deref(), + Some("https://example.com/hooks/codewhale") + ); + assert_eq!(outbox.webhook_token.as_deref(), Some("secret-token")); +} + +#[test] +fn lifecycle_outbox_toml_webhook_is_optional() { + // `path` alone enables the file outbox without any webhook. + let file_only: ConfigToml = toml::from_str( + r#" + [lifecycle_outbox] + path = "/tmp/outbox.jsonl" + "#, + ) + .expect("file-only lifecycle_outbox table"); + let outbox = file_only.lifecycle_outbox.expect("table should parse"); + assert_eq!(outbox.path, Some(PathBuf::from("/tmp/outbox.jsonl"))); + assert!(outbox.webhook_url.is_none()); + + // A webhook url without a path does not enable a file outbox; the + // consumer decides whether webhook-only delivery is meaningful, but the + // parse must stay lossless either way. + let webhook_only: ConfigToml = toml::from_str( + r#" + [lifecycle_outbox] + webhook_url = "https://example.com/hooks/codewhale" + "#, + ) + .expect("webhook-only lifecycle_outbox table"); + let outbox = webhook_only.lifecycle_outbox.expect("table should parse"); + assert!(outbox.path.is_none()); + assert_eq!( + outbox.webhook_url.as_deref(), + Some("https://example.com/hooks/codewhale") + ); + assert!(outbox.webhook_token.is_none()); +} + #[test] fn permissions_toml_deserializes_typed_ask_rules() { let permissions: PermissionsToml = toml::from_str( diff --git a/crates/hooks/Cargo.toml b/crates/hooks/Cargo.toml index 8ea246a4b3..6f03e5cd55 100644 --- a/crates/hooks/Cargo.toml +++ b/crates/hooks/Cargo.toml @@ -20,3 +20,8 @@ reqwest.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true +tracing.workspace = true + +[dev-dependencies] +tempfile.workspace = true +wiremock = "0.6" diff --git a/crates/hooks/src/lib.rs b/crates/hooks/src/lib.rs index 6d4e674402..21d5820c85 100644 --- a/crates/hooks/src/lib.rs +++ b/crates/hooks/src/lib.rs @@ -9,6 +9,13 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use tokio::io::AsyncWriteExt; +mod lifecycle_outbox; + +pub use lifecycle_outbox::{ + LifecycleEvent, LifecycleOutbox, OUTBOX_DETAIL_MAX_CHARS, OUTBOX_HEADLINE_MAX_CHARS, + OUTBOX_PREVIEW_MAX_CHARS, OUTBOX_TRUNCATION_MARKER, bounded_text, +}; + /// All events that can be emitted through the hook system. /// /// Each variant represents a distinct lifecycle or streaming event. The enum is @@ -196,16 +203,26 @@ impl HookSink for JsonlHookSink { /// The request body is `{"at": "", "event": {...}}`. /// Failed requests are retried up to 2 times with exponential back-off /// (200 ms, 400 ms). After exhausting retries the error is propagated. +#[derive(Clone)] pub struct WebhookHookSink { url: String, + /// Optional bearer token sent as `Authorization: Bearer `. + bearer_token: Option, client: reqwest::Client, } impl WebhookHookSink { /// Create a new sink that sends events to the given `url`. pub fn new(url: String) -> Self { + Self::new_with_token(url, None) + } + + /// Create a new sink that sends events to the given `url`, attaching + /// `Authorization: Bearer ` when a token is provided. + pub fn new_with_token(url: String, bearer_token: Option) -> Self { Self { url, + bearer_token, client: codewhale_release::platform_http_client_builder() .timeout(std::time::Duration::from_secs(10)) .build() @@ -216,22 +233,21 @@ impl WebhookHookSink { }), } } -} -#[async_trait] -impl HookSink for WebhookHookSink { - async fn emit(&self, event: &HookEvent) -> Result<()> { + /// POST an arbitrary JSON payload to the configured endpoint. + /// + /// This is the shared delivery path behind both [`HookSink::emit`] and + /// the lifecycle outbox fan-out. It is deliberately not part of the + /// [`HookSink`] trait: outbox events are runtime event envelopes, not + /// [`HookEvent`]s, and only the transport needs to be shared. + pub async fn post_payload(&self, payload: serde_json::Value) -> Result<()> { let mut retries = 0usize; loop { - let resp = self - .client - .post(&self.url) - .json(&json!({ - "at": Utc::now().to_rfc3339(), - "event": event, - })) - .send() - .await; + let mut request = self.client.post(&self.url).json(&payload); + if let Some(token) = self.bearer_token.as_deref().filter(|t| !t.is_empty()) { + request = request.bearer_auth(token); + } + let resp = request.send().await; match resp { Ok(response) if response.status().is_success() => return Ok(()), Ok(response) => { @@ -251,6 +267,17 @@ impl HookSink for WebhookHookSink { } } +#[async_trait] +impl HookSink for WebhookHookSink { + async fn emit(&self, event: &HookEvent) -> Result<()> { + self.post_payload(json!({ + "at": Utc::now().to_rfc3339(), + "event": event, + })) + .await + } +} + /// A [`HookSink`] that sends events over a Unix domain socket. /// /// Each event is serialized as a single JSON line (`{"at": "...", "event": {...}}\n`) diff --git a/crates/hooks/src/lifecycle_outbox.rs b/crates/hooks/src/lifecycle_outbox.rs new file mode 100644 index 0000000000..a761b78d0a --- /dev/null +++ b/crates/hooks/src/lifecycle_outbox.rs @@ -0,0 +1,802 @@ +//! Lifecycle event outbox: a local JSONL log of session/turn/subagent +//! lifecycle events plus an optional webhook fan-out. +//! +//! This is the machine-readable sibling of the TUI shell-hook system. Hooks +//! fire shell commands per event and are TUI-only; the outbox appends one +//! JSON line per event to a config-gated file and needs no per-event +//! configuration. It is additive and opt-in: with no path configured, +//! [`LifecycleOutbox::emit`] is a no-op. +//! +//! # Line schema +//! +//! Every line is a `codewhale_protocol::runtime::RuntimeEventEnvelope`: +//! +//! ```json +//! {"schema_version": 1, "seq": 3, "event": "turn_start", "kind": "turn.started", +//! "thread_id": "…", "turn_id": "…", "item_id": null, "timestamp": "…", +//! "created_at": "…", "payload": {…}} +//! ``` +//! +//! - `seq` is monotonic per outbox file. On the first write the writer +//! recovers the `seq` of the file's last complete line (bounded tail scan, +// so an outbox that grows unbounded is never re-read in full) and continues +//! from `last + 1`. +//! - `event` is the snake-case lifecycle name (`turn_start`, `turn_end`, …); +//! `kind` is the dotted kind (`turn.started`, `turn.failed`, …). +//! - Payloads are constructed by the emit sites from bounded, pre-redacted +//! fields only — never raw tool arguments, environment, or full transcript +//! text. [`bounded_text`] enforces the same ceilings as the desktop +//! notification payloads: headline ≤ 80, detail ≤ 120, preview ≤ 200 +//! characters. +//! +//! # Delivery model +//! +//! [`LifecycleOutbox::emit`] never blocks the caller: it enqueues the event +//! on an internal channel and a single writer task appends lines in order. +//! If no tokio runtime is available the event is dropped with a warning. +//! Webhook POSTs (`{"at": …, "event": …}`) are attempted after the local +//! append; failures are logged and dropped, never retried into the agent +//! loop. + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use anyhow::{Context, Result}; +use chrono::Utc; +use codewhale_protocol::runtime::{RUNTIME_EVENT_ENVELOPE_SCHEMA_VERSION, RuntimeEventEnvelope}; +use serde_json::{Value, json}; +use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; +use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender}; + +use crate::WebhookHookSink; + +/// Text-length ceilings for outbox payload fields. Mirrors the desktop +/// notification payload limits so the outbox never carries more than the +/// lock-screen-capable surface already does. +pub const OUTBOX_HEADLINE_MAX_CHARS: usize = 80; +pub const OUTBOX_DETAIL_MAX_CHARS: usize = 120; +pub const OUTBOX_PREVIEW_MAX_CHARS: usize = 200; + +/// Suffix appended when [`bounded_text`] truncates a field. +pub const OUTBOX_TRUNCATION_MARKER: &str = "…"; + +/// How far back from EOF the seq-recovery scan reads. Outbox lines are +/// bounded (payload ceilings above plus envelope overhead), so a line can +/// never approach this window and the last complete line is always inside it. +const SEQ_RECOVERY_TAIL_BYTES: u64 = 64 * 1024; + +/// One lifecycle event destined for the outbox. +/// +/// Construct one per emit site. `payload` must only contain bounded, +/// pre-redacted fields; apply [`bounded_text`] to anything free-form (error +/// messages, previews) before inserting it. +#[derive(Debug, Clone)] +pub struct LifecycleEvent { + /// Snake-case event name, e.g. `"turn_start"`. + pub event: String, + /// Dotted event kind, e.g. `"turn.started"` or `"turn.failed"`. + pub kind: String, + /// Owning session/thread id. Empty when the producer has none. + pub thread_id: String, + /// Current turn id, when known. + pub turn_id: Option, + /// Current item id, when known. + pub item_id: Option, + /// Bounded, redacted event payload. + pub payload: Value, +} + +/// The lifecycle outbox handle. +/// +/// Cheap to clone (an `Arc`). When constructed without a path the outbox is +/// disabled and every `emit` is a no-op. +#[derive(Clone)] +pub struct LifecycleOutbox { + inner: Option>, +} + +impl Default for LifecycleOutbox { + fn default() -> Self { + Self::disabled() + } +} + +impl LifecycleOutbox { + /// Create an outbox writing to `path` when set and non-empty. + /// + /// `webhook_url` optionally adds a webhook fan-out (POST `{"at", "event"}`, + /// best-effort); `webhook_token` is its optional bearer token. Webhook + /// delivery is configured independently of the file: it only ever runs + /// when `webhook_url` is set, and it never replaces the local append. + pub fn new( + path: Option, + webhook_url: Option, + webhook_token: Option, + ) -> Self { + let path = match path { + Some(path) if !path.as_os_str().is_empty() => path, + _ => return Self::disabled(), + }; + let webhook = webhook_url + .as_deref() + .map(str::trim) + .filter(|url| !url.is_empty()) + .map(|url| WebhookHookSink::new_with_token(url.to_string(), webhook_token)); + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + Self { + inner: Some(Arc::new(OutboxInner { + path, + webhook, + sender, + receiver: Mutex::new(Some(receiver)), + writer_spawned: AtomicBool::new(false), + spawn_lock: Mutex::new(()), + })), + } + } + + /// A disabled outbox that drops every event. + pub fn disabled() -> Self { + Self { inner: None } + } + + /// True when a path was configured and events will be written. + pub fn is_enabled(&self) -> bool { + self.inner.is_some() + } + + /// Emit one lifecycle event. + /// + /// Never blocks: the event is queued for the outbox's writer task (spawned + /// lazily on the current tokio runtime on first use). Events queued with + /// no runtime available — or after the writer task is gone — are dropped + /// with a warning. Delivery failures inside the writer are logged and + /// dropped as well; the outbox is observability, not control flow. + pub fn emit(&self, event: LifecycleEvent) { + let Some(inner) = self.inner.clone() else { + return; + }; + if let Err(error) = inner.enqueue(event) { + tracing::warn!(target: "lifecycle_outbox", %error, "lifecycle event dropped"); + } + } +} + +struct OutboxInner { + path: PathBuf, + webhook: Option, + sender: UnboundedSender, + /// The writer task's receive half. Taken exactly once by the writer task. + receiver: Mutex>>, + writer_spawned: AtomicBool, + /// Serializes the lazy writer-task spawn so two racing first emits cannot + /// start two writers. + spawn_lock: Mutex<()>, +} + +impl OutboxInner { + /// Queue an event and make sure the writer task exists to drain it. + /// + /// Ordering: `send` happens before the spawn so events queued before the + /// writer starts are drained first, preserving enqueue order. + fn enqueue(self: &Arc, event: LifecycleEvent) -> Result<()> { + self.sender + .send(event) + .map_err(|_| anyhow::anyhow!("lifecycle outbox writer task is gone"))?; + self.ensure_writer_spawned(); + Ok(()) + } + + fn ensure_writer_spawned(self: &Arc) { + if self.writer_spawned.load(Ordering::Acquire) { + return; + } + let _guard = self + .spawn_lock + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if self.writer_spawned.load(Ordering::Acquire) { + return; + } + let Ok(handle) = tokio::runtime::Handle::try_current() else { + tracing::warn!( + target: "lifecycle_outbox", + "no tokio runtime available; lifecycle events are queued but will not be written" + ); + return; + }; + let receiver = self + .receiver + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + let Some(receiver) = receiver else { + return; + }; + let mut state = WriterState { + path: self.path.clone(), + webhook: self.webhook.clone(), + next_seq: 0, + recovered: false, + receiver, + }; + self.writer_spawned.store(true, Ordering::Release); + handle.spawn(async move { + state.run().await; + }); + } +} + +/// The outbox writer: owns the file state and the event queue drain loop. +struct WriterState { + path: PathBuf, + webhook: Option, + /// Next seq to assign; filled in by [`Self::recover_seq`] on first use. + next_seq: u64, + recovered: bool, + receiver: UnboundedReceiver, +} + +impl WriterState { + /// Drain the queue until every sender is dropped, then exit. + async fn run(&mut self) { + while let Some(event) = self.receiver.recv().await { + if let Err(error) = self.deliver(event).await { + tracing::warn!( + target: "lifecycle_outbox", + %error, + path = %self.path.display(), + "lifecycle outbox write failed" + ); + } + } + } + + /// Assign a seq, build the envelope, append it to the outbox file, then + /// fan out to the webhook (independently of the append result). + async fn deliver(&mut self, event: LifecycleEvent) -> Result<()> { + if !self.recovered { + self.next_seq = recover_last_seq(&self.path).await?; + self.recovered = true; + } + let seq = self.next_seq; + self.next_seq = self.next_seq.saturating_add(1); + + let envelope = RuntimeEventEnvelope { + schema_version: RUNTIME_EVENT_ENVELOPE_SCHEMA_VERSION, + seq, + event: event.event, + kind: event.kind, + thread_id: event.thread_id, + turn_id: event.turn_id, + item_id: event.item_id, + timestamp: Utc::now().to_rfc3339(), + created_at: Some(Utc::now().to_rfc3339()), + payload: event.payload, + extra: Default::default(), + }; + let line = serde_json::to_string(&envelope).context("failed to encode outbox event")?; + + let append_result = self.append_line(&line).await; + + if let Some(webhook) = &self.webhook { + let payload = json!({ + "at": envelope.timestamp, + "event": envelope, + }); + if let Err(error) = webhook.post_payload(payload).await { + tracing::warn!( + target: "lifecycle_outbox", + %error, + "lifecycle webhook delivery failed (dropped)" + ); + } + } + + append_result + } + + /// Append one complete JSONL line, mirroring [`crate::JsonlHookSink`]: + /// lazy parent directories, append mode, flush before returning. The + /// writer task is the only appender for this outbox, so no extra lock is + /// needed here; the queue already serializes. + async fn append_line(&mut self, line: &str) -> Result<()> { + if let Some(parent) = self.path.parent() { + tokio::fs::create_dir_all(parent).await.with_context(|| { + format!("failed to create outbox directory {}", parent.display()) + })?; + } + let mut file = tokio::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.path) + .await + .with_context(|| format!("failed to open outbox {}", self.path.display()))?; + // Line + newline in a single `write_all`: with O_APPEND each `write` + // lands contiguously, so even a second process appending to the same + // file can interleave lines but can never splice one mid-line. + let mut record = Vec::with_capacity(line.len() + 1); + record.extend_from_slice(line.as_bytes()); + record.push(b'\n'); + file.write_all(&record) + .await + .context("failed to write outbox event")?; + file.flush().await.context("failed to flush outbox event") + } +} + +/// Recover the seq to continue from: the `seq` of the outbox file's last +/// complete line, plus 1 — or 1 for a missing/empty file. +/// +/// Only the tail of the file is read (bounded by [`SEQ_RECOVERY_TAIL_BYTES`]); +/// outbox lines are bounded far below that window, so the last complete line +/// is always within it. A partial trailing line from a crash mid-write is +/// ignored (the previous newline-terminated line wins). +async fn recover_last_seq(path: &Path) -> Result { + let mut file = match tokio::fs::File::open(path).await { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(1), + Err(error) => { + return Err(error).with_context(|| format!("failed to open outbox {}", path.display())); + } + }; + let len = file + .metadata() + .await + .with_context(|| format!("failed to stat outbox {}", path.display()))? + .len(); + if len == 0 { + return Ok(1); + } + let start = len.saturating_sub(SEQ_RECOVERY_TAIL_BYTES); + file.seek(std::io::SeekFrom::Start(start)).await?; + let mut tail = vec![0u8; (len - start) as usize]; + file.read_exact(&mut tail).await?; + + let line = match tail.iter().rposition(|byte| *byte == b'\n') { + // The bytes after the final newline are a torn trailing line from a + // crash mid-write; drop them. What remains ends at a newline, so the + // last complete line is the bytes after the previous newline. + Some(last_nl) => { + let body = &tail[..last_nl]; + match body.iter().rposition(|byte| *byte == b'\n') { + Some(idx) => &body[idx + 1..], + None => body, + } + } + // No newline at all: no complete line inside this tail (a line can + // only exceed the tail window by violating the bounded-line + // invariant). Treat the file as not-yet-writable. + None => return Ok(1), + }; + let line = std::str::from_utf8(line).context("outbox tail is not UTF-8")?; + if line.trim().is_empty() { + return Ok(1); + } + let envelope: RuntimeEventEnvelope = + serde_json::from_str(line).context("failed to parse last outbox line")?; + Ok(envelope.seq.saturating_add(1)) +} + +/// Bound free-form text to at most `max_chars` characters, stripping control +/// bytes and ANSI escape sequences and collapsing whitespace runs first. +/// +/// The limit counts Unicode scalar values, not bytes, so multi-byte text gets +/// the same ceiling as ASCII. The result is safe to embed in an outbox +/// payload. Callers remain responsible for only ever passing non-secret +/// fields (error messages, previews, model/provider labels — never raw tool +/// arguments, environment, or full transcript text), the same discipline the +/// desktop notification payloads enforce. +pub fn bounded_text(text: &str, max_chars: usize) -> String { + let cleaned: String = text + .chars() + .filter(|ch| !ch.is_control()) + .collect::() + .split_whitespace() + .collect::>() + .join(" "); + let mut truncated = false; + let mut out = String::new(); + let mut char_count = 0usize; + for ch in cleaned.chars() { + if char_count + 1 > max_chars { + truncated = true; + break; + } + out.push(ch); + char_count += 1; + } + if truncated { + // Make room for the marker while staying under the character ceiling. + let marker_chars = OUTBOX_TRUNCATION_MARKER.chars().count(); + while char_count + marker_chars > max_chars { + out.pop(); + char_count -= 1; + } + out.push_str(OUTBOX_TRUNCATION_MARKER); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + fn temp_outbox_path(name: &str) -> (tempfile::TempDir, PathBuf) { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join(name); + (dir, path) + } + + fn event(name: &str, kind: &str) -> LifecycleEvent { + LifecycleEvent { + event: name.to_string(), + kind: kind.to_string(), + thread_id: "session-1".to_string(), + turn_id: Some("turn-1".to_string()), + item_id: None, + payload: json!({"status": "completed"}), + } + } + + async fn deliver_all(state: &mut WriterState, events: Vec) { + for event in events { + state.deliver(event).await.expect("deliver"); + } + } + + async fn read_lines(path: &Path) -> Vec { + let text = tokio::fs::read_to_string(path).await.expect("read outbox"); + text.lines() + .map(|line| serde_json::from_str::(line).expect("json line")) + .collect() + } + + #[tokio::test] + async fn appends_one_jsonl_line_per_event_with_envelope_schema() { + let (_dir, path) = temp_outbox_path("schema.jsonl"); + let mut state = WriterState { + path: path.clone(), + webhook: None, + next_seq: 0, + recovered: false, + receiver: tokio::sync::mpsc::unbounded_channel().1, + }; + deliver_all(&mut state, vec![event("turn_start", "turn.started")]).await; + + let lines = read_lines(&path).await; + assert_eq!(lines.len(), 1); + let line = &lines[0]; + assert_eq!(line["schema_version"], 1); + assert_eq!(line["seq"], 1); + assert_eq!(line["event"], "turn_start"); + assert_eq!(line["kind"], "turn.started"); + assert_eq!(line["thread_id"], "session-1"); + assert_eq!(line["turn_id"], "turn-1"); + assert_eq!(line["item_id"], Value::Null); + assert!(line["timestamp"].as_str().is_some()); + assert!(line["payload"]["status"].as_str() == Some("completed")); + } + + /// Every emit site now carries `payload.workspace` (and subagent events + /// additionally `payload.subagent`) for consumer-side routing. The writer + /// must preserve those fields verbatim through the envelope round trip + /// for every event type. + #[tokio::test] + async fn payload_workspace_and_subagent_fields_survive_the_round_trip() { + let (_dir, path) = temp_outbox_path("routing-fields.jsonl"); + let mut state = WriterState { + path: path.clone(), + webhook: None, + next_seq: 0, + recovered: false, + receiver: tokio::sync::mpsc::unbounded_channel().1, + }; + let workspace = "/home/cw/wt-lane"; + let subagent = "explore-1"; + let subagent_payload = json!({ "workspace": workspace, "subagent": subagent }); + deliver_all( + &mut state, + vec![ + LifecycleEvent { + event: "session_start".to_string(), + kind: "session.started".to_string(), + thread_id: "session-1".to_string(), + turn_id: None, + item_id: None, + payload: json!({ "workspace": workspace }), + }, + LifecycleEvent { + event: "turn_start".to_string(), + kind: "turn.started".to_string(), + thread_id: "session-1".to_string(), + turn_id: Some("turn-1".to_string()), + item_id: None, + payload: json!({ "workspace": workspace }), + }, + LifecycleEvent { + event: "turn_end".to_string(), + kind: "turn.completed".to_string(), + thread_id: "session-1".to_string(), + turn_id: Some("turn-1".to_string()), + item_id: None, + payload: json!({ "workspace": workspace }), + }, + LifecycleEvent { + event: "turn_stalled".to_string(), + kind: "turn.stalled".to_string(), + thread_id: "session-1".to_string(), + turn_id: Some("turn-1".to_string()), + item_id: None, + payload: json!({ "workspace": workspace }), + }, + LifecycleEvent { + event: "subagent_spawn".to_string(), + kind: "subagent.spawned".to_string(), + thread_id: "session-1".to_string(), + turn_id: Some("turn-1".to_string()), + item_id: None, + payload: subagent_payload.clone(), + }, + LifecycleEvent { + event: "subagent_complete".to_string(), + kind: "subagent.completed".to_string(), + thread_id: "session-1".to_string(), + turn_id: Some("turn-1".to_string()), + item_id: None, + payload: subagent_payload.clone(), + }, + LifecycleEvent { + event: "session_end".to_string(), + kind: "session.ended".to_string(), + thread_id: "session-1".to_string(), + turn_id: None, + item_id: None, + payload: json!({ "workspace": workspace }), + }, + ], + ) + .await; + + let lines = read_lines(&path).await; + let events: Vec<&str> = lines + .iter() + .map(|line| line["event"].as_str().expect("event")) + .collect(); + assert_eq!( + events, + vec![ + "session_start", + "turn_start", + "turn_end", + "turn_stalled", + "subagent_spawn", + "subagent_complete", + "session_end", + ], + "the routing-field contract must cover every lifecycle event type" + ); + for line in &lines { + assert_eq!( + line["payload"]["workspace"], + json!(workspace), + "workspace must survive the round trip for event {}", + line["event"] + ); + } + for event in ["subagent_spawn", "subagent_complete"] { + let line = lines + .iter() + .find(|line| line["event"] == event) + .expect(event); + assert_eq!( + line["payload"]["subagent"], + json!(subagent), + "subagent must survive the round trip for event {event}" + ); + } + } + + #[tokio::test] + async fn seq_is_monotonic_and_recovers_across_reopen() { + let (_dir, path) = temp_outbox_path("seq.jsonl"); + let mut state = WriterState { + path: path.clone(), + webhook: None, + next_seq: 0, + recovered: false, + receiver: tokio::sync::mpsc::unbounded_channel().1, + }; + deliver_all( + &mut state, + vec![ + event("session_start", "session.started"), + event("turn_start", "turn.started"), + event("turn_end", "turn.completed"), + ], + ) + .await; + + // A fresh writer (new process, same file) continues the sequence. + let mut reopened = WriterState { + path: path.clone(), + webhook: None, + next_seq: 0, + recovered: false, + receiver: tokio::sync::mpsc::unbounded_channel().1, + }; + deliver_all(&mut reopened, vec![event("turn_start", "turn.started")]).await; + + let lines = read_lines(&path).await; + let seqs: Vec = lines + .iter() + .map(|line| line["seq"].as_u64().expect("seq")) + .collect(); + assert_eq!(seqs, vec![1, 2, 3, 4]); + } + + #[tokio::test] + async fn missing_and_empty_files_start_at_seq_1() { + let (_dir, path) = temp_outbox_path("empty.jsonl"); + assert_eq!(recover_last_seq(&path).await.expect("missing file"), 1); + + tokio::fs::write(&path, "").await.expect("empty file"); + assert_eq!(recover_last_seq(&path).await.expect("empty file"), 1); + } + + #[tokio::test] + async fn partial_trailing_line_is_ignored_during_recovery() { + let (_dir, path) = temp_outbox_path("partial.jsonl"); + tokio::fs::write( + &path, + format!( + "{}\n{}\n{{\"schema_version\":1,\"seq\":3,\"event\":\"turn_", + r#"{"schema_version":1,"seq":1,"event":"session_start","kind":"session.started","thread_id":"s","turn_id":null,"item_id":null,"timestamp":"t","payload":{}}"#, + r#"{"schema_version":1,"seq":2,"event":"turn_start","kind":"turn.started","thread_id":"s","turn_id":null,"item_id":null,"timestamp":"t","payload":{}}"#, + ), + ) + .await + .expect("write partial outbox"); + // The torn trailing line is not a complete record; recovery continues + // from the last complete line's seq (2) → next seq 3. + assert_eq!(recover_last_seq(&path).await.expect("recover"), 3); + } + + #[tokio::test] + async fn emit_queues_and_writes_in_order_without_blocking() { + let (_dir, path) = temp_outbox_path("emit.jsonl"); + let outbox = LifecycleOutbox::new(Some(path.clone()), None, None); + assert!(outbox.is_enabled()); + + outbox.emit(event("session_start", "session.started")); + outbox.emit(event("turn_start", "turn.started")); + outbox.emit(event("turn_end", "turn.completed")); + + // The writer task drains asynchronously; wait for the lines to land. + for _ in 0..100 { + if tokio::fs::metadata(&path) + .await + .is_ok_and(|meta| meta.len() > 0) + && read_lines(&path).await.len() >= 3 + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + let lines = read_lines(&path).await; + assert_eq!(lines.len(), 3, "expected all queued events to be written"); + let events: Vec<&str> = lines + .iter() + .map(|line| line["event"].as_str().expect("event")) + .collect(); + assert_eq!(events, vec!["session_start", "turn_start", "turn_end"]); + let seqs: Vec = lines + .iter() + .map(|line| line["seq"].as_u64().expect("seq")) + .collect(); + assert_eq!(seqs, vec![1, 2, 3], "seq must be assigned in emit order"); + } + + #[test] + fn disabled_outbox_drops_events_and_reports_disabled() { + let outbox = LifecycleOutbox::new(None, None, None); + assert!(!outbox.is_enabled()); + outbox.emit(event("turn_start", "turn.started")); // must not panic + + let empty_path = LifecycleOutbox::new(Some(PathBuf::new()), None, None); + assert!(!empty_path.is_enabled()); + + let default = LifecycleOutbox::default(); + assert!(!default.is_enabled()); + } + + #[test] + fn webhook_only_configures_without_a_file_path() { + // `webhook_url` without `path` is stored losslessly in config; the + // outbox handle itself only activates on a path. + let outbox = LifecycleOutbox::new( + None, + Some("https://example.com/hook".to_string()), + Some("token".to_string()), + ); + assert!(!outbox.is_enabled()); + } + + #[test] + fn bounded_text_truncates_to_limit_with_marker() { + assert_eq!(bounded_text("short", 80), "short"); + let long = "x".repeat(200); + let bounded = bounded_text(&long, OUTBOX_DETAIL_MAX_CHARS); + assert_eq!(bounded.chars().count(), OUTBOX_DETAIL_MAX_CHARS); + assert!(bounded.ends_with(OUTBOX_TRUNCATION_MARKER)); + assert!(bounded.starts_with('x')); + } + + #[test] + fn bounded_text_strips_controls_and_collapses_whitespace() { + assert_eq!( + bounded_text("line\x1b[31m one\n\n two\t", 80), + "line[31m one two" + ); + assert_eq!(bounded_text("", 80), ""); + assert_eq!(bounded_text(" \n\t ", 80), ""); + } + + #[test] + fn bounded_text_respects_utf8_boundaries() { + // 30 multi-byte emoji (4 bytes each) = 120 bytes but only 30 chars. + let emoji = "🦈".repeat(30); + let bounded = bounded_text(&emoji, OUTBOX_DETAIL_MAX_CHARS); + assert!(bounded.chars().count() <= OUTBOX_DETAIL_MAX_CHARS); + assert!(bounded.starts_with('🦈')); + } + + /// The webhook transport must POST `{"at", "event"}` JSON and, when a + /// token is configured, send it as `Authorization: Bearer `. + #[tokio::test] + async fn webhook_posts_at_event_payload_with_bearer_token() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/hook")) + .and(wiremock::matchers::header( + "authorization", + "Bearer secret-token", + )) + .and(wiremock::matchers::body_partial_json(json!({ + "event": {"kind": "turn.started"} + }))) + .respond_with(wiremock::ResponseTemplate::new(200)) + .mount(&server) + .await; + + let webhook = WebhookHookSink::new_with_token( + format!("{}/hook", server.uri()), + Some("secret-token".to_string()), + ); + webhook + .post_payload(json!( + {"at": "2026-08-19T00:00:00Z", "event": {"kind": "turn.started"}} + )) + .await + .expect("webhook delivery"); + + let requests = server.received_requests().await.expect("requests"); + assert_eq!(requests.len(), 1, "exactly one webhook POST"); + } + + /// A webhook that always fails must surface its error to the caller + /// (which logs and drops it) — never panic, never retry forever. + #[tokio::test] + async fn webhook_failure_is_an_error_not_a_panic() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .respond_with(wiremock::ResponseTemplate::new(500)) + .mount(&server) + .await; + + let webhook = WebhookHookSink::new_with_token(format!("{}/hook", server.uri()), None); + let result = webhook.post_payload(json!({})).await; + assert!(result.is_err(), "expected the failure to be reported"); + } +} diff --git a/crates/tui/Cargo.toml b/crates/tui/Cargo.toml index 5d0e4c0077..96151a3d26 100644 --- a/crates/tui/Cargo.toml +++ b/crates/tui/Cargo.toml @@ -34,6 +34,7 @@ codewhale-config = { path = "../config", version = "0.9.11" } codewhale-command-contract = { path = "../command-contract", version = "0.9.11" } codewhale-core = { path = "../core", version = "0.9.11" } codewhale-execpolicy = { path = "../execpolicy", version = "0.9.11" } +codewhale-hooks = { path = "../hooks", version = "0.9.11" } codewhale-lane = { path = "../lane", version = "0.9.11" } codewhale-paths = { path = "../paths", version = "0.9.11" } codewhale-protocol = { path = "../protocol", version = "0.9.11" } diff --git a/crates/tui/src/config.rs b/crates/tui/src/config.rs index 6aa53665e4..34df417c6f 100644 --- a/crates/tui/src/config.rs +++ b/crates/tui/src/config.rs @@ -2987,6 +2987,12 @@ pub struct Config { #[serde(default)] pub hooks: Option, + /// Lifecycle event outbox (`[lifecycle_outbox]`). Opt-in: an unset or + /// empty `path` disables the feature and leaves behavior unchanged. + /// Fires for interactive TUI sessions and headless `codewhale exec` runs. + #[serde(default)] + pub lifecycle_outbox: Option, + /// Provider-specific credentials and defaults shared with the `codewhale` facade. #[serde(default)] pub providers: Option, @@ -10175,6 +10181,7 @@ fn merge_config(base: Config, override_cfg: Config) -> Config { tui: override_cfg.tui.or(base.tui), transcript: override_cfg.transcript.or(base.transcript), hooks: override_cfg.hooks.or(base.hooks), + lifecycle_outbox: override_cfg.lifecycle_outbox.or(base.lifecycle_outbox), providers: merge_providers(base.providers, override_cfg.providers), features: merge_features(base.features, override_cfg.features), notifications: override_cfg.notifications.or(base.notifications), diff --git a/crates/tui/src/config/tests.rs b/crates/tui/src/config/tests.rs index 747fc8c397..f0a3639af3 100644 --- a/crates/tui/src/config/tests.rs +++ b/crates/tui/src/config/tests.rs @@ -744,6 +744,36 @@ web_search = true Ok(()) } +#[test] +fn tui_config_parses_lifecycle_outbox_table() { + let raw = r#" +[lifecycle_outbox] +path = "~/.codewhale/notifications/outbox.jsonl" +webhook_url = "https://example.com/hooks/codewhale" +webhook_token = "secret-token" +"#; + let parsed: ConfigFile = toml::from_str(raw).expect("parse lifecycle_outbox config"); + + let outbox = parsed + .base + .lifecycle_outbox + .expect("lifecycle_outbox table should parse"); + assert_eq!( + outbox.path, + Some(PathBuf::from("~/.codewhale/notifications/outbox.jsonl")) + ); + assert_eq!( + outbox.webhook_url.as_deref(), + Some("https://example.com/hooks/codewhale") + ); + assert_eq!(outbox.webhook_token.as_deref(), Some("secret-token")); + + // Off by default: a config without the table leaves the feature off. + let absent: ConfigFile = + toml::from_str("model = \"demo\"").expect("parse config without outbox table"); + assert!(absent.base.lifecycle_outbox.is_none()); +} + #[test] fn tui_config_parses_hotbar_bindings() { let raw = r#" diff --git a/crates/tui/src/exec_agent.rs b/crates/tui/src/exec_agent.rs new file mode 100644 index 0000000000..ca4ce5d9a9 --- /dev/null +++ b/crates/tui/src/exec_agent.rs @@ -0,0 +1,1066 @@ +//! Non-interactive exec agent assembly: the `run_exec_agent` pipeline +//! that resolves the CLI route, builds the engine configuration, spawns +//! the engine, and drives the exec output stream to completion. +//! +//! Extracted verbatim from `lib.rs` (#5586, the issue's prescribed +//! engine-config-assembly cut). The two functions were crate-private in +//! the root and are `pub(crate)` here purely so the root's glob re-export +//! keeps the dispatch site and tests resolving unchanged. + +use super::*; + +pub(crate) fn exec_max_steps(max_turns: Option) -> u32 { + max_turns.unwrap_or(u32::MAX) +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn run_exec_agent( + config: &Config, + model: &str, + prompt: &str, + workspace: PathBuf, + max_subagents: usize, + auto_approve: bool, + allow_sandbox_elevation: bool, + explicit_sandbox: Option<&str>, + trust_mode: bool, + json_output: bool, + resume_session: Option, + force_configured_route: bool, + output_format: ExecOutputFormat, + max_turns: u32, + max_tool_calls: Option, + allowed_tools: Option>, + disallowed_tools: Option>, + append_system_prompt: Option, + tool_authority_json: Option, + plugin_registry: std::sync::Arc, +) -> Result<()> { + use crate::compaction::CompactionConfig; + use crate::core::engine::{EngineConfig, spawn_engine}; + use crate::core::events::Event; + use crate::core::ops::Op; + use crate::tools::plan::new_shared_plan_state; + use crate::tools::todo::new_shared_todo_list; + use crate::tui::app::AppMode; + + validate_exec_tool_authority_resume(tool_authority_json.as_deref(), resume_session.is_some())?; + let fleet_authority = tool_authority_json + .as_deref() + .map(crate::tools::spec::ToolAuthorityEnvelope::from_json) + .transpose() + .map_err(anyhow::Error::msg)?; + let fleet_authority_active = fleet_authority.is_some(); + let outer_network_access = fleet_authority + .as_ref() + .and_then(|authority| authority.network_access); + let outer_shell_authority = fleet_authority + .as_ref() + .map(|authority| authority.shell) + .unwrap_or_default(); + if let Some(envelope) = fleet_authority { + crate::tools::spec::install_process_tool_authority(envelope).map_err(anyhow::Error::msg)?; + } + + let route = resolve_cli_exec_route(config, model, prompt, force_configured_route).await?; + let execution_config = config_for_cli_route(config, &route); + let auto_model = route.auto_model; + let effective_provider = route.provider; + let effective_model = route.model; + let validated_route = crate::route_runtime::resolve_runtime_route( + &execution_config, + effective_provider, + Some(&effective_model), + ) + .map_err(anyhow::Error::msg)? + .validate() + .map_err(anyhow::Error::msg)?; + let effective_provider_name = validated_route.identity.key.clone(); + let effective_provider_id = validated_route.identity.exact_id.clone(); + let (effective_provider_kind, effective_stream_provider_id) = + exec_stream_provider_route(&validated_route.identity); + let route_source = if auto_model { + "auto_resolver" + } else { + "explicit_or_configured" + } + .to_string(); + let exec_started = Instant::now(); + let prompt_sha256 = format!("sha256:{}", crate::hashing::sha256_hex(prompt.as_bytes())); + let binary_sha256 = current_binary_sha256(); + let approval_posture = if auto_approve { "auto_tools" } else { "ask" }.to_string(); + let sandbox_posture = explicit_sandbox.unwrap_or("configured_default").to_string(); + let active_route_limits = + crate::route_budget::known_route_limits(validated_route.candidate.limits()); + let max_subagents = if max_subagents == config.max_subagents_for_provider(config.api_provider()) + { + execution_config + .max_subagents_for_provider(effective_provider) + .clamp(1, MAX_SUBAGENTS) + } else { + max_subagents + }; + // A FIXED model with `--reasoning-effort auto` (the exact shape a Fleet + // worker subprocess launches with: `--model --reasoning-effort + // auto`) is still Auto. `auto_model` is a *model* decision and is false + // here, so deriving the auto flag from it left this path both raw and + // non-auto: the literal string `"auto"` travelled to the engine while the + // receipt claimed no Auto was in play. + let reasoning_effort_auto = route.auto_controls_reasoning; + // Resolve Auto against this run's prompt at the CLI boundary, exactly like + // `run_one_shot`/`run_one_shot_json` and the interactive launch path do, + // so the tier the engine (and the receipt below) sees is concrete. + let effective_reasoning_effort = route.reasoning_effort.and_then(|effort| { + cli_reasoning_effort_value_for_prompt(&execution_config, &effective_model, effort, prompt) + }); + + let settings = crate::settings::Settings::load().unwrap_or_default(); + let auto_compact_enabled = if crate::settings::Settings::auto_compact_explicitly_configured() { + settings.auto_compact + } else { + crate::route_budget::auto_compact_default_for_route( + effective_provider, + &effective_model, + active_route_limits, + ) + }; + let compaction = CompactionConfig { + enabled: auto_compact_enabled, + model: effective_model.clone(), + effective_context_window: Some(crate::route_budget::route_context_window_tokens( + effective_provider, + &effective_model, + active_route_limits, + )), + token_threshold: crate::route_budget::compaction_threshold_for_route_at_percent( + effective_provider, + &effective_model, + active_route_limits, + settings.auto_compact_threshold_percent, + ), + ..Default::default() + }; + + let network_policy = exec_network_policy(&execution_config, outer_network_access); + + let lsp_config = (!fleet_authority_active) + .then(|| { + execution_config + .lsp + .clone() + .map(crate::config::LspConfigToml::into_runtime) + }) + .flatten(); + let mut engine_features = execution_config.features(); + apply_fleet_engine_feature_caps( + &mut engine_features, + fleet_authority_active, + outer_network_access, + outer_shell_authority, + ); + if crate::core::allowlist_is_native_file_and_shell_only(allowed_tools.as_deref()) { + engine_features.disable(crate::features::Feature::Mcp); + } + let engine_plugin_registry = if fleet_authority_active { + std::sync::Arc::new(crate::plugins::PluginRegistry::empty(&workspace)) + } else { + plugin_registry + }; + let exec_allow_shell = crate::tools::spec::fleet_exec_shell_enabled( + fleet_authority_active, + outer_shell_authority, + disallowed_tools.as_deref(), + ) || (!fleet_authority_active + && (auto_approve || execution_config.allow_shell())); + let persist_services_enabled = cfg!(unix) + && !fleet_authority_active + && exec_allow_shell + && explicit_sandbox + .is_some_and(|sandbox| sandbox.eq_ignore_ascii_case("danger-full-access")); + let exec_shell_manager = crate::tools::shell::new_shared_shell_manager(workspace.clone()); + let runtime_services = crate::tools::spec::RuntimeToolServices { + shell_manager: Some(exec_shell_manager.clone()), + persist_services_enabled, + ..crate::tools::spec::RuntimeToolServices::default() + }; + + let engine_config = EngineConfig { + model: effective_model.clone(), + active_route_limits, + workspace: workspace.clone(), + subagent_state_root: None, + plugin_registry: Some(std::sync::Arc::clone(&engine_plugin_registry)), + allow_shell: exec_allow_shell, + trust_mode, + notes_path: execution_config.notes_path(), + mcp_config_path: execution_config.mcp_config_path(), + skills_dir: execution_config.skills_dir(), + skills_scan_codewhale_only: execution_config.skills_config().scan_codewhale_only(), + instructions: { + let mut instrs: Vec = execution_config + .instructions_paths() + .into_iter() + .map(Into::into) + .collect(); + if let Some(ref extra) = append_system_prompt { + instrs.push(crate::prompts::InstructionSource::Inline { + name: "cli:append-system-prompt".into(), + content: extra.clone(), + }); + } + instrs + }, + project_context_pack_enabled: execution_config.project_context_pack_enabled(), + translation_enabled: false, + max_steps: max_turns, + max_subagents, + max_admitted_subagents: execution_config + .max_admitted_subagents_for_provider(effective_provider) + .max(max_subagents), + launch_concurrency: execution_config.launch_concurrency_for_provider(effective_provider), + subagents_enabled: !fleet_authority_active + && execution_config.subagents_enabled_for_provider(effective_provider), + features: engine_features, + auto_review_policy: execution_config.auto_review_policy(), + compaction: compaction.clone(), + todos: new_shared_todo_list(), + plan_state: new_shared_plan_state(), + goal_state: crate::tools::goal::new_shared_goal_state(), + max_spawn_depth: if fleet_authority_active { + 0 + } else { + execution_config.subagent_max_spawn_depth_for_provider(effective_provider) + }, + subagent_token_budget: execution_config + .subagent_token_budget_for_provider(effective_provider), + network_policy, + snapshots_enabled: !fleet_authority_active && execution_config.snapshots_config().enabled, + snapshots_max_workspace_bytes: execution_config + .snapshots_config() + .max_workspace_gb + .saturating_mul(1024 * 1024 * 1024), + lsp_config, + runtime_services, + subagent_model_overrides: execution_config.subagent_model_overrides(), + fleet_roster: std::sync::Arc::new(crate::fleet::identity::load_effective_roster( + &execution_config.fleet_config(), + &workspace, + Some(engine_plugin_registry.as_ref()), + )), + subagent_api_timeout: std::time::Duration::from_secs( + execution_config.subagent_api_timeout_secs_for_provider(effective_provider), + ), + stream_chunk_timeout: std::time::Duration::from_secs( + execution_config.stream_chunk_timeout_secs(), + ), + subagent_heartbeat_timeout: std::time::Duration::from_secs( + execution_config.subagent_heartbeat_timeout_secs_for_provider(effective_provider), + ), + prefer_bwrap: execution_config.prefer_bwrap.unwrap_or(false), + bwrap_extensions: crate::sandbox::BwrapMountExtensions { + read_only_roots: execution_config.bwrap_ro_roots.clone(), + device_roots: execution_config.bwrap_dev_roots.clone(), + }, + denied_read_subpaths: execution_config.sandbox_denied_read_paths.clone(), + memory_enabled: execution_config.memory_enabled(), + memory_path: execution_config.memory_path(), + speech_output_dir: execution_config.speech_output_dir(), + vision_config: execution_config.vision_model_config(), + strict_tool_mode: execution_config.strict_tool_mode.unwrap_or(false), + goal_objective: None, + goal_token_budget: None, + goal_status: crate::tools::goal::GoalStatus::Active, + goal_max_continuations: execution_config.goal_max_continuations(), + goal_continuation_delay_seconds: execution_config.goal_continuation_delay_seconds(), + allowed_tools: allowed_tools.clone(), + disallowed_tools: disallowed_tools.clone(), + max_tool_calls, + hook_executor: None, + locale_tag: crate::localization::resolve_locale(&settings.locale) + .tag() + .to_string(), + workshop: { + crate::tools::large_output_router::WorkshopConfig::install_active( + config.workshop.as_ref(), + ); + config.workshop.clone() + }, + search_provider: execution_config.search_provider(), + search_api_key: execution_config + .search + .as_ref() + .and_then(|s| s.api_key.clone()), + search_base_url: execution_config + .search + .as_ref() + .and_then(|s| s.base_url.clone()), + tools_always_load: if fleet_authority_active { + std::collections::HashSet::new() + } else { + execution_config.tools_always_load() + }, + tools: if fleet_authority_active { + None + } else { + execution_config.tools.clone() + }, + verbosity: execution_config.verbosity.clone(), + workspace_follow_symlinks: settings.workspace_follow_symlinks, + exec_policy_engine: execution_config.exec_policy_engine.clone(), + terminal_chrome_enabled: false, + advisor_config: execution_config + .advisor + .as_ref() + .map(crate::tools::subagent::AdvisorConfig::from_toml) + .unwrap_or_else(crate::tools::subagent::AdvisorConfig::disabled), + }; + + let engine_handle = spawn_engine(engine_config, &execution_config); + let mode = if auto_approve { + AppMode::Yolo + } else { + AppMode::Agent + }; + + let resuming_session = resume_session.is_some(); + let mut loaded_session_id = None; + if let Some(saved) = resume_session { + let saved_id = saved.metadata.id.clone(); + if saved.metadata.workspace != workspace && output_format == ExecOutputFormat::Text { + eprintln!( + "Warning: session {} was created in a different workspace ({}). Resuming anyway.", + truncate_id(&saved_id), + saved.metadata.workspace.display(), + ); + } + + engine_handle + .send(Op::SyncSession { + session_id: Some(saved_id.clone()), + messages: saved.messages, + system_prompt: saved.system_prompt.map(SystemPrompt::Text), + system_prompt_override: false, + model: saved.metadata.model, + workspace: saved.metadata.workspace, + mode, + }) + .await?; + loaded_session_id = Some(saved_id.clone()); + if output_format == ExecOutputFormat::Text && !json_output { + eprintln!("{}", exec_resumed_session_line(&saved_id)); + } + } + + // Lifecycle outbox (`[lifecycle_outbox]`): headless `codewhale exec` + // gets the same turn boundaries as the interactive TUI. Disabled + // (all emits no-op) when the config has no path. + let lifecycle_outbox = config + .lifecycle_outbox + .as_ref() + .map(|outbox| { + codewhale_hooks::LifecycleOutbox::new( + outbox.path.clone(), + outbox.webhook_url.clone(), + outbox.webhook_token.clone(), + ) + }) + .unwrap_or_else(codewhale_hooks::LifecycleOutbox::disabled); + // Wall clock for the outbox `turn_end` duration. `exec` never receives + // a TurnStarted engine event, so the start is marked at the same + // `Op::SendMessage` boundary where `turn_start` is emitted below. + let exec_turn_started_at = Instant::now(); + + engine_handle + .send(Op::SendMessage { + content: prompt.to_string(), + mode, + route: Box::new(validated_route.into_resolved()), + compaction: Box::new(compaction.clone()), + goal_objective: None, + goal_token_budget: None, + goal_status: crate::tools::goal::GoalStatus::Active, + allowed_tools: allowed_tools.clone(), + dynamic_tools: Vec::new(), + hook_executor: None, + reasoning_effort: effective_reasoning_effort, + reasoning_effort_auto, + auto_model, + allow_shell: auto_approve || execution_config.allow_shell(), + trust_mode, + auto_approve, + translation_enabled: false, + approval_mode: if auto_approve { + crate::tui::approval::ApprovalMode::Bypass + } else { + execution_config + .approval_policy + .as_deref() + .and_then(crate::tui::approval::ApprovalMode::from_config_value) + .unwrap_or_default() + }, + verbosity: execution_config.verbosity.clone(), + provenance: crate::core::ops::UserInputProvenance::ExternalUser, + }) + .await?; + + // Lifecycle outbox: the clean headless turn-start boundary. `exec` has + // no TurnStarted engine event; the message submission above is exactly + // where the engine begins the turn. No-op when the feature is disabled. + lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "turn_start".to_string(), + kind: "turn.started".to_string(), + thread_id: loaded_session_id.clone().unwrap_or_default(), + turn_id: None, + item_id: None, + payload: serde_json::json!({ + "model": codewhale_hooks::bounded_text( + &effective_model, + codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + ), + "workspace": workspace.display().to_string(), + }), + }); + + let mut summary = ExecSummary { + mode: "agent".to_string(), + provider: effective_provider_name.clone(), + model: effective_model.clone(), + prompt: prompt.to_string(), + ..ExecSummary::default() + }; + let can_elevate_sandbox = + exec_sandbox_elevation_authorized(allow_sandbox_elevation, explicit_sandbox); + let mut sandbox_denied = false; + let mut approval_required = false; + let mut tool_error_seen = false; + let mut last_error_category = None; + let mut reported_sandbox_contract = false; + + let should_persist_session = resuming_session || output_format == ExecOutputFormat::StreamJson; + let mut latest_session_id = loaded_session_id; + let mut latest_messages: Vec = Vec::new(); + let mut latest_system_prompt: Option = None; + let mut latest_model = effective_model; + let mut latest_workspace = workspace.clone(); + let mut tool_starts: HashMap = HashMap::new(); + let mut turn_usage_seq: u32 = 0; + + let mut stdout = io::stdout(); + let mut ends_with_newline = false; + loop { + let event = { + let mut rx = engine_handle.rx_event.write().await; + rx.recv().await + }; + + let Some(event) = event else { + break; + }; + + match event { + Event::MessageDelta { content, .. } => { + summary.output.push_str(&content); + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::Content { content })?; + } else if !json_output { + print!("{content}"); + stdout.flush()?; + } + ends_with_newline = summary.output.ends_with('\n'); + } + Event::MessageComplete { .. } + if output_format == ExecOutputFormat::Text + && !json_output + && !ends_with_newline => + { + println!(); + } + Event::ThinkingDelta { .. } => { + // Exec stream-json intentionally omits reasoning deltas; the + // TUI transcript retains its existing Activity Detail surface. + } + Event::ToolCallStarted { id, name, input } => { + let started_at = chrono::Utc::now().to_rfc3339(); + tool_starts.insert(id.clone(), (Instant::now(), started_at.clone())); + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::ToolUse { + name, + id, + input, + started_at, + })?; + } else if !json_output { + let summary = summarize_tool_args(&input); + if let Some(summary) = summary { + eprintln!("tool: {name} ({summary})"); + } else { + eprintln!("tool: {name}"); + } + } + } + Event::ToolCallComplete { + id, name, result, .. + } => { + let (duration_ms, started_at) = tool_starts + .remove(&id) + .map(|(started, timestamp)| { + ( + u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), + timestamp, + ) + }) + .unwrap_or_else(|| (0, chrono::Utc::now().to_rfc3339())); + let receipt_name = name.clone(); + match result { + Ok(output) => { + tool_error_seen |= !output.success; + summary.tools.push(ExecToolEntry { + name: name.clone(), + success: output.success, + output: output.content.clone(), + }); + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::ToolResult { + id, + name: receipt_name, + output: output.content, + status: if output.success { + "success".to_string() + } else { + "error".to_string() + }, + started_at, + completed_at: chrono::Utc::now().to_rfc3339(), + duration_ms, + side_effect_status: output + .metadata + .as_ref() + .and_then(|metadata| metadata.get("side_effect_status")) + .and_then(serde_json::Value::as_str) + .unwrap_or("unknown") + .to_string(), + error_category: (!output.success).then(|| { + output + .metadata + .as_ref() + .and_then(|metadata| metadata.get("error_category")) + .and_then(serde_json::Value::as_str) + .unwrap_or("tool_reported_failure") + .to_string() + }), + truncated: output + .metadata + .as_ref() + .and_then(|metadata| metadata.get("truncated")) + .and_then(serde_json::Value::as_bool), + artifact: tool_artifact_receipt(output.metadata.as_ref()), + result_metadata: output.metadata, + })?; + } else if !json_output { + if name == "exec_shell" && !output.content.trim().is_empty() { + eprintln!("tool {name} completed"); + eprintln!( + "--- stdout/stderr ---\n{}\n---------------------", + output.content + ); + } else { + eprintln!( + "tool {name} completed: {}", + summarize_tool_output(&output.content) + ); + } + } + } + Err(err) => { + tool_error_seen = true; + let error_text = err.to_string(); + summary.tools.push(ExecToolEntry { + name: name.clone(), + success: false, + output: error_text.clone(), + }); + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::ToolResult { + id, + name: receipt_name, + output: error_text, + status: "error".to_string(), + started_at, + completed_at: chrono::Utc::now().to_rfc3339(), + duration_ms, + side_effect_status: "not_started_or_unknown".to_string(), + error_category: Some(tool_error_receipt_category(&err).to_string()), + truncated: None, + artifact: None, + result_metadata: None, + })?; + } else if !json_output { + eprintln!("tool {name} failed: {err}"); + } + } + } + } + Event::AgentSpawned { id, prompt, .. } + if output_format == ExecOutputFormat::Text && !json_output => + { + eprintln!("sub-agent {id} spawned: {}", summarize_tool_output(&prompt)); + } + Event::AgentProgress { id, status, .. } + if output_format == ExecOutputFormat::Text && !json_output => + { + eprintln!("sub-agent {id}: {status}"); + } + Event::AgentComplete { id, result, .. } + if output_format == ExecOutputFormat::Text && !json_output => + { + eprintln!( + "sub-agent {id} completed: {}", + summarize_tool_output(&result) + ); + } + Event::AgentSpawned { + id, + parent_run_id, + spawn_depth, + model, + route_source, + .. + } if output_format == ExecOutputFormat::StreamJson => { + emit_exec_stream_event(&ExecStreamEvent::AgentSpawned { + id, + model, + spawn_depth, + parent_run_id, + route_source, + })?; + } + Event::AgentSpawned { .. } + | Event::AgentProgress { .. } + | Event::AgentComplete { .. } => {} + Event::WorkflowUi { run_id, event, .. } + if output_format == ExecOutputFormat::StreamJson => + { + emit_exec_stream_event(&ExecStreamEvent::WorkflowEvent { run_id, event })?; + } + Event::ApprovalRequired { id, .. } => { + if auto_approve { + let _ = engine_handle.approve_tool_call(id).await; + } else { + approval_required = true; + let _ = engine_handle.deny_tool_call(id).await; + } + } + Event::ElevationRequired { + tool_id, + tool_name, + denial_reason, + .. + } => { + if can_elevate_sandbox { + let policy = crate::sandbox::SandboxPolicy::DangerFullAccess; + let _ = engine_handle.retry_tool_with_policy(tool_id, policy).await; + } else { + sandbox_denied = true; + approval_required = true; + summary.outcomes.push(ExecOutcome { + kind: "sandbox_denied".to_string(), + outcome: "approval_required".to_string(), + tool_name: tool_name.clone(), + reason: denial_reason.clone(), + }); + if !reported_sandbox_contract { + eprintln!( + "sandbox denied {tool_name}: {denial_reason}; --auto approves tools but does not elevate sandbox access — use --sandbox danger-full-access or --allow-sandbox-elevation to opt in" + ); + reported_sandbox_contract = true; + } + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::SandboxDenied { + tool_id: tool_id.clone(), + tool_name, + reason: denial_reason, + outcome: "approval_required".to_string(), + })?; + } + let _ = engine_handle.deny_tool_call(tool_id).await; + } + } + Event::Error { + envelope, + recoverable: _, + } => { + // Only a non-recoverable envelope may force the run summary + // into failure. Recoverable warnings (stream-stall notices, + // transient retry noise) are still streamed for visibility, + // but the terminal TurnComplete event carries the + // authoritative turn outcome — letting a warning set + // `summary.error` here would exit an otherwise-successful + // `exec` run non-zero. + if exec_error_event_is_fatal(&envelope) { + last_error_category = Some(envelope.category); + summary.error_category = Some(envelope.category.to_string()); + summary.error = Some(envelope.message.clone()); + } + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::Error { + error: envelope.message, + })?; + } else if !json_output { + eprintln!("error: {}", envelope.message); + } + } + Event::TurnUsage { + usage, duration_ms, .. + } => { + if output_format == ExecOutputFormat::StreamJson { + turn_usage_seq = turn_usage_seq.saturating_add(1); + emit_exec_stream_event(&ExecStreamEvent::TurnUsage { + turn: turn_usage_seq, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + reasoning_tokens: usage.reasoning_tokens, + prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens, + prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens, + prompt_cache_write_tokens: usage.prompt_cache_write_tokens, + reasoning_replay_tokens: usage.reasoning_replay_tokens, + duration_ms, + })?; + } + } + Event::TurnComplete { + status, + error, + usage, + tool_catalog, + .. + } => { + let (terminal_status, terminal_error) = (status, error); + #[cfg(unix)] + let (mut terminal_status, mut terminal_error) = (terminal_status, terminal_error); + if matches!( + terminal_status, + crate::core::events::TurnOutcomeStatus::Completed + ) && terminal_error.is_none() + { + #[cfg(unix)] + match exec_shell_manager.lock() { + Ok(mut manager) => match manager.commit_persistent_services() { + Ok(receipts) => { + for receipt in &receipts { + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event( + &ExecStreamEvent::ServiceReleased { + task_id: receipt.task_id.clone(), + pid: receipt.pid, + process_group_id: receipt.process_group_id, + ownership: receipt.ownership.clone(), + }, + )?; + } else if !json_output { + eprintln!( + "persistent service released: {} pid={} pgid={} ownership={}", + receipt.task_id, + receipt.pid, + receipt.process_group_id, + receipt.ownership + ); + } + } + summary.released_services.extend(receipts); + } + Err(error) => { + manager.abort_persistent_services(); + terminal_status = crate::core::events::TurnOutcomeStatus::Failed; + terminal_error = Some(format!( + "Persistent service ownership transfer failed: {error}" + )); + } + }, + Err(_) => { + terminal_status = crate::core::events::TurnOutcomeStatus::Failed; + terminal_error = Some( + "Persistent service ownership transfer failed: shell manager lock poisoned" + .to_string(), + ); + } + } + } else if let Ok(mut manager) = exec_shell_manager.lock() { + manager.abort_persistent_services(); + } + summary.status = Some(format!("{terminal_status:?}").to_lowercase()); + if terminal_error.is_some() { + summary.error = terminal_error; + } + if sandbox_denied + && summary.error.is_none() + && matches!( + terminal_status, + crate::core::events::TurnOutcomeStatus::Failed + ) + { + summary.error = Some( + "exec turn failed after sandbox denial; explicit sandbox elevation was not authorized" + .to_string(), + ); + } + // Lifecycle outbox: the clean headless turn-end boundary. + // `terminal_status` is authoritative here — persistent-service + // handoff failures above already demoted it to Failed, and + // `summary.error` includes the sandbox-denial augmentation. + // No-op when the feature is disabled. + { + let outbox_status = format!("{terminal_status:?}").to_lowercase(); + let kind = match terminal_status { + crate::core::events::TurnOutcomeStatus::Completed => "turn.completed", + crate::core::events::TurnOutcomeStatus::Failed => "turn.failed", + crate::core::events::TurnOutcomeStatus::Interrupted => "turn.interrupted", + }; + lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "turn_end".to_string(), + kind: kind.to_string(), + thread_id: latest_session_id.clone().unwrap_or_default(), + turn_id: None, + item_id: None, + payload: serde_json::json!({ + "status": outbox_status, + "duration_ms": exec_turn_started_at.elapsed().as_millis() as u64, + "workspace": latest_workspace.display().to_string(), + "error": summary.error.as_deref().map(|message| { + codewhale_hooks::bounded_text( + message, + codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + ) + }), + }), + }); + } + if last_error_category.is_none() { + last_error_category = summary + .error + .as_deref() + .map(crate::error_taxonomy::classify_error_message); + summary.error_category = + last_error_category.map(|category| category.to_string()); + } + let termination_reason = crate::core::termination::classify_turn_termination( + terminal_status, + last_error_category, + tool_error_seen, + approval_required, + ); + summary.termination_reason = Some(termination_reason.as_str().to_string()); + // State the exit class here rather than inferring it later + // from the process exit code: `Canceled` exits 130, the same + // value the SIGINT path uses, so a code-based derivation would + // report every Esc-cancelled turn as a signal. A no-op unless + // this process was armed. + if !termination_reason.is_success() { + codewhale_telemetry::set_exit_class(codewhale_telemetry::ExitClass::Error); + } + let saved_session_id = if should_persist_session && !latest_messages.is_empty() { + match persist_exec_session( + &latest_messages, + &latest_model, + PersistedProviderRoute { + kind: effective_provider.as_str(), + id: effective_provider_id.as_deref(), + }, + &latest_workspace, + &latest_system_prompt, + latest_session_id.as_deref(), + u64::from(usage.input_tokens) + u64::from(usage.output_tokens), + ) { + Ok(id) => { + if output_format == ExecOutputFormat::Text && !json_output { + eprintln!("{}", exec_saved_session_line(&id)); + } + Some(id) + } + Err(err) => { + if output_format == ExecOutputFormat::Text && !json_output { + eprintln!("warning: failed to save exec session: {err}"); + } + latest_session_id.clone() + } + } + } else { + latest_session_id.clone() + }; + if output_format == ExecOutputFormat::StreamJson { + if let Some(id) = saved_session_id.as_ref() { + emit_exec_stream_event(&ExecStreamEvent::SessionCapture { + content: exec_stream_session_ref(id), + })?; + } + // Resolved output ceiling and its provenance, surfaced so a + // wrong ceiling is visible in the receipt rather than + // requiring packet capture. + let codewhale_max_output_tokens = + crate::route_budget::effective_max_output_tokens_for_route( + effective_provider, + &latest_model, + active_route_limits, + ); + let codewhale_max_output_tokens_source = + crate::route_budget::output_ceiling_source( + effective_provider, + &latest_model, + ) + .as_str(); + emit_exec_stream_event(&ExecStreamEvent::Metadata { + meta: Box::new(ExecStreamMeta { + receipt_kind: "terminal", + provider: effective_provider_kind.clone(), + provider_id: effective_stream_provider_id.clone(), + model: latest_model.clone(), + route_source: route_source.clone(), + input_tokens: Some(usage.input_tokens), + output_tokens: Some(usage.output_tokens), + prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens, + prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens, + prompt_cache_write_tokens: usage.prompt_cache_write_tokens, + reasoning_tokens: usage.reasoning_tokens, + codewhale_max_output_tokens: Some(codewhale_max_output_tokens), + codewhale_max_output_tokens_source: Some( + codewhale_max_output_tokens_source, + ), + duration_ms: u64::try_from(exec_started.elapsed().as_millis()) + .unwrap_or(u64::MAX), + retry_count: None, + approval_posture: approval_posture.clone(), + sandbox_posture: sandbox_posture.clone(), + binary_sha256: binary_sha256.clone(), + config_sha256: None, + prompt_sha256: prompt_sha256.clone(), + tool_catalog_sha256: tool_catalog.as_ref().and_then(|catalog| { + serde_json::to_vec(catalog).ok().map(|bytes| { + format!("sha256:{}", crate::hashing::sha256_hex(&bytes)) + }) + }), + input_analysis: exec_stream_input_analysis( + &latest_messages, + latest_system_prompt.as_ref(), + ), + visible_final_answer_chars: summary.output.chars().count(), + resume_command: saved_session_id + .as_deref() + .map(exec_stream_resume_hint) + .unwrap_or_default(), + session_id: saved_session_id + .as_deref() + .map(exec_stream_session_ref) + .unwrap_or_default(), + workspace: latest_workspace.display().to_string(), + message_count: latest_messages.len(), + status: summary.status.clone(), + termination_reason: summary.termination_reason.clone(), + error_category: summary.error_category.clone(), + error: summary.error.clone(), + }), + })?; + emit_exec_stream_event(&ExecStreamEvent::Done)?; + } + let _ = engine_handle.send(Op::Shutdown).await; + break; + } + Event::SessionUpdated { + session_id, + messages, + system_prompt, + model, + workspace, + } => { + latest_session_id = Some(session_id); + latest_messages = messages; + latest_system_prompt = system_prompt; + latest_model = model; + latest_workspace = workspace; + } + // #3027: surface the engine's max-steps notice in text mode so a + // --max-turns run that stops early says why instead of going quiet. + Event::Status { message } + if output_format == ExecOutputFormat::Text + && !json_output + && message.contains("Maximum model steps") => + { + eprintln!("{message}"); + } + _ => {} + } + } + + if summary.status.is_none() { + if let Ok(mut manager) = exec_shell_manager.lock() { + manager.abort_persistent_services(); + } + let error = summary.error.clone().unwrap_or_else(|| { + "Engine event channel closed before a terminal turn receipt".to_string() + }); + let category = last_error_category + .unwrap_or_else(|| crate::error_taxonomy::classify_error_message(&error)); + let termination_reason = crate::core::termination::classify_turn_termination( + crate::core::events::TurnOutcomeStatus::Failed, + Some(category), + tool_error_seen, + approval_required, + ); + summary.status = Some("failed".to_string()); + summary.error_category = Some(category.to_string()); + summary.termination_reason = Some(termination_reason.as_str().to_string()); + summary.error = Some(error.clone()); + // Lifecycle outbox: the engine channel closed before a terminal + // turn receipt. Every emitted `turn_start` still gets its matching + // `turn_end` so a supervisor never sees an orphaned in-progress + // turn. No-op when the feature is disabled. + lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "turn_end".to_string(), + kind: "turn.failed".to_string(), + thread_id: latest_session_id.clone().unwrap_or_default(), + turn_id: None, + item_id: None, + payload: serde_json::json!({ + "status": "failed", + "duration_ms": exec_turn_started_at.elapsed().as_millis() as u64, + "workspace": latest_workspace.display().to_string(), + "error": codewhale_hooks::bounded_text( + &error, + codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + ), + }), + }); + if output_format == ExecOutputFormat::StreamJson { + emit_exec_stream_event(&ExecStreamEvent::Error { error })?; + } + } + + if json_output { + println!("{}", serde_json::to_string_pretty(&summary)?); + } + + if let Some(error) = summary.error.as_ref() + && !error.trim().is_empty() + { + // Distinguish retryable infrastructure failures (provider/transport, + // after all in-session retries are exhausted) from genuine task + // failures so supervisors and bench harnesses can tell them apart at + // the process level without parsing the stream. Genuine failures + // keep the historical `bail!` → exit 1 path. + let exit_code = exec_failure_exit_code(summary.error_category.as_deref()); + if exit_code != 1 { + eprintln!("Error: exec turn failed: {error}"); + let _ = io::stdout().flush(); + std::process::exit(exit_code); + } + bail!("exec turn failed: {error}"); + } + + if matches!( + summary.status.as_deref(), + Some("failed" | "canceled" | "interrupted") + ) { + let status = summary.status.as_deref().unwrap_or("unknown"); + bail!("exec turn ended with status {status}"); + } + + Ok(()) +} diff --git a/crates/tui/src/lib.rs b/crates/tui/src/lib.rs index 3765194851..21d6185524 100644 --- a/crates/tui/src/lib.rs +++ b/crates/tui/src/lib.rs @@ -11573,973 +11573,10 @@ fn spawn_parent_death_watch() { .expect("spawn parent-death watch thread"); } -fn exec_max_steps(max_turns: Option) -> u32 { - max_turns.unwrap_or(u32::MAX) -} - -#[allow(clippy::too_many_arguments)] -async fn run_exec_agent( - config: &Config, - model: &str, - prompt: &str, - workspace: PathBuf, - max_subagents: usize, - auto_approve: bool, - allow_sandbox_elevation: bool, - explicit_sandbox: Option<&str>, - trust_mode: bool, - json_output: bool, - resume_session: Option, - force_configured_route: bool, - output_format: ExecOutputFormat, - max_turns: u32, - max_tool_calls: Option, - allowed_tools: Option>, - disallowed_tools: Option>, - append_system_prompt: Option, - tool_authority_json: Option, - plugin_registry: std::sync::Arc, -) -> Result<()> { - use crate::compaction::CompactionConfig; - use crate::core::engine::{EngineConfig, spawn_engine}; - use crate::core::events::Event; - use crate::core::ops::Op; - use crate::tools::plan::new_shared_plan_state; - use crate::tools::todo::new_shared_todo_list; - use crate::tui::app::AppMode; - - validate_exec_tool_authority_resume(tool_authority_json.as_deref(), resume_session.is_some())?; - let fleet_authority = tool_authority_json - .as_deref() - .map(crate::tools::spec::ToolAuthorityEnvelope::from_json) - .transpose() - .map_err(anyhow::Error::msg)?; - let fleet_authority_active = fleet_authority.is_some(); - let outer_network_access = fleet_authority - .as_ref() - .and_then(|authority| authority.network_access); - let outer_shell_authority = fleet_authority - .as_ref() - .map(|authority| authority.shell) - .unwrap_or_default(); - if let Some(envelope) = fleet_authority { - crate::tools::spec::install_process_tool_authority(envelope).map_err(anyhow::Error::msg)?; - } - - let route = resolve_cli_exec_route(config, model, prompt, force_configured_route).await?; - let execution_config = config_for_cli_route(config, &route); - let auto_model = route.auto_model; - let effective_provider = route.provider; - let effective_model = route.model; - let validated_route = crate::route_runtime::resolve_runtime_route( - &execution_config, - effective_provider, - Some(&effective_model), - ) - .map_err(anyhow::Error::msg)? - .validate() - .map_err(anyhow::Error::msg)?; - let effective_provider_name = validated_route.identity.key.clone(); - let effective_provider_id = validated_route.identity.exact_id.clone(); - let (effective_provider_kind, effective_stream_provider_id) = - exec_stream_provider_route(&validated_route.identity); - let route_source = if auto_model { - "auto_resolver" - } else { - "explicit_or_configured" - } - .to_string(); - let exec_started = Instant::now(); - let prompt_sha256 = format!("sha256:{}", crate::hashing::sha256_hex(prompt.as_bytes())); - let binary_sha256 = current_binary_sha256(); - let approval_posture = if auto_approve { "auto_tools" } else { "ask" }.to_string(); - let sandbox_posture = explicit_sandbox.unwrap_or("configured_default").to_string(); - let active_route_limits = - crate::route_budget::known_route_limits(validated_route.candidate.limits()); - let max_subagents = if max_subagents == config.max_subagents_for_provider(config.api_provider()) - { - execution_config - .max_subagents_for_provider(effective_provider) - .clamp(1, MAX_SUBAGENTS) - } else { - max_subagents - }; - // A FIXED model with `--reasoning-effort auto` (the exact shape a Fleet - // worker subprocess launches with: `--model --reasoning-effort - // auto`) is still Auto. `auto_model` is a *model* decision and is false - // here, so deriving the auto flag from it left this path both raw and - // non-auto: the literal string `"auto"` travelled to the engine while the - // receipt claimed no Auto was in play. - let reasoning_effort_auto = route.auto_controls_reasoning; - // Resolve Auto against this run's prompt at the CLI boundary, exactly like - // `run_one_shot`/`run_one_shot_json` and the interactive launch path do, - // so the tier the engine (and the receipt below) sees is concrete. - let effective_reasoning_effort = route.reasoning_effort.and_then(|effort| { - cli_reasoning_effort_value_for_prompt(&execution_config, &effective_model, effort, prompt) - }); - - let settings = crate::settings::Settings::load().unwrap_or_default(); - let auto_compact_enabled = if crate::settings::Settings::auto_compact_explicitly_configured() { - settings.auto_compact - } else { - crate::route_budget::auto_compact_default_for_route( - effective_provider, - &effective_model, - active_route_limits, - ) - }; - let compaction = CompactionConfig { - enabled: auto_compact_enabled, - model: effective_model.clone(), - effective_context_window: Some(crate::route_budget::route_context_window_tokens( - effective_provider, - &effective_model, - active_route_limits, - )), - token_threshold: crate::route_budget::compaction_threshold_for_route_at_percent( - effective_provider, - &effective_model, - active_route_limits, - settings.auto_compact_threshold_percent, - ), - ..Default::default() - }; - - let network_policy = exec_network_policy(&execution_config, outer_network_access); - - let lsp_config = (!fleet_authority_active) - .then(|| { - execution_config - .lsp - .clone() - .map(crate::config::LspConfigToml::into_runtime) - }) - .flatten(); - let mut engine_features = execution_config.features(); - apply_fleet_engine_feature_caps( - &mut engine_features, - fleet_authority_active, - outer_network_access, - outer_shell_authority, - ); - if crate::core::allowlist_is_native_file_and_shell_only(allowed_tools.as_deref()) { - engine_features.disable(crate::features::Feature::Mcp); - } - let engine_plugin_registry = if fleet_authority_active { - std::sync::Arc::new(crate::plugins::PluginRegistry::empty(&workspace)) - } else { - plugin_registry - }; - let exec_allow_shell = crate::tools::spec::fleet_exec_shell_enabled( - fleet_authority_active, - outer_shell_authority, - disallowed_tools.as_deref(), - ) || (!fleet_authority_active - && (auto_approve || execution_config.allow_shell())); - let persist_services_enabled = cfg!(unix) - && !fleet_authority_active - && exec_allow_shell - && explicit_sandbox - .is_some_and(|sandbox| sandbox.eq_ignore_ascii_case("danger-full-access")); - let exec_shell_manager = crate::tools::shell::new_shared_shell_manager(workspace.clone()); - let runtime_services = crate::tools::spec::RuntimeToolServices { - shell_manager: Some(exec_shell_manager.clone()), - persist_services_enabled, - ..crate::tools::spec::RuntimeToolServices::default() - }; - - let engine_config = EngineConfig { - model: effective_model.clone(), - active_route_limits, - workspace: workspace.clone(), - subagent_state_root: None, - plugin_registry: Some(std::sync::Arc::clone(&engine_plugin_registry)), - allow_shell: exec_allow_shell, - trust_mode, - notes_path: execution_config.notes_path(), - mcp_config_path: execution_config.mcp_config_path(), - skills_dir: execution_config.skills_dir(), - skills_scan_codewhale_only: execution_config.skills_config().scan_codewhale_only(), - instructions: { - let mut instrs: Vec = execution_config - .instructions_paths() - .into_iter() - .map(Into::into) - .collect(); - if let Some(ref extra) = append_system_prompt { - instrs.push(crate::prompts::InstructionSource::Inline { - name: "cli:append-system-prompt".into(), - content: extra.clone(), - }); - } - instrs - }, - project_context_pack_enabled: execution_config.project_context_pack_enabled(), - translation_enabled: false, - max_steps: max_turns, - max_subagents, - max_admitted_subagents: execution_config - .max_admitted_subagents_for_provider(effective_provider) - .max(max_subagents), - launch_concurrency: execution_config.launch_concurrency_for_provider(effective_provider), - subagents_enabled: !fleet_authority_active - && execution_config.subagents_enabled_for_provider(effective_provider), - features: engine_features, - auto_review_policy: execution_config.auto_review_policy(), - compaction: compaction.clone(), - todos: new_shared_todo_list(), - plan_state: new_shared_plan_state(), - goal_state: crate::tools::goal::new_shared_goal_state(), - max_spawn_depth: if fleet_authority_active { - 0 - } else { - execution_config.subagent_max_spawn_depth_for_provider(effective_provider) - }, - subagent_token_budget: execution_config - .subagent_token_budget_for_provider(effective_provider), - network_policy, - snapshots_enabled: !fleet_authority_active && execution_config.snapshots_config().enabled, - snapshots_max_workspace_bytes: execution_config - .snapshots_config() - .max_workspace_gb - .saturating_mul(1024 * 1024 * 1024), - lsp_config, - runtime_services, - subagent_model_overrides: execution_config.subagent_model_overrides(), - fleet_roster: std::sync::Arc::new(crate::fleet::identity::load_effective_roster( - &execution_config.fleet_config(), - &workspace, - Some(engine_plugin_registry.as_ref()), - )), - subagent_api_timeout: std::time::Duration::from_secs( - execution_config.subagent_api_timeout_secs_for_provider(effective_provider), - ), - stream_chunk_timeout: std::time::Duration::from_secs( - execution_config.stream_chunk_timeout_secs(), - ), - subagent_heartbeat_timeout: std::time::Duration::from_secs( - execution_config.subagent_heartbeat_timeout_secs_for_provider(effective_provider), - ), - prefer_bwrap: execution_config.prefer_bwrap.unwrap_or(false), - bwrap_extensions: crate::sandbox::BwrapMountExtensions { - read_only_roots: execution_config.bwrap_ro_roots.clone(), - device_roots: execution_config.bwrap_dev_roots.clone(), - }, - denied_read_subpaths: execution_config.sandbox_denied_read_paths.clone(), - memory_enabled: execution_config.memory_enabled(), - memory_path: execution_config.memory_path(), - speech_output_dir: execution_config.speech_output_dir(), - vision_config: execution_config.vision_model_config(), - strict_tool_mode: execution_config.strict_tool_mode.unwrap_or(false), - goal_objective: None, - goal_token_budget: None, - goal_status: crate::tools::goal::GoalStatus::Active, - goal_max_continuations: execution_config.goal_max_continuations(), - goal_continuation_delay_seconds: execution_config.goal_continuation_delay_seconds(), - allowed_tools: allowed_tools.clone(), - disallowed_tools: disallowed_tools.clone(), - max_tool_calls, - hook_executor: None, - locale_tag: crate::localization::resolve_locale(&settings.locale) - .tag() - .to_string(), - workshop: { - crate::tools::large_output_router::WorkshopConfig::install_active( - config.workshop.as_ref(), - ); - config.workshop.clone() - }, - search_provider: execution_config.search_provider(), - search_api_key: execution_config - .search - .as_ref() - .and_then(|s| s.api_key.clone()), - search_base_url: execution_config - .search - .as_ref() - .and_then(|s| s.base_url.clone()), - tools_always_load: if fleet_authority_active { - std::collections::HashSet::new() - } else { - execution_config.tools_always_load() - }, - tools: if fleet_authority_active { - None - } else { - execution_config.tools.clone() - }, - verbosity: execution_config.verbosity.clone(), - workspace_follow_symlinks: settings.workspace_follow_symlinks, - exec_policy_engine: execution_config.exec_policy_engine.clone(), - terminal_chrome_enabled: false, - advisor_config: execution_config - .advisor - .as_ref() - .map(crate::tools::subagent::AdvisorConfig::from_toml) - .unwrap_or_else(crate::tools::subagent::AdvisorConfig::disabled), - }; - - let engine_handle = spawn_engine(engine_config, &execution_config); - let mode = if auto_approve { - AppMode::Yolo - } else { - AppMode::Agent - }; - - let resuming_session = resume_session.is_some(); - let mut loaded_session_id = None; - if let Some(saved) = resume_session { - let saved_id = saved.metadata.id.clone(); - if saved.metadata.workspace != workspace && output_format == ExecOutputFormat::Text { - eprintln!( - "Warning: session {} was created in a different workspace ({}). Resuming anyway.", - truncate_id(&saved_id), - saved.metadata.workspace.display(), - ); - } - - engine_handle - .send(Op::SyncSession { - session_id: Some(saved_id.clone()), - messages: saved.messages, - system_prompt: saved.system_prompt.map(SystemPrompt::Text), - system_prompt_override: false, - model: saved.metadata.model, - workspace: saved.metadata.workspace, - mode, - }) - .await?; - loaded_session_id = Some(saved_id.clone()); - if output_format == ExecOutputFormat::Text && !json_output { - eprintln!("{}", exec_resumed_session_line(&saved_id)); - } - } - - engine_handle - .send(Op::SendMessage { - content: prompt.to_string(), - mode, - route: Box::new(validated_route.into_resolved()), - compaction: Box::new(compaction.clone()), - goal_objective: None, - goal_token_budget: None, - goal_status: crate::tools::goal::GoalStatus::Active, - allowed_tools: allowed_tools.clone(), - dynamic_tools: Vec::new(), - hook_executor: None, - reasoning_effort: effective_reasoning_effort, - reasoning_effort_auto, - auto_model, - allow_shell: auto_approve || execution_config.allow_shell(), - trust_mode, - auto_approve, - translation_enabled: false, - approval_mode: if auto_approve { - crate::tui::approval::ApprovalMode::Bypass - } else { - execution_config - .approval_policy - .as_deref() - .and_then(crate::tui::approval::ApprovalMode::from_config_value) - .unwrap_or_default() - }, - verbosity: execution_config.verbosity.clone(), - provenance: crate::core::ops::UserInputProvenance::ExternalUser, - }) - .await?; - - let mut summary = ExecSummary { - mode: "agent".to_string(), - provider: effective_provider_name.clone(), - model: effective_model.clone(), - prompt: prompt.to_string(), - ..ExecSummary::default() - }; - let can_elevate_sandbox = - exec_sandbox_elevation_authorized(allow_sandbox_elevation, explicit_sandbox); - let mut sandbox_denied = false; - let mut approval_required = false; - let mut tool_error_seen = false; - let mut last_error_category = None; - let mut reported_sandbox_contract = false; - - let should_persist_session = resuming_session || output_format == ExecOutputFormat::StreamJson; - let mut latest_session_id = loaded_session_id; - let mut latest_messages: Vec = Vec::new(); - let mut latest_system_prompt: Option = None; - let mut latest_model = effective_model; - let mut latest_workspace = workspace.clone(); - let mut tool_starts: HashMap = HashMap::new(); - let mut turn_usage_seq: u32 = 0; - - let mut stdout = io::stdout(); - let mut ends_with_newline = false; - loop { - let event = { - let mut rx = engine_handle.rx_event.write().await; - rx.recv().await - }; - - let Some(event) = event else { - break; - }; - - match event { - Event::MessageDelta { content, .. } => { - summary.output.push_str(&content); - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::Content { content })?; - } else if !json_output { - print!("{content}"); - stdout.flush()?; - } - ends_with_newline = summary.output.ends_with('\n'); - } - Event::MessageComplete { .. } - if output_format == ExecOutputFormat::Text - && !json_output - && !ends_with_newline => - { - println!(); - } - Event::ThinkingDelta { .. } => { - // Exec stream-json intentionally omits reasoning deltas; the - // TUI transcript retains its existing Activity Detail surface. - } - Event::ToolCallStarted { id, name, input } => { - let started_at = chrono::Utc::now().to_rfc3339(); - tool_starts.insert(id.clone(), (Instant::now(), started_at.clone())); - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::ToolUse { - name, - id, - input, - started_at, - })?; - } else if !json_output { - let summary = summarize_tool_args(&input); - if let Some(summary) = summary { - eprintln!("tool: {name} ({summary})"); - } else { - eprintln!("tool: {name}"); - } - } - } - Event::ToolCallComplete { - id, name, result, .. - } => { - let (duration_ms, started_at) = tool_starts - .remove(&id) - .map(|(started, timestamp)| { - ( - u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), - timestamp, - ) - }) - .unwrap_or_else(|| (0, chrono::Utc::now().to_rfc3339())); - let receipt_name = name.clone(); - match result { - Ok(output) => { - tool_error_seen |= !output.success; - summary.tools.push(ExecToolEntry { - name: name.clone(), - success: output.success, - output: output.content.clone(), - }); - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::ToolResult { - id, - name: receipt_name, - output: output.content, - status: if output.success { - "success".to_string() - } else { - "error".to_string() - }, - started_at, - completed_at: chrono::Utc::now().to_rfc3339(), - duration_ms, - side_effect_status: output - .metadata - .as_ref() - .and_then(|metadata| metadata.get("side_effect_status")) - .and_then(serde_json::Value::as_str) - .unwrap_or("unknown") - .to_string(), - error_category: (!output.success).then(|| { - output - .metadata - .as_ref() - .and_then(|metadata| metadata.get("error_category")) - .and_then(serde_json::Value::as_str) - .unwrap_or("tool_reported_failure") - .to_string() - }), - truncated: output - .metadata - .as_ref() - .and_then(|metadata| metadata.get("truncated")) - .and_then(serde_json::Value::as_bool), - artifact: tool_artifact_receipt(output.metadata.as_ref()), - result_metadata: output.metadata, - })?; - } else if !json_output { - if name == "exec_shell" && !output.content.trim().is_empty() { - eprintln!("tool {name} completed"); - eprintln!( - "--- stdout/stderr ---\n{}\n---------------------", - output.content - ); - } else { - eprintln!( - "tool {name} completed: {}", - summarize_tool_output(&output.content) - ); - } - } - } - Err(err) => { - tool_error_seen = true; - let error_text = err.to_string(); - summary.tools.push(ExecToolEntry { - name: name.clone(), - success: false, - output: error_text.clone(), - }); - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::ToolResult { - id, - name: receipt_name, - output: error_text, - status: "error".to_string(), - started_at, - completed_at: chrono::Utc::now().to_rfc3339(), - duration_ms, - side_effect_status: "not_started_or_unknown".to_string(), - error_category: Some(tool_error_receipt_category(&err).to_string()), - truncated: None, - artifact: None, - result_metadata: None, - })?; - } else if !json_output { - eprintln!("tool {name} failed: {err}"); - } - } - } - } - Event::AgentSpawned { id, prompt, .. } - if output_format == ExecOutputFormat::Text && !json_output => - { - eprintln!("sub-agent {id} spawned: {}", summarize_tool_output(&prompt)); - } - Event::AgentProgress { id, status, .. } - if output_format == ExecOutputFormat::Text && !json_output => - { - eprintln!("sub-agent {id}: {status}"); - } - Event::AgentComplete { id, result, .. } - if output_format == ExecOutputFormat::Text && !json_output => - { - eprintln!( - "sub-agent {id} completed: {}", - summarize_tool_output(&result) - ); - } - Event::AgentSpawned { - id, - parent_run_id, - spawn_depth, - model, - route_source, - .. - } if output_format == ExecOutputFormat::StreamJson => { - emit_exec_stream_event(&ExecStreamEvent::AgentSpawned { - id, - model, - spawn_depth, - parent_run_id, - route_source, - })?; - } - Event::AgentSpawned { .. } - | Event::AgentProgress { .. } - | Event::AgentComplete { .. } => {} - Event::WorkflowUi { run_id, event, .. } - if output_format == ExecOutputFormat::StreamJson => - { - emit_exec_stream_event(&ExecStreamEvent::WorkflowEvent { run_id, event })?; - } - Event::ApprovalRequired { id, .. } => { - if auto_approve { - let _ = engine_handle.approve_tool_call(id).await; - } else { - approval_required = true; - let _ = engine_handle.deny_tool_call(id).await; - } - } - Event::ElevationRequired { - tool_id, - tool_name, - denial_reason, - .. - } => { - if can_elevate_sandbox { - let policy = crate::sandbox::SandboxPolicy::DangerFullAccess; - let _ = engine_handle.retry_tool_with_policy(tool_id, policy).await; - } else { - sandbox_denied = true; - approval_required = true; - summary.outcomes.push(ExecOutcome { - kind: "sandbox_denied".to_string(), - outcome: "approval_required".to_string(), - tool_name: tool_name.clone(), - reason: denial_reason.clone(), - }); - if !reported_sandbox_contract { - eprintln!( - "sandbox denied {tool_name}: {denial_reason}; --auto approves tools but does not elevate sandbox access — use --sandbox danger-full-access or --allow-sandbox-elevation to opt in" - ); - reported_sandbox_contract = true; - } - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::SandboxDenied { - tool_id: tool_id.clone(), - tool_name, - reason: denial_reason, - outcome: "approval_required".to_string(), - })?; - } - let _ = engine_handle.deny_tool_call(tool_id).await; - } - } - Event::Error { - envelope, - recoverable: _, - } => { - // Only a non-recoverable envelope may force the run summary - // into failure. Recoverable warnings (stream-stall notices, - // transient retry noise) are still streamed for visibility, - // but the terminal TurnComplete event carries the - // authoritative turn outcome — letting a warning set - // `summary.error` here would exit an otherwise-successful - // `exec` run non-zero. - if exec_error_event_is_fatal(&envelope) { - last_error_category = Some(envelope.category); - summary.error_category = Some(envelope.category.to_string()); - summary.error = Some(envelope.message.clone()); - } - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::Error { - error: envelope.message, - })?; - } else if !json_output { - eprintln!("error: {}", envelope.message); - } - } - Event::TurnUsage { - usage, duration_ms, .. - } => { - if output_format == ExecOutputFormat::StreamJson { - turn_usage_seq = turn_usage_seq.saturating_add(1); - emit_exec_stream_event(&ExecStreamEvent::TurnUsage { - turn: turn_usage_seq, - input_tokens: usage.input_tokens, - output_tokens: usage.output_tokens, - reasoning_tokens: usage.reasoning_tokens, - prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens, - prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens, - prompt_cache_write_tokens: usage.prompt_cache_write_tokens, - reasoning_replay_tokens: usage.reasoning_replay_tokens, - duration_ms, - })?; - } - } - Event::TurnComplete { - status, - error, - usage, - tool_catalog, - .. - } => { - let (terminal_status, terminal_error) = (status, error); - #[cfg(unix)] - let (mut terminal_status, mut terminal_error) = (terminal_status, terminal_error); - if matches!( - terminal_status, - crate::core::events::TurnOutcomeStatus::Completed - ) && terminal_error.is_none() - { - #[cfg(unix)] - match exec_shell_manager.lock() { - Ok(mut manager) => match manager.commit_persistent_services() { - Ok(receipts) => { - for receipt in &receipts { - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event( - &ExecStreamEvent::ServiceReleased { - task_id: receipt.task_id.clone(), - pid: receipt.pid, - process_group_id: receipt.process_group_id, - ownership: receipt.ownership.clone(), - }, - )?; - } else if !json_output { - eprintln!( - "persistent service released: {} pid={} pgid={} ownership={}", - receipt.task_id, - receipt.pid, - receipt.process_group_id, - receipt.ownership - ); - } - } - summary.released_services.extend(receipts); - } - Err(error) => { - manager.abort_persistent_services(); - terminal_status = crate::core::events::TurnOutcomeStatus::Failed; - terminal_error = Some(format!( - "Persistent service ownership transfer failed: {error}" - )); - } - }, - Err(_) => { - terminal_status = crate::core::events::TurnOutcomeStatus::Failed; - terminal_error = Some( - "Persistent service ownership transfer failed: shell manager lock poisoned" - .to_string(), - ); - } - } - } else if let Ok(mut manager) = exec_shell_manager.lock() { - manager.abort_persistent_services(); - } - summary.status = Some(format!("{terminal_status:?}").to_lowercase()); - if terminal_error.is_some() { - summary.error = terminal_error; - } - if sandbox_denied - && summary.error.is_none() - && matches!( - terminal_status, - crate::core::events::TurnOutcomeStatus::Failed - ) - { - summary.error = Some( - "exec turn failed after sandbox denial; explicit sandbox elevation was not authorized" - .to_string(), - ); - } - if last_error_category.is_none() { - last_error_category = summary - .error - .as_deref() - .map(crate::error_taxonomy::classify_error_message); - summary.error_category = - last_error_category.map(|category| category.to_string()); - } - let termination_reason = crate::core::termination::classify_turn_termination( - terminal_status, - last_error_category, - tool_error_seen, - approval_required, - ); - summary.termination_reason = Some(termination_reason.as_str().to_string()); - // State the exit class here rather than inferring it later - // from the process exit code: `Canceled` exits 130, the same - // value the SIGINT path uses, so a code-based derivation would - // report every Esc-cancelled turn as a signal. A no-op unless - // this process was armed. - if !termination_reason.is_success() { - codewhale_telemetry::set_exit_class(codewhale_telemetry::ExitClass::Error); - } - let saved_session_id = if should_persist_session && !latest_messages.is_empty() { - match persist_exec_session( - &latest_messages, - &latest_model, - PersistedProviderRoute { - kind: effective_provider.as_str(), - id: effective_provider_id.as_deref(), - }, - &latest_workspace, - &latest_system_prompt, - latest_session_id.as_deref(), - u64::from(usage.input_tokens) + u64::from(usage.output_tokens), - ) { - Ok(id) => { - if output_format == ExecOutputFormat::Text && !json_output { - eprintln!("{}", exec_saved_session_line(&id)); - } - Some(id) - } - Err(err) => { - if output_format == ExecOutputFormat::Text && !json_output { - eprintln!("warning: failed to save exec session: {err}"); - } - latest_session_id.clone() - } - } - } else { - latest_session_id.clone() - }; - if output_format == ExecOutputFormat::StreamJson { - if let Some(id) = saved_session_id.as_ref() { - emit_exec_stream_event(&ExecStreamEvent::SessionCapture { - content: exec_stream_session_ref(id), - })?; - } - // Resolved output ceiling and its provenance, surfaced so a - // wrong ceiling is visible in the receipt rather than - // requiring packet capture. - let codewhale_max_output_tokens = - crate::route_budget::effective_max_output_tokens_for_route( - effective_provider, - &latest_model, - active_route_limits, - ); - let codewhale_max_output_tokens_source = - crate::route_budget::output_ceiling_source( - effective_provider, - &latest_model, - ) - .as_str(); - emit_exec_stream_event(&ExecStreamEvent::Metadata { - meta: Box::new(ExecStreamMeta { - receipt_kind: "terminal", - provider: effective_provider_kind.clone(), - provider_id: effective_stream_provider_id.clone(), - model: latest_model.clone(), - route_source: route_source.clone(), - input_tokens: Some(usage.input_tokens), - output_tokens: Some(usage.output_tokens), - prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens, - prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens, - prompt_cache_write_tokens: usage.prompt_cache_write_tokens, - reasoning_tokens: usage.reasoning_tokens, - codewhale_max_output_tokens: Some(codewhale_max_output_tokens), - codewhale_max_output_tokens_source: Some( - codewhale_max_output_tokens_source, - ), - duration_ms: u64::try_from(exec_started.elapsed().as_millis()) - .unwrap_or(u64::MAX), - retry_count: None, - approval_posture: approval_posture.clone(), - sandbox_posture: sandbox_posture.clone(), - binary_sha256: binary_sha256.clone(), - config_sha256: None, - prompt_sha256: prompt_sha256.clone(), - tool_catalog_sha256: tool_catalog.as_ref().and_then(|catalog| { - serde_json::to_vec(catalog).ok().map(|bytes| { - format!("sha256:{}", crate::hashing::sha256_hex(&bytes)) - }) - }), - input_analysis: exec_stream_input_analysis( - &latest_messages, - latest_system_prompt.as_ref(), - ), - visible_final_answer_chars: summary.output.chars().count(), - resume_command: saved_session_id - .as_deref() - .map(exec_stream_resume_hint) - .unwrap_or_default(), - session_id: saved_session_id - .as_deref() - .map(exec_stream_session_ref) - .unwrap_or_default(), - workspace: latest_workspace.display().to_string(), - message_count: latest_messages.len(), - status: summary.status.clone(), - termination_reason: summary.termination_reason.clone(), - error_category: summary.error_category.clone(), - error: summary.error.clone(), - }), - })?; - emit_exec_stream_event(&ExecStreamEvent::Done)?; - } - let _ = engine_handle.send(Op::Shutdown).await; - break; - } - Event::SessionUpdated { - session_id, - messages, - system_prompt, - model, - workspace, - } => { - latest_session_id = Some(session_id); - latest_messages = messages; - latest_system_prompt = system_prompt; - latest_model = model; - latest_workspace = workspace; - } - // #3027: surface the engine's max-steps notice in text mode so a - // --max-turns run that stops early says why instead of going quiet. - Event::Status { message } - if output_format == ExecOutputFormat::Text - && !json_output - && message.contains("Maximum model steps") => - { - eprintln!("{message}"); - } - _ => {} - } - } - - if summary.status.is_none() { - if let Ok(mut manager) = exec_shell_manager.lock() { - manager.abort_persistent_services(); - } - let error = summary.error.clone().unwrap_or_else(|| { - "Engine event channel closed before a terminal turn receipt".to_string() - }); - let category = last_error_category - .unwrap_or_else(|| crate::error_taxonomy::classify_error_message(&error)); - let termination_reason = crate::core::termination::classify_turn_termination( - crate::core::events::TurnOutcomeStatus::Failed, - Some(category), - tool_error_seen, - approval_required, - ); - summary.status = Some("failed".to_string()); - summary.error_category = Some(category.to_string()); - summary.termination_reason = Some(termination_reason.as_str().to_string()); - summary.error = Some(error.clone()); - if output_format == ExecOutputFormat::StreamJson { - emit_exec_stream_event(&ExecStreamEvent::Error { error })?; - } - } - - if json_output { - println!("{}", serde_json::to_string_pretty(&summary)?); - } - - if let Some(error) = summary.error.as_ref() - && !error.trim().is_empty() - { - // Distinguish retryable infrastructure failures (provider/transport, - // after all in-session retries are exhausted) from genuine task - // failures so supervisors and bench harnesses can tell them apart at - // the process level without parsing the stream. Genuine failures - // keep the historical `bail!` → exit 1 path. - let exit_code = exec_failure_exit_code(summary.error_category.as_deref()); - if exit_code != 1 { - eprintln!("Error: exec turn failed: {error}"); - let _ = io::stdout().flush(); - std::process::exit(exit_code); - } - bail!("exec turn failed: {error}"); - } - - if matches!( - summary.status.as_deref(), - Some("failed" | "canceled" | "interrupted") - ) { - let status = summary.status.as_deref().unwrap_or("unknown"); - bail!("exec turn ended with status {status}"); - } - - Ok(()) -} +// The non-interactive exec agent assembly lives in `exec_agent`; the +// glob re-export keeps the dispatch and test references unchanged (#5586). +mod exec_agent; +pub(crate) use exec_agent::*; #[cfg(test)] mod serve_bind_host_tests { diff --git a/crates/tui/src/tui/app.rs b/crates/tui/src/tui/app.rs index 282c956dc6..28def2a755 100644 --- a/crates/tui/src/tui/app.rs +++ b/crates/tui/src/tui/app.rs @@ -1688,6 +1688,9 @@ pub struct App { pub api_key_env_only: bool, // Hooks system pub hooks: HookExecutor, + /// Lifecycle event outbox (`[lifecycle_outbox]` config). Disabled + /// (all emits no-ops) when no path is configured. + pub lifecycle_outbox: codewhale_hooks::LifecycleOutbox, #[allow(dead_code)] pub yolo: bool, /// One-shot YOLO→Act+Bypass migration notice for this session (#0.8.68 M6). diff --git a/crates/tui/src/tui/app/init.rs b/crates/tui/src/tui/app/init.rs index 70995a5712..23ca624a67 100644 --- a/crates/tui/src/tui/app/init.rs +++ b/crates/tui/src/tui/app/init.rs @@ -612,6 +612,20 @@ impl App { ); let hooks = HookExecutor::new(hooks_config, workspace.clone()); + // Initialize the lifecycle event outbox (`[lifecycle_outbox]`). + // Disabled (all emits no-op) when the config has no path. + let lifecycle_outbox = config + .lifecycle_outbox + .as_ref() + .map(|outbox| { + codewhale_hooks::LifecycleOutbox::new( + outbox.path.clone(), + outbox.webhook_url.clone(), + outbox.webhook_token.clone(), + ) + }) + .unwrap_or_else(codewhale_hooks::LifecycleOutbox::disabled); + // Initialize plan state let plan_state = new_shared_plan_state(); let todos = new_shared_todo_list(); @@ -890,6 +904,7 @@ impl App { onboarding_had_trust_step: !was_onboarded && needs_workspace_trust, api_key_env_only, hooks, + lifecycle_outbox, yolo: yolo_compat, yolo_compat_notified: false, startup_defaults: Default::default(), diff --git a/crates/tui/src/tui/ui/event_loop.rs b/crates/tui/src/tui/ui/event_loop.rs index fc3468e570..b1e094461e 100644 --- a/crates/tui/src/tui/ui/event_loop.rs +++ b/crates/tui/src/tui/ui/event_loop.rs @@ -525,6 +525,12 @@ pub async fn run_tui( // Fire session start hook { let context = app.base_hook_context(); + // Captured before the hook executor moves `context` into its blocking + // task; the outbox emit below needs the same session identity. + let outbox_thread_id = context.session_id.clone().unwrap_or_default(); + let outbox_mode = context.mode.clone(); + let outbox_model = context.model.clone(); + let outbox_workspace = context.workspace.clone(); let hooks = app.hooks.clone(); if let Err(error) = tokio::task::spawn_blocking(move || hooks.execute(HookEvent::SessionStart, &context)) @@ -533,6 +539,23 @@ pub async fn run_tui( tracing::error!(target: "hooks", %error, "session_start executor task was lost"); app.status_message = Some("session_start hook executor did not run".to_string()); } + // Lifecycle outbox (`[lifecycle_outbox]`): fires alongside the + // session_start hook, with the same session identity. No-op when + // the feature is disabled. + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "session_start".to_string(), + kind: "session.started".to_string(), + thread_id: outbox_thread_id, + turn_id: None, + item_id: None, + payload: serde_json::json!({ + "mode": outbox_mode, + "model": outbox_model, + "workspace": outbox_workspace + .as_ref() + .map(|path| path.display().to_string()), + }), + }); } // Spawn the persistence actor so checkpoint/session-save I/O stays off @@ -621,6 +644,22 @@ pub async fn run_tui( { let context = app.base_hook_context(); let _ = app.execute_hooks(HookEvent::SessionEnd, &context); + // Lifecycle outbox (`[lifecycle_outbox]`): fires alongside the + // session_end hook, with the same session identity. No-op when + // the feature is disabled. + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "session_end".to_string(), + kind: "session.ended".to_string(), + thread_id: context.session_id.clone().unwrap_or_default(), + turn_id: None, + item_id: None, + payload: serde_json::json!({ + "workspace": context.workspace + .as_ref() + .map(|path| path.display().to_string()), + "total_tokens": context.total_tokens, + }), + }); } // Flush the persistence actor: clear this session's checkpoint, collect @@ -1526,6 +1565,23 @@ pub(crate) async fn run_event_loop( app.last_reasoning = None; app.pending_tool_uses.clear(); last_status_frame = Instant::now(); + // Lifecycle outbox (`[lifecycle_outbox]`): the turn + // boundary the shell-hook system deliberately lacks. + // No-op when the feature is disabled. + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "turn_start".to_string(), + kind: "turn.started".to_string(), + thread_id: app.hooks.session_id().to_string(), + turn_id: app.runtime_turn_id.clone(), + item_id: None, + payload: serde_json::json!({ + "model": codewhale_hooks::bounded_text( + &app.model, + codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + ), + "workspace": app.workspace.display().to_string(), + }), + }); } EngineEvent::ToolRequestSnapshot { snapshot } => { app.session.last_tool_request_snapshot = Some(snapshot); @@ -2035,6 +2091,41 @@ pub(crate) async fn run_event_loop( surface_observer_hook_submission_failure(app, error); } + // Lifecycle outbox (`[lifecycle_outbox]`): one + // `turn_end` event per completed turn, with the kind + // projected from the turn status — `turn.failed` for + // failed turns, `turn.completed` for completed ones, + // `turn.interrupted` for locally cancelled ones. + // No-op when the feature is disabled. + { + let outbox_status = + app.runtime_turn_status.as_deref().unwrap_or("unknown"); + let kind = match outbox_status { + "completed" => "turn.completed", + "failed" => "turn.failed", + "interrupted" => "turn.interrupted", + _ => "turn.ended", + }; + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "turn_end".to_string(), + kind: kind.to_string(), + thread_id: app.hooks.session_id().to_string(), + turn_id: app.runtime_turn_id.clone(), + item_id: None, + payload: serde_json::json!({ + "status": outbox_status, + "duration_ms": turn_elapsed.as_millis() as u64, + "workspace": app.workspace.display().to_string(), + "error": error + .as_deref() + .map(|message| codewhale_hooks::bounded_text( + message, + codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + )), + }), + }); + } + if queued_to_send.is_none() { queued_to_send = app.pop_queued_message(); } diff --git a/crates/tui/src/tui/ui/observer_hooks.rs b/crates/tui/src/tui/ui/observer_hooks.rs index 456d1aa3d1..2a3b1e94ca 100644 --- a/crates/tui/src/tui/ui/observer_hooks.rs +++ b/crates/tui/src/tui/ui/observer_hooks.rs @@ -12,11 +12,60 @@ pub(super) fn execute_subagent_observer_hook( text_field: &str, text: &str, ) -> Result<(), String> { + let (preview, truncated) = bounded_subagent_hook_preview(text); + + // Lifecycle outbox (`[lifecycle_outbox]`): fires even when no shell hook + // is configured for this event — the outbox is independent of the hook + // command list. Preview is bounded (preview ceiling) and only ever the + // preview text, never the raw prompt/result. No-op when disabled. + match &event { + HookEvent::SubagentSpawn => { + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "subagent_spawn".to_string(), + kind: "subagent.spawned".to_string(), + thread_id: app.hooks.session_id().to_string(), + turn_id: app.runtime_turn_id.clone(), + item_id: None, + payload: serde_json::json!({ + "agent_id": agent_id, + "subagent": agent_id, + "workspace": app.workspace.display().to_string(), + "prompt_preview": codewhale_hooks::bounded_text( + &preview, + codewhale_hooks::OUTBOX_PREVIEW_MAX_CHARS, + ), + "prompt_truncated": truncated, + }), + }); + } + HookEvent::SubagentComplete => { + let status = subagent_completion_status(text).unwrap_or_else(|| "unknown".to_string()); + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "subagent_complete".to_string(), + kind: "subagent.completed".to_string(), + thread_id: app.hooks.session_id().to_string(), + turn_id: app.runtime_turn_id.clone(), + item_id: None, + payload: serde_json::json!({ + "agent_id": agent_id, + "subagent": agent_id, + "workspace": app.workspace.display().to_string(), + "status": status, + "result_preview": codewhale_hooks::bounded_text( + &preview, + codewhale_hooks::OUTBOX_PREVIEW_MAX_CHARS, + ), + "result_truncated": truncated, + }), + }); + } + _ => {} + } + if !app.hooks.has_hooks_for_event(event) { return Ok(()); } - let (preview, truncated) = bounded_subagent_hook_preview(text); let context = app.base_hook_context().with_message(&preview); let mut payload = serde_json::json!({ "event": event.as_str(), diff --git a/crates/tui/src/tui/ui/session_state.rs b/crates/tui/src/tui/ui/session_state.rs index 5588234b2f..b1553d7477 100644 --- a/crates/tui/src/tui/ui/session_state.rs +++ b/crates/tui/src/tui/ui/session_state.rs @@ -251,6 +251,10 @@ pub(crate) fn maybe_throttled_recovery_snapshot( } pub(crate) fn recover_stalled_runtime_turn(app: &mut App, message: &str, level: StatusToastLevel) { + // Capture the turn identity before the reset below clears it; the + // outbox event must name the turn that stalled. + let stalled_turn_id = app.runtime_turn_id.clone(); + let stalled_session_id = app.hooks.session_id().to_string(); // Finalize in-flight thinking / assistant / tool cells so the // transcript doesn't show permanent spinners after recovery. streaming_thinking::finalize_current(app); @@ -279,6 +283,24 @@ pub(crate) fn recover_stalled_runtime_turn(app: &mut App, message: &str, level: // Per-turn scroll lock — clear so the next turn auto-scrolls. app.user_scrolled_during_stream = false; app.push_status_toast(message, level, None); + // Lifecycle outbox (`[lifecycle_outbox]`): the first scriptable stall + // signal. Until now a wedged turn was only visible as this toast; with + // the outbox enabled a supervisor can react to the same moment. + // No-op when the feature is disabled. + app.lifecycle_outbox.emit(codewhale_hooks::LifecycleEvent { + event: "turn_stalled".to_string(), + kind: "turn.stalled".to_string(), + thread_id: stalled_session_id, + turn_id: stalled_turn_id, + item_id: None, + payload: serde_json::json!({ + "message": codewhale_hooks::bounded_text( + message, + codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + ), + "workspace": app.workspace.display().to_string(), + }), + }); } pub(crate) fn recover_engine_event_disconnect(app: &mut App) -> bool { @@ -1057,3 +1079,112 @@ mod derived_title_tests { assert_eq!(derive_session_title(&[user("\u{1b}\u{7}\u{200b}")]), None); } } + +#[cfg(test)] +mod stall_outbox_tests { + use super::*; + use crate::tui::app::TuiOptions; + + /// `recover_stalled_runtime_turn` must emit a `turn_stalled` lifecycle + /// outbox event naming the wedged turn — the first scriptable stall + /// signal. The outbox is opt-in, so the test enables it through config. + #[tokio::test] + async fn stalled_turn_emits_turn_stalled_outbox_event() { + let _lock = crate::test_support::lock_test_env(); + let dir = tempfile::tempdir().expect("tempdir"); + let outbox_path = dir.path().join("outbox.jsonl"); + + let config = Config { + lifecycle_outbox: Some(codewhale_config::LifecycleOutboxToml { + path: Some(outbox_path.clone()), + webhook_url: None, + webhook_token: None, + }), + ..Default::default() + }; + let options = TuiOptions { + start_in_agent_mode: true, + ..crate::test_support::test_tui_options(dir.path()) + }; + let mut app = App::new(options, &config); + assert!(app.lifecycle_outbox.is_enabled()); + let expected_workspace = app.workspace.display().to_string(); + + app.runtime_turn_id = Some("turn-1".to_string()); + app.runtime_turn_status = Some("in_progress".to_string()); + app.is_loading = true; + recover_stalled_runtime_turn( + &mut app, + "Turn stalled — no completion signal received", + StatusToastLevel::Error, + ); + + // The outbox writer task drains asynchronously; wait for the line. + let mut lines = Vec::new(); + for _ in 0..200 { + if let Ok(text) = tokio::fs::read_to_string(&outbox_path).await { + lines = text + .lines() + .map(|line| serde_json::from_str::(line).expect("json")) + .collect(); + if !lines.is_empty() { + break; + } + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + + assert_eq!(lines.len(), 1, "expected one turn_stalled outbox line"); + let line = &lines[0]; + assert_eq!(line["event"], "turn_stalled"); + assert_eq!(line["kind"], "turn.stalled"); + assert_eq!(line["turn_id"], "turn-1"); + assert_eq!(line["schema_version"], 1); + assert_eq!(line["seq"], 1); + // Every payload carries the workspace for consumer-side routing. + assert_eq!( + line["payload"]["workspace"], + serde_json::json!(expected_workspace) + ); + // The stall message is engine-authored and safe, but still bounded + // and never raw tool/environment content. + let message = line["payload"]["message"].as_str().expect("message"); + assert!(message.contains("stalled")); + assert!( + message.chars().count() <= codewhale_hooks::OUTBOX_DETAIL_MAX_CHARS, + "stall message must be bounded" + ); + } + + /// A disabled outbox (config without a path) must make stall recovery + /// behave exactly as before: the toast still lands, no file is written. + #[tokio::test] + async fn stalled_turn_without_outbox_config_writes_nothing() { + let _lock = crate::test_support::lock_test_env(); + let dir = tempfile::tempdir().expect("tempdir"); + let options = TuiOptions { + start_in_agent_mode: true, + ..crate::test_support::test_tui_options(dir.path()) + }; + let mut app = App::new(options, &Config::default()); + assert!(!app.lifecycle_outbox.is_enabled()); + + app.runtime_turn_id = Some("turn-1".to_string()); + app.runtime_turn_status = Some("in_progress".to_string()); + app.is_loading = true; + recover_stalled_runtime_turn( + &mut app, + "Turn stalled — no completion signal received", + StatusToastLevel::Error, + ); + + // Recovery still clears the wedged turn state and posts the toast. + assert!(app.runtime_turn_id.is_none()); + assert!(!app.is_loading); + assert!(!app.status_toasts.is_empty()); + assert!( + !dir.path().join("outbox.jsonl").exists(), + "no outbox file must be created when the feature is off" + ); + } +} diff --git a/crates/tui/tests/integration/lifecycle_outbox_exec.rs b/crates/tui/tests/integration/lifecycle_outbox_exec.rs new file mode 100644 index 0000000000..0bbea10fa4 --- /dev/null +++ b/crates/tui/tests/integration/lifecycle_outbox_exec.rs @@ -0,0 +1,343 @@ +//! End-to-end contract for the `[lifecycle_outbox]` feature on headless +//! `codewhale exec`: with a path configured, a run appends one JSONL +//! `RuntimeEventEnvelope` line per turn boundary (`turn_start` at message +//! dispatch, `turn_end` at the terminal receipt), the per-file `seq` recovers +//! across processes, and with no path configured no file is ever created. +//! +//! A `wiremock` OpenAI-compatible endpoint stands in for the provider, so the +//! run is a real `exec` process end to end — same loader, same engine, same +//! outbox writer — with no external network. + +#![cfg(unix)] + +use std::io::Read; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::time::Duration; + +use serde_json::{Value, json}; +use tempfile::TempDir; +use wait_timeout::ChildExt; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const TEST_MODEL: &str = "lifecycle-outbox-model"; +const RUN_TIMEOUT: Duration = Duration::from_secs(60); + +/// Placeholder in `outbox_toml` replaced with the isolated home's absolute +/// outbox path (so callers can read the file back after the run). +const OUTBOX_PATH_TOKEN: &str = "__OUTBOX_PATH__"; + +fn sse_chunk(value: Value) -> String { + format!( + "data: {}\n\n", + serde_json::to_string(&value).expect("SSE JSON") + ) +} + +/// Final-answer SSE: one content delta, then a clean stop. +fn answer_sse(answer: &str) -> String { + [ + sse_chunk(json!({ + "id": "chatcmpl-outbox", + "object": "chat.completion.chunk", + "model": TEST_MODEL, + "choices": [{"index": 0, "delta": {"content": answer}, "finish_reason": null}] + })), + sse_chunk(json!({ + "id": "chatcmpl-outbox", + "object": "chat.completion.chunk", + "model": TEST_MODEL, + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}] + })), + "data: [DONE]\n\n".to_string(), + ] + .join("") +} + +async fn start_mock_llm() -> MockServer { + let server = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/v1/models")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "application/json") + .set_body_json(json!({ + "object": "list", + "data": [{ "id": TEST_MODEL, "object": "model" }] + })), + ) + .mount(&server) + .await; + + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .insert_header("cache-control", "no-cache") + .set_body_string(answer_sse("ok")), + ) + .mount(&server) + .await; + + server +} + +fn preserve_host_env(command: &mut Command) { + command.env_clear(); + for key in [ + "PATH", + "PATHEXT", + "SystemRoot", + "SystemDrive", + "WINDIR", + "COMSPEC", + "TEMP", + "TMP", + "TERM", + "COLORTERM", + "LANG", + "LC_ALL", + ] { + if let Some(value) = std::env::var_os(key) { + command.env(key, value); + } + } +} + +/// Run `codewhale exec` against the mock provider with the given +/// `[lifecycle_outbox]` config block (already TOML-formatted, may be empty). +/// Any `__OUTBOX_PATH__` token in it is replaced with the isolated home's +/// absolute outbox path. Returns the isolated home and workspace dirs (the +/// latter so callers can assert the outbox `payload.workspace` exactly). +fn run_exec_with_outbox_config(server: &MockServer, outbox_toml: &str) -> (TempDir, TempDir) { + let workspace = TempDir::new().expect("workspace tempdir"); + let home = TempDir::new().expect("home tempdir"); + let outbox_path = home_outbox_path(&home); + let outbox_toml = outbox_toml.replace(OUTBOX_PATH_TOKEN, &outbox_path.display().to_string()); + + std::fs::create_dir_all(home.path().join(".codewhale")).expect("create codewhale config dir"); + std::fs::create_dir_all(home.path().join(".deepseek")).expect("create deepseek config dir"); + std::fs::write( + home.path().join(".codewhale").join("config.toml"), + format!("provider = \"deepseek\"\nmodel = \"{TEST_MODEL}\"\n{outbox_toml}"), + ) + .expect("write exec config"); + + let mut command = Command::new(codewhale_tui_binary()); + preserve_host_env(&mut command); + command + .current_dir(workspace.path()) + .arg("--workspace") + .arg(workspace.path()) + .arg("--no-project-config") + .arg("exec") + .arg("--auto") + .arg("--model") + .arg(TEST_MODEL) + .arg("answer briefly") + .env("HOME", home.path()) + .env("USERPROFILE", home.path()) + .env("XDG_CONFIG_HOME", home.path().join(".config")) + .env("XDG_DATA_HOME", home.path().join(".local").join("share")) + .env("XDG_CACHE_HOME", home.path().join(".cache")) + .env( + "CODEWHALE_CONFIG_PATH", + home.path().join(".codewhale").join("config.toml"), + ) + .env( + "DEEPSEEK_CONFIG_PATH", + home.path().join(".deepseek").join("config.toml"), + ) + .env("DEEPSEEK_API_KEY", "ci-test-key-not-real") + .env("DEEPSEEK_BASE_URL", server.uri()) + .env("CODEWHALE_BASE_URL", server.uri()) + .env("DEEPSEEK_MODEL", TEST_MODEL) + .env("CODEWHALE_MODEL", TEST_MODEL) + .env("RUST_LOG", "warn") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let mut child = command.spawn().expect("spawn codewhale-tui exec"); + let stdout_reader = read_pipe_in_background(child.stdout.take().expect("stdout pipe")); + let stderr_reader = read_pipe_in_background(child.stderr.take().expect("stderr pipe")); + + let status = match child + .wait_timeout(RUN_TIMEOUT) + .expect("wait for codewhale-tui") + { + Some(status) => status, + None => { + let _ = child.kill(); + let _ = child.wait(); + let stdout = join_pipe_reader(stdout_reader, "stdout"); + let stderr = join_pipe_reader(stderr_reader, "stderr"); + panic!( + "codewhale-tui exec timed out after {RUN_TIMEOUT:?}\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&stdout), + String::from_utf8_lossy(&stderr) + ); + } + }; + + let stdout = join_pipe_reader(stdout_reader, "stdout"); + let stderr = join_pipe_reader(stderr_reader, "stderr"); + assert!( + status.success(), + "codewhale-tui exec failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&stdout), + String::from_utf8_lossy(&stderr) + ); + + (home, workspace) +} + +fn read_pipe_in_background(mut reader: R) -> std::thread::JoinHandle>> +where + R: Read + Send + 'static, +{ + std::thread::spawn(move || { + let mut output = Vec::new(); + reader.read_to_end(&mut output).map(|_| output) + }) +} + +fn join_pipe_reader( + handle: std::thread::JoinHandle>>, + stream_name: &str, +) -> Vec { + handle + .join() + .expect("pipe reader join") + .unwrap_or_else(|err| panic!("failed to read {stream_name}: {err}")) +} + +fn read_outbox_lines(path: &Path) -> Vec { + let text = std::fs::read_to_string(path).expect("read outbox file"); + text.lines() + .map(|line| { + serde_json::from_str(line) + .unwrap_or_else(|err| panic!("outbox line should parse: {err}\nline: {line}")) + }) + .collect() +} + +fn codewhale_tui_binary() -> PathBuf { + if let Some(path) = option_env!("CARGO_BIN_EXE_codewhale-tui") { + return PathBuf::from(path); + } + if let Ok(path) = std::env::var("CARGO_BIN_EXE_codewhale-tui") { + return PathBuf::from(path); + } + + let mut path = std::env::current_exe().expect("current test executable path"); + path.pop(); + if path.ends_with("deps") { + path.pop(); + } + path.push(format!("codewhale-tui{}", std::env::consts::EXE_SUFFIX)); + path +} + +fn home_outbox_path(home: &TempDir) -> PathBuf { + home.path() + .join(".codewhale") + .join("notifications") + .join("outbox.jsonl") +} + +#[tokio::test(flavor = "multi_thread")] +async fn exec_emits_turn_start_and_turn_end_to_the_configured_outbox() { + let server = start_mock_llm().await; + let (home, workspace) = run_exec_with_outbox_config( + &server, + &format!("[lifecycle_outbox]\npath = {}\n", json!(OUTBOX_PATH_TOKEN)), + ); + + let outbox_path = home_outbox_path(&home); + assert!(outbox_path.exists(), "outbox file must be created"); + let lines = read_outbox_lines(&outbox_path); + assert_eq!( + lines.len(), + 2, + "one turn_start and one turn_end line: {lines:#?}" + ); + + let start = &lines[0]; + assert_eq!(start["event"], "turn_start"); + assert_eq!(start["kind"], "turn.started"); + assert_eq!(start["schema_version"], 1); + assert_eq!(start["seq"], 1); + assert!(start["timestamp"].as_str().is_some()); + // Headless exec has no engine turn id and (for a fresh run) no session + // id yet — both are honest absences, never fabricated. + assert!(start["turn_id"].is_null()); + // The model field is bounded and never the raw prompt. + assert_eq!(start["payload"]["model"], TEST_MODEL); + // Every payload carries the workspace for consumer-side routing; exec + // runs with `--workspace `, so the emitted path must match it. + assert_eq!( + start["payload"]["workspace"], + json!(workspace.path().to_string_lossy().as_ref()), + "turn_start must carry the workspace" + ); + + let end = &lines[1]; + assert_eq!(end["event"], "turn_end"); + assert_eq!(end["kind"], "turn.completed"); + assert_eq!(end["seq"], 2); + assert_eq!(end["payload"]["status"], "completed"); + assert!(end["payload"]["error"].is_null()); + assert!(end["payload"]["duration_ms"].as_u64().is_some()); + assert_eq!( + end["payload"]["workspace"], + json!(workspace.path().to_string_lossy().as_ref()), + "turn_end must carry the workspace" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn exec_without_outbox_config_writes_no_file() { + let server = start_mock_llm().await; + let (home, _workspace) = run_exec_with_outbox_config(&server, ""); + + assert!( + !home_outbox_path(&home).exists(), + "no outbox file must be created when [lifecycle_outbox] is unset" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn outbox_seq_recovers_across_processes() { + let server = start_mock_llm().await; + + // First run writes seq 1 (turn_start) and 2 (turn_end). + let (home, _workspace) = run_exec_with_outbox_config( + &server, + &format!("[lifecycle_outbox]\npath = {}\n", json!(OUTBOX_PATH_TOKEN)), + ); + let shared_outbox = home_outbox_path(&home); + + // Second process, pointing at the SAME file: seq must continue at 3. + let (_second_home, _second_workspace) = run_exec_with_outbox_config( + &server, + &format!( + "[lifecycle_outbox]\npath = {}\n", + json!(shared_outbox.display().to_string()) + ), + ); + + let lines = read_outbox_lines(&shared_outbox); + assert_eq!(lines.len(), 4, "two runs, four lines: {lines:#?}"); + let seqs: Vec = lines + .iter() + .map(|line| line["seq"].as_u64().expect("seq")) + .collect(); + assert_eq!( + seqs, + vec![1, 2, 3, 4], + "seq must be monotonic across processes" + ); +} diff --git a/crates/tui/tests/integration/main.rs b/crates/tui/tests/integration/main.rs index b089913348..02fd3bc3cd 100644 --- a/crates/tui/tests/integration/main.rs +++ b/crates/tui/tests/integration/main.rs @@ -102,6 +102,7 @@ mod exec_persistent_service; mod exec_stream_drop_acceptance; mod exec_turn_usage; mod integration_mock_llm; +mod lifecycle_outbox_exec; mod palette_audit; mod protocol_recovery; mod reasoning_content_replayed_after_tool_call; diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 6c283be28e..2ff8a0a576 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -2415,6 +2415,70 @@ iTerm2, WezTerm, Ghostty, and kitty are matched first and use their own notification protocols, and `method = "osc9"` / `"bel"` / `"off"` opt out of the `osascript` path explicitly. +## Lifecycle Outbox (`[lifecycle_outbox]`) + +The lifecycle outbox is an opt-in, machine-readable stream of session, +turn, and sub-agent lifecycle events. With a path configured, Codewhale +appends one JSON line per event to that file — for interactive TUI +sessions *and* headless `codewhale exec` runs — so a supervisor +(terminal multiplexer wrapper, automation harness, alerting setup) can +react to what happened without scraping the screen or installing per-hook +shell commands. Unset or empty `path` = the feature is **off** and +behavior is unchanged. + +```toml +[lifecycle_outbox] +path = "~/.codewhale/notifications/outbox.jsonl" # unset/empty = OFF +webhook_url = "" # optional; POSTs events as JSON when set +webhook_token = "" # optional bearer token for webhook_url +``` + +### Events emitted + +| Event | Kind | Fired at | +|---|---|---| +| `turn_start` | `turn.started` | a new turn begins (TUI TurnStarted; `exec` at message dispatch) | +| `turn_end` | `turn.completed` / `turn.failed` / `turn.interrupted` | turn completion, kind projected from the turn status | +| `turn_stalled` | `turn.stalled` | the stall watchdog recovers a wedged turn | +| `subagent_spawn` | `subagent.spawned` | a sub-agent is spawned | +| `subagent_complete` | `subagent.completed` | a sub-agent reaches a terminal state | +| `session_start` | `session.started` | interactive session start | +| `session_end` | `session.ended` | interactive session end | + +### File contract + +Each line is a `RuntimeEventEnvelope`: + +```json +{"schema_version": 1, "seq": 3, "event": "turn_start", "kind": "turn.started", + "thread_id": "…", "turn_id": "…", "item_id": null, "timestamp": "…", + "created_at": "…", "payload": {…}} +``` + +- `seq` is monotonic per outbox file and recovers from the last written + line when a new process opens the file. +- Lines are written one complete JSON line per append, serialized by an + internal writer task and flushed before the next event; concurrent + sessions writing the same path do not interleave bytes mid-line, but + separate processes each continue from their own recovered `seq`, so seqs + can repeat across processes sharing one file — prefer one file per + process for strict uniqueness. +- Parent directories are created lazily on the first event. +- Payloads are constructed from bounded, pre-redacted fields only — never + raw tool arguments, environment, or full transcript text. Free-form + fields (error messages, previews) are capped at the notification limits + (80 headline / 120 detail / 200 preview characters) and stripped of + control bytes. + +### Webhook delivery + +With `webhook_url` set, every event is additionally POSTed as +`{"at": "", "event": {…}}` with +`Authorization: Bearer ` when a token is configured. +Delivery is best-effort: failures are logged and dropped, never retried +into the agent loop, and a failing webhook never blocks the local file +append. + ## Tool Catalog Codewhale loads a small core native tool catalog by default and leaves less diff --git a/docs/changelog-lifecycle-outbox.md b/docs/changelog-lifecycle-outbox.md new file mode 100644 index 0000000000..8b81511ee9 --- /dev/null +++ b/docs/changelog-lifecycle-outbox.md @@ -0,0 +1,90 @@ +# Changelog — lifecycle outbox (`[lifecycle_outbox]`) + +Changelog for the general lifecycle event outbox (target: upstream +PR). Feature-complete against the v0.9.9 baseline (`6f3850c3d`). + +## Added + +- **Config**: new `[lifecycle_outbox]` table with three optional keys: + - `path` — JSONL outbox file. Unset/empty = feature **off**, behavior + unchanged (the whole feature is additive and opt-in). + - `webhook_url` — optional webhook endpoint; POSTs fire only when set. + - `webhook_token` — optional bearer token for `webhook_url`. + Documented in `docs/CONFIGURATION.md` and `config.example.toml`. The + documented example default is `~/.codewhale/notifications/outbox.jsonl`; + the config key drives the real path. +- **Writer** (`crates/hooks/src/lifecycle_outbox.rs`): appends one JSONL line + per event to the configured path — lazy parent dirs, append+flush, single + internal writer task serializing emits in order. Line shape is the existing + `RuntimeEventEnvelope` (`schema_version, seq, event, kind, thread_id, + turn_id, item_id, timestamp, created_at, payload`). `seq` is monotonic per + outbox file and recovers from the last complete line's `seq` on open + (bounded 64 KiB tail scan; a torn trailing line from a crash is ignored). + Payloads are constructed from bounded, pre-redacted fields only — never + raw tool args, environment, or transcript text — with free-form fields + capped at the notification limits (headline ≤ 80, detail ≤ 120, + preview ≤ 200 chars) and stripped of control bytes. +- **Webhook**: `WebhookHookSink` (previously dead code with no config + surface) now supports an optional bearer token and is wired to outbox + events when `webhook_url` is set — POST `{"at", "event"}`. Delivery is + best-effort: failures are logged and dropped, never retried into the + agent loop, and a failing webhook never blocks the local append. + +## Events emitted + +| Event | Kind | Site | +|---|---|---| +| `turn_start` | `turn.started` | TUI `EngineEvent::TurnStarted`; headless `exec` at `Op::SendMessage` | +| `turn_end` | `turn.completed` / `turn.failed` / `turn.interrupted` | TUI `TurnComplete` processing; headless `exec` `TurnComplete` (kind projected from status) | +| `turn_stalled` | `turn.stalled` | `recover_stalled_runtime_turn` — the first scriptable stall signal | +| `subagent_spawn` | `subagent.spawned` | subagent observer site (fires even with no hooks configured) | +| `subagent_complete` | `subagent.completed` | subagent observer site (fires even with no hooks configured) | +| `session_start` | `session.started` | TUI session-start hook fire site | +| `session_end` | `session.ended` | TUI session-end hook fire site | + +Headless `codewhale exec` coverage: `turn_start` at message dispatch and +`turn_end` at the terminal `TurnComplete` — **and** at the "engine channel +closed before a terminal receipt" path, so every emitted `turn_start` has a +matching `turn_end` and a supervisor never sees an orphaned in-progress +turn. `exec` has no TurnStarted engine event, so `turn_id` is absent there; +`thread_id` is the resumed session id when `--continue` was used, empty for +fresh runs (the session id is only minted at persistence time). + +## File contract + +- One JSON object per line; every line is a complete + `RuntimeEventEnvelope`; appended and flushed per event. +- `seq` counts up per file, starting at 1 for a new file, recovering from + the last complete line after a restart. +- Cross-process: appends use O_APPEND with the line + newline in a single + write, so two processes sharing one file can interleave *lines* but never + splice a line mid-record. Seq uniqueness is per process recovery, so + sharing one file across processes can repeat seq values — use one file + per process for strict uniqueness. + +## Tests + +- `crates/hooks`: append/schema shape, seq recovery across reopen, missing/ + empty file, torn trailing line, emit ordering under the writer task, + disabled-outbox no-ops, `bounded_text` ceilings (incl. UTF-8 boundaries). +- `crates/config`: `[lifecycle_outbox]` off-by-default, webhook optional, + full-table parse. +- `crates/tui`: TUI config parse of the table; stall-recovery emit-site + tests (enabled outbox writes one `turn_stalled` line naming the wedged + turn; disabled outbox writes nothing and recovery behavior is unchanged). + +## Not changed + +- With `[lifecycle_outbox]` unset, zero behavior change: the outbox handle + is a disabled no-op and no file or HTTP request is ever made. +- No new runtime dependencies (JSONL append uses tokio fs; webhook reuses + the existing `reqwest` client builder). + +## Remaining / follow-ups + +- Session ids for fresh headless `exec` runs are empty on `turn_start` + (minted only when the run is persisted); a supervisor correlating runs + can key on the process + file. +- Webhook-only configuration (url without path) parses losslessly but does + not activate the outbox handle today — the file path is the feature gate. + Documented; can be lifted later if webhook-only delivery is wanted.