diff --git a/crates/runtime/src/adapter/claude_protocol/adapter.rs b/crates/runtime/src/adapter/claude_protocol/adapter.rs index 340cb765..1e5da9c4 100644 --- a/crates/runtime/src/adapter/claude_protocol/adapter.rs +++ b/crates/runtime/src/adapter/claude_protocol/adapter.rs @@ -1,20 +1,26 @@ //! The real [`Adapter`] implementation for `AdapterMode::Protocol`: //! spawns `claude` with [`super::launch::build_argv`]'s fixed launch -//! argv after [`super::trust::workspace_trust_accepted`] passes, delivers the -//! initial prompt as the first `stream-json` input message, hands its -//! stdout/stdin to [`super::reader::drive_turn`] for the whole turn, and -//! reconciles what that turn saw against claude's own durable transcript -//! via [`super::reconcile::find_gaps`] once it ends. +//! argv after [`super::trust::workspace_trust_accepted`] passes, hands its +//! stdout/stdin to [`super::reader::drive_turn`] for the whole turn -- +//! including the initial prompt, which `drive_turn` itself delivers +//! only once its own `initialize` control-channel handshake completes, +//! not before -- and reconciles what that turn saw against claude's own +//! durable transcript via [`super::reconcile::find_gaps`] once it ends. //! //! This is the first real caller for three modules this spike built //! ahead of it: `reconcile::find_gaps` (previously dead code, reachable //! only from its own tests and the external drop-seam test), //! `approval_bridge`'s whole module (previously `#![allow(dead_code)]`), -//! and `reader::drive_turn` itself. +//! and `reader::drive_turn` itself. It is also the first real caller of +//! [`pane::PaneSupport`]: when a `TuiSupport` bundle was ever supplied +//! to the registry, [`Self::attach_pane`] opens a crew-rendered pane for +//! the turn, best-effort -- see `pane`'s own module doc comment for the +//! format and the provenance/redaction notes worth reading before +//! citing this pane's own text as evidence of anything. //! //! **Not in scope here** (see this spike's own PR body for the full -//! non-goals list): a renderer beyond bare legibility, `resume`/ -//! `--continue`, the other three vendors, and performance work. +//! non-goals list): `resume`/`--continue`, the other three vendors, and +//! performance work. //! //! `--model` is threaded through as a plain, already-resolved parameter //! this adapter does not itself resolve -- it is handed exactly @@ -33,8 +39,10 @@ use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::Arc; +use std::time::Duration; -use tokio::io::AsyncWriteExt; +use nix::sys::signal::{Signal, kill}; +use nix::unistd::Pid; use tokio::sync::Mutex as AsyncMutex; use crew_protocol::{Classified, ContentClass}; @@ -47,11 +55,11 @@ use crate::adapter::r#trait::{ Adapter, AdapterMessage, AdapterSnapshot, CancelScope, ProbeResult, StartSpec, VendorSessionRef, }; use crate::approval::ApprovalService; -use crate::supervisor::EnvironmentPolicy; +use crate::supervisor::{EnvironmentPolicy, EscalationTimings, TerminationOutcome}; use super::approval_bridge::ProtocolApprovalCallback; use super::reader::{self, RunIdentity}; -use super::{reconcile, trust}; +use super::{pane, reconcile, trust}; /// Everything [`ClaudeProtocolAdapter`] needs that only exists after the /// daemon's own `ApprovalService` is constructed -- the server-owned @@ -98,6 +106,22 @@ pub(crate) struct ClaudeProtocolAdapter { /// [`Adapter::cancel`] can reach it. `None` before the first /// [`Adapter::start`] call and after the turn ends. child: AsyncMutex>, + /// Present only when a `TuiSupport` bundle was ever supplied to the + /// registry -- see [`pane::PaneSupport`]'s own doc comment. `None` + /// means this adapter runs with no pane at all, never a refusal to + /// start. + pane_support: Option, + /// How long [`settle_after_turn`] waits at each step of its + /// SIGINT -> SIGTERM -> SIGKILL escalation -- production's own + /// [`EscalationTimings::default`] (5s/5s, same as every other + /// supervised process in this daemon); a test overrides it to keep + /// an escalation test's own runtime bounded. + escalation: EscalationTimings, + /// How long [`settle_after_turn`] waits, unsignaled, for the + /// ordinary case (claude exiting on its own) before it starts + /// escalating at all -- production's own [`SELF_EXIT_GRACE`]; a test + /// overrides it for the same reason `escalation` is overridable. + self_exit_grace: Duration, } impl ClaudeProtocolAdapter { @@ -107,6 +131,7 @@ impl ClaudeProtocolAdapter { environment_allowlist: Vec, model: Option, bundle: ProtocolBundle, + pane_support: Option, ) -> Self { Self { repo_root, @@ -116,6 +141,9 @@ impl ClaudeProtocolAdapter { claude_json_path: trust::default_claude_json_path(), bin: "claude".to_string(), child: AsyncMutex::new(None), + pane_support, + escalation: EscalationTimings::default(), + self_exit_grace: SELF_EXIT_GRACE, } } @@ -130,6 +158,20 @@ impl ClaudeProtocolAdapter { self } + /// Overrides the escalation timings and self-exit grace window + /// [`settle_after_turn`] waits out -- a test-only seam, so an + /// escalation test does not have to pay production's real windows. + #[cfg(test)] + fn with_escalation_timings( + mut self, + self_exit_grace: Duration, + escalation: EscalationTimings, + ) -> Self { + self.self_exit_grace = self_exit_grace; + self.escalation = escalation; + self + } + fn env(&self) -> std::collections::HashMap { let current: std::collections::HashMap = std::env::vars().collect(); EnvironmentPolicy::baseline().build(¤t, &self.environment_allowlist) @@ -196,6 +238,73 @@ impl ClaudeProtocolAdapter { tracing::warn!(error = %err, run_id = %ids.run_id, "failed to journal claude-protocol reconciliation"); } } + + /// Attaches this run's pane, if [`Self::pane_support`] was ever + /// supplied -- best-effort: a failure here is logged and this + /// adapter proceeds with no pane, never fails the turn (see + /// [`pane::PaneSupport`]'s own doc comment on why this adapter's + /// pane is a convenience, not a control surface). + async fn attach_pane(&self, ids: RunIdentity) -> Option { + let support = self.pane_support.as_ref()?; + let (target, output_tx) = pane::PaneAttachTarget::new(); + let target = Arc::new(target); + let socket_path = support.panes_dir.join(format!("{}.sock", ids.run_id)); + let attach_server = match crate::display::AttachServer::start( + socket_path, + Arc::clone(&target) as Arc, + target.on_user_input(), + ) { + Ok(server) => server, + Err(err) => { + tracing::warn!(error = %err, run_id = %ids.run_id, "failed to start this run's attach server; continuing without a pane"); + return None; + } + }; + let outcome = support + .pane_coordinator + .attach(crate::display::PaneAttachRequest { + run_id: ids.run_id, + worker_id: ids.worker_id, + adapter: self.kind().to_string(), + placement: support.placement, + forced_backend: support.forced_backend, + launch_program: support.launch_program, + }) + .await; + // The banner is the operational form of this module's own + // provenance note -- a viewer must never be able to read even + // one line before knowing what this pane is (see + // `pane::attach_banner`'s own doc comment). + let _ = output_tx.send(pane::attach_banner()); + Some(AttachedPane { + attach_server, + outcome, + output_tx, + }) + } + + /// Releases the pane [`Self::attach_pane`] set up, if any. Called on + /// every exit path from [`Self::start`], success or error. + async fn detach_pane(&self, pane: AttachedPane, succeeded: bool) { + if let Some(support) = &self.pane_support { + support + .pane_coordinator + .detach(&pane.outcome, succeeded, support.close_on_exit) + .await; + } + pane.attach_server.stop(); + } +} + +/// What [`ClaudeProtocolAdapter::attach_pane`] set up, kept alive for the +/// duration of one turn so [`ClaudeProtocolAdapter::detach_pane`] can +/// release it afterward. `output_tx` is also read from directly, by +/// [`ClaudeProtocolAdapter::start`], to hand `reader::drive_turn` a +/// place to push formatted lines into. +struct AttachedPane { + attach_server: crate::display::AttachServer, + outcome: crate::display::PaneAttachOutcome, + output_tx: tokio::sync::broadcast::Sender>, } /// `~/.claude/projects//.jsonl` @@ -215,6 +324,105 @@ fn transcript_path(canonical_repo_root: &Path, session_id: &str) -> PathBuf { .join(format!("{session_id}.jsonl")) } +/// How long a completed turn's own claude process gets to exit on its +/// own before [`settle_after_turn`] starts signaling it. `drive_turn` +/// dropping its own `stdin`/`stdout` handles when it returns closes +/// this adapter's write half of the pipe, which SHOULD make claude see +/// EOF and exit -- a live capture's own call 9 reasoned this, but never +/// independently tested it (that call's own harness never closed its +/// side of stdin, unlike this adapter's `drive_turn`, so it could not +/// observe this specifically). This window exists so that reasoning is +/// never load-bearing: whether or not it holds, a process still running +/// after it gets escalated exactly like a wedged process anywhere else +/// in this daemon would. +const SELF_EXIT_GRACE: Duration = Duration::from_secs(5); + +/// Waits for `child` to exit on its own, escalating +/// SIGINT -> SIGTERM -> SIGKILL on `escalation`'s own timings if it does +/// not -- the same discipline +/// `crate::supervisor::process::ManagedProcess::terminate` applies to +/// every other supervised process in this daemon, reused here as free +/// functions over a bare `tokio::process::Child` rather than by +/// adopting `ManagedProcess`/`Supervisor` themselves: `drive_turn`'s own +/// generic `AsyncRead`/`AsyncWrite` interface (and the fast, in-memory +/// `tokio::io::duplex`-based tests built on it) is the reason this +/// adapter still spawns via a bare `tokio::process::Command` rather than +/// `Supervisor::spawn` -- a real gap against this module's own stated +/// invariant ("every adapter launches its supervised vendor process +/// through this module"), carried forward rather than fixed here: fixing +/// it would mean rebuilding `drive_turn` over `ManagedProcess`'s framed +/// stdout/queued-stdin API instead, which is a larger change than "give +/// `child.wait()` a deadline" asks for. +/// +/// Signals are sent to `child`'s own pid directly, never a process +/// group: unlike `Supervisor::spawn`'s children, this spawn was never +/// given a process group of its own (see the gap above), and claude's +/// own `-p` invocation is a single non-interactive process, not an +/// interactive shell expected to leave orphaned, signal-ignoring +/// grandchildren behind the way a PTY-hosted shell can. +async fn settle_after_turn( + child: &mut tokio::process::Child, + self_exit_grace: Duration, + escalation: EscalationTimings, +) -> TerminationOutcome { + if let Ok(Some(status)) = child.try_wait() { + return TerminationOutcome::Exited { + code: status.code(), + }; + } + + if let Some(outcome) = wait_step(child, self_exit_grace).await { + return outcome; + } + + let Some(pid) = child.id().map(|id| id as i32) else { + // Already reaped by something else between the checks above and + // here -- nothing left to signal; `wait()`'s own error, if any, + // is reported as an exit of unknown code rather than escalated + // further, matching `ManagedProcess::terminate`'s own precedent + // for this situation. + return match child.wait().await { + Ok(status) => TerminationOutcome::Exited { + code: status.code(), + }, + Err(_) => TerminationOutcome::Exited { code: None }, + }; + }; + + let _ = kill(Pid::from_raw(pid), Signal::SIGINT); + if let Some(outcome) = wait_step(child, escalation.sigint_to_sigterm).await { + return outcome; + } + + let _ = kill(Pid::from_raw(pid), Signal::SIGTERM); + if let Some(outcome) = wait_step(child, escalation.sigterm_to_sigkill).await { + return outcome; + } + + let _ = kill(Pid::from_raw(pid), Signal::SIGKILL); + let _ = child.wait().await; + TerminationOutcome::Killed +} + +/// Waits out `duration` for `child` to exit, without signaling it. +/// `Some` only on a confirmed exit within the window; `None` for either +/// a timeout (still running) or a `wait()` error (state unknown) -- +/// neither is a confirmed exit, so both mean the caller should keep +/// escalating, matching +/// `crate::supervisor::process::ManagedProcess::wait_out_step`'s own +/// stance on the identical question. +async fn wait_step( + child: &mut tokio::process::Child, + duration: Duration, +) -> Option { + match tokio::time::timeout(duration, child.wait()).await { + Ok(Ok(status)) => Some(TerminationOutcome::Exited { + code: status.code(), + }), + Ok(Err(_)) | Err(_) => None, + } +} + impl Adapter for ClaudeProtocolAdapter { fn kind(&self) -> &str { "claude" @@ -284,105 +492,108 @@ impl Adapter for ClaudeProtocolAdapter { )); } - let model = self - .model - .as_deref() - .map(str::trim) - .filter(|model| !model.is_empty()); - let argv = super::launch::build_argv(model); - let env = self.env(); - - let mut command = tokio::process::Command::new(&self.bin); - command - .args(&argv) - .current_dir(&canonical_repo_root) - .env_clear() - .envs(&env) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::null()) - .kill_on_drop(true); - let mut child = command - .spawn() - .map_err(|e| AdapterError::process(self.kind(), "start", e.to_string()))?; - let pid = child.id().unwrap_or(0); - let mut stdin = child - .stdin - .take() - .expect("stdin was requested as piped at spawn"); - let stdout = child - .stdout - .take() - .expect("stdout was requested as piped at spawn"); - - let first_message = serde_json::json!({ - "type": "user", - "message": { - "role": "user", - "content": [{"type": "text", "text": spec.prompt}], - }, - }); - let mut first_line = serde_json::to_string(&first_message) - .expect("a constructed value always serializes"); - first_line.push('\n'); - stdin - .write_all(first_line.as_bytes()) - .await - .map_err(|e| AdapterError::process(self.kind(), "start", e.to_string()))?; + // Attached before the process spawns, so a viewer sees every + // line from the very first one -- best-effort: unlike a + // `TuiAdapter`, this adapter's pane is a convenience view, not + // its control surface, so a failure to attach never fails the + // turn, only logs and proceeds without one. + let pane = self.attach_pane(ids).await; + let pane_output = pane.as_ref().map(|p| &p.output_tx); - *self.child.lock().await = Some(child); + let turn: Result<(), AdapterError> = async { + let model = self + .model + .as_deref() + .map(str::trim) + .filter(|model| !model.is_empty()); + let argv = super::launch::build_argv(model); + let env = self.env(); - sink.emit(AdapterEvent { - run_id: ids.run_id, - task_id: ids.task_id, - worker_id: ids.worker_id, - payload: AdapterEventPayload::ProcessStarted { pid }, - cursor: None, - }) - .await?; - - let outcome = reader::drive_turn( - stdout, - stdin, - &sink, - &self.bundle.approval_service, - &self.bundle.callback, - ids, - ) - .await?; - - let status = { - let mut guard = self.child.lock().await; - let child = guard.as_mut().ok_or_else(|| { - AdapterError::invalid_vendor_state( - self.kind(), - "start", - "the spawned child process handle was missing at reap time", - ) - })?; - child - .wait() - .await - .map_err(|e| AdapterError::process(self.kind(), "start", e.to_string()))? - }; - *self.child.lock().await = None; + let mut command = tokio::process::Command::new(&self.bin); + command + .args(&argv) + .current_dir(&canonical_repo_root) + .env_clear() + .envs(&env) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true); + let mut child = command + .spawn() + .map_err(|e| AdapterError::process(self.kind(), "start", e.to_string()))?; + let pid = child.id().unwrap_or(0); + let stdin = child + .stdin + .take() + .expect("stdin was requested as piped at spawn"); + let stdout = child + .stdout + .take() + .expect("stdout was requested as piped at spawn"); - sink.emit(AdapterEvent { - run_id: ids.run_id, - task_id: ids.task_id, - worker_id: ids.worker_id, - payload: AdapterEventPayload::ProcessExited { - exit_code: status.code(), - signal: None, - }, - cursor: None, - }) - .await?; + *self.child.lock().await = Some(child); - self.reconcile_turn(&canonical_repo_root, &outcome, &sink, ids) - .await; + sink.emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload: AdapterEventPayload::ProcessStarted { pid }, + cursor: None, + }) + .await?; - Ok(()) + let outcome = reader::drive_turn( + stdout, + stdin, + &sink, + &self.bundle.approval_service, + &self.bundle.callback, + ids, + pane_output, + &spec.prompt, + ) + .await?; + + let termination = { + let mut guard = self.child.lock().await; + let child = guard.as_mut().ok_or_else(|| { + AdapterError::invalid_vendor_state( + self.kind(), + "start", + "the spawned child process handle was missing at reap time", + ) + })?; + settle_after_turn(child, self.self_exit_grace, self.escalation).await + }; + *self.child.lock().await = None; + let (exit_code, signal) = termination.exit_signals(); + + sink.emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload: AdapterEventPayload::ProcessExited { exit_code, signal }, + cursor: None, + }) + .await?; + + self.reconcile_turn(&canonical_repo_root, &outcome, &sink, ids) + .await; + + Ok(()) + } + .await; + + // Detached on every exit path from here, success or error -- + // `PaneCoordinator::detach` is what releases the live-pane + // slot and journals `DisplayPaneDetached`; leaving it + // unreached on an error path would leak both. + if let Some(pane) = pane { + self.detach_pane(pane, turn.is_ok()).await; + } + + turn }) } @@ -510,7 +721,7 @@ mod tests { r#"#!/bin/sh printf '%s\n' "$@" > "{argv_path}" read -r _first_line -echo '{{"type":"system","subtype":"init","session_id":"sess-argv-test"}}' +echo '{{"type":"system","subtype":"init","session_id":"sess-argv-test","claude_code_version":"2.1.268","permissionMode":"auto"}}' echo '{{"type":"result","subtype":"success"}}' "#, argv_path = argv_path.display(), @@ -581,6 +792,7 @@ echo '{{"type":"result","subtype":"success"}}' approval_service, callback, }, + None, ) .with_test_overrides(claude_json_path, bin.to_string_lossy().to_string()); @@ -612,4 +824,328 @@ echo '{{"type":"result","subtype":"success"}}' db.shutdown().await.ok(); } + + /// A pane backend that always succeeds with a non-empty pane ref -- + /// mirrors `tests/tui_claude_registry.rs`'s own `FakeBackend` + /// fixture exactly, for the identical reason: `HiddenDisplay`'s own + /// `pane_ref` is empty, which would make "a real pane attached" + /// indistinguishable from "no real pane at all" in this test's own + /// assertions. + struct FakePaneBackend; + + impl crate::display::DisplayBackendTrait for FakePaneBackend { + fn backend_name(&self) -> &str { + "tmux" + } + + fn is_available(&self) -> bool { + true + } + + fn activate(&mut self) -> Result<(), String> { + Ok(()) + } + + fn status(&self) -> crew_protocol::DisplayStatus { + crew_protocol::DisplayStatus::new(crew_protocol::DisplayBackend::Tmux, true, false) + } + + fn create_pane( + &self, + req: crate::display::PaneRequest, + ) -> crate::display::DisplayFuture<'_, crate::display::PaneHandle> { + let handle = crate::display::PaneHandle { + backend: crew_protocol::DisplayBackend::Tmux, + pane_ref: "fake-pane-1".to_string(), + placement: req.placement, + }; + Box::pin(async move { Ok(handle) }) + } + + fn close_pane( + &self, + _handle: &crate::display::PaneHandle, + ) -> crate::display::DisplayFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + } + + /// End-to-end proof that `Self::attach_pane`/`Self::detach_pane` + /// actually reach a real `PaneCoordinator`: drives a real turn (the + /// same fake-script harness as the model-argv test above) with a + /// real pane wired in, then asserts a real `DisplayPaneAttached` and + /// `DisplayPaneDetached` were journaled with the fake backend's own + /// non-empty pane ref -- not that the two pieces were built to + /// agree, but that wiring them together actually reaches the + /// journal. + #[tokio::test] + async fn a_real_turn_attaches_and_detaches_a_real_pane() { + // `tempdir_in("/tmp")`, not the bare `TempDir::new()` default + // (`std::env::temp_dir()`, a deeply-nested path on macOS): the + // attach socket this test binds lives under this directory, and + // a `run_id`-length UUID appended to that deep a path overflows + // the platform `sun_path` limit -- the same reason + // `tests/tui_claude_registry.rs`'s own harness uses this exact + // override. + let dir = tempfile::Builder::new() + .prefix("bat-pane-") + .tempdir_in("/tmp") + .unwrap(); + let repo_root = dir.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + let argv_path = dir.path().join("argv.txt"); + let bin = write_argv_recording_fake_claude(dir.path(), &argv_path); + let claude_json_path = trusted_claude_json(dir.path(), &repo_root); + + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = crew_protocol::ProjectId::new(); + let approval_service = StdArc::new(ApprovalService::new( + StdArc::clone(&db), + project_id, + StdArc::new(NoopApprovalCallback) as StdArc, + broadcast::channel(64).0, + )); + let callback = StdArc::new(ProtocolApprovalCallback::new()); + + let mut display_registry = crate::display::DisplayRegistry::new(); + display_registry.register(Box::new(FakePaneBackend)); + let (events_tx, _events_rx) = broadcast::channel(64); + let pane_coordinator = Arc::new(crate::display::PaneCoordinator::new( + Arc::new(display_registry), + StdArc::clone(&db), + project_id, + events_tx, + std::path::PathBuf::from("/opt/crew/bin/crewd"), + dir.path().to_path_buf(), + repo_root.clone(), + crate::security::redaction::Redactor::new(), + )); + let panes_dir = dir.path().join("panes"); + std::fs::create_dir_all(&panes_dir).unwrap(); + + let adapter = ClaudeProtocolAdapter::new( + repo_root.clone(), + Vec::new(), + None, + ProtocolBundle { + approval_service, + callback, + }, + Some(pane::PaneSupport { + pane_coordinator, + panes_dir, + placement: crew_protocol::DisplayPlacement::SplitRight, + forced_backend: None, + launch_program: None, + close_on_exit: crate::config::crew::CloseOnExit::Always, + }), + ) + .with_test_overrides(claude_json_path, bin.to_string_lossy().to_string()); + + let events = StdArc::new(parking_lot::Mutex::new(Vec::new())); + let sink: Arc = Arc::new(RecordingSink { events }); + + // `PaneCoordinator::attach`/`detach` journal against a real + // `runs` row (a foreign-key reference, invariant 3) -- unlike + // `RecordingSink`, they write through the real `db` this test + // constructed, so the row has to actually exist first, the same + // seeding `tests/tui_claude_registry.rs`'s own + // `seed_worker_and_run` does. + let task_id = TaskId::new(); + let worker_id = WorkerId::new(); + let run_id = RunId::new(); + db.run_domain_op(Box::new({ + let task_id = task_id.to_string(); + let worker_id = worker_id.to_string(); + let run_id = run_id.to_string(); + let project_id = project_id.to_string(); + move |conn| { + conn.execute( + "INSERT INTO tasks (task_id, project_id, owner_client_instance_id, revision, created_at, updated_at) \ + VALUES (?1, ?2, 'test-owner', 1, ?3, ?3)", + rusqlite::params![task_id, project_id, "2026-01-01T00:00:00Z"], + )?; + conn.execute( + "INSERT INTO worker_profiles (id, fingerprint, adapter, model, permission_envelope) \ + VALUES (?1, 'sha256:test', 'claude', 'test-model', '{}')", + rusqlite::params![worker_id.clone()], + )?; + conn.execute( + "INSERT INTO workers (worker_id, project_id, profile_id, resolved_profile_json, created_at) \ + VALUES (?1, ?2, ?1, '{}', ?3)", + rusqlite::params![worker_id, project_id, "2026-01-01T00:00:00Z"], + )?; + conn.execute( + "INSERT INTO runs (run_id, task_id, worker_id, state, created_at) \ + VALUES (?1, ?2, ?3, 'queued', ?4)", + rusqlite::params![run_id, task_id, worker_id, "2026-01-01T00:00:00Z"], + )?; + Ok(serde_json::Value::Null) + } + })) + .await + .expect("seed task/worker/run"); + + adapter + .start( + StartSpec { + run_id, + task_id, + worker_id, + prompt: "hello".to_string(), + resume: None, + }, + sink, + ) + .await + .expect("start must succeed against the fake binary and fake pane backend"); + + let run_id_string = run_id.to_string(); + let dump: Vec = db + .run_domain_op(Box::new(move |conn| { + let mut stmt = conn + .prepare("SELECT event_json FROM events WHERE run_id = ?1 ORDER BY sequence")?; + let rows = stmt + .query_map(rusqlite::params![run_id_string], |row| { + row.get::<_, String>(0) + })? + .collect::, _>>()?; + Ok(serde_json::json!(rows)) + })) + .await + .unwrap() + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap().to_string()) + .collect(); + + assert!( + dump.iter() + .any(|e| e.contains("displayPaneAttached") && e.contains("fake-pane-1")), + "expected a DisplayPaneAttached event naming the fake backend's own pane ref, got: {dump:#?}" + ); + assert!( + dump.iter().any(|e| e.contains("displayPaneDetached")), + "expected a DisplayPaneDetached event once the turn settled, got: {dump:#?}" + ); + + db.shutdown().await.ok(); + } + + /// A `claude` that ignores SIGINT and SIGTERM outright and never + /// exits on its own -- [`settle_after_turn`]'s own escalation ladder + /// is the only thing that ever reaps it, and only reaches SIGKILL + /// after climbing through both prior steps: this is what makes the + /// call 9 stdin-drop reasoning a non-assumption, per the maintainer's + /// own ruling on this ("do not rely on it; bound the wait and + /// escalate instead"). + fn write_wedged_fake_claude(dir: &std::path::Path) -> PathBuf { + let script = r#"#!/bin/sh +trap '' INT TERM +read -r _first_line +echo '{"type":"system","subtype":"init","session_id":"sess-wedged","claude_code_version":"2.1.268","permissionMode":"auto"}' +echo '{"type":"result","subtype":"success"}' +while true; do sleep 1; done +"#; + let path = dir.join("wedged-claude.sh"); + std::fs::write(&path, script).expect("write wedged fake claude script"); + let mut perms = std::fs::metadata(&path).unwrap().permissions(); + perms.set_mode(0o755); + std::fs::set_permissions(&path, perms).unwrap(); + path + } + + /// End to end proof that a claude process which never exits on its + /// own, and ignores both SIGINT and SIGTERM, still gets reaped: the + /// turn completes normally (the protocol's own `result` line + /// arrived), and `Adapter::start` still returns rather than hanging + /// forever on `child.wait()`, with the final `ProcessExited` event + /// reporting the SIGKILL escalation actually needed. + #[tokio::test] + async fn a_wedged_process_is_escalated_to_sigkill_rather_than_hung_on() { + let dir = tempfile::TempDir::new().unwrap(); + let repo_root = dir.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + let bin = write_wedged_fake_claude(dir.path()); + let claude_json_path = trusted_claude_json(dir.path(), &repo_root); + + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = crew_protocol::ProjectId::new(); + let approval_service = StdArc::new(ApprovalService::new( + StdArc::clone(&db), + project_id, + StdArc::new(NoopApprovalCallback) as StdArc, + broadcast::channel(64).0, + )); + let callback = StdArc::new(ProtocolApprovalCallback::new()); + + let adapter = ClaudeProtocolAdapter::new( + repo_root.clone(), + Vec::new(), + None, + ProtocolBundle { + approval_service, + callback, + }, + None, + ) + .with_test_overrides(claude_json_path, bin.to_string_lossy().to_string()) + .with_escalation_timings( + Duration::from_millis(50), + EscalationTimings { + sigint_to_sigterm: Duration::from_millis(50), + sigterm_to_sigkill: Duration::from_millis(50), + }, + ); + + let events = StdArc::new(parking_lot::Mutex::new(Vec::new())); + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::clone(&events), + }); + + tokio::time::timeout( + Duration::from_secs(10), + adapter.start( + StartSpec { + run_id: RunId::new(), + task_id: TaskId::new(), + worker_id: WorkerId::new(), + prompt: "hello".to_string(), + resume: None, + }, + sink, + ), + ) + .await + .expect("start must not hang past the escalation ladder's own bounded windows") + .expect("start must still succeed: the turn itself completed normally"); + + { + let recorded = events.lock(); + let exited = recorded.iter().find_map(|payload| match payload { + AdapterEventPayload::ProcessExited { exit_code, signal } => { + Some((*exit_code, signal.clone())) + } + _ => None, + }); + assert_eq!( + exited, + Some((None, Some("SIGKILL".to_string()))), + "expected a SIGKILL-escalated exit, got: {recorded:#?}" + ); + } + + db.shutdown().await.ok(); + } } diff --git a/crates/runtime/src/adapter/claude_protocol/approval_bridge.rs b/crates/runtime/src/adapter/claude_protocol/approval_bridge.rs index 7c0cba20..d1b2bfcf 100644 --- a/crates/runtime/src/adapter/claude_protocol/approval_bridge.rs +++ b/crates/runtime/src/adapter/claude_protocol/approval_bridge.rs @@ -31,6 +31,18 @@ pub(crate) struct PermissionRequest { pub(crate) request_id: String, pub(crate) tool_name: String, pub(crate) input: serde_json::Value, + /// Claude's own id for the specific tool call this request is + /// blocked on (`toolu_...`) -- distinct from `request_id`, which is + /// this control-channel round trip's own correlation id, generated + /// fresh per request. A live capture's final `result` message + /// reports a denied tool call by THIS id, in `permission_denials`, + /// never by `request_id` -- so reconciling a denial against + /// whichever requests this reader actually bridged needs this + /// field, not `request_id`. `None` when absent -- observed present + /// on every real `can_use_tool` request this spike captured, but + /// parsed as optional rather than required, matching this module's + /// own tolerant-parse stance for every other field here. + pub(crate) tool_use_id: Option, } /// Parses one `can_use_tool` control-channel message. Tolerant the same @@ -52,6 +64,10 @@ pub(crate) fn parse_permission_request(value: &serde_json::Value) -> Option` without spawning anything, the //! same shape `claude_protocol::approval_bridge`'s parse/build pair and //! `claude_protocol::reconcile::find_gaps` already take. +//! +//! **Departs from the maintainer-approved launch-posture wording**, which did not +//! name a `--permission-mode`: a live capture found that without one, +//! the effective mode is inherited from the operator's own +//! `~/.claude.json`/settings default, silently, including a value +//! (`bypassPermissions`) that would make `can_use_tool` never reach +//! this adapter's own bridge at all -- permission mode is load-bearing +//! for permissions, and the standing rule is that crew passes anything +//! load-bearing explicitly rather than inherits it. The maintainer +//! ruled on this departure directly: pin it, and pin a permissive +//! value. [`PINNED_PERMISSION_MODE`] is that value. + +/// The permission mode this adapter pins explicitly -- `"auto"`, not +/// the `--allow-dangerously-skip-permissions` flag: that flag only +/// makes bypass *available* (its own `--help` text: "without it being +/// enabled by default"), it does not set a mode, so passing it alone +/// would still leave the effective mode inherited from the operator's +/// own settings -- exactly the defect this pin exists to close. +/// `"auto"`'s own semantics are themselves unmeasured by this spike (a +/// finding of their own, not assumed); what is pinned here is the +/// SPELLING crew passes, not a claim about what it permits. +/// +/// A named constant, not an inline literal, specifically so +/// [`super::posture`]'s own direct assertion +/// (`system/init`'s reported `permissionMode` equals what crew passed) +/// compares against the SAME value crew actually sent, never a second, +/// independently-typed copy of the same string that could drift from +/// this one. +pub(crate) const PINNED_PERMISSION_MODE: &str = "auto"; + +/// `--permission-prompts` is pinned too, on the same ground as +/// `PINNED_PERMISSION_MODE` -- this is the flag that decides whether +/// the host is consulted about a permission decision AT ALL, load- +/// bearing for permissions by definition. Pinning it changes nothing +/// today (`"host"` is already the CLI's own default), which is exactly +/// the point: today's default staying the default is not something +/// this adapter should depend on staying true by accident. +pub(crate) const PINNED_PERMISSION_PROMPTS: &str = "host"; + +/// `--permission-prompt-tool stdio` -- the actual enabler of the +/// `can_use_tool` control-channel requests `claude_protocol::approval_bridge` +/// bridges into crew's own approval ledger. Not documented under this +/// flag's own `--help` entry (that text describes it as naming an MCP +/// tool); found instead by reading the SDK's own source that constructs +/// it -- the Python SDK's `subprocess_cli.py` passes exactly the literal +/// `"stdio"` sentinel whenever a caller supplies a `can_use_tool` +/// callback (`types.py` documents the same shape). The TS SDK is +/// inferred to do the same from the Python side's own comments, not +/// read directly -- this repo ships no TypeScript SDK source to read. +/// Confirmed empirically, not just from source: a live call without +/// this flag produced zero `can_use_tool` control-channel frames across +/// seven calls and two forcing conditions; the identical settings and +/// prompt, with only this flag added, produced a real `can_use_tool` +/// request, and a real reply built from +/// `super::approval_bridge::build_permission_response`'s own unmodified +/// output then let the tool actually execute. +/// +/// A named constant for the same reason [`PINNED_PERMISSION_MODE`] is +/// one: nothing else in this adapter needs a second, independently +/// typed copy of this spelling. +pub(crate) const PINNED_PERMISSION_PROMPT_TOOL: &str = "stdio"; /// Builds the argv `claude` is launched with, everything after the /// binary name itself. `model`, when given, is placed exactly where @@ -37,6 +98,12 @@ pub(crate) fn build_argv(model: Option<&str>) -> Vec { "--verbose".to_string(), "--setting-sources".to_string(), "user,project,local".to_string(), + "--permission-mode".to_string(), + PINNED_PERMISSION_MODE.to_string(), + "--permission-prompts".to_string(), + PINNED_PERMISSION_PROMPTS.to_string(), + "--permission-prompt-tool".to_string(), + PINNED_PERMISSION_PROMPT_TOOL.to_string(), ]; if let Some(model) = model { argv.push("--model".to_string()); @@ -69,6 +136,55 @@ mod tests { assert!(!argv.iter().any(|arg| arg.contains("disableAllHooks"))); } + /// The maintainer-ruled departure from the original launch-posture wording, + /// proven directly: `--permission-mode` is always pinned to + /// [`PINNED_PERMISSION_MODE`], never left to be inherited from the + /// operator's own settings. + #[test] + fn permission_mode_is_always_pinned() { + let argv = build_argv(None); + assert!( + argv.windows(2) + .any(|w| w == ["--permission-mode", PINNED_PERMISSION_MODE]) + ); + } + + /// The same pin, for the flag that decides whether the host is + /// consulted at all -- a consequence of the same rule, not a second decision (see this + /// module's own doc comment). + #[test] + fn permission_prompts_is_always_pinned() { + let argv = build_argv(None); + assert!( + argv.windows(2) + .any(|w| w == ["--permission-prompts", PINNED_PERMISSION_PROMPTS]) + ); + } + + /// `--allow-dangerously-skip-permissions` only makes bypass + /// available; it does not pin a mode. Never on this argv, on + /// purpose -- see this module's own doc comment on why it would + /// satisfy the letter of "pass a flag" and none of the actual rule. + #[test] + fn never_passes_the_skip_permissions_flag() { + let argv = build_argv(None); + assert!(!argv.iter().any(|arg| arg.contains("skip-permissions"))); + } + + /// The sentinel that actually enables `can_use_tool` control + /// requests -- see [`PINNED_PERMISSION_PROMPT_TOOL`]'s own doc + /// comment for how that was confirmed. Proven pinned the same way + /// as the other two flags: always present, never left to whatever + /// `claude`'s own default for this flag might be. + #[test] + fn permission_prompt_tool_is_always_pinned_to_stdio() { + let argv = build_argv(None); + assert!( + argv.windows(2) + .any(|w| w == ["--permission-prompt-tool", PINNED_PERMISSION_PROMPT_TOOL]) + ); + } + #[test] fn no_model_flag_when_none_is_given() { let argv = build_argv(None); diff --git a/crates/runtime/src/adapter/claude_protocol/mod.rs b/crates/runtime/src/adapter/claude_protocol/mod.rs index f9324702..75932aa0 100644 --- a/crates/runtime/src/adapter/claude_protocol/mod.rs +++ b/crates/runtime/src/adapter/claude_protocol/mod.rs @@ -14,10 +14,11 @@ //! or hook-suppressing flag -- the workspace-trust pre-check //! ([`trust::workspace_trust_accepted`]), the control-channel //! reader ([`reader::drive_turn`]), the approval bridge -//! ([`approval_bridge`]), and reconciliation against claude's own -//! transcript ([`reconcile::find_gaps`]). Still not built: the pane -//! renderer beyond bare legibility, `resume`/`--continue`, and the -//! other three vendors. +//! ([`approval_bridge`]), reconciliation against claude's own +//! transcript ([`reconcile::find_gaps`]), a crew-rendered pane +//! ([`pane::PaneSupport`]), and an init-time posture assertion +//! ([`posture::version_gate`]). Still not built: `resume`/`--continue` +//! and the other three vendors. pub(crate) mod adapter; // `pub`, not `pub(crate)`: `ProtocolApprovalCallback`'s own doc comment @@ -26,6 +27,8 @@ pub(crate) mod adapter; // already set, for a different property). pub mod approval_bridge; pub(crate) mod launch; +pub(crate) mod pane; +pub(crate) mod posture; pub(crate) mod reader; pub(crate) mod trust; diff --git a/crates/runtime/src/adapter/claude_protocol/pane.rs b/crates/runtime/src/adapter/claude_protocol/pane.rs new file mode 100644 index 00000000..b0ce8318 --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/pane.rs @@ -0,0 +1,175 @@ +//! The pane renderer: a crew-authored, self-formatted view of a +//! protocol-mode turn, wired through the existing +//! `crate::display::attach`/`crate::display::coordinator` machinery +//! unchanged. Claude's own screen does not exist in protocol mode -- +//! there is no PTY -- so a viewer attaching to this run's pane sees +//! lines this module writes, one per normalized event, never a byte +//! claude itself painted. +//! +//! **This is a crew-authored report, not a recording of the vendor's +//! own output**, and that distinction matters for two reasons stated +//! plainly rather than left for a reader to infer: +//! +//! - **Provenance.** [`crate::adapter::tui`]'s own pane shows the +//! vendor's raw PTY bytes verbatim; this pane shows text crew parsed +//! and formatted from the control channel. A human attached to this +//! pane is watching crew's report of the turn, not the turn itself -- +//! the banner [`attach_banner`] writes on every attach says so, and a +//! caller citing this pane as "what claude printed" would be wrong in +//! a way citing the TUI pane's own text never is. +//! - **Redaction.** The durable journal redacts every free-text field +//! before it becomes part of a committed `RuntimeEvent` (invariant 4). +//! This pane's lines are built from the SAME normalized text before +//! that boundary, so they are NOT redacted -- exactly like the TUI +//! pane's raw PTY bytes are not redacted either, for the identical +//! reason (a live view is not the durable record). Recorded here so +//! the next person asking "is this pane also redacted" finds the +//! answer already written rather than re-deriving it. +//! +//! `write_input` on [`PaneAttachTarget`] is a documented no-op: this +//! adapter has no path from a viewer's keystroke back into claude's own +//! stdin mid-turn (`declared_capabilities().steering` is already +//! `None`). A keystroke a viewer sends is never silently swallowed, +//! though -- see [`PaneAttachTarget::on_user_input`]'s own doc comment +//! for why it gets an immediate, local, unredacted notice rather than a +//! journaled `OutOfBandInput` (that event asserts the journal may be +//! incomplete, which cannot be true here: nothing the viewer typed ever +//! reached the vendor). + +use std::sync::Arc; + +use tokio::sync::broadcast; + +use crew_protocol::{DisplayBackend, DisplayPlacement, HostProgramHint}; + +use crate::config::crew::CloseOnExit; +use crate::display::{AttachError, AttachTarget, PaneCoordinator}; + +/// Everything [`super::adapter::ClaudeProtocolAdapter`] needs to attach +/// a pane for one run -- present only when a [`crate::adapter::tui::TuiSupport`] +/// bundle was ever supplied to the registry (`AdapterRegistry::set_tui_support`); +/// `None` otherwise, in which case the adapter runs with no pane at all +/// rather than refusing to start. Unlike a `TuiAdapter`, this adapter's +/// pane is a convenience view, not its control surface -- the turn +/// completes identically whether or not anyone is watching. +pub(crate) struct PaneSupport { + pub(crate) pane_coordinator: Arc, + pub(crate) panes_dir: std::path::PathBuf, + pub(crate) placement: DisplayPlacement, + pub(crate) forced_backend: Option, + pub(crate) launch_program: Option, + pub(crate) close_on_exit: CloseOnExit, +} + +/// The banner [`PaneAttachTarget`]'s owner writes as the pane's first +/// line, before any turn content -- the operational form of this +/// module's own provenance note: a viewer must not be able to attach +/// and read even one line without already knowing what they are +/// looking at. +pub(crate) fn attach_banner() -> Vec { + render_line( + "crew-rendered view -- not claude's own screen. The vendor's own transcript is the record.", + ) +} + +/// Encodes one line for the pane: `text` plus a CRLF terminator. CRLF, +/// not a bare `\n`: `crewd attach`'s own client puts the local terminal +/// into raw mode (`cli.rs`'s `cfmakeraw` wrapper) before pumping bytes, +/// so nothing downstream supplies the carriage return a cooked terminal +/// would have added for free. +pub(crate) fn render_line(text: &str) -> Vec { + let mut line = text.as_bytes().to_vec(); + line.extend_from_slice(b"\r\n"); + line +} + +/// The [`AttachTarget`] this adapter's pane wiring hands to +/// [`crate::display::AttachServer::start`]. Holds the one +/// `broadcast::Sender` [`super::reader::drive_turn`]'s caller pushes +/// formatted lines into as the turn progresses -- `subscribe_output` +/// hands out receivers off that same sender, exactly +/// `tests/attach.rs`'s own `FakeTarget` precedent for a non-PTY target. +pub(crate) struct PaneAttachTarget { + output_tx: broadcast::Sender>, +} + +impl PaneAttachTarget { + pub(crate) fn new() -> (Self, broadcast::Sender>) { + let (output_tx, _) = broadcast::channel(256); + ( + Self { + output_tx: output_tx.clone(), + }, + output_tx, + ) + } + + /// The `on_user_input` callback [`crate::display::AttachServer::start`] + /// takes: writes a single, immediate, local notice back into this + /// same pane's own broadcast rather than journaling + /// `RuntimeEvent::OutOfBandInput`. That event's own meaning is "a + /// human bypassed the adapter, so the journal may not reflect + /// everything the vendor received" -- untrue here by construction, + /// since [`AttachTarget::write_input`] on this type never reaches + /// claude at all. Journaling it anyway would raise a flag asserting + /// a condition that cannot have occurred; feedback exactly where the + /// keystroke went is the correct answer, not a misfiled event. + pub(crate) fn on_user_input(&self) -> Box) + Send + Sync> { + let output_tx = self.output_tx.clone(); + Box::new(move |_bytes: Vec| { + // No alternative path is offered here: this adapter declares + // `steering: None` and `send` refuses every message kind + // unconditionally (`declared_capabilities()`'s own doc + // comment) -- there is no working way to steer this run yet, + // and the notice must not imply one exists. + let _ = output_tx.send(render_line( + "this pane is a read-only view -- what was typed here was discarded, not sent \ + to claude. This run cannot be steered.", + )); + }) + } +} + +impl AttachTarget for PaneAttachTarget { + /// A documented no-op -- see this module's own doc comment for why. + fn write_input<'a>( + &'a self, + _bytes: Vec, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async move { Ok(()) }) + } + + fn subscribe_output(&self) -> broadcast::Receiver> { + self.output_tx.subscribe() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rendered_lines_are_crlf_terminated() { + let line = render_line("hello"); + assert_eq!(line, b"hello\r\n"); + } + + #[tokio::test] + async fn write_input_is_a_no_op_that_never_errors() { + let (target, _tx) = PaneAttachTarget::new(); + target.write_input(b"anything".to_vec()).await.unwrap(); + } + + #[tokio::test] + async fn on_user_input_writes_a_read_only_notice_never_an_out_of_band_event() { + let (target, _tx) = PaneAttachTarget::new(); + let mut rx = target.subscribe_output(); + let callback = target.on_user_input(); + callback(b"someone typed here".to_vec()); + let received = rx.try_recv().expect("a notice must be sent"); + let text = String::from_utf8(received).unwrap(); + assert!(text.contains("read-only")); + assert!(text.contains("discarded")); + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/posture.rs b/crates/runtime/src/adapter/claude_protocol/posture.rs new file mode 100644 index 00000000..b8b670ef --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/posture.rs @@ -0,0 +1,127 @@ +//! The init-time posture assertion: proves the process actually +//! launched matches what this adapter intended, rather than trusting +//! [`super::launch::build_argv`]'s own argv to have taken effect just +//! because nothing errored at spawn time. +//! +//! **What a live capture settled, and what it left open.** A real +//! `system/init` message (`claude_code_version` `2.1.268`, three +//! captures) carries no field naming which `--setting-sources` were +//! actually applied -- so there is nothing to assert about settings +//! application directly, and the CLI-version range check below is not +//! the fallback half of a choice, it is the only posture assertion +//! `system/init` offers for that question. `system/init` does carry +//! `permissionMode`, which the same capture found was NOT pinned by +//! this adapter's own launch argv originally -- an operator's personal +//! `~/.claude.json`/settings default silently governed it instead, +//! including a value that would make `can_use_tool` never reach this +//! adapter's own bridge at all. The maintainer ruled on that departure +//! from the original launch-posture wording directly: +//! [`super::launch::PINNED_PERMISSION_MODE`] is now always passed, and +//! [`permission_mode_gate`] asserts `system/init` actually reports it +//! back -- a DIRECT check, not a proxy: it verifies the pin took +//! effect, rather than inferring posture from a version being in +//! range. Both checks run; they catch different classes (an +//! incompatible binary vs. a setting that failed to apply), and +//! neither replaces the other. + +/// The `claude --version` range this adapter's fixed argv was actually +/// launched against and observed to behave as expected -- mirrors +/// `crate::adapter::tui::claude`'s own `MIN_TESTED_VERSION`/ +/// `MAX_TESTED_VERSION` precedent and its same honesty about what part +/// of the range is real: the one version this spike's own live capture +/// confirmed is `2.1.268`; the rest of the range is an untested +/// extrapolation, not a second data point. +const MIN_TESTED_VERSION: (u32, u32, u32) = (2, 1, 0); +const MAX_TESTED_VERSION: (u32, u32, u32) = (2, 99, 99); + +/// Parses the leading `MAJOR.MINOR.PATCH` out of `system/init`'s own +/// `claude_code_version` field. Returns `None` for anything that does +/// not start with three dot-separated integers -- [`version_gate`] +/// treats an unparseable string as incompatible, never as a silent +/// pass, the same stance `tui::claude`'s own version parser takes. +fn parse_leading_version(reported: &str) -> Option<(u32, u32, u32)> { + let mut parts = reported.split('.'); + let major = parts.next()?.parse().ok()?; + let minor = parts.next()?.parse().ok()?; + let patch = parts.next()?.parse().ok()?; + Some((major, minor, patch)) +} + +/// Asserts `reported` (`system/init`'s own `claude_code_version` +/// field) falls within the tested range. `Err` names the reported +/// version and the tested range -- `reported` itself is safe to +/// interpolate here, unlike `tui::claude::version_gate`'s own probed +/// `--version` stdout: `system/init` is a structured JSON field this +/// adapter itself parsed, never raw untrusted CLI output that could +/// carry a stack trace or secret-shaped text. +pub(crate) fn version_gate(reported: &str) -> Result<(), String> { + match parse_leading_version(reported) { + Some(version) if version >= MIN_TESTED_VERSION && version <= MAX_TESTED_VERSION => Ok(()), + Some(version) => Err(format!( + "claude {version:?} is outside the tested range {MIN_TESTED_VERSION:?}..={MAX_TESTED_VERSION:?}" + )), + None => Err(format!( + "system/init's own claude_code_version field ({reported:?}) does not look like a version" + )), + } +} + +/// Asserts `system/init`'s own reported `permissionMode` equals +/// [`super::launch::PINNED_PERMISSION_MODE`] -- the direct check this +/// module's own doc comment describes: it proves the pin actually took +/// effect, never just that this adapter believes it sent one. +/// `reported` is safe to interpolate for the same reason +/// [`version_gate`]'s own doc comment gives: a structured field this +/// adapter itself parsed, never raw untrusted CLI output. +pub(crate) fn permission_mode_gate(reported: &str) -> Result<(), String> { + if reported == super::launch::PINNED_PERMISSION_MODE { + Ok(()) + } else { + Err(format!( + "system/init reported permissionMode {reported:?}, expected the pinned {:?} -- the \ + --permission-mode flag this adapter passes did not take effect", + super::launch::PINNED_PERMISSION_MODE + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_the_one_version_this_spike_actually_captured() { + assert_eq!(version_gate("2.1.268"), Ok(())); + } + + #[test] + fn accepts_a_version_inside_the_tested_range() { + assert_eq!(version_gate("2.5.0"), Ok(())); + } + + #[test] + fn rejects_a_version_below_the_tested_range() { + assert!(version_gate("1.9.9").is_err()); + } + + #[test] + fn rejects_an_unparseable_string() { + let err = version_gate("not-a-version").unwrap_err(); + assert!(err.contains("not-a-version")); + } + + #[test] + fn accepts_the_pinned_mode_reported_back() { + assert_eq!( + permission_mode_gate(super::super::launch::PINNED_PERMISSION_MODE), + Ok(()) + ); + } + + #[test] + fn rejects_a_different_reported_mode() { + let err = permission_mode_gate("plan").unwrap_err(); + assert!(err.contains("plan")); + assert!(err.contains(super::super::launch::PINNED_PERMISSION_MODE)); + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/reader.rs b/crates/runtime/src/adapter/claude_protocol/reader.rs index c17a5cdf..265659fa 100644 --- a/crates/runtime/src/adapter/claude_protocol/reader.rs +++ b/crates/runtime/src/adapter/claude_protocol/reader.rs @@ -26,6 +26,7 @@ use std::collections::HashSet; use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; +use tokio::sync::broadcast; use crew_protocol::{Classified, ContentClass, RunId, TaskId, WorkerId}; @@ -61,23 +62,90 @@ enum StreamLine { name: String, }, ControlRequest(serde_json::Value), + /// `type: "control_response"`: a reply to a control request THIS + /// reader itself sent -- today, the only one it ever sends is + /// [`INITIALIZE_REQUEST_ID`]'s own handshake, so `request_id` is + /// checked against that one constant; a future second outbound + /// request would need this reader to track more than one pending id + /// to disambiguate, which nothing here does yet because nothing + /// here sends a second one. `request_id` is read from + /// `response.request_id` -- the same nesting claude's own reply to + /// crew's `can_use_tool` answers uses + /// ([`approval_bridge::build_permission_response`]'s own shape), + /// confirmed live for the initialize handshake specifically in the + /// same capture that confirmed `can_use_tool` itself. + ControlResponse { + request_id: Option, + }, /// `type: "result"`: claude's own turn-complete marker. The read /// loop stops here rather than waiting for stdout to close -- /// waiting for EOF as well would work too (claude exits after this), /// but would leave a turn's own completion depending on process /// exit timing rather than the protocol's own explicit signal. - TurnComplete, + /// Carries `permission_denials` straight off the same message -- a + /// live capture confirmed the shape (`tool_name`, `tool_use_id`, + /// `tool_input` per entry) -- so [`drive_turn`] can check each one + /// against BOTH of [`TurnOutcome::bridged_tool_use_ids`] and + /// [`TurnOutcome::observed_permission_denials`] before ever + /// reporting the turn complete: explained by either one is + /// expected; explained by neither is the one case nothing on the + /// stream accounts for. + TurnComplete { + permission_denials: Vec, + }, /// `type: "system", subtype: "init"`: the one message this reader /// reads a `session_id` off of, for /// [`TurnOutcome::session_id`] -- claude's own transcript file is /// named `.jsonl`, and reconciliation cannot find the - /// right file without it. + /// right file without it. Also the one message + /// [`super::posture::version_gate`]/[`super::posture::permission_mode_gate`] + /// check `claude_code_version`/`permissionMode` against -- see that + /// module's own doc comment for what a live capture confirmed + /// `system/init` does and does not report. SystemInit { session_id: Option, + claude_code_version: Option, + permission_mode: Option, + }, + /// `type: "system", subtype: "permission_denied"`: a post-hoc + /// notification that a tool call was denied WITHOUT ever raising a + /// `can_use_tool` control request -- the shape + /// `permission-denied-frame.jsonl` exists to document (a hard `deny` + /// rule, or an `ask` rule matched with no host present, both deny + /// and report rather than ask). Its own `tool_use_id` is what makes + /// [`TurnOutcome::observed_permission_denials`] the second leg of + /// the ledger-reconciliation check at [`StreamLine::TurnComplete`]: + /// a denial explained by ONE of this or + /// [`TurnOutcome::bridged_tool_use_ids`] is expected; a denial + /// explained by NEITHER is the one case nothing on the whole stream + /// accounts for. + PermissionDeniedNotice { + tool_use_id: Option, }, Other, } +/// One entry off the final `result` message's own `permission_denials` +/// array -- a tool call claude denied without this reader's own +/// `StreamLine::ControlRequest` handling ever seeing it approved (a +/// hard `deny` rule denies without asking anyone; an `ask` rule with no +/// host present, per the documented hostless behavior, denies and +/// reports rather than silently allowing). Field names mirror the real +/// capture exactly; both optional, tolerant of a shape variation this +/// reader has not seen, matching every other parse in this module. +#[derive(Debug, PartialEq, Clone)] +struct PermissionDenial { + tool_name: Option, + tool_use_id: Option, +} + +/// The fixed correlation id this reader sends its own `initialize` +/// control request under -- see [`drive_turn`]'s own doc comment for +/// why a literal constant is sufficient (this reader only ever sends +/// one outbound control request per turn, so nothing else could collide +/// with it). +const INITIALIZE_REQUEST_ID: &str = "crew-initialize"; + /// What one call to [`drive_turn`] actually observed, for its caller to /// reconcile against the vendor's own durable transcript afterward. #[derive(Debug, Default)] @@ -94,6 +162,31 @@ pub(crate) struct TurnOutcome { /// reported one -- `None` when the stream closed before that /// message ever arrived (nothing to reconcile against without it). pub(crate) session_id: Option, + /// Every `tool_use_id` this reader actually bridged into crew's own + /// approval ledger this turn, via + /// [`approval_bridge::handle_permission_request`] -- checking + /// membership in this set at [`StreamLine::TurnComplete`] time IS + /// checking the ledger, not a proxy for it: + /// `ApprovalService::request` durably commits the approval row + /// before `handle_permission_request` ever returns, and nothing + /// removes a row from `approvals` for a run this reader is + /// currently driving, so this in-memory set and a fresh query + /// against the database can never diverge within one call to + /// [`drive_turn`]. + pub(crate) bridged_tool_use_ids: HashSet, + /// Every `tool_use_id` this reader saw named on a real + /// `system`/`permission_denied` notification this turn -- the + /// second leg of the ledger-reconciliation check at + /// [`StreamLine::TurnComplete`]. This is what makes that check + /// three-way rather than two: a denial explained by THIS set is an + /// operator's own rule denying without asking anyone (legitimate, + /// not a sentinel failure); a denial explained by neither this set + /// nor [`Self::bridged_tool_use_ids`] is the one case nothing on + /// the stream accounts for. Keying on the presence of this frame, + /// not on its `decision_reason_type` field (an undocumented, + /// single-observation vendor enum a future release could silently + /// add a new value to), is deliberate. + pub(crate) observed_permission_denials: HashSet, } /// Classifies one already-parsed JSON line from claude's stdout. Never @@ -106,7 +199,34 @@ pub(crate) struct TurnOutcome { fn classify_line(value: &serde_json::Value) -> StreamLine { match value.get("type").and_then(serde_json::Value::as_str) { Some("control_request") => StreamLine::ControlRequest(value.clone()), - Some("result") => StreamLine::TurnComplete, + Some("control_response") => StreamLine::ControlResponse { + request_id: value + .get("response") + .and_then(|response| response.get("request_id")) + .and_then(serde_json::Value::as_str) + .map(str::to_string), + }, + Some("result") => StreamLine::TurnComplete { + permission_denials: value + .get("permission_denials") + .and_then(serde_json::Value::as_array) + .map(|entries| { + entries + .iter() + .map(|entry| PermissionDenial { + tool_name: entry + .get("tool_name") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + tool_use_id: entry + .get("tool_use_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + }) + .collect() + }) + .unwrap_or_default(), + }, Some("system") if value.get("subtype").and_then(serde_json::Value::as_str) == Some("init") => { @@ -115,6 +235,25 @@ fn classify_line(value: &serde_json::Value) -> StreamLine { .get("session_id") .and_then(serde_json::Value::as_str) .map(str::to_string), + claude_code_version: value + .get("claude_code_version") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + permission_mode: value + .get("permissionMode") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + } + } + Some("system") + if value.get("subtype").and_then(serde_json::Value::as_str) + == Some("permission_denied") => + { + StreamLine::PermissionDeniedNotice { + tool_use_id: value + .get("tool_use_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string), } } Some("assistant") => { @@ -164,16 +303,65 @@ fn classify_line(value: &serde_json::Value) -> StreamLine { /// bridging any `can_use_tool` control request through `approval_service`/ /// `callback`, writing the reply back to `stdin`. /// +/// The very first thing written to `stdin` is an `initialize` control +/// request (`request_id` [`INITIALIZE_REQUEST_ID`]) -- "as the SDK +/// does," per the maintainer's own ruling on this: the SDK always sends +/// one, even though it declares nothing about permissions itself. Once +/// this reader sees the matching `control_response` (which may arrive +/// before or after `system/init` -- a live capture found both orderings +/// possible, and this loop's own dispatch does not depend on either), +/// `prompt` is delivered as the ordinary `stream-json` user message. +/// Any line arriving before that point (a `system`/`hook_*` line, the +/// operator's own SessionStart hooks, `system/init` itself) is still +/// classified and acted on through the exact same dispatch below -- this +/// is not a separate handshake-only loop bolted in front of the real +/// one. +/// /// Returns once the turn is complete (or the stream closed) -- it never /// loops forever on a stream that legitimately ends, and it never treats -/// EOF as an error: a process that exits after its own `result` line is -/// the ordinary case, not a fault. +/// EOF as an error, with one exception: a stream that closes before the +/// initialize handshake ever completed means `prompt` was never sent at +/// all, so nothing this turn was meant to do ever ran -- that is +/// reported as a protocol error, not folded into an ordinary empty +/// [`TurnOutcome`]. Once the prompt has been sent, a stream that later +/// closes without an explicit `result` line is the ordinary case again +/// (claude exited after signalling completion, or crashed): `TurnOutcome` +/// is returned as `Ok`, matching this function's original stance, +/// leaving `Adapter::start`'s own caller to observe the process exit and +/// reconcile against the transcript regardless of how the stream ended. +/// +/// `pane_output`, when given, gets one formatted line per normalized +/// event pushed into it as the turn progresses -- see +/// `claude_protocol::pane`'s own module doc comment for the format and +/// why it is a crew-authored report, never the vendor's own output. A +/// send failing (no receiver currently attached, the ordinary case when +/// nobody is watching) is silently ignored: the pane is a convenience +/// view, never load-bearing for the turn itself. /// /// # Errors /// Propagates a stdout read failure, a stdin write failure, or /// [`approval_bridge::handle_permission_request`]'s own error (a run no /// longer `working`, or this adapter torn down mid-approval) as a -/// [`crate::adapter::error::AdapterError::process`]/`::protocol`. +/// [`crate::adapter::error::AdapterError::process`]/`::protocol`. Also +/// returns `::protocol` if the stream closes before the prompt was ever +/// sent, or if claude's own final `result` reports a permission denial +/// explained by NEITHER [`TurnOutcome::bridged_tool_use_ids`] (the host +/// was asked, over the control channel, and answered) NOR +/// [`TurnOutcome::observed_permission_denials`] (an operator's own rule +/// denied it without asking anyone, announced on the stream by its own +/// `system`/`permission_denied` notice) -- the one observable signal +/// that the host was never consulted for it AT ALL, which is what a +/// broken/renamed `--permission-prompt-tool` sentinel would cause. A +/// denial explained by either set is expected and does not fail the +/// turn -- an operator's own hard `deny` rule is legitimate, ordinary +/// behavior, not a sentinel failure. +// The `prompt` parameter added for the initialize handshake is the +// eighth -- matching the existing precedent elsewhere in this codebase +// (`build_tui_adapter`, `TuiAdapter::fail_start`) of allowing this +// specific lint on a function whose every parameter is load-bearing and +// already documented above, rather than introducing a params struct +// for one field. +#[allow(clippy::too_many_arguments)] pub(crate) async fn drive_turn( stdout: R, mut stdin: W, @@ -181,6 +369,8 @@ pub(crate) async fn drive_turn( approval_service: &ApprovalService, callback: &ProtocolApprovalCallback, ids: RunIdentity, + pane_output: Option<&broadcast::Sender>>, + prompt: &str, ) -> Result where R: AsyncRead + Unpin, @@ -188,7 +378,27 @@ where { use crate::adapter::error::AdapterError; + let render = |text: &str| { + if let Some(tx) = pane_output { + let _ = tx.send(super::pane::render_line(text)); + } + }; + + let init_request = serde_json::json!({ + "type": "control_request", + "request_id": INITIALIZE_REQUEST_ID, + "request": {"subtype": "initialize"}, + }); + let mut init_line = + serde_json::to_string(&init_request).expect("a constructed value always serializes"); + init_line.push('\n'); + stdin + .write_all(init_line.as_bytes()) + .await + .map_err(|e| AdapterError::process("claude", "write_stdin", e.to_string()))?; + let mut outcome = TurnOutcome::default(); + let mut prompt_sent = false; let mut lines = BufReader::new(stdout).lines(); loop { let line = lines @@ -196,6 +406,19 @@ where .await .map_err(|e| AdapterError::process("claude", "read_stdout", e.to_string()))?; let Some(line) = line else { + if !prompt_sent { + // The stream closed before this adapter's own + // initialize handshake ever completed -- the prompt was + // never sent, so no turn ran at all. Distinct from the + // ordinary "closed after result" case below: there, + // something happened; here, nothing did. + return Err(AdapterError::protocol( + "claude", + "start", + "claude's stdout closed before this adapter's own initialize handshake \ + completed -- the prompt was never sent, so this turn never ran", + )); + } // Stream closed without an explicit `result` line -- claude // exited (or crashed) before signalling turn completion. // Not this loop's own error to raise: `Adapter::start`'s @@ -234,10 +457,59 @@ where outcome.entry_ids.insert(uuid.to_string()); } match classify_line(&value) { - StreamLine::SystemInit { session_id } => { + StreamLine::SystemInit { + session_id, + claude_code_version, + permission_mode, + } => { outcome.session_id = session_id; + // Both checked as soon as `system/init` arrives, before + // this adapter trusts anything the rest of the turn + // reports -- they catch different classes (an + // incompatible binary vs. a pinned setting that failed + // to apply) and either one failing must abort the turn, + // not merely be noted afterward. + let reported_version = claude_code_version.as_deref().unwrap_or(""); + if let Err(detail) = super::posture::version_gate(reported_version) { + return Err(AdapterError::incompatible_version( + "claude", "start", detail, + )); + } + let reported_mode = permission_mode.as_deref().unwrap_or(""); + if let Err(detail) = super::posture::permission_mode_gate(reported_mode) { + return Err(AdapterError::protocol("claude", "start", detail)); + } } + StreamLine::ControlResponse { request_id } if !prompt_sent => { + if request_id.as_deref() != Some(INITIALIZE_REQUEST_ID) { + // Not the handshake's own response (or carries no + // `request_id` at all) -- this reader has nothing + // else pending to match it against yet, since the + // prompt (and anything that could raise a + // `can_use_tool` request) has not been sent. + continue; + } + let mut prompt_line = serde_json::json!({ + "type": "user", + "message": { + "role": "user", + "content": [{"type": "text", "text": prompt}], + }, + }) + .to_string(); + prompt_line.push('\n'); + stdin + .write_all(prompt_line.as_bytes()) + .await + .map_err(|e| AdapterError::process("claude", "write_stdin", e.to_string()))?; + prompt_sent = true; + } + // Once the handshake is done, any further `control_response` + // is one this reader never sends a matching request for + // today -- nothing to act on. + StreamLine::ControlResponse { .. } => {} StreamLine::AssistantText(text) => { + render(&format!("assistant: {text}")); sink.emit(AdapterEvent { run_id: ids.run_id, task_id: ids.task_id, @@ -254,6 +526,7 @@ where .await?; } StreamLine::ToolStarted { tool_call_id, name } => { + render(&format!("tool: {name}")); sink.emit(AdapterEvent { run_id: ids.run_id, task_id: ids.task_id, @@ -263,6 +536,18 @@ where }) .await?; } + StreamLine::PermissionDeniedNotice { tool_use_id } => { + render("permission denied without asking"); + // Recorded regardless of whether `tool_use_id` is + // present -- a `None` here just means this particular + // notice can never explain a denial by id, not that it + // should be discarded outright (the same tolerant + // stance every other optional field in this module + // takes). + if let Some(tool_use_id) = tool_use_id { + outcome.observed_permission_denials.insert(tool_use_id); + } + } StreamLine::ControlRequest(request) => { let Some(parsed) = approval_bridge::parse_permission_request(&request) else { // A control request this reader does not recognize @@ -274,6 +559,18 @@ where // unrecognized transcript entry is in `reconcile`. continue; }; + render(&format!("permission requested: {}", parsed.tool_name)); + // Recorded before the request is bridged, not after the + // reply comes back -- `ApprovalService::request` (called + // inside `handle_permission_request`) durably commits the + // approval row before that call ever returns, so this + // set and the ledger it stands in for agree at every + // point after this line, not just once the whole request + // finishes (which may block a long time on a human + // decision). + if let Some(tool_use_id) = parsed.tool_use_id.clone() { + outcome.bridged_tool_use_ids.insert(tool_use_id); + } let reply = approval_bridge::handle_permission_request( approval_service, callback, @@ -287,8 +584,46 @@ where .write_all(&reply) .await .map_err(|e| AdapterError::process("claude", "write_stdin", e.to_string()))?; + render("permission decided"); + } + StreamLine::TurnComplete { permission_denials } => { + render("turn complete"); + // The ledger-reconciliation check, three-way: a denial + // is expected if EITHER this reader bridged it into + // crew's own approval ledger (`bridged_tool_use_ids` -- + // the host was asked and answered, deny included) OR a + // real `system`/`permission_denied` notice named it on + // the stream (`observed_permission_denials` -- an + // operator's own rule denied it without asking anyone, + // which is legitimate and not a sentinel failure). + // Explained by NEITHER is the one case nothing on the + // whole stream accounts for -- the observable proof + // that the host was never consulted for it at all. + // Fail-closed, not merely logged: a run whose own + // approval ledger cannot account for one of its own + // permission decisions is not one this adapter can + // vouch for having driven correctly. + for denial in &permission_denials { + let explained = denial.tool_use_id.as_deref().is_some_and(|id| { + outcome.bridged_tool_use_ids.contains(id) + || outcome.observed_permission_denials.contains(id) + }); + if !explained { + return Err(AdapterError::protocol( + "claude", + "start", + format!( + "claude's own result reported a permission denial for {} with \ + no corresponding entry in this turn's own approval ledger, and \ + no matching system/permission_denied notice on the stream -- \ + the host was never consulted for it", + denial.tool_name.as_deref().unwrap_or("") + ), + )); + } + } + return Ok(outcome); } - StreamLine::TurnComplete => return Ok(outcome), StreamLine::Other => {} } } @@ -299,7 +634,6 @@ mod tests { use super::*; use std::sync::Arc as StdArc; use tokio::io::AsyncReadExt; - use tokio::sync::broadcast; use crew_protocol::{ DecidedBy, ProjectId, Redacted, Run, RunFlags, RunState, TaskRef, Timestamp, Worker, @@ -354,7 +688,53 @@ mod tests { #[test] fn classifies_the_result_marker() { let value = serde_json::json!({"type": "result", "subtype": "success"}); - assert_eq!(classify_line(&value), StreamLine::TurnComplete); + assert_eq!( + classify_line(&value), + StreamLine::TurnComplete { + permission_denials: Vec::new(), + } + ); + } + + /// `result`'s own `permission_denials` array, parsed off the exact + /// shape a live capture confirmed (`tool_name`, `tool_use_id`, + /// `tool_input` per entry -- `tool_input` itself unused here, this + /// reader only needs the other two to reconcile against + /// `TurnOutcome::bridged_tool_use_ids`). + #[test] + fn classifies_the_result_markers_own_permission_denials() { + let value = serde_json::json!({ + "type": "result", + "subtype": "success", + "permission_denials": [ + {"tool_name": "Bash", "tool_use_id": "toolu_1", "tool_input": {}}, + ], + }); + assert_eq!( + classify_line(&value), + StreamLine::TurnComplete { + permission_denials: vec![PermissionDenial { + tool_name: Some("Bash".to_string()), + tool_use_id: Some("toolu_1".to_string()), + }], + } + ); + } + + /// `control_response`'s own `request_id`, nested under `response` -- + /// the same nesting the reply this reader itself writes uses. + #[test] + fn classifies_a_control_response_and_reads_its_nested_request_id() { + let value = serde_json::json!({ + "type": "control_response", + "response": {"subtype": "success", "request_id": "crew-initialize", "response": {}}, + }); + assert_eq!( + classify_line(&value), + StreamLine::ControlResponse { + request_id: Some("crew-initialize".to_string()), + } + ); } #[test] @@ -374,13 +754,20 @@ mod tests { } #[test] - fn classifies_system_init_and_carries_its_session_id() { - let value = - serde_json::json!({"type": "system", "subtype": "init", "session_id": "sess-1"}); + fn classifies_system_init_and_carries_its_session_id_version_and_permission_mode() { + let value = serde_json::json!({ + "type": "system", + "subtype": "init", + "session_id": "sess-1", + "claude_code_version": "2.1.268", + "permissionMode": "auto", + }); assert_eq!( classify_line(&value), StreamLine::SystemInit { - session_id: Some("sess-1".to_string()) + session_id: Some("sess-1".to_string()), + claude_code_version: Some("2.1.268".to_string()), + permission_mode: Some("auto".to_string()), } ); } @@ -417,14 +804,15 @@ mod tests { /// End to end against real in-memory pipes (`tokio::io::duplex`), /// no subprocess involved: proves the loop reads assistant text and /// a tool-start off a simulated stdout, stops at `result`, and never - /// touches stdin when no control request arrives. + /// writes anything to stdin beyond its own initialize handshake when + /// no `can_use_tool` control request ever arrives. #[tokio::test] async fn drives_a_turn_with_no_approval_needed() { let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); let (adapter_stdin, mut stdin_capture) = tokio::io::duplex(4096); let script = concat!( - r#"{"type":"system","subtype":"init","session_id":"sess-1"}"#, + r#"{"type":"system","subtype":"init","session_id":"sess-1","claude_code_version":"2.1.268","permissionMode":"auto"}"#, "\n", r#"{"type":"assistant","uuid":"u1","message":{"content":[{"type":"text","text":"hello"}]}}"#, "\n", @@ -463,6 +851,8 @@ mod tests { &approval_service, &callback, ids(), + None, + "hello", ) .await .expect("must not error"); @@ -485,11 +875,204 @@ mod tests { )); } - // Nothing should have been written back to stdin: no control - // request ever arrived to answer. + // The only thing ever written to stdin is this reader's own + // initialize handshake -- this script never sends back a + // matching `control_response`, so the prompt itself is never + // sent either, and (with no control request arriving) nothing + // else is written after that. let mut written = Vec::new(); stdin_capture.read_to_end(&mut written).await.unwrap(); - assert!(written.is_empty()); + let written = String::from_utf8(written).expect("valid utf-8"); + let mut written_lines = written.lines(); + let first_line: serde_json::Value = + serde_json::from_str(written_lines.next().expect("at least one line")).unwrap(); + assert_eq!(first_line["type"], "control_request"); + assert_eq!(first_line["request_id"], INITIALIZE_REQUEST_ID); + assert_eq!(first_line["request"]["subtype"], "initialize"); + assert!(written_lines.next().is_none(), "no second line expected"); + + db.shutdown().await.ok(); + } + + /// An incompatible (or missing) `claude_code_version` aborts the + /// turn as soon as `system/init` arrives -- before any assistant + /// text or tool call is ever processed, not merely reported + /// afterward. Proven by a script whose `system/init` claims a + /// version far outside the tested range, then keeps sending + /// content `drive_turn` must never reach. + #[tokio::test] + async fn an_incompatible_reported_version_aborts_the_turn_immediately() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + + let script = concat!( + r#"{"type":"system","subtype":"init","session_id":"sess-1","claude_code_version":"0.1.0"}"#, + "\n", + r#"{"type":"assistant","message":{"content":[{"type":"text","text":"should never be read"}]}}"#, + "\n", + r#"{"type":"result","subtype":"success"}"#, + "\n", + ); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = ApprovalService::new( + StdArc::clone(&db), + project_id, + StdArc::new(crate::approval::NoopApprovalCallback) as StdArc, + broadcast::channel(64).0, + ); + + let result = drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + None, + "should never be sent", + ) + .await; + + let err = result.expect_err("an out-of-range version must abort the turn"); + assert_eq!( + err.error_code(), + crate::adapter::AdapterErrorCode::IncompatibleVersion + ); + + db.shutdown().await.ok(); + } + + /// The direct permission-mode assertion, proven the same way: a + /// `system/init` reporting a mode other than the pinned one aborts + /// the turn immediately, before any content is processed. + #[tokio::test] + async fn a_mismatched_reported_permission_mode_aborts_the_turn_immediately() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + + let script = concat!( + r#"{"type":"system","subtype":"init","session_id":"sess-1","claude_code_version":"2.1.268","permissionMode":"plan"}"#, + "\n", + r#"{"type":"assistant","message":{"content":[{"type":"text","text":"should never be read"}]}}"#, + "\n", + r#"{"type":"result","subtype":"success"}"#, + "\n", + ); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = ApprovalService::new( + StdArc::clone(&db), + project_id, + StdArc::new(crate::approval::NoopApprovalCallback) as StdArc, + broadcast::channel(64).0, + ); + + let result = drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + None, + "should never be sent", + ) + .await; + + let err = result.expect_err("a mismatched permission mode must abort the turn"); + assert_eq!(err.error_code(), crate::adapter::AdapterErrorCode::Protocol); + assert!(err.detail().contains("plan")); + + db.shutdown().await.ok(); + } + + /// `pane_output`, when given, actually receives one formatted line + /// per normalized event -- proven by subscribing to it before the + /// turn runs and reading real lines back, not by inspecting + /// `drive_turn`'s own source for a call site that looks right. + #[tokio::test] + async fn pane_output_receives_one_formatted_line_per_event() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + + let script = concat!( + r#"{"type":"assistant","message":{"content":[{"type":"text","text":"hello"}]}}"#, + "\n", + r#"{"type":"result","subtype":"success"}"#, + "\n", + ); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = ApprovalService::new( + StdArc::clone(&db), + project_id, + StdArc::new(crate::approval::NoopApprovalCallback) as StdArc, + broadcast::channel(64).0, + ); + + let (pane_tx, mut pane_rx) = broadcast::channel(16); + + drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + Some(&pane_tx), + "hello", + ) + .await + .expect("must not error"); + + let first = pane_rx.try_recv().expect("a line for the assistant text"); + assert_eq!(first, super::super::pane::render_line("assistant: hello")); + let second = pane_rx.try_recv().expect("a line for turn completion"); + assert_eq!(second, super::super::pane::render_line("turn complete")); + assert!( + pane_rx.try_recv().is_err(), + "no third line should have been sent" + ); db.shutdown().await.ok(); } @@ -591,6 +1174,8 @@ mod tests { task_id, worker_id, }, + None, + "hello", ) .await } @@ -598,7 +1183,7 @@ mod tests { let request = concat!( r#"{"type":"control_request","request_id":"req-1","request":"#, - r#"{"subtype":"can_use_tool","tool_name":"Write","input":{}}}"#, + r#"{"subtype":"can_use_tool","tool_name":"Write","input":{},"tool_use_id":"toolu_1"}}"#, "\n", ); test_side.write_all(request.as_bytes()).await.unwrap(); @@ -650,11 +1235,423 @@ mod tests { let mut written = Vec::new(); stdin_capture.read_to_end(&mut written).await.unwrap(); let written = String::from_utf8(written).expect("valid utf-8"); - let first_line = written.lines().next().expect("at least one reply line"); - let reply: serde_json::Value = serde_json::from_str(first_line).expect("valid json line"); - assert_eq!(reply["response"]["request_id"], "req-1"); + // The first line is this reader's own initialize handshake + // (never answered by this test's script), so the reply to + // "req-1" is found by its own `request_id`, not by position. + let reply = written + .lines() + .map(|line| serde_json::from_str::(line).expect("valid json line")) + .find(|value| value["response"]["request_id"] == "req-1") + .expect("a reply naming req-1 must have been written"); assert_eq!(reply["response"]["response"]["behavior"], "allow"); db.shutdown().await.ok(); } + + fn noop_approval_service( + db: &StdArc, + project_id: ProjectId, + ) -> ApprovalService { + ApprovalService::new( + StdArc::clone(db), + project_id, + StdArc::new(crate::approval::NoopApprovalCallback) as StdArc, + broadcast::channel(64).0, + ) + } + + /// The initialize handshake actually gates when the prompt goes out: + /// a hook-shaped line arriving first is processed and ignored (falls + /// through to `Other`, like today), and the prompt itself is not + /// written until the matching `control_response` arrives -- proven + /// by reading stdin back in order, not by inspecting source. + #[tokio::test] + async fn the_prompt_is_sent_only_after_the_initialize_handshake_completes() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, mut stdin_capture) = tokio::io::duplex(4096); + + let script = concat!( + r#"{"type":"system","subtype":"hook_started"}"#, + "\n", + r#"{"type":"control_response","response":{"subtype":"success","request_id":"crew-initialize","response":{}}}"#, + "\n", + r#"{"type":"result","subtype":"success"}"#, + "\n", + ); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = noop_approval_service(&db, project_id); + + drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + None, + "the real prompt", + ) + .await + .expect("must not error"); + + let mut written = Vec::new(); + stdin_capture.read_to_end(&mut written).await.unwrap(); + let written = String::from_utf8(written).expect("valid utf-8"); + let lines: Vec = written + .lines() + .map(|line| serde_json::from_str(line).expect("valid json line")) + .collect(); + assert_eq!(lines.len(), 2, "expected the handshake, then the prompt"); + assert_eq!(lines[0]["request_id"], INITIALIZE_REQUEST_ID); + assert_eq!(lines[1]["type"], "user"); + assert_eq!(lines[1]["message"]["content"][0]["text"], "the real prompt"); + + db.shutdown().await.ok(); + } + + /// A stream that closes before the handshake's own `control_response` + /// ever arrives means the prompt was never sent -- distinct from an + /// ordinary post-`result` close, and reported as an error rather + /// than an empty, ordinary-looking [`TurnOutcome`]. + #[tokio::test] + async fn a_stream_closed_before_the_handshake_completes_is_an_error() { + let (test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = noop_approval_service(&db, project_id); + + let result = drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + None, + "never sent", + ) + .await; + + let err = result.expect_err("closing before the handshake completes must error"); + assert_eq!(err.error_code(), crate::adapter::AdapterErrorCode::Protocol); + + db.shutdown().await.ok(); + } + + /// The ledger-reconciliation check, fail-closed direction: a + /// `permission_denials` entry whose `tool_use_id` this reader never + /// bridged into `ApprovalService` AND never saw named on a real + /// `system`/`permission_denied` notice -- explained by neither leg + /// of the three-way check -- aborts the turn: the observable proof + /// the host was never consulted for it. + #[tokio::test] + async fn an_unexplained_permission_denial_fails_the_turn() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + + let script = concat!( + r#"{"type":"result","subtype":"success","permission_denials":"#, + r#"[{"tool_name":"Bash","tool_use_id":"toolu_never_bridged"}]}"#, + "\n", + ); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = noop_approval_service(&db, project_id); + + let result = drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + None, + "hello", + ) + .await; + + let err = result.expect_err("an unexplained denial must fail the turn"); + assert_eq!(err.error_code(), crate::adapter::AdapterErrorCode::Protocol); + assert!(err.detail().contains("Bash")); + + db.shutdown().await.ok(); + } + + /// The real recorded shape, as a fixture rather than an inline + /// literal -- `fixtures/adapters/claude-protocol/permission-denied-frame.jsonl` + /// is the actual live capture (scrubbed) this reader's own + /// `system`/`permission_denied` classification exists to parse. + fn permission_denied_frame_fixture() -> String { + let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../fixtures/adapters/claude-protocol/permission-denied-frame.jsonl"); + std::fs::read_to_string(&path) + .unwrap_or_else(|err| panic!("reading fixture {path:?}: {err}")) + } + + /// Classifies the real fixture, not a hand-typed lookalike. + #[test] + fn classifies_the_real_permission_denied_frame_fixture() { + let line = permission_denied_frame_fixture(); + let line = line.lines().next().expect("fixture has at least one line"); + let value: serde_json::Value = serde_json::from_str(line).expect("valid json line"); + assert_eq!( + classify_line(&value), + StreamLine::PermissionDeniedNotice { + tool_use_id: Some("toolu_00000000000000000000000001".to_string()), + } + ); + } + + /// The ledger-reconciliation check's second leg, proven against the + /// real fixture: a `permission_denials` entry whose `tool_use_id` + /// this reader never bridged, but DID see named on a real + /// `system`/`permission_denied` notice on the stream, is explained + /// -- an operator's own rule denying without asking anyone, not a + /// sentinel failure -- and the turn completes normally. + #[tokio::test] + async fn a_denial_matching_an_observed_permission_denied_notice_does_not_fail_the_turn() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + + let mut script = permission_denied_frame_fixture(); + if !script.ends_with('\n') { + script.push('\n'); + } + script.push_str(concat!( + r#"{"type":"result","subtype":"success","permission_denials":"#, + r#"[{"tool_name":"Bash","tool_use_id":"toolu_00000000000000000000000001"}]}"#, + "\n", + )); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let callback = ProtocolApprovalCallback::new(); + let approval_service = noop_approval_service(&db, project_id); + + let outcome = drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + None, + "hello", + ) + .await + .expect("a denial explained by an observed notice must not fail the turn"); + + assert!( + outcome + .observed_permission_denials + .contains("toolu_00000000000000000000000001") + ); + + db.shutdown().await.ok(); + } + + /// The other direction of the same check: a `permission_denials` + /// entry whose `tool_use_id` DOES match a request this reader + /// bridged through `ApprovalService` -- crew's own ledger accounts + /// for it (a human, or crew's own policy, decided to deny it) -- and + /// the turn completes normally rather than failing closed. + #[tokio::test] + async fn a_denial_matching_a_bridged_request_does_not_fail_the_turn() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, _stdin_capture) = tokio::io::duplex(4096); + + let state_dir = tempfile::TempDir::new().unwrap(); + let db = StdArc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = ProjectId::new(); + let task_id = TaskId::new(); + let worker_id = WorkerId::new(); + let run_id = RunId::new(); + db.run_domain_op(Box::new(move |conn| { + let mut repo = DomainRepository::new(conn, project_id); + repo.upsert_task( + task_id, + &TaskRef { + owner_client_instance_id: "omp-1".into(), + revision: 1, + }, + )?; + repo.create_worker(&Worker { + worker_id, + profile_ref: WorkerProfileRef { + id: worker_id, + fingerprint: "sha256:fake".into(), + adapter: "claude".into(), + model: "test".into(), + permission_envelope: serde_json::json!({}), + }, + parent_worker_id: None, + created_at: Timestamp::now(), + })?; + repo.submit_run( + &Run { + run_id, + task_id, + worker_id, + state: RunState::try_from("queued").unwrap(), + flags: RunFlags::default(), + vendor_session_id: None, + started_at: None, + completed_at: None, + }, + None, + None, + )?; + Ok(serde_json::json!({})) + })) + .await + .expect("seed"); + for state in ["starting", "working"] { + let to = RunState::try_from(state).unwrap(); + db.run_domain_op(Box::new(move |conn| { + DomainRepository::new(conn, project_id) + .transition_run(run_id, &to, None) + .map(|_| serde_json::json!({})) + })) + .await + .unwrap(); + } + + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::new(parking_lot::Mutex::new(Vec::new())), + }); + let callback = StdArc::new(ProtocolApprovalCallback::new()); + let approval_service = ApprovalService::new( + StdArc::clone(&db), + project_id, + StdArc::clone(&callback) as StdArc, + broadcast::channel(64).0, + ); + + let drive = tokio::spawn({ + let callback = StdArc::clone(&callback); + async move { + drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + RunIdentity { + run_id, + task_id, + worker_id, + }, + None, + "hello", + ) + .await + } + }); + + let request = concat!( + r#"{"type":"control_request","request_id":"req-1","request":"#, + r#"{"subtype":"can_use_tool","tool_name":"Bash","input":{},"tool_use_id":"toolu_denied"}}"#, + "\n", + ); + test_side.write_all(request.as_bytes()).await.unwrap(); + + tokio::task::yield_now().await; + let approval_id = { + let value = db + .run_domain_op(Box::new(move |conn| { + conn.query_row( + "SELECT approval_id FROM approvals WHERE run_id = ?1", + [run_id.to_string()], + |row| row.get::<_, String>(0), + ) + .map(|id| serde_json::json!({ "id": id })) + .map_err(Into::into) + })) + .await + .expect("approval row must exist"); + crew_protocol::ApprovalId::parse(value["id"].as_str().unwrap()).unwrap() + }; + let decide_service = ApprovalService::new( + StdArc::clone(&db), + project_id, + callback as StdArc, + broadcast::channel(64).0, + ); + decide_service + .decide( + approval_id, + "omp-1", + "deny", + &Redacted::assert_runtime_authored("denied for the test"), + DecidedBy::Human, + ) + .await + .expect("decide must succeed"); + + let result_line = concat!( + r#"{"type":"result","subtype":"success","permission_denials":"#, + r#"[{"tool_name":"Bash","tool_use_id":"toolu_denied"}]}"#, + "\n", + ); + test_side.write_all(result_line.as_bytes()).await.unwrap(); + drop(test_side); + + drive + .await + .expect("task must not panic") + .expect("a denial crew itself decided must not fail the turn"); + + db.shutdown().await.ok(); + } } diff --git a/crates/runtime/src/adapter/claude_protocol/reconcile.rs b/crates/runtime/src/adapter/claude_protocol/reconcile.rs index f17f6cfa..921e0220 100644 --- a/crates/runtime/src/adapter/claude_protocol/reconcile.rs +++ b/crates/runtime/src/adapter/claude_protocol/reconcile.rs @@ -12,15 +12,18 @@ //! stop repeating (screen recognition's own "the accumulator still says //! the gate is on screen" shape, one layer up). //! -//! **The entry schema below is provisional**, not yet checked against a -//! live capture: this spike's own stop conditions ask that of the -//! control channel, and the same caution applies here. What every entry -//! needs to carry for this module's OWN purpose is narrow regardless of -//! the vendor's exact JSON shape: an identifier stable enough to -//! correlate against what this run already journaled, and a kind -//! coarse enough to describe a gap without ever carrying vendor -//! content into a closed-set field (`TranscriptEntryKind`'s own doc -//! comment). +//! **Checked against a live capture** (three real turns against +//! `claude_code_version` `2.1.268`): the entry shape below -- `uuid` as +//! the correlation id, `type` as the kind discriminant -- is exactly +//! what a real transcript carries for `user`/`assistant` entries, which +//! is all this module's own purpose needs regardless of the vendor's +//! full JSON shape. One thing the same capture found that this module +//! did not originally account for: a real transcript also carries +//! non-conversational bookkeeping entries (a `SessionStart` hook's own +//! error, observed as `type: "attachment"`) that DO carry a `uuid` but +//! never appear on the live control-channel stream at all -- see +//! [`TranscriptEntryKind::Attachment`]'s own doc comment for why that +//! matters to [`find_gaps`] specifically. use std::collections::HashSet; @@ -38,12 +41,36 @@ pub enum TranscriptEntryKind { User, /// An assistant-authored turn: text, tool use, or both. Assistant, + /// A vendor-authored bookkeeping/diagnostic record -- observed as + /// `type: "attachment"` (a `SessionStart` hook's own error output, + /// in the live capture that found this) -- written to the durable + /// transcript but confirmed, across every capture taken so far, + /// never to appear on the live control-channel stream at all. + /// [`find_gaps`] excludes this kind from its own gap set for + /// exactly that reason: an entry that never travels the stream by + /// design is not a gap when it is absent from what the stream + /// delivered. Still counted in `examined`, same as every other + /// entry this module can parse -- only the gap set narrows, not the + /// count of what was looked at. + Attachment, /// Anything this module does not yet classify -- session metadata, /// a summary entry, or a future entry type the vendor adds. Grouped /// under one variant rather than one per unknown shape so a vendor /// addition never requires touching this enum's own callers; the /// PARSED `kind` this module reads directly (a small string) is /// enough to distinguish them without one Rust variant each. + /// + /// Deliberately NOT excluded from [`find_gaps`]'s own gap set, on + /// the same basis [`Attachment`](Self::Attachment) IS excluded: this + /// module only ever denylists a kind it has positively confirmed + /// never streams. A kind it has not yet seen is `Other`, not + /// `Attachment`, and stays gappable -- noisy (a false-positive gap + /// the first time a genuinely new non-streaming kind appears) is + /// the correct failure direction for a mechanism whose whole + /// purpose is catching silence; silently excluding every unknown + /// kind (an allowlist instead of a denylist) would make a real + /// future drop of an unrecognized kind invisible by construction, + /// which is the failure this module exists to prevent. Other, } @@ -96,6 +123,7 @@ fn parse_transcript_line_uninjected(line: &[u8]) -> Option { let kind = match value.get("type").and_then(serde_json::Value::as_str) { Some("user") => TranscriptEntryKind::User, Some("assistant") => TranscriptEntryKind::Assistant, + Some("attachment") => TranscriptEntryKind::Attachment, _ => TranscriptEntryKind::Other, }; Some(TranscriptEntry { entry_id, kind }) @@ -184,6 +212,14 @@ pub fn find_gaps( let examined = entries.len() as u64; let gaps: Vec = entries .into_iter() + // `examined` above already counted every entry this module + // could parse, `Attachment` included -- only the gap set + // narrows here. A denylist (name the kinds confirmed never to + // stream), not an allowlist (name the kinds allowed to be + // gaps): see `TranscriptEntryKind::Other`'s own doc comment for + // why an unrecognized future kind must stay gappable rather + // than silently excluded. + .filter(|entry| entry.kind != TranscriptEntryKind::Attachment) .filter(|entry| !journaled_entry_ids.contains(&entry.entry_id)) .collect(); (examined, gaps) @@ -210,6 +246,39 @@ mod tests { assert_eq!(gaps.len(), 2); } + /// The denylist, proven directly: an `attachment`-kind entry never + /// journaled live is counted in `examined` (it was successfully + /// parsed) but never reported as a gap (it is a confirmed + /// never-streams kind) -- the exact case a live capture found real + /// transcripts carry (a `SessionStart` hook's own error record). + #[test] + fn an_attachment_entry_is_examined_but_never_reported_as_a_gap() { + let transcript = [entry("u1", "user"), entry("hook1", "attachment")].join("\n"); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &HashSet::new()); + assert_eq!(examined, 2, "attachment entries are parsed, so they count"); + assert_eq!( + gaps, + vec![TranscriptEntry { + entry_id: "u1".to_string(), + kind: TranscriptEntryKind::User, + }], + "only the user entry may be a gap; the attachment entry must not be" + ); + } + + /// The denylist's other half: a kind this module does not recognize + /// at all (never `Other`'s own doc comment) still stays gappable -- + /// proving the choice is a denylist of confirmed non-streaming + /// kinds, not an allowlist of `User`/`Assistant` alone. + #[test] + fn an_unrecognized_kind_still_counts_as_a_gap() { + let transcript = entry("mystery1", "some-future-kind-this-module-has-never-seen"); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &HashSet::new()); + assert_eq!(examined, 1); + assert_eq!(gaps.len(), 1, "an unrecognized kind must remain gappable"); + assert_eq!(gaps[0].kind, TranscriptEntryKind::Other); + } + /// The ordinary case: everything the transcript recorded was also /// journaled live, so reconciliation finds nothing to repair -- /// still `examined > 0`, distinguishing "checked and agreed" from diff --git a/crates/runtime/src/adapter/registry.rs b/crates/runtime/src/adapter/registry.rs index adb7e7f8..9de062de 100644 --- a/crates/runtime/src/adapter/registry.rs +++ b/crates/runtime/src/adapter/registry.rs @@ -1528,18 +1528,12 @@ fn build_adapter( }; return match kind { super::AdapterKind::Claude => { - // `db`/`project_id`/`events_tx`/`display`/`resume_cursor` - // are unused here: unlike a `TuiAdapter`, this adapter - // builds no internal pane/sink of its own -- the caller - // (`run_one`/`resume_one`) builds the shared - // `DomainAdapterEventSink` from those same values and - // hands it to `Adapter::start` separately, exactly the - // path every adapter this registry constructs goes - // through. `resume_cursor` in particular is always - // `None` here in practice: `crate::recovery`'s own - // `AdapterMode::Protocol` arm already refuses resume - // before this function is ever reached for one. - // + // `resume_cursor` is always `None` here in practice: + // `crate::recovery`'s own `AdapterMode::Protocol` arm + // already refuses resume before this function is ever + // reached for one -- there is no seam here to hand it + // to, so it is intentionally never consumed. + let _ = resume_cursor; // `profile.model` passed straight through, trimmed -- // see `super::claude_protocol::adapter`'s own module doc // comment on why the boot-config-precedence @@ -1551,6 +1545,33 @@ fn build_adapter( // guess at emptiness this adapter has any reason to // expect. let model = Some(profile.model.trim().to_string()).filter(|m| !m.is_empty()); + // Pane support, on the same terms `build_tui_adapter` + // gets it: only when a `TuiSupport` bundle was ever + // supplied. `None` here means this adapter runs with no + // pane at all -- never a refusal to start, unlike TUI + // mode's own `TuiModeUnavailable` -- because the pane is + // this adapter's convenience view, not its control + // surface (see `claude_protocol::pane`'s own module doc + // comment). + let pane_support = tui.map(|tui| { + let pane_coordinator = + build_pane_coordinator(&tui, db, project_id, events_tx, repo_root); + let placement = display + .as_ref() + .map(|selection| selection.placement) + .unwrap_or(DisplayPlacement::SplitRight); + let launch_program = display + .as_ref() + .and_then(|selection| selection.launch_program); + super::claude_protocol::pane::PaneSupport { + pane_coordinator, + panes_dir: tui.panes_dir.clone(), + placement, + forced_backend: tui.forced_backend, + launch_program, + close_on_exit: tui.close_on_exit, + } + }); Ok(Arc::new( super::claude_protocol::adapter::ClaudeProtocolAdapter::new( repo_root.to_path_buf(), @@ -1560,6 +1581,7 @@ fn build_adapter( approval_service: Arc::clone(&protocol.approval_service), callback: Arc::clone(&protocol.callback), }, + pane_support, ), )) } @@ -1644,6 +1666,41 @@ fn build_adapter( /// [`Adapter::resume`] re-tails from exactly where the journal says the /// previous incarnation stopped; the transcript path itself is derived /// deterministically inside the adapter from the vendor's own layout. +/// Builds the one [`PaneCoordinator`] a run's pane wiring needs from a +/// [`TuiSupport`] bundle -- shared by [`build_tui_adapter`] and +/// `build_adapter`'s `AdapterMode::Protocol` branch, so the two never +/// carry independently-maintained copies of the same construction +/// sequence (the same drift hazard two separately-maintained +/// `AdapterMode` enums already are, one level down, and worth avoiding +/// here rather than repeating it). +fn build_pane_coordinator( + tui: &Arc, + db: Arc, + project_id: crew_protocol::ProjectId, + events_tx: tokio::sync::broadcast::Sender, + repo_root: &std::path::Path, +) -> Arc { + // Same patterns already validated once at startup + // (`lifecycle.rs`'s fail-closed `Redactor::with_org_rules` call) -- + // a compile error building this second instance from the same + // config can only be a bug. + let redactor = crate::security::redaction::Redactor::with_org_rules(&tui.org_security_patterns) + .expect("org_security_patterns already validated at startup"); + Arc::new( + PaneCoordinator::new( + Arc::clone(&tui.display_registry), + db, + project_id, + events_tx, + tui.crewd_path.clone(), + tui.state_dir.clone(), + repo_root.to_path_buf(), + redactor, + ) + .with_force_hidden_displays(tui.force_hidden_displays), + ) +} + #[allow(clippy::too_many_arguments)] fn build_tui_adapter( vendor: V, @@ -1684,21 +1741,7 @@ fn build_tui_adapter( // (`lifecycle.rs`'s fail-closed `Redactor::with_org_rules` call) -- // a compile error building this second instance from the same // config can only be a bug. - let redactor = crate::security::redaction::Redactor::with_org_rules(&tui.org_security_patterns) - .expect("org_security_patterns already validated at startup"); - let pane_coordinator = Arc::new( - PaneCoordinator::new( - Arc::clone(&tui.display_registry), - db, - project_id, - events_tx, - tui.crewd_path.clone(), - tui.state_dir.clone(), - repo_root.to_path_buf(), - redactor, - ) - .with_force_hidden_displays(tui.force_hidden_displays), - ); + let pane_coordinator = build_pane_coordinator(tui, db, project_id, events_tx, repo_root); let placement = display .as_ref() .map(|selection| selection.placement) diff --git a/crates/runtime/src/supervisor/mod.rs b/crates/runtime/src/supervisor/mod.rs index c67a251a..39590ebe 100644 --- a/crates/runtime/src/supervisor/mod.rs +++ b/crates/runtime/src/supervisor/mod.rs @@ -5,6 +5,17 @@ //! `tokio::process::Command` directly, so every worker gets the same //! process-group, bounding, and escalation guarantees regardless of which //! adapter owns it. +//! +//! **One documented exception**: `crate::adapter::claude_protocol` spawns +//! via a bare `tokio::process::Command`, not `Supervisor::spawn` -- its +//! own `settle_after_turn` reuses this module's [`EscalationTimings`]/ +//! [`TerminationOutcome`] to get the same escalation discipline, but +//! without a process group of its own (a real gap, not a design choice: +//! `claude_protocol::reader::drive_turn` reads/writes over generic +//! `AsyncRead`/`AsyncWrite`, which [`ManagedProcess`]'s framed-stdout/ +//! queued-stdin API does not satisfy without a larger rewrite than that +//! adapter's own spike scope covers). Carried forward, not silently +//! divergent: see that adapter's own `settle_after_turn` doc comment. mod environment; mod output; diff --git a/fixtures/adapters/claude-protocol/README.md b/fixtures/adapters/claude-protocol/README.md new file mode 100644 index 00000000..ce0a33d1 --- /dev/null +++ b/fixtures/adapters/claude-protocol/README.md @@ -0,0 +1,109 @@ +# `claude-protocol` discriminating-experiment fixtures: provenance + +Three files, from a live-capture sequence that answered whether +`can_use_tool` approval requests travel over claude's `stream-json` stdout +the way `claude_protocol::approval_bridge` assumes. The first two are from +an early round that turned out to be missing the flag actually needed to +enable that control channel at all; the third is from the corrected call +that supplied it and got a real, answered `can_use_tool` round trip. All +three are kept for the real wire shapes they captured. None is a raw, +byte-for-byte scrub of the real capture the way `claude-tui/session.jsonl` +is -- all are **condensed and scrubbed**: every identifying field +(username, home path, session id, message id, request id, tool-use id, +timestamp, the actual echoed test string) was replaced with a stable +placeholder, and for `ask-rule-no-control-frame.jsonl` specifically, the +surrounding SessionStart hook lines and the real +`tools`/`mcp_servers`/`agents`/`skills`/`plugins` arrays (which in the raw +capture ran to several hundred real, environment-specific entries) were +trimmed to a representative few. The real captures underneath (nine live +`claude -p` calls against `claude_code_version` `2.1.268`, in a real trusted +repository, real billed API calls) are not committed anywhere and exist only +in this session's own local scratch. + +## `permission-denied-frame.jsonl` + +One line: the `type: "system", subtype: "permission_denied"` shape a real +`rm` call produced when denied by a hard `deny` rule. NOT evidence about +`can_use_tool`/the control channel -- the underlying live call was later +found to be confounded (the `rm` command matched the operator's own personal +deny rule, so no approval request was ever generated to observe; a hard deny +never asks anyone). + +**No longer a curiosity -- this is load-bearing test input.** +`claude_protocol::reader::classify_line` originally fell through to +`Other` for this shape, and this fixture documented that as a confirmed +gap. It now parses it (`StreamLine::PermissionDeniedNotice`), because +`can-use-tool-round-trip.jsonl`'s own three `tool_use_id` correlations +(assistant `tool_use` -> this notice -> the closing `result`'s own +`permission_denials` entry) showed a hard deny announces itself on the +stream and correlates to its own denial entry by that id -- the second +leg of `drive_turn`'s ledger-reconciliation check, alongside what it +bridged into `ApprovalService`. `crates/runtime/src/adapter/claude_protocol/reader.rs`'s +own test suite reads this exact file as its fixture for that +classification, not a hand-typed lookalike. + +## `ask-rule-no-control-frame.jsonl` + +The full turn from a calibration call: `--permission-mode manual` +(reports back as `permissionMode: "default"`), a custom `ask` rule for +one specific `Bash` command via `--settings`, and a prompt invoking +exactly that command. `permissionMode` was reported back and matched +the mode this call actually passed (`manual` was passed, `"default"` +came back, consistent with what was already measured) -- this fixture +does NOT show the production `auto` pin; a reader reaching for it to +exercise the posture assertion against the pinned production value +will find `default` here and should not conclude the assertion is +broken. + +**What this fixture does NOT establish, and how that was resolved.** The +tool ran immediately with `permission_denials: []`, and no +`control_request`/`control_response`/`can_use_tool`-shaped line appears +anywhere in the sequence. Two things came to light after this capture +that both bore on what that null result meant -- both are now resolved, +not open: + +1. An empty `permission_denials` on an `ask`-rule call is *consistent + with the rule never having matched at all* -- the documented + hostless behaviour is deny-and-report, not silent allow, so a + genuine ask-match with no host present should show up as a denial, + not run clean; on its own, this capture cannot distinguish "the rule + matched and was allowed" from "the rule never matched". **Eliminated** + by a positive control, not left open: the corrected call that produced + `can-use-tool-round-trip.jsonl` ran a hard `deny` rule alongside the + `ask` rule in the same turn, and the hard-denied command came back + denied and not run -- proof that rule matching itself was live in + that call, the exact discriminator this capture lacked. +2. Neither this call nor any of the seven in this round passed + `--permission-prompt-tool stdio` -- the flag believed (from SDK + source, not CLI docs) to be the actual enabler of `can_use_tool` + control frames at all. Absent that flag, a null result here says + nothing about whether the control channel exists; it may only mean + the mechanism was never turned on. **Confirmed as the actual cause**: + adding only that flag, with everything else unchanged, is what + produced `can-use-tool-round-trip.jsonl`'s own real `can_use_tool` + request. + +Kept as the recorded shape of a `manual`-mode, ask-configured, +allowed-outcome turn -- not as evidence for or against the approval +bridge's own premise; see `can-use-tool-round-trip.jsonl` below for that. + +## `can-use-tool-round-trip.jsonl` + +The corrected call: the same two rules (`ask` on one `Bash` command, +hard `deny` on another) plus `--permission-prompt-tool stdio`. This is +the only one of the three fixtures here that proves the reply side, +live, against the real CLI, rather than against this codebase's own +fake stream: the assistant tries the denied command first (denied and +not run, the positive control described above), then tries the +ask-configured command, which produces a real `control_request` +(`can_use_tool`); a reply built from +`claude_protocol::approval_bridge::build_permission_response`'s own +unmodified shape -- `request_id` only, never `tool_use_id` -- was sent +back, and the command then actually ran, with its real stdout appearing +in both the `tool_result` and the assistant's own final summary. The +closing `result` message's `permission_denials` names only the +hard-denied command, not the ask-configured one that ran -- direct +confirmation that a request `approval_bridge` answers is not also +reported as a denial. Settles, live, that `build_permission_response` +needs no code changes and that `request_id` alone is sufficient for the +CLI's own correlation. diff --git a/fixtures/adapters/claude-protocol/ask-rule-no-control-frame.jsonl b/fixtures/adapters/claude-protocol/ask-rule-no-control-frame.jsonl new file mode 100644 index 00000000..782634d6 --- /dev/null +++ b/fixtures/adapters/claude-protocol/ask-rule-no-control-frame.jsonl @@ -0,0 +1,5 @@ +{"type":"system","subtype":"init","cwd":"/workspace/crew","session_id":"00000000-0000-4000-8000-000000000010","tools":["Bash","Read","Write"],"mcp_servers":[],"model":"claude-sonnet-5","permissionMode":"default","apiKeySource":"none","claude_code_version":"2.1.268","output_style":"default","capabilities":["interrupt_receipt_v1","interrupt_cancel_queued_v1","msg_lifecycle_v1"],"analytics_disabled":false,"product_feedback_disabled":false,"uuid":"00000000-0000-4000-8000-000000000011","memory_paths":{"auto":"/workspace/crew/.claude-memory/"},"messaging_socket_path":"/tmp/cc-socks/0.sock","fast_mode_state":"off","fast_mode_disabled_reason":"sdk_opt_in_required"} +{"type":"assistant","message":{"model":"claude-sonnet-5","id":"msg_000000000000000000000001","type":"message","role":"assistant","content":[{"type":"tool_use","id":"toolu_00000000000000000000000002","name":"Bash","input":{"command":"echo fixture-marker","description":"Echo the requested string"},"caller":{"type":"direct"}}]},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000010","uuid":"00000000-0000-4000-8000-000000000012","timestamp":"2026-01-01T00:00:00.000Z","request_id":"req_000000000000000000000001"} +{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_00000000000000000000000002","type":"tool_result","content":"fixture-marker","is_error":false}]},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000010","uuid":"00000000-0000-4000-8000-000000000013","timestamp":"2026-01-01T00:00:01.000Z","tool_use_result":{"stdout":"fixture-marker","stderr":"","interrupted":false,"isImage":false,"noOutputExpected":false}} +{"type":"assistant","message":{"model":"claude-sonnet-5","id":"msg_000000000000000000000002","type":"message","role":"assistant","content":[{"type":"text","text":"fixture-marker"}]},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000010","uuid":"00000000-0000-4000-8000-000000000014","timestamp":"2026-01-01T00:00:02.000Z","request_id":"req_000000000000000000000002"} +{"type":"result","subtype":"success","session_id":"00000000-0000-4000-8000-000000000010","permission_denials":[],"is_error":false,"num_turns":2,"result":"fixture-marker","uuid":"00000000-0000-4000-8000-000000000015"} diff --git a/fixtures/adapters/claude-protocol/can-use-tool-round-trip.jsonl b/fixtures/adapters/claude-protocol/can-use-tool-round-trip.jsonl new file mode 100644 index 00000000..0c2e2d89 --- /dev/null +++ b/fixtures/adapters/claude-protocol/can-use-tool-round-trip.jsonl @@ -0,0 +1,9 @@ +{"type":"assistant","message":{"model":"claude-sonnet-5","id":"","type":"message","role":"assistant","content":[{"type":"tool_use","id":"","name":"Bash","input":{"command":"echo denyprobe one","description":"Print denyprobe one"}}]}} +{"type":"system","subtype":"permission_denied","tool_name":"Bash","tool_use_id":"","decision_reason_type":"subcommandResults","message":"Permission to use Bash with command echo denyprobe one has been denied.","uuid":"","session_id":""} +{"type":"user","message":{"role":"user","content":[{"type":"tool_result","content":"Permission to use Bash with command echo denyprobe one has been denied.","is_error":true,"tool_use_id":""}]}} +{"type":"assistant","message":{"model":"claude-sonnet-5","id":"","type":"message","role":"assistant","content":[{"type":"tool_use","id":"","name":"Bash","input":{"command":"echo askprobe two","description":"Print askprobe two"}}]}} +{"type":"control_request","request_id":"","request":{"subtype":"can_use_tool","tool_name":"Bash","display_name":"Bash","input":{"command":"echo askprobe two","description":"Print askprobe two"},"description":"Print askprobe two","decision_reason_type":"rule","tool_use_id":""}} +{"type":"control_response","response":{"subtype":"success","request_id":"","response":{"behavior":"allow"}}} +{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"","content":"askprobe two","is_error":false}]}} +{"type":"assistant","message":{"model":"claude-sonnet-5","id":"","type":"message","role":"assistant","content":[{"type":"text","text":"The first command was denied by permissions:\n\n- `echo denyprobe one` — denied, not run.\n- `echo askprobe two` — ran, output: `askprobe two`"}]}} +{"type":"result","subtype":"success","is_error":false,"num_turns":3,"session_id":"","permission_denials":[{"tool_name":"Bash","tool_use_id":"","tool_input":{"command":"echo denyprobe one","description":"Print denyprobe one"}}]} diff --git a/fixtures/adapters/claude-protocol/permission-denied-frame.jsonl b/fixtures/adapters/claude-protocol/permission-denied-frame.jsonl new file mode 100644 index 00000000..b7286d77 --- /dev/null +++ b/fixtures/adapters/claude-protocol/permission-denied-frame.jsonl @@ -0,0 +1 @@ +{"type":"system","subtype":"permission_denied","tool_name":"Bash","tool_use_id":"toolu_00000000000000000000000001","decision_reason_type":"subcommandResults","message":"Permission to use Bash with command rm /workspace/scratch/throwaway.txt has been denied.","uuid":"00000000-0000-4000-8000-000000000001","session_id":"00000000-0000-4000-8000-000000000000"}