diff --git a/crates/protocol/src/event.rs b/crates/protocol/src/event.rs index f9fe9c92..20baa319 100644 --- a/crates/protocol/src/event.rs +++ b/crates/protocol/src/event.rs @@ -905,6 +905,27 @@ pub enum RuntimeEvent { output_tokens: u64, cost_usd: Option, }, + /// A protocol-first adapter reconciled its journal against the + /// vendor's own durable transcript for this run. Carries no free + /// text, only counts. `examined == 0` is itself a finding, not a + /// clean pass: a reconciliation that never looked at anything is + /// indistinguishable from one that never ran at all, and this is the + /// field that lets a reader (or a future check) tell those two + /// apart. No repair mechanism exists yet, so `gapsRepaired` is + /// always `0` today; `gapsFound` and `gapsRepaired` being unequal is + /// the current, expected state, not a defect, until an actual repair + /// step is built. + AdapterReconciliationEvent { + run_id: RunId, + task_id: TaskId, + worker_id: WorkerId, + #[ts(type = "number")] + examined: u64, + #[ts(type = "number")] + gaps_found: u64, + #[ts(type = "number")] + gaps_repaired: u64, + }, /// An artifact produced by a worker adapter. AdapterArtifactEvent { run_id: RunId, @@ -1651,11 +1672,14 @@ mod redaction_enumeration { ( "RuntimeEvent::EscalationRaised.reason", "A machine-assigned code from a closed set, chosen by the runtime and never \ - caller- or vendor-derived: the three production construction sites pass \ + caller- or vendor-derived: the four production construction sites pass \ the literals `repeated_failure` \ (crates/runtime/src/adapter/run_lifecycle.rs), `write_violation` \ - (crates/runtime/src/domain/repository.rs), and `vendorFirstRunGate` \ - (crates/runtime/src/adapter/event_sink.rs). This reason previously \ + (crates/runtime/src/domain/repository.rs), `vendorFirstRunGate` \ + (crates/runtime/src/adapter/event_sink.rs), and `claudeWorkspaceTrustPending` \ + (crates/runtime/src/adapter/event_sink.rs, a protocol-mode Claude adapter's own \ + pre-spawn trust check -- distinct from `vendorFirstRunGate` because that run has \ + no live pane a human could answer the gate from). This reason previously \ offered two supports and neither existed -- it cited \"the field's own doc\", \ which had no doc, and said the worker's text travels in the sibling \ `question`, which no production site populates before `vendorFirstRunGate`. \ diff --git a/crates/runtime/src/adapter/claude_protocol/adapter.rs b/crates/runtime/src/adapter/claude_protocol/adapter.rs new file mode 100644 index 00000000..340cb765 --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/adapter.rs @@ -0,0 +1,615 @@ +//! 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. +//! +//! 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. +//! +//! **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. +//! +//! `--model` is threaded through as a plain, already-resolved parameter +//! this adapter does not itself resolve -- it is handed exactly +//! `profile.model`, the run's own resolved model, the same field the +//! fix for `build_tui_adapter`'s own model-resolution gap (that +//! adapter's boot-config snapshot silently outliving a later per-run +//! choice) threads through as `run_model`. There is no boot-loaded +//! per-vendor config for protocol mode to override in the first place +//! (no `crew.json` adapters map is ever read here), so that fix's +//! three-row precedence has nothing to collide with: this adapter's own +//! rule is simply "use it, trimmed, when non-empty; omit `--model` +//! otherwise and let claude's own default apply" -- [`Self::start`]'s +//! own test proves this against a REAL launched process's real argv, +//! not a configured value, mirroring that fix's own reproduction shape. + +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::sync::Arc; + +use tokio::io::AsyncWriteExt; +use tokio::sync::Mutex as AsyncMutex; + +use crew_protocol::{Classified, ContentClass}; + +use crate::adapter::AdapterFuture; +use crate::adapter::capability::AdapterCapabilities; +use crate::adapter::error::AdapterError; +use crate::adapter::event_sink::{AdapterEvent, AdapterEventPayload, AdapterEventSink}; +use crate::adapter::r#trait::{ + Adapter, AdapterMessage, AdapterSnapshot, CancelScope, ProbeResult, StartSpec, VendorSessionRef, +}; +use crate::approval::ApprovalService; +use crate::supervisor::EnvironmentPolicy; + +use super::approval_bridge::ProtocolApprovalCallback; +use super::reader::{self, RunIdentity}; +use super::{reconcile, trust}; + +/// Everything [`ClaudeProtocolAdapter`] needs that only exists after the +/// daemon's own `ApprovalService` is constructed -- the server-owned +/// instance `approval/decide` dispatches through, and the callback map +/// [`build_argv`]'s caller registered as `ServerConfig::approval_callback` +/// in its place. One instance is shared by every `ClaudeProtocolAdapter` +/// this daemon ever constructs -- an approval is looked up by its own +/// id, never by which adapter instance created it, so sharing one +/// callback map across runs is correct, not merely convenient (see +/// `approval_bridge::ProtocolApprovalCallback`'s own doc comment on why +/// a per-adapter registry keyed by worker id was rejected). +#[derive(Clone)] +pub(crate) struct ProtocolBundle { + pub(crate) approval_service: Arc, + pub(crate) callback: Arc, +} + +/// Drives one `claude` process over its `stream-json` control channel. +pub(crate) struct ClaudeProtocolAdapter { + repo_root: PathBuf, + /// `WorkerProfile::environmentAllowlist` -- mirrors + /// `ClaudeTuiVendor`'s own field of the same name exactly. + environment_allowlist: Vec, + /// The already-resolved model to launch with, if any -- see this + /// module's own doc comment on why resolving it is not this + /// adapter's job. + model: Option, + bundle: ProtocolBundle, + /// Where to read claude's own trust record from -- + /// `trust::default_claude_json_path()` in production; a temp path in + /// a test. `None` only when `HOME` could not be resolved, in which + /// case [`Self::start`] treats the workspace as untrusted (fail + /// closed, never guessed). + claude_json_path: Option, + /// The binary [`Self::start`] spawns -- `"claude"` (resolved via + /// `PATH`) in production, a fake script's path in a test. A plain + /// field, injected at construction, rather than a `PATH`-shadowing + /// trick or a `#[cfg(test)]` branch: the same shape + /// `AdapterConfig.bin` already gives `ClaudeTuiVendor`, so a test can + /// prove what actually reached `Command::new` without touching + /// global process state. + bin: String, + /// The live child process, held for the duration of one turn so + /// [`Adapter::cancel`] can reach it. `None` before the first + /// [`Adapter::start`] call and after the turn ends. + child: AsyncMutex>, +} + +impl ClaudeProtocolAdapter { + #[must_use] + pub(crate) fn new( + repo_root: PathBuf, + environment_allowlist: Vec, + model: Option, + bundle: ProtocolBundle, + ) -> Self { + Self { + repo_root, + environment_allowlist, + model, + bundle, + claude_json_path: trust::default_claude_json_path(), + bin: "claude".to_string(), + child: AsyncMutex::new(None), + } + } + + /// Overrides where [`Self::start`] reads the trust record from and + /// which binary it spawns -- a test-only seam, never reachable from + /// production construction (`Self::new` above is the only + /// `pub(crate)` constructor `super::super::registry` can see). + #[cfg(test)] + fn with_test_overrides(mut self, claude_json_path: PathBuf, bin: String) -> Self { + self.claude_json_path = Some(claude_json_path); + self.bin = bin; + self + } + + fn env(&self) -> std::collections::HashMap { + let current: std::collections::HashMap = std::env::vars().collect(); + EnvironmentPolicy::baseline().build(¤t, &self.environment_allowlist) + } + + /// Reconciles what [`reader::drive_turn`] actually saw against + /// claude's own transcript, journaling the result through `sink`. + /// Best-effort: a failure here is logged, never propagated -- the + /// turn itself already completed (or failed) by the time this runs, + /// and a reconciliation-journaling failure must not turn an + /// otherwise-successful turn into a failed run. + async fn reconcile_turn( + &self, + canonical_repo_root: &Path, + outcome: &reader::TurnOutcome, + sink: &Arc, + ids: RunIdentity, + ) { + let transcript = outcome.session_id.as_deref().and_then(|session_id| { + std::fs::read(transcript_path(canonical_repo_root, session_id)).ok() + }); + let (examined, gaps) = match &transcript { + Some(bytes) => reconcile::find_gaps(bytes, &outcome.entry_ids), + None => (0, Vec::new()), + }; + // `examined == 0` is itself a finding (`reconcile::find_gaps`'s + // own doc comment): a turn this adapter cannot reconcile at all + // (no session id ever arrived, the transcript file does not + // exist yet, or every line in it failed to parse) must never be + // journaled as an ordinary, successful reconciliation -- it is + // reported as a distinguishable protocol-health failure instead, + // never silently folded into `gaps_found: 0`, which would read + // identically to "checked, and found nothing wrong." + let payload = if examined == 0 { + AdapterEventPayload::ProtocolHealthChanged { + healthy: false, + detail: Classified { + class: ContentClass::Visible, + value: "could not reconcile this turn against claude's own transcript \ + (no entries were readable)" + .to_string(), + }, + } + } else { + AdapterEventPayload::ReconciliationCompleted { + examined, + gaps_found: gaps.len() as u64, + // No repair mechanism exists yet -- this adapter only + // detects and reports a gap, it does not yet re-journal + // a missed entry. Left at `0` rather than guessed. + gaps_repaired: 0, + } + }; + if let Err(err) = sink + .emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload, + cursor: None, + }) + .await + { + tracing::warn!(error = %err, run_id = %ids.run_id, "failed to journal claude-protocol reconciliation"); + } + } +} + +/// `~/.claude/projects//.jsonl` +/// -- the same on-disk layout +/// `crate::adapter::tui::claude::ClaudeTuiVendor::transcript_root` +/// resolves for the TUI path, reused verbatim rather than +/// re-derived: both adapters read the identical vendor-owned directory, +/// just at different times (this one once, after the turn; that one +/// continuously, while it runs). `HOME` unresolved falls back to +/// `/root`, matching that function's own precedent. +fn transcript_path(canonical_repo_root: &Path, session_id: &str) -> PathBuf { + let home = std::env::var("HOME").unwrap_or_else(|_| "/root".to_string()); + PathBuf::from(home) + .join(".claude") + .join("projects") + .join(crate::adapter::tui::claude::slug_cwd(canonical_repo_root)) + .join(format!("{session_id}.jsonl")) +} + +impl Adapter for ClaudeProtocolAdapter { + fn kind(&self) -> &str { + "claude" + } + + fn capabilities(&self) -> AdapterCapabilities { + super::declared_capabilities() + } + + /// Spawns `claude --version` directly -- no model call, matching + /// `TuiAdapter::probe`'s own committed shape (a synchronous + /// `std::process::Command` inside this async fn) exactly, for the + /// same reason: a probe is inherently a one-shot version check, not + /// a long-lived stream this adapter needs to hold open. + fn probe(&self) -> AdapterFuture<'_, ProbeResult> { + Box::pin(async move { + let output = std::process::Command::new(&self.bin) + .arg("--version") + .output() + .map_err(|e| AdapterError::unavailable(self.kind(), "probe", e.to_string()))?; + if !output.status.success() { + return Err(AdapterError::unavailable( + self.kind(), + "probe", + "claude --version exited non-zero", + )); + } + let version = String::from_utf8_lossy(&output.stdout).trim().to_string(); + Ok(ProbeResult { + version: Some(version), + auth_ready: true, + capabilities: self.capabilities(), + inventory_incomplete: true, + }) + }) + } + + fn start(&self, spec: StartSpec, sink: Arc) -> AdapterFuture<'_, ()> { + Box::pin(async move { + let ids = RunIdentity { + run_id: spec.run_id, + task_id: spec.task_id, + worker_id: spec.worker_id, + }; + let canonical_repo_root = + std::fs::canonicalize(&self.repo_root).unwrap_or_else(|_| self.repo_root.clone()); + + let trusted = self + .claude_json_path + .as_deref() + .is_some_and(|path| trust::workspace_trust_accepted(path, &canonical_repo_root)); + if !trusted { + sink.emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload: AdapterEventPayload::WorkspaceTrustPending, + cursor: None, + }) + .await?; + return Err(AdapterError::auth_required( + self.kind(), + "start", + "claude's one-time workspace-trust prompt for this repository has not been \ + accepted yet; trust it once from an interactive claude session in this \ + repository, then resubmit", + )); + } + + 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()))?; + + *self.child.lock().await = Some(child); + + 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; + + 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.reconcile_turn(&canonical_repo_root, &outcome, &sink, ids) + .await; + + Ok(()) + }) + } + + fn resume( + &self, + _session: VendorSessionRef, + _sink: Arc, + ) -> AdapterFuture<'_, ()> { + Box::pin(async move { Err(AdapterError::capability_unsupported(self.kind(), "resume")) }) + } + + /// Committed scope is one worker, one turn: no steering, no + /// follow-up, no out-of-band answer delivery yet, matching + /// `declared_capabilities().steering`'s own `None`. + fn send(&self, _message: AdapterMessage) -> AdapterFuture<'_, ()> { + Box::pin(async move { Err(AdapterError::capability_unsupported(self.kind(), "send")) }) + } + + /// The real bridge exists (`super::approval_bridge`), reached from + /// `super::reader::drive_turn` while a turn is live -- this method + /// is the same trait-level seam every other adapter's own + /// `respond_to_approval` is (nothing calls it in production; + /// `ApprovalService::decide` calls `ApprovalCallback::acknowledge` + /// directly instead, per that trait's own doc comment). Kept a typed + /// refusal here rather than duplicating the bridge behind a second + /// entry point. + fn respond_to_approval(&self, _approval_id: &str, _decision: &str) -> AdapterFuture<'_, ()> { + Box::pin(async move { + Err(AdapterError::capability_unsupported( + self.kind(), + "respond_to_approval", + )) + }) + } + + fn cancel(&self, _scope: CancelScope) -> AdapterFuture<'_, ()> { + Box::pin(async move { + let mut guard = self.child.lock().await; + if let Some(child) = guard.as_mut() { + child + .kill() + .await + .map_err(|e| AdapterError::process(self.kind(), "cancel", e.to_string()))?; + } + Ok(()) + }) + } + + fn snapshot(&self) -> AdapterFuture<'_, AdapterSnapshot> { + Box::pin(async move { + let running = self.child.lock().await.is_some(); + Ok(AdapterSnapshot { + state_summary: if running { + "running".to_string() + } else { + "idle".to_string() + }, + children: Vec::new(), + usage: None, + artifacts: Vec::new(), + }) + }) + } + + fn dispose(&self) -> AdapterFuture<'_, ()> { + Box::pin(async move { + let mut guard = self.child.lock().await; + if let Some(child) = guard.as_mut() { + let _ = child.kill().await; + } + *guard = None; + Ok(()) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::fs::PermissionsExt; + use std::sync::Arc as StdArc; + + use tokio::sync::broadcast; + + use crew_protocol::{RunId, TaskId, WorkerId}; + + use crate::approval::NoopApprovalCallback; + use crate::db::DatabaseHandle; + + /// A minimal in-memory sink that just records every payload it was + /// handed -- the same small fixture `reader.rs`'s own test module + /// defines, duplicated here rather than exported across a private + /// module boundary for one shared use. + struct RecordingSink { + events: StdArc>>, + } + + impl AdapterEventSink for RecordingSink { + fn emit(&self, event: AdapterEvent) -> AdapterFuture<'_, u64> { + self.events.lock().push(event.payload); + Box::pin(async { Ok(0) }) + } + + fn note_real_user_turn(&self, _run_id: RunId) -> AdapterFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + } + + /// A fake `claude` that dumps its own received argv (one entry per + /// line) to `argv_path` before doing anything else, then emits just + /// enough `stream-json` to let [`Adapter::start`] complete: a + /// `system/init` line (so reconciliation has a session id to look + /// for, even though no transcript file will exist for it) and a + /// `result` line. Mirrors + /// `crates/runtime/tests/tui_claude_registry.rs`'s own + /// `write_argv_recording_claude_script` -- the same proof shape + /// (read back what was actually executed, not a stored value that + /// could look right while the real spawn used something else), a + /// stream-json body instead of a PTY transcript. + fn write_argv_recording_fake_claude( + dir: &std::path::Path, + argv_path: &std::path::Path, + ) -> PathBuf { + let script = format!( + r#"#!/bin/sh +printf '%s\n' "$@" > "{argv_path}" +read -r _first_line +echo '{{"type":"system","subtype":"init","session_id":"sess-argv-test"}}' +echo '{{"type":"result","subtype":"success"}}' +"#, + argv_path = argv_path.display(), + ); + let path = dir.join("fake-claude.sh"); + std::fs::write(&path, script).expect("write fake claude script"); + let mut perms = std::fs::metadata(&path).unwrap().permissions(); + perms.set_mode(0o755); + std::fs::set_permissions(&path, perms).unwrap(); + path + } + + /// Keyed on the CANONICALIZED repo root -- `Self::start` canonicalizes + /// before checking (matching `ClaudeTuiVendor::transcript_root`'s own + /// precedent that a real recording proved necessary: a tempdir under + /// `/tmp`/`/var` is frequently a symlink to `/private/tmp`/ + /// `/private/var` on macOS, so the raw and canonical paths can + /// genuinely differ even for a path this test itself just created). + fn trusted_claude_json(dir: &std::path::Path, repo_root: &std::path::Path) -> PathBuf { + let canonical = + std::fs::canonicalize(repo_root).unwrap_or_else(|_| repo_root.to_path_buf()); + let path = dir.join(".claude.json"); + let contents = serde_json::json!({ + "projects": { + canonical.to_string_lossy().to_string(): { "hasTrustDialogAccepted": true }, + } + }); + std::fs::write(&path, contents.to_string()).unwrap(); + path + } + + /// The same reproduction shape the `build_tui_adapter` model-resolution + /// fix used: assert against the ARGV THE REAL SPAWN RECEIVED, read + /// back from a file the fake binary itself wrote, never a + /// configured/stored value that could look right while the actual + /// launch used something else. Proves `--model` reaches the real + /// launch for this adapter the same way that fix proved it for + /// `TuiAdapter`. + #[tokio::test] + async fn the_resolved_model_reaches_the_real_launched_argv() { + let dir = tempfile::TempDir::new().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 adapter = ClaudeProtocolAdapter::new( + repo_root.clone(), + Vec::new(), + Some("claude-sonnet-5".to_string()), + ProtocolBundle { + approval_service, + callback, + }, + ) + .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: StdArc::clone(&events), + }); + + 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 succeed against the fake binary"); + + let recorded_argv = std::fs::read_to_string(&argv_path).expect("fake binary must have run"); + let argv: Vec<&str> = recorded_argv.lines().collect(); + assert!( + argv.windows(2).any(|w| w == ["--model", "claude-sonnet-5"]), + "expected --model claude-sonnet-5 in the real launched argv, got: {argv:?}" + ); + + 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 new file mode 100644 index 00000000..7c0cba20 --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/approval_bridge.rs @@ -0,0 +1,468 @@ +//! Bridges a `can_use_tool` request off claude's control channel into +//! crew's own approval ledger (`crate::approval::ApprovalService`) -- +//! the existing, durable, replayable mechanism no adapter has ever +//! actually called into before this one (`Adapter::respond_to_approval` +//! is `capability_unsupported` everywhere else today). No new ledger +//! type: `ApprovalRequest`/`ApprovalDecision` already carry request, +//! decision, who decided, and when. +//! +//! **The control-channel JSON shapes below are provisional**, not yet +//! checked against a live capture -- the same caution +//! `claude_protocol::reconcile`'s own transcript entry schema carries, +//! and for the identical reason: this spike's own stop conditions ask +//! that of the control channel specifically, and confirming it is part +//! of the spike's own job, not something to guess past. + +use std::collections::HashMap; +use std::sync::Mutex as StdMutex; + +use crew_protocol::{ApprovalId, ApprovalRequest, RunId, TaskId, Timestamp}; +use tokio::sync::oneshot; + +use crate::approval::{ApprovalCallback, ApprovalService, CallbackFuture}; + +/// One parsed `can_use_tool` request from claude's control channel. +/// Provisional field names -- see this module's own doc comment. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct PermissionRequest { + /// The control channel's own correlation id for this request -- + /// echoed back verbatim in the reply so claude can match it to the + /// pending tool call it is blocked on. + pub(crate) request_id: String, + pub(crate) tool_name: String, + pub(crate) input: serde_json::Value, +} + +/// Parses one `can_use_tool` control-channel message. Tolerant the same +/// way `claude_protocol::reconcile`'s transcript parser is: a shape this +/// function does not recognize returns `None` rather than panicking or +/// guessing -- an unrecognized control message is a fact for the reader +/// loop to act on (most likely: fail the run, since a permission +/// request it cannot answer blocks claude indefinitely), not something +/// this parser should paper over. +pub(crate) fn parse_permission_request(value: &serde_json::Value) -> Option { + let request = value.get("request")?; + if request.get("subtype")?.as_str()? != "can_use_tool" { + return None; + } + Some(PermissionRequest { + request_id: value.get("request_id")?.as_str()?.to_string(), + tool_name: request.get("tool_name")?.as_str()?.to_string(), + input: request + .get("input") + .cloned() + .unwrap_or(serde_json::Value::Null), + }) +} + +/// Builds the control-channel reply bytes that answer `request_id`, +/// unblocking claude's own pending tool call. `decision` is crew's own +/// wire vocabulary (`"approve"`/`"deny"`, matching +/// `crew_protocol::ApprovalDecision::decision`) -- translated here into +/// the shape claude's own protocol expects, exactly once, at this one +/// boundary, so nowhere else in this adapter has to know both +/// vocabularies. +pub(crate) fn build_permission_response(request_id: &str, decision: &str) -> Vec { + let behavior = if decision == "approve" { + "allow" + } else { + "deny" + }; + let mut line = serde_json::json!({ + "type": "control_response", + "response": { + "subtype": "success", + "request_id": request_id, + "response": { "behavior": behavior }, + }, + }) + .to_string(); + line.push('\n'); + line.into_bytes() +} + +/// Bridges `ApprovalService::decide`'s callback into a specific in-flight +/// control-channel request via a one-shot channel captured at request +/// time -- deliberately NOT a `worker_id -> Arc` registry +/// lookup. An approval belongs to a specific in-flight request, not to +/// an adapter instance: a registry lookup would happily deliver a +/// decision to a NEW instance after a restart, one with no pending tool +/// call and no idea what the decision refers to. The channel makes that +/// coupling explicit and lets it die with the request, which is what the +/// domain actually looks like. +pub struct ProtocolApprovalCallback { + pending: StdMutex>>, +} + +impl ProtocolApprovalCallback { + /// `pub`, not `pub(crate)`: `crates/runtime/tests/tui_claude_registry.rs` + /// constructs one directly, as an ordinary external dependency, to + /// install the REAL daemon-shaped `ApprovalService` wiring + /// (`ProtocolApprovalCallback` as its callback, exactly as + /// `lifecycle.rs` wires it) ahead of a TUI run, and prove that run's + /// own behavior is unaffected by it -- the same "widen specifically + /// so an external test can prove a production property" precedent + /// `claude_protocol::reconcile`'s own module already set. + #[must_use] + pub fn new() -> Self { + Self { + pending: StdMutex::new(HashMap::new()), + } + } + + /// Registers a one-shot receiver for `approval_id`, to be resolved + /// once `ApprovalService::decide` calls this callback back. Callers + /// must register BEFORE calling `ApprovalService::request`, so a + /// decision racing in immediately after cannot be missed. + pub(crate) fn register(&self, approval_id: ApprovalId) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + self.pending + .lock() + .expect("approval-callback mutex never poisoned") + .insert(approval_id, tx); + rx + } +} + +impl Default for ProtocolApprovalCallback { + fn default() -> Self { + Self::new() + } +} + +impl ApprovalCallback for ProtocolApprovalCallback { + fn acknowledge(&self, approval_id: ApprovalId, decision: &str) -> CallbackFuture<'static> { + let sender = self + .pending + .lock() + .expect("approval-callback mutex never poisoned") + .remove(&approval_id); + let decision = decision.to_string(); + Box::pin(async move { + if let Some(sender) = sender { + // A dropped receiver -- the run was cancelled, this + // adapter was torn down, or claude's own process already + // exited -- is a normal outcome here, not an error: the + // decision is already durably recorded in the ledger by + // `ApprovalService::decide` before this callback ever + // runs, and the vendor call this reply would have + // unblocked is already moot. The same shape as a + // resolved race elsewhere in this ADR (a second answerer + // arriving after the first already won) being treated as + // a normal outcome, not an error, one layer up -- `send`'s + // own `Result` is deliberately discarded, not propagated + // as this callback's own failure. + let _ = sender.send(decision); + } + Ok(()) + }) + } +} + +/// Bridges one parsed `can_use_tool` request all the way to the reply +/// bytes claude's control channel needs: creates the approval (which +/// atomically pauses the run at `waitingUser`), registers this specific +/// request's one-shot receiver BEFORE submitting it (never after -- +/// registering after would race a decision that resolves before the +/// receiver exists), and awaits the decision to build the reply. +/// +/// # Errors +/// Propagates `ApprovalService::request`'s own error (e.g. the run is +/// not `working`), or a closed-channel error if the callback's sender is +/// dropped without ever sending -- which +/// [`ProtocolApprovalCallback::acknowledge`] does not do on its own +/// initiative (a `None` lookup there means this exact request id was +/// never registered, not that a decision resolved and chose to say +/// nothing); reaching this error means the whole `ProtocolApprovalCallback` +/// was dropped (this adapter's own teardown) while a decision was still +/// pending. +pub(crate) async fn handle_permission_request( + approval_service: &ApprovalService, + callback: &ProtocolApprovalCallback, + run_id: RunId, + task_id: TaskId, + request: PermissionRequest, +) -> Result, String> { + let approval_id = ApprovalId::new(); + let receiver = callback.register(approval_id); + let approval = ApprovalRequest { + approval_id, + run_id, + task_id, + action: request.tool_name, + arguments: request.input, + human_required: true, + policy_reason: "claude requested permission to use a tool over its control channel" + .to_string(), + created_at: Timestamp::now(), + decided_at: None, + decision: None, + decided_by: None, + reason: None, + }; + approval_service + .request(approval) + .await + .map_err(|err| err.to_string())?; + let decision = receiver + .await + .map_err(|_| "the approval callback was dropped before a decision arrived".to_string())?; + Ok(build_permission_response(&request.request_id, &decision)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crew_protocol::{ + DecidedBy, Redacted, Run, RunFlags, RunState, TaskRef, Worker, WorkerId, WorkerProfileRef, + }; + use std::sync::Arc; + use tokio::sync::broadcast; + + use crate::db::DatabaseHandle; + use crate::domain::DomainRepository; + + /// Seeds one task/worker/run and drives it to `working` -- the same + /// shape `run_flags_lost_update.rs`'s own `seed_pending_approval` + /// uses, minus the approval itself: this module's own + /// `handle_permission_request` is what creates that, which is the + /// thing under test here. + async fn seed_working_run( + db: &DatabaseHandle, + project_id: crew_protocol::ProjectId, + ) -> (RunId, TaskId) { + 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, + }, + )?; + let 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.create_worker(&worker)?; + let run = Run { + run_id, + task_id, + worker_id, + state: RunState::try_from("queued").expect("queued is a valid state"), + flags: RunFlags::default(), + vendor_session_id: None, + started_at: None, + completed_at: None, + }; + repo.submit_run(&run, None, None)?; + Ok(serde_json::json!({})) + })) + .await + .expect("seed task/worker/run"); + + for state in ["starting", "working"] { + let to = RunState::try_from(state).expect("valid state"); + db.run_domain_op(Box::new(move |conn| { + let mut repo = DomainRepository::new(conn, project_id); + repo.transition_run(run_id, &to, None) + .map(|_| serde_json::json!({})) + })) + .await + .unwrap_or_else(|e| panic!("drive to {state} failed: {e}")); + } + (run_id, task_id) + } + + /// The full bridge, end to end, against a REAL `ApprovalService` (not + /// a fake): parses a `can_use_tool` request, submits it through + /// `handle_permission_request` (which creates the approval and parks + /// on this specific request's own receiver), decides it exactly the + /// way `approval/decide` would for a human answering a dialog, and + /// asserts the reply bytes this adapter would write back to claude's + /// control channel are correct -- proving the whole round trip, not + /// just its two halves in isolation. + #[tokio::test] + async fn a_real_approval_round_trip_produces_the_correct_control_channel_reply() { + let state_dir = tempfile::TempDir::new().unwrap(); + let db = Arc::new( + DatabaseHandle::start(state_dir.path().join("runtime.db")) + .await + .unwrap(), + ); + let project_id = crew_protocol::ProjectId::new(); + let (run_id, task_id) = seed_working_run(&db, project_id).await; + + let callback = Arc::new(ProtocolApprovalCallback::new()); + let service = ApprovalService::new( + Arc::clone(&db), + project_id, + Arc::clone(&callback) as Arc, + broadcast::channel(64).0, + ); + + let request = parse_permission_request(&serde_json::json!({ + "type": "control_request", + "request_id": "req-42", + "request": { + "subtype": "can_use_tool", + "tool_name": "Write", + "input": { "path": "/tmp/x" }, + }, + })) + .expect("must parse"); + + let handled = tokio::spawn({ + let callback = Arc::clone(&callback); + async move { handle_permission_request(&service, &callback, run_id, task_id, request).await } + }); + + // Give `handle_permission_request` a chance to create the + // approval and register its receiver before deciding it -- + // mirroring the real timing (claude blocks on this reply; crew + // decides only once the approval genuinely exists). + 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(|approval_id| serde_json::json!({ "approvalId": approval_id })) + .map_err(Into::into) + })) + .await + .expect("the approval must have been created"); + let approval_id = value["approvalId"].as_str().expect("a string column"); + ApprovalId::parse(approval_id).expect("a valid approval id") + }; + + let decide_service = ApprovalService::new( + Arc::clone(&db), + project_id, + callback as Arc, + broadcast::channel(64).0, + ); + decide_service + .decide( + approval_id, + "omp-1", + "approve", + &Redacted::assert_runtime_authored("a human approved it"), + DecidedBy::Human, + ) + .await + .expect("decide must succeed"); + + let reply = handled + .await + .expect("the task must not panic") + .expect("handle_permission_request must succeed"); + let value: serde_json::Value = serde_json::from_slice(&reply).expect("valid json line"); + assert_eq!(value["response"]["request_id"], "req-42"); + assert_eq!(value["response"]["response"]["behavior"], "allow"); + + db.shutdown().await.ok(); + } + + #[test] + fn parses_a_well_formed_can_use_tool_request() { + let value = serde_json::json!({ + "type": "control_request", + "request_id": "req-1", + "request": { + "subtype": "can_use_tool", + "tool_name": "Write", + "input": { "path": "/tmp/x", "content": "hi" }, + }, + }); + let parsed = parse_permission_request(&value).expect("must parse"); + assert_eq!(parsed.request_id, "req-1"); + assert_eq!(parsed.tool_name, "Write"); + assert_eq!( + parsed.input, + serde_json::json!({ "path": "/tmp/x", "content": "hi" }) + ); + } + + /// A different control-channel message shape (not a permission + /// request at all) must not be misparsed as one -- silently treating + /// an unrelated control message as `can_use_tool` would fabricate an + /// approval nobody asked for. + #[test] + fn a_non_permission_control_message_does_not_parse() { + let value = serde_json::json!({ + "type": "control_request", + "request_id": "req-2", + "request": { "subtype": "set_model", "model": "opus" }, + }); + assert_eq!(parse_permission_request(&value), None); + } + + #[test] + fn approve_maps_to_allow_and_names_the_request_id() { + let bytes = build_permission_response("req-1", "approve"); + let value: serde_json::Value = serde_json::from_slice(&bytes).expect("valid json line"); + assert_eq!(value["response"]["request_id"], "req-1"); + assert_eq!(value["response"]["response"]["behavior"], "allow"); + } + + #[test] + fn deny_maps_to_deny() { + let bytes = build_permission_response("req-1", "deny"); + let value: serde_json::Value = serde_json::from_slice(&bytes).expect("valid json line"); + assert_eq!(value["response"]["response"]["behavior"], "deny"); + } + + /// The Q2 ruling, proven directly: acknowledging a decision for an + /// approval id nobody registered (the receiver was already dropped, + /// or never existed) must not error -- it is a normal outcome, the + /// same shape as losing a race, not a fault. + #[tokio::test] + async fn acknowledging_an_unregistered_or_dropped_approval_never_errors() { + let callback = ProtocolApprovalCallback::new(); + let result = callback.acknowledge(ApprovalId::new(), "approve").await; + assert!(result.is_ok(), "must not error: {result:?}"); + + // Registered, then the receiver is dropped before a decision -- + // simulating the run being cancelled or this adapter torn down + // while an approval was still pending. + let approval_id = ApprovalId::new(); + let receiver = callback.register(approval_id); + drop(receiver); + let result = callback.acknowledge(approval_id, "approve").await; + assert!( + result.is_ok(), + "a dropped receiver must not error: {result:?}" + ); + } + + /// The ordinary path: register, acknowledge, and the registered + /// receiver actually gets the decision. + #[tokio::test] + async fn acknowledging_a_registered_approval_delivers_the_decision() { + let callback = ProtocolApprovalCallback::new(); + let approval_id = ApprovalId::new(); + let receiver = callback.register(approval_id); + callback + .acknowledge(approval_id, "approve") + .await + .expect("must not error"); + let decision = receiver.await.expect("the receiver must get the decision"); + assert_eq!(decision, "approve"); + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/launch.rs b/crates/runtime/src/adapter/claude_protocol/launch.rs new file mode 100644 index 00000000..2cc0380e --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/launch.rs @@ -0,0 +1,83 @@ +//! The fixed argv this adapter launches `claude` with -- the launch +//! posture settled before any code here existed: plain `-p`, both stream-json +//! formats, an explicit `--setting-sources` naming every source claude +//! would already consult on its own, and nothing that suppresses the +//! vendor's own hooks or hides its settings from it. No `--bare`, no +//! `disableAllHooks`: this spike drives a real, normally-configured +//! claude, not a stripped-down one -- the terminal-path adapters make +//! the identical choice for the same reason (`ClaudeTuiVendor`'s own +//! module doc comment: "interactive `claude`, never a mode that hides +//! its own settings from it"). +//! +//! A pure function, deliberately: every flag this adapter passes is +//! assertable against a `Vec` without spawning anything, the +//! same shape `claude_protocol::approval_bridge`'s parse/build pair and +//! `claude_protocol::reconcile::find_gaps` already take. + +/// Builds the argv `claude` is launched with, everything after the +/// binary name itself. `model`, when given, is placed exactly where +/// [`crate::adapter::tui::claude::ClaudeTuiVendor::base_args`] places its +/// own `--model` -- but where that function reads `cfg.model` (the +/// boot-loaded config's value only; a run's own resolved model choice +/// is a known, separately-tracked gap that function still has, one this +/// adapter is known to inherit until a fix for it lands on the shared +/// construction path and is merged into this branch too), this one +/// takes it as a plain parameter: the precedence between a run's own +/// resolved model and the boot config is that fix's to establish, not +/// this adapter's to reinvent, and this function has no opinion on +/// which value it was handed -- it only places it. +#[must_use] +pub(crate) fn build_argv(model: Option<&str>) -> Vec { + let mut argv = vec![ + "-p".to_string(), + "--input-format".to_string(), + "stream-json".to_string(), + "--output-format".to_string(), + "stream-json".to_string(), + "--verbose".to_string(), + "--setting-sources".to_string(), + "user,project,local".to_string(), + ]; + if let Some(model) = model { + argv.push("--model".to_string()); + argv.push(model.to_string()); + } + argv +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_d6_posture_flags_are_always_present_and_never_bare() { + let argv = build_argv(None); + assert!(argv.contains(&"-p".to_string())); + assert!( + argv.windows(2) + .any(|w| w == ["--input-format", "stream-json"]) + ); + assert!( + argv.windows(2) + .any(|w| w == ["--output-format", "stream-json"]) + ); + assert!( + argv.windows(2) + .any(|w| w == ["--setting-sources", "user,project,local"]) + ); + assert!(!argv.contains(&"--bare".to_string())); + assert!(!argv.iter().any(|arg| arg.contains("disableAllHooks"))); + } + + #[test] + fn no_model_flag_when_none_is_given() { + let argv = build_argv(None); + assert!(!argv.contains(&"--model".to_string())); + } + + #[test] + fn a_given_model_is_placed_verbatim() { + let argv = build_argv(Some("claude-sonnet-5")); + assert!(argv.windows(2).any(|w| w == ["--model", "claude-sonnet-5"])); + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/mod.rs b/crates/runtime/src/adapter/claude_protocol/mod.rs new file mode 100644 index 00000000..f9324702 --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/mod.rs @@ -0,0 +1,80 @@ +//! The claude protocol adapter: drives `claude -p +//! --input-format stream-json --output-format stream-json` and its +//! control channel directly, rather than a PTY -- the terminal becomes a +//! self-rendered view, never the control surface. Selected by +//! `AdapterMode::Protocol`, claude only for this spike; every other +//! reserved kind under that mode stays typed-rejected the same way +//! `AdapterMode::Headless` is today. +//! +//! [`adapter::ClaudeProtocolAdapter`] is the real [`super::Adapter`] +//! implementation [`super::registry::gate_profile`]'s +//! `AdapterMode::Protocol` branch was built ahead of: subprocess spawn +//! with the settled fixed argv ([`launch::build_argv`]) -- plain `-p`, +//! both stream-json formats, explicit `--setting-sources`, no `--bare` +//! 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. + +pub(crate) mod adapter; +// `pub`, not `pub(crate)`: `ProtocolApprovalCallback`'s own doc comment +// explains why an external test needs to reach it (the same "widen to +// let a test prove a production property" precedent `reconcile` below +// already set, for a different property). +pub mod approval_bridge; +pub(crate) mod launch; +pub(crate) mod reader; +pub(crate) mod trust; + +// `pub`, not `pub(crate)`: `crates/runtime/tests/claude_protocol_reconcile.rs` +// links this crate as an ordinary external dependency specifically to +// reach `find_gaps` under a compilation where `cfg(test)` is NOT applied +// to this crate's own code -- the one place able to prove the +// production (never-injected) parse path actually runs, on the same +// rule `adapter::tui::grid`'s own module doc comment states for +// `TerminalGrid`. The test-only injection seam itself +// (`reconcile::arm_drop_for_test`) stays private to this module: an +// external, production-shaped caller must never be able to reach it, +// only to prove it is not there. +pub mod reconcile; + +use super::capability::{ + AdapterCapabilities, ApprovalsCapability, DurabilityCapability, NativeViewCapability, + NestedCapability, ProtocolKind, ResumeCapability, SteeringCapability, UsageCapability, + WorkspaceControlCapability, +}; + +/// The capabilities this adapter *intends* to support, for +/// [`super::registry::GatedCapabilities::Unproven`] to carry until a real +/// conformance suite exists to prove them. Deliberately conservative -- +/// committed scope only, not aspirational: this spike proves one worker, +/// one turn, one approval, no resume, so `resume`/`steering`/`usage` are +/// declared absent rather than guessed at. `approvals: Controllable` is +/// the one claim this spike is actually built to demonstrate: a real +/// vendor permission request answered through crew's own approval +/// ledger, not merely observed. +/// +/// A free function, not a method on some disposable adapter instance +/// (the pattern `TuiAdapter::capabilities()` uses): there is no cheap, +/// side-effect-free way to construct this adapter yet (it will need +/// `Arc` and friends), and this value does not depend +/// on any of that -- it is a fact about what the module intends, not +/// about a particular instance's runtime state. +#[must_use] +pub fn declared_capabilities() -> AdapterCapabilities { + AdapterCapabilities { + protocol: ProtocolKind::Structured, + resume: ResumeCapability::None, + steering: SteeringCapability::None, + approvals: ApprovalsCapability::Controllable, + structured_result: true, + usage: UsageCapability::None, + nested: NestedCapability::None, + native_view: NativeViewCapability::None, + workspace_control: WorkspaceControlCapability::Write, + durability: DurabilityCapability::ParentScoped, + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/reader.rs b/crates/runtime/src/adapter/claude_protocol/reader.rs new file mode 100644 index 00000000..c17a5cdf --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/reader.rs @@ -0,0 +1,660 @@ +//! The control-channel reader: the loop that actually turns claude's +//! `stream-json` stdout into normalized [`AdapterEventPayload`]s and +//! turns a `can_use_tool` control request into a reply written back to +//! its stdin -- the piece `claude_protocol::reconcile`'s `find_gaps` and +//! `claude_protocol::approval_bridge`'s whole module were built ahead +//! of, and were dead code without. +//! +//! Framing mirrors `crate::coordination::mcp::SocketConnection`'s own +//! idiom (newline-delimited JSON, no length prefix), role-reversed: that +//! type is the server side of an NDJSON socket, this is the client side +//! of an NDJSON pipe pair, but the framing itself -- one JSON value per +//! line, no other delimiter -- is identical. +//! +//! **The message shapes classified below are the same ones +//! `crate::adapter::tui::claude`'s transcript tailer already parses** +//! (`type: "assistant"` with a `message.content` block array) -- +//! claude's own stream-json output and its durable transcript share one +//! wire format. What is genuinely new here, and still provisional +//! pending a live capture (this spike's own stop condition, the same +//! one `reconcile`'s and `approval_bridge`'s own module doc comments +//! carry): the `type: "control_request"`/`"control_response"` shapes +//! [`super::approval_bridge`] parses and builds, which only exist on +//! this streaming path, never in the vendor's own transcript file. + +use std::collections::HashSet; +use std::sync::Arc; + +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; + +use crew_protocol::{Classified, ContentClass, RunId, TaskId, WorkerId}; + +use crate::approval::ApprovalService; + +use super::approval_bridge::{self, ProtocolApprovalCallback}; +use crate::adapter::event_sink::{AdapterEvent, AdapterEventPayload, AdapterEventSink}; + +/// The three correlated ids every emitted event and every approval this +/// turn raises must carry -- bundled so `drive_turn`'s own signature +/// does not repeat them three parameters at a time. +#[derive(Debug, Clone, Copy)] +pub(crate) struct RunIdentity { + pub(crate) run_id: RunId, + pub(crate) task_id: TaskId, + pub(crate) worker_id: WorkerId, +} + +/// One classified `stream-json` line, reduced to what [`drive_turn`] +/// needs to act on. `Other` is deliberately one variant for every shape +/// this reader does not (yet) act on -- a `system`/`init` line, a +/// `thinking` block, a `user`-typed tool-result echo -- rather than one +/// variant per shape this module has no caller for yet: the same +/// "unknown/uninteresting collapses to one variant" choice +/// `claude_protocol::reconcile::TranscriptEntryKind::Other` already +/// makes, for the identical reason (a vendor addition should never force +/// a change here). +#[derive(Debug, PartialEq)] +enum StreamLine { + AssistantText(String), + ToolStarted { + tool_call_id: String, + name: String, + }, + ControlRequest(serde_json::Value), + /// `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, + /// `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. + SystemInit { + session_id: Option, + }, + Other, +} + +/// What one call to [`drive_turn`] actually observed, for its caller to +/// reconcile against the vendor's own durable transcript afterward. +#[derive(Debug, Default)] +pub(crate) struct TurnOutcome { + /// Every entry id (`uuid`) seen on the live stream, regardless of + /// whether this reader emitted a normalized event for it -- a + /// `thinking` block or a `system` line contributes its id here even + /// though neither produces an [`AdapterEventPayload`], because + /// `claude_protocol::reconcile::find_gaps` must not call an entry + /// this adapter genuinely saw a "gap" just because it chose not to + /// surface it as a message. + pub(crate) entry_ids: HashSet, + /// The vendor session id this turn ran under, if `system/init` + /// reported one -- `None` when the stream closed before that + /// message ever arrived (nothing to reconcile against without it). + pub(crate) session_id: Option, +} + +/// Classifies one already-parsed JSON line from claude's stdout. Never +/// panics on an unexpected shape -- an unrecognized `type`, a present +/// `type` with a missing expected child field, or no `type` at all all +/// fall through to [`StreamLine::Other`], matching +/// `claude_protocol::reconcile`'s own tolerant-parse stance: a shape +/// this function does not recognize is a fact for the caller to decide +/// what to do with (here: nothing), never a crash. +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("system") + if value.get("subtype").and_then(serde_json::Value::as_str) == Some("init") => + { + StreamLine::SystemInit { + session_id: value + .get("session_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + } + } + Some("assistant") => { + let Some(content) = value + .get("message") + .and_then(|message| message.get("content")) + .and_then(serde_json::Value::as_array) + else { + return StreamLine::Other; + }; + // The first recognized block wins: a `text` block becomes a + // message, a `tool_use` block (absent any `text` block + // before it) becomes a tool start. A `thinking` block is + // skipped outright -- never surfaced as an + // `AdapterEventPayload` at all, the same filtering this + // module's own doc comment (quoting `event_sink.rs`'s) + // requires of every adapter before content ever reaches the + // sink. + for block in content { + match block.get("type").and_then(serde_json::Value::as_str) { + Some("text") => { + if let Some(text) = block.get("text").and_then(serde_json::Value::as_str) { + return StreamLine::AssistantText(text.to_string()); + } + } + Some("tool_use") => { + let id = block.get("id").and_then(serde_json::Value::as_str); + let name = block.get("name").and_then(serde_json::Value::as_str); + if let (Some(id), Some(name)) = (id, name) { + return StreamLine::ToolStarted { + tool_call_id: id.to_string(), + name: name.to_string(), + }; + } + } + _ => {} + } + } + StreamLine::Other + } + _ => StreamLine::Other, + } +} + +/// Reads `stdout` line by line until claude's own turn-complete marker +/// or a closed stream, normalizing what it recognizes through `sink` and +/// bridging any `can_use_tool` control request through `approval_service`/ +/// `callback`, writing the reply back to `stdin`. +/// +/// 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. +/// +/// # 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`. +pub(crate) async fn drive_turn( + stdout: R, + mut stdin: W, + sink: &Arc, + approval_service: &ApprovalService, + callback: &ProtocolApprovalCallback, + ids: RunIdentity, +) -> Result +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + use crate::adapter::error::AdapterError; + + let mut outcome = TurnOutcome::default(); + let mut lines = BufReader::new(stdout).lines(); + loop { + let line = lines + .next_line() + .await + .map_err(|e| AdapterError::process("claude", "read_stdout", e.to_string()))?; + let Some(line) = line else { + // 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 + // caller observes the process exit and reconciles against + // the transcript regardless of how the stream ended. + return Ok(outcome); + }; + if line.trim().is_empty() { + continue; + } + let Ok(value) = serde_json::from_str::(&line) else { + // A malformed line on claude's own stdout is a protocol + // fact worth surfacing, not a parse error to swallow -- + // but never fatal to the turn: the next line may well be + // well-formed. + sink.emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload: AdapterEventPayload::ProtocolHealthChanged { + healthy: false, + detail: Classified { + class: ContentClass::Visible, + value: "received a non-JSON line on claude's stdout".to_string(), + }, + }, + cursor: None, + }) + .await?; + continue; + }; + // Recorded regardless of classification -- see `TurnOutcome::entry_ids`'s + // own doc comment for why an entry crew chose not to emit an + // event for must still count as "seen". + if let Some(uuid) = value.get("uuid").and_then(serde_json::Value::as_str) { + outcome.entry_ids.insert(uuid.to_string()); + } + match classify_line(&value) { + StreamLine::SystemInit { session_id } => { + outcome.session_id = session_id; + } + StreamLine::AssistantText(text) => { + sink.emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload: AdapterEventPayload::MessageFinal { + role: "assistant".to_string(), + text: Classified { + class: ContentClass::Visible, + value: text, + }, + }, + cursor: None, + }) + .await?; + } + StreamLine::ToolStarted { tool_call_id, name } => { + sink.emit(AdapterEvent { + run_id: ids.run_id, + task_id: ids.task_id, + worker_id: ids.worker_id, + payload: AdapterEventPayload::ToolStarted { tool_call_id, name }, + cursor: None, + }) + .await?; + } + StreamLine::ControlRequest(request) => { + let Some(parsed) = approval_bridge::parse_permission_request(&request) else { + // A control request this reader does not recognize + // (claude's control channel carries more request + // subtypes than just `can_use_tool` -- e.g. + // `set_model`/`interrupt` acknowledgements this + // spike never sends and so never expects a request + // for). Never guessed at: skipped, exactly like an + // unrecognized transcript entry is in `reconcile`. + continue; + }; + let reply = approval_bridge::handle_permission_request( + approval_service, + callback, + ids.run_id, + ids.task_id, + parsed, + ) + .await + .map_err(|e| AdapterError::protocol("claude", "respond_to_approval", e))?; + stdin + .write_all(&reply) + .await + .map_err(|e| AdapterError::process("claude", "write_stdin", e.to_string()))?; + } + StreamLine::TurnComplete => return Ok(outcome), + StreamLine::Other => {} + } + } +} + +#[cfg(test)] +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, + WorkerProfileRef, + }; + + use crate::adapter::event_sink::{AdapterEvent, AdapterEventSink}; + use crate::approval::ApprovalCallback; + use crate::db::DatabaseHandle; + use crate::domain::DomainRepository; + + #[test] + fn classifies_assistant_text() { + let value = serde_json::json!({ + "type": "assistant", + "message": {"content": [{"type": "text", "text": "hi there"}]}, + }); + assert_eq!( + classify_line(&value), + StreamLine::AssistantText("hi there".to_string()) + ); + } + + #[test] + fn classifies_tool_use() { + let value = serde_json::json!({ + "type": "assistant", + "message": {"content": [ + {"type": "tool_use", "id": "t1", "name": "Read", "input": {}}, + ]}, + }); + assert_eq!( + classify_line(&value), + StreamLine::ToolStarted { + tool_call_id: "t1".to_string(), + name: "Read".to_string(), + } + ); + } + + /// A thinking block, alone, must never surface as a message -- the + /// filtering this module's own doc comment requires. + #[test] + fn a_thinking_only_block_is_other() { + let value = serde_json::json!({ + "type": "assistant", + "message": {"content": [{"type": "thinking", "thinking": "hmm"}]}, + }); + assert_eq!(classify_line(&value), StreamLine::Other); + } + + #[test] + fn classifies_the_result_marker() { + let value = serde_json::json!({"type": "result", "subtype": "success"}); + assert_eq!(classify_line(&value), StreamLine::TurnComplete); + } + + #[test] + fn an_unrecognized_type_is_other() { + let value = serde_json::json!({"type": "queued_command", "command": "noop"}); + assert_eq!(classify_line(&value), StreamLine::Other); + } + + /// `system`'s own `subtype` gates recognition: only `"init"` is + /// classified, any other subtype (a future addition this reader + /// does not yet know about) falls through to `Other` like any + /// other unrecognized shape. + #[test] + fn a_system_line_with_a_different_subtype_is_other() { + let value = serde_json::json!({"type": "system", "subtype": "compact_boundary"}); + assert_eq!(classify_line(&value), StreamLine::Other); + } + + #[test] + fn classifies_system_init_and_carries_its_session_id() { + let value = + serde_json::json!({"type": "system", "subtype": "init", "session_id": "sess-1"}); + assert_eq!( + classify_line(&value), + StreamLine::SystemInit { + session_id: Some("sess-1".to_string()) + } + ); + } + + /// A minimal in-memory [`AdapterEventSink`] that just records every + /// payload it was handed -- proving `drive_turn`'s own dispatch + /// without a real `DomainAdapterEventSink`/database. + struct RecordingSink { + events: StdArc>>, + } + + impl AdapterEventSink for RecordingSink { + fn emit(&self, event: AdapterEvent) -> crate::adapter::AdapterFuture<'_, u64> { + self.events.lock().push(event.payload); + Box::pin(async { Ok(0) }) + } + + fn note_real_user_turn(&self, _run_id: RunId) -> crate::adapter::AdapterFuture<'_, ()> { + // Unexercised by this module's own tests: `drive_turn` never + // calls it (that signal belongs to a human typing directly + // into a pane, which protocol mode has none of). + Box::pin(async { Ok(()) }) + } + } + + fn ids() -> RunIdentity { + RunIdentity { + run_id: RunId::new(), + task_id: TaskId::new(), + worker_id: WorkerId::new(), + } + } + + /// 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. + #[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"}"#, + "\n", + r#"{"type":"assistant","uuid":"u1","message":{"content":[{"type":"text","text":"hello"}]}}"#, + "\n", + r#"{"type":"assistant","uuid":"u2","message":{"content":[{"type":"tool_use","id":"t1","name":"Read","input":{}}]}}"#, + "\n", + r#"{"type":"result","subtype":"success"}"#, + "\n", + ); + test_side.write_all(script.as_bytes()).await.unwrap(); + drop(test_side); + + let events = StdArc::new(parking_lot::Mutex::new(Vec::new())); + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::clone(&events), + }); + + 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 outcome = drive_turn( + adapter_stdout, + adapter_stdin, + &sink, + &approval_service, + &callback, + ids(), + ) + .await + .expect("must not error"); + + assert_eq!(outcome.session_id.as_deref(), Some("sess-1")); + assert_eq!( + outcome.entry_ids, + ["u1".to_string(), "u2".to_string()].into_iter().collect() + ); + + { + let recorded = events.lock(); + assert!(matches!( + recorded[0], + AdapterEventPayload::MessageFinal { .. } + )); + assert!(matches!( + recorded[1], + AdapterEventPayload::ToolStarted { .. } + )); + } + + // Nothing should have been written back to stdin: no control + // request ever arrived to answer. + let mut written = Vec::new(); + stdin_capture.read_to_end(&mut written).await.unwrap(); + assert!(written.is_empty()); + + db.shutdown().await.ok(); + } + + /// A `can_use_tool` control request is bridged through a REAL + /// `ApprovalService` and the reply is written back onto the + /// adapter's own stdin pipe -- proving `drive_turn` actually + /// completes the round trip end to end, not just that it calls into + /// `approval_bridge`. + #[tokio::test] + async fn bridges_a_control_request_and_writes_the_reply_to_stdin() { + let (mut test_side, adapter_stdout) = tokio::io::duplex(4096); + let (adapter_stdin, mut 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 events = StdArc::new(parking_lot::Mutex::new(Vec::new())); + let sink: Arc = Arc::new(RecordingSink { + events: StdArc::clone(&events), + }); + 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, + }, + ) + .await + } + }); + + let request = concat!( + r#"{"type":"control_request","request_id":"req-1","request":"#, + r#"{"subtype":"can_use_tool","tool_name":"Write","input":{}}}"#, + "\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", + "approve", + &Redacted::assert_runtime_authored("approved for the test"), + DecidedBy::Human, + ) + .await + .expect("decide must succeed"); + + test_side + .write_all(b"{\"type\":\"result\",\"subtype\":\"success\"}\n") + .await + .unwrap(); + drop(test_side); + + drive + .await + .expect("task must not panic") + .expect("drive_turn 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 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"); + assert_eq!(reply["response"]["response"]["behavior"], "allow"); + + db.shutdown().await.ok(); + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/reconcile.rs b/crates/runtime/src/adapter/claude_protocol/reconcile.rs new file mode 100644 index 00000000..f17f6cfa --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/reconcile.rs @@ -0,0 +1,320 @@ +//! Reconciles a run's journal against claude's own durable transcript +//! (`~/.claude/projects//.jsonl`) -- the audit path this +//! adapter relies on precisely because the live control-channel stream +//! carries no sequence number a dropped message would be missed +//! against, so "the stream went quiet" and "nothing happened" are +//! indistinguishable from the stream alone. The vendor's own transcript +//! is written to disk continuously during the turn and is read here as +//! a plain file, never through the same in-process channel this +//! adapter already consumed events from -- reading the same live +//! stream twice would make this reconciliation agree with itself by +//! construction, which is the exact failure this mechanism exists to +//! 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). + +use std::collections::HashSet; + +/// The coarse kind of one transcript entry -- closed, and never +/// widened to carry the entry's own content: this is the field a +/// journaled `ReconciliationCompleted` count could eventually be broken +/// down by, and the same "closed-set, no interpolated content" rule +/// `TerminalGrid::mark_unsupported`'s own `class` field follows applies +/// here for the identical reason (a caller-visible field, not a debug +/// aid). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum TranscriptEntryKind { + /// A user-authored turn -- the prompt this adapter itself injected, + /// or (for a resumed session) prior conversation. + User, + /// An assistant-authored turn: text, tool use, or both. + Assistant, + /// 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. + Other, +} + +/// One entry from claude's own transcript, reduced to what +/// reconciliation needs: an identity to correlate against what this run +/// already journaled, and a coarse kind to describe a gap by. Never +/// carries the entry's own text -- reconciliation counts and repairs, +/// it does not re-surface vendor content a second time outside the +/// path that already redacts it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TranscriptEntry { + /// The vendor's own per-entry identifier (`uuid` in every entry + /// shape observed so far) -- the correlation key against what this + /// run already journaled. Provisional field name pending a live + /// capture, per this module's own doc comment. + pub entry_id: String, + pub kind: TranscriptEntryKind, +} + +/// Parses claude's transcript into entries this module can reconcile, +/// tolerant of one malformed line the same way +/// `adapter::tui::verify::verify_recorded_prompt` is: a line that is +/// not valid JSON, or valid JSON this module does not recognize the +/// shape of, is skipped rather than failing the whole pass. A vendor +/// transcript is append-only and machine-written; a single torn write +/// at the tail (the file caught mid-flush) must not make an otherwise +/// intact transcript unreadable. +fn parse_transcript(bytes: &[u8]) -> Vec { + bytes + .split(|byte| *byte == b'\n') + .filter(|line| !line.is_empty()) + .filter_map(parse_transcript_line) + .collect() +} + +#[cfg(not(test))] +fn parse_transcript_line(line: &[u8]) -> Option { + parse_transcript_line_uninjected(line) +} + +/// The real parse, shared by both cfg bodies below -- the test build's +/// only addition is the drop seam wrapped around this, never a second +/// implementation of the parse itself. Mirrors +/// `TerminalGrid::mark_unsupported`'s own split for the identical +/// reason: the two builds must do the same real work and differ only in +/// the one property under test. +fn parse_transcript_line_uninjected(line: &[u8]) -> Option { + let value: serde_json::Value = serde_json::from_slice(line).ok()?; + let entry_id = value.get("uuid")?.as_str()?.to_string(); + let kind = match value.get("type").and_then(serde_json::Value::as_str) { + Some("user") => TranscriptEntryKind::User, + Some("assistant") => TranscriptEntryKind::Assistant, + _ => TranscriptEntryKind::Other, + }; + Some(TranscriptEntry { entry_id, kind }) +} + +#[cfg(test)] +thread_local! { + /// Set by a test via [`arm_drop_for_test`], consumed by the very + /// next parsed entry whose kind matches -- exactly once, so a test + /// proves reconciliation catches ONE dropped event without having + /// to also prove it does not swallow every subsequent one of the + /// same kind. Thread-local, not a shared global: `cargo test` runs + /// this crate's tests on a shared process, and a global flag would + /// make one test's injection visible to another running + /// concurrently on a different thread. + static DROP_NEXT: std::cell::Cell> = const { std::cell::Cell::new(None) }; +} + +/// Arms the test-only drop seam: the next transcript line whose parsed +/// kind equals `kind`, on THIS thread, is dropped as though it had +/// never been written -- simulating exactly the failure this +/// reconciliation mechanism exists to catch (a live event the vendor +/// recorded that this run's own event stream never delivered). Consumed +/// on first match; call again to drop a second one. +/// +/// `#[cfg(test)]` only -- there is no non-test version of this function +/// at all, not a no-op body. A release build has no symbol, no thread +/// local, and no branch checking one: nothing here for a build flag or +/// config value to ever re-arm. See +/// `crates/runtime/tests/claude_protocol_reconcile.rs` for the proof +/// that this is true of the actual shipping binary, not merely of this +/// crate's own test compilation (`cfg!(test)` is true throughout that +/// compilation regardless, so a test living in THIS file could never +/// tell the difference on its own -- the same structural gap +/// `TerminalGrid::mark_unsupported`'s own doc comment names, closed the +/// same way: an external integration test linking this crate as an +/// ordinary dependency, where the `test` cfg is not applied). +#[cfg(test)] +pub(crate) fn arm_drop_for_test(kind: TranscriptEntryKind) { + DROP_NEXT.with(|cell| cell.set(Some(kind))); +} + +#[cfg(test)] +fn parse_transcript_line(line: &[u8]) -> Option { + let entry = parse_transcript_line_uninjected(line)?; + let armed = DROP_NEXT.with(std::cell::Cell::get); + if armed == Some(entry.kind) { + DROP_NEXT.with(|cell| cell.set(None)); + return None; + } + Some(entry) +} + +/// Compares `transcript_bytes` against `journaled_entry_ids` (every +/// vendor entry id this run has already journaled, from whichever +/// events this adapter already emitted while the turn was live) and +/// reports every transcript entry NOT in that set as a gap -- present +/// in the vendor's own durable record, missing from what this run +/// actually journaled. Pure and side-effect-free: the caller owns +/// deciding how to repair each gap (journaling a +/// [`super::super::event_sink::AdapterEventPayload`] for it) and +/// counting how many of the attempts actually succeeded, since only the +/// caller holds the sink and the run/task/worker identity a repair +/// needs. +/// +/// `examined` is the number of entries this pass actually parsed -- +/// deliberately not the number of lines in the file (a malformed line +/// is not "examined", it could not be) -- so a transcript that parsed +/// as entirely empty (a truncated read, a wrong path, a vendor format +/// change) is visible as `examined == 0` rather than looking identical +/// to a transcript with nothing to reconcile. +/// +/// **Not yet enforced, and it is the caller's obligation once it +/// exists, not this function's:** an `examined == 0` result must never +/// be journaled as an ordinary, successful `ReconciliationCompleted` -- +/// refuse to journal it, or journal it distinguishably as a failure. A +/// pass that looked at nothing and a pass that found nothing wrong must +/// never read the same in the durable record. This function only +/// SURFACES the count; nothing yet acts on it, and it must before this +/// module has a real caller. +pub fn find_gaps( + transcript_bytes: &[u8], + journaled_entry_ids: &HashSet, +) -> (u64, Vec) { + let entries = parse_transcript(transcript_bytes); + let examined = entries.len() as u64; + let gaps: Vec = entries + .into_iter() + .filter(|entry| !journaled_entry_ids.contains(&entry.entry_id)) + .collect(); + (examined, gaps) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry(uuid: &str, kind: &str) -> String { + format!(r#"{{"uuid":"{uuid}","type":"{kind}","message":{{}}}}"#) + } + + /// The floor every other test here depends on: a transcript with + /// nothing this run already journaled must report every entry as a + /// gap, and `examined` must equal the entry count, not the gap + /// count -- the two are the same only by coincidence in this one + /// test, never by construction. + #[test] + fn every_entry_is_a_gap_against_an_empty_journaled_set() { + let transcript = [entry("u1", "user"), entry("a1", "assistant")].join("\n"); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &HashSet::new()); + assert_eq!(examined, 2); + assert_eq!(gaps.len(), 2); + } + + /// 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 + /// "never checked". + #[test] + fn nothing_is_a_gap_when_the_journal_already_has_every_entry() { + let transcript = [entry("u1", "user"), entry("a1", "assistant")].join("\n"); + let journaled: HashSet = ["u1".to_string(), "a1".to_string()].into_iter().collect(); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &journaled); + assert_eq!(examined, 2); + assert!(gaps.is_empty(), "must find no gaps: {gaps:?}"); + } + + /// A malformed line (a torn write at the tail, the shape this + /// module does not recognize) is skipped, not fatal to the pass -- + /// and, the part worth pinning explicitly, it does NOT count toward + /// `examined`: a line that could not be parsed was not examined. + #[test] + fn a_malformed_line_is_skipped_and_never_counted_as_examined() { + let transcript = format!( + "{}\nnot json at all\n{}", + entry("u1", "user"), + entry("a1", "assistant") + ); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &HashSet::new()); + assert_eq!( + examined, 2, + "the malformed line must not inflate the examined count" + ); + assert_eq!(gaps.len(), 2); + } + + /// An empty transcript must report `examined == 0`, never silently + /// look identical to "reconciled and found nothing wrong" -- the + /// distinction `ReconciliationCompleted`'s own doc comment exists + /// for, proven here at the level this module owns. + #[test] + fn an_empty_transcript_examines_nothing() { + let (examined, gaps) = find_gaps(b"", &HashSet::new()); + assert_eq!(examined, 0); + assert!(gaps.is_empty()); + } + + /// The drop seam, exercised directly: arm it for one kind, and the + /// very next entry of that kind is absent from the parsed result -- + /// not merely absent from `journaled_entry_ids`, absent from + /// PARSING, exactly simulating a live event the adapter's own + /// stream never delivered. A second entry of the SAME kind + /// afterward must survive untouched, proving the seam consumes + /// itself rather than dropping every match. + #[test] + fn the_drop_seam_removes_exactly_the_next_matching_entry() { + arm_drop_for_test(TranscriptEntryKind::Assistant); + let transcript = [ + entry("u1", "user"), + entry("a1", "assistant"), + entry("a2", "assistant"), + ] + .join("\n"); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &HashSet::new()); + // The dropped entry was never parsed at all, so it does not + // count as examined either -- from this pass's own point of + // view, a dropped live event and a line the vendor never wrote + // are indistinguishable, which is the whole point: this proves + // the SAME reconciliation logic that would need to catch a real + // gap actually does. + assert_eq!( + examined, 2, + "the dropped entry must not be counted as examined" + ); + let gap_ids: Vec<&str> = gaps.iter().map(|g| g.entry_id.as_str()).collect(); + assert_eq!( + gap_ids, + vec!["u1", "a2"], + "a1 must be the one missing entry" + ); + } + + /// The realistic end-to-end shape this module exists for: an entry + /// IS in the vendor's transcript, the live stream dropped it (so it + /// is NOT in `journaled_entry_ids`), and reconciliation must report + /// it as a gap -- not merely absent from the raw parse (the + /// previous test) but absent from what this run believed it had + /// journaled, which is the actual failure mode this whole module + /// exists to catch. + #[test] + fn a_transcript_entry_missing_from_the_journal_is_reported_as_a_gap() { + let transcript = [ + entry("u1", "user"), + entry("a1", "assistant"), + entry("a2", "assistant"), + ] + .join("\n"); + // The live stream delivered u1 and a1, but a2 (say, a second + // assistant chunk arriving right as the process exited) never + // reached this run's own event pipeline. + let journaled: HashSet = ["u1".to_string(), "a1".to_string()].into_iter().collect(); + let (examined, gaps) = find_gaps(transcript.as_bytes(), &journaled); + assert_eq!(examined, 3); + assert_eq!( + gaps, + vec![TranscriptEntry { + entry_id: "a2".to_string(), + kind: TranscriptEntryKind::Assistant, + }] + ); + } +} diff --git a/crates/runtime/src/adapter/claude_protocol/trust.rs b/crates/runtime/src/adapter/claude_protocol/trust.rs new file mode 100644 index 00000000..40fff05c --- /dev/null +++ b/crates/runtime/src/adapter/claude_protocol/trust.rs @@ -0,0 +1,121 @@ +//! The one-time workspace-trust pre-check this adapter must run before +//! ever spawning `claude`: unlike the TUI path (which discovers a trust +//! gate by pattern-matching text claude itself painted, after already +//! spawning it), protocol mode has no terminal to observe -- the check +//! has to happen first, by reading claude's own trust record directly. +//! +//! Crew never writes `~/.claude.json`. This module only ever reads it, +//! and treats every way that read can go wrong (missing file, unreadable, +//! unparseable, no entry for this repo, an entry present but not `true`) +//! identically: not yet trusted. A first-run prompt claude has not yet +//! answered and a config file this process cannot read are the same fact +//! from this adapter's own perspective -- it must not launch either way. + +use std::path::Path; + +/// Reads `claude_json_path` (in production, `~/.claude.json`) and reports +/// whether `repo_root` has already had its one-time workspace-trust +/// dialog accepted, per claude's own +/// `.projects[""].hasTrustDialogAccepted` record. +/// +/// `repo_root` must already be canonicalized by the caller -- this +/// function does no path resolution of its own, matching +/// `crate::adapter::tui::claude::slug_cwd`'s own precondition on its +/// `cwd` argument. Fails closed on every error shape (missing file, +/// unreadable, malformed JSON, no entry for this repo, or an entry whose +/// flag is present but not literally `true`): the caller must treat +/// anything other than a proven `true` as "not yet trusted," never guess +/// in claude's favor. +#[must_use] +pub(crate) fn workspace_trust_accepted(claude_json_path: &Path, repo_root: &Path) -> bool { + let Ok(contents) = std::fs::read_to_string(claude_json_path) else { + return false; + }; + let Ok(value) = serde_json::from_str::(&contents) else { + return false; + }; + let key = repo_root.to_string_lossy(); + value + .get("projects") + .and_then(|projects| projects.get(key.as_ref())) + .and_then(|project| project.get("hasTrustDialogAccepted")) + .and_then(serde_json::Value::as_bool) + == Some(true) +} + +/// The real, per-user `~/.claude.json` path this check reads in +/// production -- the same file [`super::super::event_sink`]'s +/// `first_run_gate_question` already names for the TUI-path equivalent +/// of this same one-time step. `None` when `HOME` cannot be resolved; +/// the caller must treat that the same as "not trusted" rather than +/// guessing a path. +#[must_use] +pub(crate) fn default_claude_json_path() -> Option { + dirs_home().map(|home| home.join(".claude.json")) +} + +/// The one place this module reads `HOME` -- a thin seam so tests never +/// have to mutate real process environment (which races every other +/// test in the same binary) to prove the "cannot resolve HOME" arm. +fn dirs_home() -> Option { + std::env::var_os("HOME").map(std::path::PathBuf::from) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::PathBuf; + + fn write_claude_json(dir: &tempfile::TempDir, contents: &str) -> PathBuf { + let path = dir.path().join(".claude.json"); + std::fs::write(&path, contents).unwrap(); + path + } + + #[test] + fn a_missing_file_is_not_trusted() { + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join("does-not-exist.json"); + assert!(!workspace_trust_accepted(&path, Path::new("/repo"))); + } + + #[test] + fn malformed_json_is_not_trusted() { + let dir = tempfile::TempDir::new().unwrap(); + let path = write_claude_json(&dir, "not json at all"); + assert!(!workspace_trust_accepted(&path, Path::new("/repo"))); + } + + #[test] + fn no_entry_for_this_repo_is_not_trusted() { + let dir = tempfile::TempDir::new().unwrap(); + let path = write_claude_json( + &dir, + r#"{"projects":{"/some/other/repo":{"hasTrustDialogAccepted":true}}}"#, + ); + assert!(!workspace_trust_accepted(&path, Path::new("/repo"))); + } + + /// A `false` flag is a claude-authored fact ("asked, and declined" or + /// "asked, not yet answered persistently"), not an absence -- still + /// not trusted, on the same fail-closed basis. + #[test] + fn a_present_but_false_flag_is_not_trusted() { + let dir = tempfile::TempDir::new().unwrap(); + let path = write_claude_json( + &dir, + r#"{"projects":{"/repo":{"hasTrustDialogAccepted":false}}}"#, + ); + assert!(!workspace_trust_accepted(&path, Path::new("/repo"))); + } + + #[test] + fn a_true_flag_for_this_exact_repo_is_trusted() { + let dir = tempfile::TempDir::new().unwrap(); + let path = write_claude_json( + &dir, + r#"{"projects":{"/repo":{"hasTrustDialogAccepted":true},"/other":{"hasTrustDialogAccepted":false}}}"#, + ); + assert!(workspace_trust_accepted(&path, Path::new("/repo"))); + } +} diff --git a/crates/runtime/src/adapter/event_sink.rs b/crates/runtime/src/adapter/event_sink.rs index 6e06ec72..85f988a0 100644 --- a/crates/runtime/src/adapter/event_sink.rs +++ b/crates/runtime/src/adapter/event_sink.rs @@ -155,6 +155,48 @@ pub enum AdapterEventPayload { FirstRunGateDetected { kind: crew_protocol::FirstRunGateKind, }, + /// A protocol-first adapter reconciled its journal against + /// the vendor's own durable transcript for this run -- the audit path + /// a live event stream alone cannot guarantee, since no vendor's + /// wire carries a sequence number a dropped message would be missed + /// against. Carries no free text, only counts: `examined` is how + /// many transcript entries this pass actually looked at, and + /// `examined == 0` is itself a finding, not a clean pass -- a + /// reconciliation that never looked at anything is indistinguishable + /// from one that never ran, and a boolean "did it run" flag cannot + /// tell those apart the way a real count can. + /// + /// **Correction**: this field's own doc previously said `gaps_found` + /// and `gaps_repaired` are expected to be equal. No repair mechanism + /// exists yet -- `claude_protocol::reconcile::TranscriptEntry` + /// carries only an id and a coarse kind, deliberately never the + /// entry's own content, so there is nothing a repair step could + /// re-journal from it today without reading the raw transcript line + /// a second time. The one real caller + /// (`claude_protocol::adapter::ClaudeProtocolAdapter`) always emits + /// `gaps_repaired: 0`; the two counts being unequal is the current, + /// expected state, not a defect, until an actual repair step is + /// built. + ReconciliationCompleted { + examined: u64, + gaps_found: u64, + gaps_repaired: u64, + }, + /// A protocol-first Claude adapter checked claude's own trust record + /// (`~/.claude.json`) before ever spawning it and found this + /// repository not yet accepted -- the protocol-mode equivalent of + /// `FirstRunGateDetected { kind: ClaudeWorkspaceTrust }`, but + /// detected by reading a file up front rather than by pattern- + /// matching text a live PTY painted, since protocol mode has no + /// terminal to observe. Carries no fields: the fact is a single, + /// closed one. Maps directly to `RuntimeEvent::EscalationRaised { + /// reason: "claudeWorkspaceTrustPending" }` -- a new reason distinct + /// from `"vendorFirstRunGate"`, precisely because the run this + /// escalates has no live pane a human could type into to resolve + /// it: the run fails outright (see + /// `claude_protocol::adapter::ClaudeProtocolAdapter::start`), and + /// this reason's own remediation text says so. + WorkspaceTrustPending, } /// Adapters push ordered normalized events into the runtime journal @@ -440,10 +482,53 @@ impl DomainAdapterEventSink { kind, } } + AdapterEventPayload::ReconciliationCompleted { + examined, + gaps_found, + gaps_repaired, + } => RuntimeEvent::AdapterReconciliationEvent { + run_id, + task_id, + worker_id, + examined, + gaps_found, + gaps_repaired, + }, + // The primary event this emit call journals IS the + // escalation itself -- unlike `FirstRunGateDetected` above, + // there is no separate "detected" fact distinct from + // "escalated" here (the same shape `write_violation`'s own + // construction site in `DomainRepository` already uses: one + // event, not a detection event plus a paired escalation). + AdapterEventPayload::WorkspaceTrustPending => RuntimeEvent::EscalationRaised { + run_id, + task_id, + worker_id, + reason: "claudeWorkspaceTrustPending".to_string(), + question: Some(workspace_trust_pending_question()), + }, } } } +/// The runtime-authored escalation question for +/// [`AdapterEventPayload::WorkspaceTrustPending`] -- a fixed template, +/// nothing captured, on the same +/// [`crew_protocol::Redacted::assert_runtime_authored`] basis +/// `first_run_gate_question` uses. Distinct wording from that +/// function's own `ClaudeWorkspaceTrust` case: this run has no live +/// pane to point a human at (protocol mode has no terminal at all), so +/// the remediation is "run claude yourself once, then resubmit," never +/// "answer it in the worker's pane." +fn workspace_trust_pending_question() -> crew_protocol::Redacted { + crew_protocol::Redacted::assert_runtime_authored( + "Claude has not yet had its one-time workspace-trust prompt accepted for this \ + repository, and this run has no terminal to answer it from. Trust the repository \ + once by running claude interactively here yourself, then resubmit this run. Crew \ + will not accept the prompt for you.", + ) +} + /// The runtime-authored escalation question for a recognized first-run /// gate, built via [`crew_protocol::Redacted::assert_runtime_authored`]: /// every word here is a fixed template plus a closed-enum-derived phrase, diff --git a/crates/runtime/src/adapter/mod.rs b/crates/runtime/src/adapter/mod.rs index 2e025885..b589e685 100644 --- a/crates/runtime/src/adapter/mod.rs +++ b/crates/runtime/src/adapter/mod.rs @@ -6,6 +6,7 @@ //! [`crate::domain::DomainRepository`] directly. mod activity; mod capability; +pub mod claude_protocol; mod error; mod event_sink; pub mod mcp_config; @@ -36,7 +37,8 @@ pub use profile::{ }; pub use profile_store::{ProfileStore, ProfileStoreError}; pub use registry::{ - AdapterAuthorization, AdapterRegistry, FixtureAuthorization, RegistryError, ResumeSupport, + AdapterAuthorization, AdapterRegistry, FixtureAuthorization, GatedCapabilities, + ProtocolSupport, RegistryError, ResumeSupport, }; pub use run_lifecycle::RunLifecycleSink; pub use r#trait::{ diff --git a/crates/runtime/src/adapter/profile.rs b/crates/runtime/src/adapter/profile.rs index 7fb7fdb0..161ebb3e 100644 --- a/crates/runtime/src/adapter/profile.rs +++ b/crates/runtime/src/adapter/profile.rs @@ -195,16 +195,21 @@ impl StartupOptions { } } -/// Whether a supervised adapter process runs attached to a terminal UI or -/// fully headless. Defaults to `Headless` so a profile serialized before -/// this field existed still deserializes -- the same wire-compat pattern -/// `ApprovalEvent.reason` uses in `crew_protocol::event`. +/// Whether a supervised adapter process runs attached to a terminal UI, +/// fully headless, or driven over its own structured protocol with the +/// terminal reduced to a self-rendered view. Defaults to `Headless` so a +/// profile serialized before this field existed still deserializes -- +/// the same wire-compat pattern `ApprovalEvent.reason` uses in +/// `crew_protocol::event`. See `config::crew::AdapterMode`'s own doc +/// comment (the config-layer twin of this enum) for `Protocol`'s own +/// meaning. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub enum AdapterMode { Tui, #[default] Headless, + Protocol, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] diff --git a/crates/runtime/src/adapter/registry.rs b/crates/runtime/src/adapter/registry.rs index 306556cb..adb7e7f8 100644 --- a/crates/runtime/src/adapter/registry.rs +++ b/crates/runtime/src/adapter/registry.rs @@ -93,13 +93,23 @@ pub trait AdapterAuthorization: Send + Sync { /// `effective_capabilities`' undowngraded-`Skipped` guarantee exist to /// close. /// + /// **A coarser version of the same rule, at the whole-mode grain + /// rather than the per-scenario one above:** `effective_capabilities` + /// is a [`GatedCapabilities`], not a bare [`AdapterCapabilities`], + /// specifically so a mode with no conformance suite to run AT ALL + /// (`AdapterMode::Protocol`, which has none yet -- see + /// [`GatedCapabilities::Unproven`]) cannot reach this function + /// looking identical to a suite that actually ran. The binding + /// constraint above extends to it: a real capability check must + /// deny on `Unproven`, never treat it as `Proven`'s equivalent. + /// /// # Errors /// Returns a human-readable denial reason. The run is never started /// when this returns `Err`. fn authorize( &self, profile: &WorkerProfile, - effective_capabilities: &AdapterCapabilities, + effective_capabilities: &GatedCapabilities, policy: Option<&crate::config::RuntimePolicy>, ) -> Result<(), String>; @@ -109,6 +119,47 @@ pub trait AdapterAuthorization: Send + Sync { fn release(&self); } +/// The capabilities [`AdapterAuthorization::authorize`] actually sees -- +/// wrapped, not passed alongside a `proven: bool`, so a caller cannot +/// drop the distinction by accident and every `authorize` implementation +/// is forced to say which one it has. See that trait method's own doc +/// comment for the binding constraint this exists to let a future +/// capability check honor. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum GatedCapabilities { + /// A conformance suite (fixture or live) actually ran for this + /// profile's `(kind, mode)`; these are its own `effective_capabilities`. + Proven(AdapterCapabilities), + /// No conformance suite exists for this profile's adapter mode at + /// all yet -- today, only `AdapterMode::Protocol` + /// (`crate::adapter::claude_protocol`, spike-scoped, tracked by a + /// card for the real suite). Carries the adapter's own DECLARED + /// capabilities, never independently verified. An + /// [`AdapterAuthorization::authorize`] implementation that grows a + /// real capability check must deny on this variant, not treat it as + /// equivalent to [`Self::Proven`]. + Unproven(AdapterCapabilities), +} + +impl GatedCapabilities { + /// The wrapped capabilities regardless of provenance -- for a caller + /// (today, only [`crate::policy::PolicyEvaluator`]) that genuinely + /// reads zero capability fields and therefore has no reason to + /// branch on which variant it received. + #[must_use] + pub fn capabilities(&self) -> &AdapterCapabilities { + match self { + Self::Proven(capabilities) | Self::Unproven(capabilities) => capabilities, + } + } + + /// Whether these capabilities were actually conformance-verified. + #[must_use] + pub fn is_proven(&self) -> bool { + matches!(self, Self::Proven(_)) + } +} + /// A deterministic allow/deny fixture for tests. Production callers must /// supply a real policy, per the plan's "do not ship a permissive /// production authorization implementation." @@ -120,7 +171,7 @@ impl AdapterAuthorization for FixtureAuthorization { fn authorize( &self, _profile: &WorkerProfile, - _effective_capabilities: &AdapterCapabilities, + _effective_capabilities: &GatedCapabilities, _policy: Option<&crate::config::RuntimePolicy>, ) -> Result<(), String> { if self.allow { @@ -158,6 +209,18 @@ pub enum RegistryError { /// until its TUI vendor impl lands). #[error("adapter {0} has no TUI-mode implementation yet; mode: \"tui\" is unavailable for it")] TuiModeUnavailable(String), + /// `mode: "protocol"` was requested but + /// [`AdapterRegistry::set_protocol_support`] was never called -- + /// the same defense-in-depth shape as [`Self::TuiModeUnavailable`]: + /// `gate_profile` already refuses a non-Claude kind under this mode + /// before `build_adapter` is ever reached, so a caller hitting this + /// variant means Claude itself was requested in a window before this + /// registry's own protocol support was wired up. + #[error( + "adapter {0} has no protocol-mode implementation yet; mode: \"protocol\" is unavailable \ + for it" + )] + ProtocolModeUnavailable(String), /// [`AdapterRegistry::resume_run`] was called before the caller ever /// supplied [`ResumeSupport`] via /// [`AdapterRegistry::set_resume_support`]. Fail closed: a resume @@ -234,6 +297,30 @@ pub struct ResumeSupport { pub events_tx: tokio::sync::broadcast::Sender, } +/// Everything `build_adapter`'s `AdapterMode::Protocol` branch needs +/// beyond `super::claude_protocol::adapter::ClaudeProtocolAdapter`'s own +/// vendor-facing fields: the server-owned `ApprovalService` `respond_to_approval`'s +/// real bridge submits requests to, and the callback map that same +/// bridge registers its one-shot receivers into. A post-construction +/// setter bundle -- exactly like [`TuiSupport`]/[`ResumeSupport`], and +/// for the identical reason: `ApprovalService` is constructed inside +/// `crate::ipc::Server::bind` (from `ServerConfig::approval_callback`), +/// which happens after this registry must already be handed to +/// [`crate::ipc::ServerConfig::run_driver`]. +/// +/// `pub`, not `pub(crate)`, on the same basis as +/// `super::claude_protocol::approval_bridge::ProtocolApprovalCallback`'s +/// own doc comment: `crates/runtime/tests/tui_claude_registry.rs` +/// installs this bundle on a registry driving a TUI run, to prove that +/// run's behavior is unaffected by the real daemon-shaped approval +/// wiring being present -- it never touches `mode: "tui"`'s own +/// dispatch, but the earlier version of this claim was asserted, not +/// proven. +pub struct ProtocolSupport { + pub approval_service: Arc, + pub callback: Arc, +} + /// Implements [`RunDriver`] against the four real worker adapters. /// /// Always constructed behind an `Arc` in practice (exactly like every @@ -276,6 +363,13 @@ pub struct AdapterRegistry { /// makes every `resume_run` a typed [`RegistryError::ResumeUnsupported`] /// refusal, never a silently unwired resume. resume_support: Mutex>>, + /// `AdapterMode::Protocol` support (see [`ProtocolSupport`]'s own doc + /// comment). `None` until [`Self::set_protocol_support`] is called + /// (or permanently, for callers -- chiefly tests -- that never call + /// it): `build_adapter`'s `AdapterMode::Protocol` branch gets a typed + /// [`RegistryError::ProtocolModeUnavailable`] refusal in that window, + /// the same shape [`Self::tui`]'s absence gives `mode: "tui"`. + protocol: Mutex>>, running: Arc>>>, /// Org security patterns for redaction. org_security_patterns: Vec, @@ -308,6 +402,7 @@ impl AdapterRegistry { broker: Mutex::new(None), tui: Mutex::new(None), resume_support: Mutex::new(None), + protocol: Mutex::new(None), activity: Mutex::new(None), max_live_sessions: Mutex::new(None), running: Arc::new(Mutex::new(HashMap::new())), @@ -354,6 +449,14 @@ impl AdapterRegistry { *self.resume_support.lock() = Some(support); } + /// Supplies the [`ProtocolSupport`] bundle `build_adapter`'s + /// `AdapterMode::Protocol` branch needs. A post-construction setter + /// for the same reason `set_resume_support` is -- see + /// [`ProtocolSupport`]'s own doc comment. + pub fn set_protocol_support(&self, support: Arc) { + *self.protocol.lock() = Some(support); + } + /// Supplies the shared [`ActivityClock`] every run's lifecycle sink /// touches and lifecycle's timeout sweep reads. A post-construction /// setter -- exactly like `set_broker` -- because the clock must be @@ -521,6 +624,7 @@ impl AdapterRegistry { self.mcp.clone(), self.broker.lock().clone(), self.tui.lock().clone(), + self.protocol.lock().clone(), Arc::clone(&support.db), support.project_id, support.events_tx.clone(), @@ -542,7 +646,7 @@ impl AdapterRegistry { support.project_id, support.events_tx.clone(), self.org_security_patterns.clone(), - effective_capabilities.nested != NestedCapability::Managed, + effective_capabilities.capabilities().nested != NestedCapability::Managed, Arc::clone(&support.violation_service), // Resume support carries no workspace context; a resumed run // is treated as shared (the conservative side for the @@ -583,6 +687,7 @@ impl RunDriver for AdapterRegistry { let mcp = self.mcp.clone(); let broker = self.broker.lock().clone(); let tui = self.tui.lock().clone(); + let protocol = self.protocol.lock().clone(); let org_security_patterns = self.org_security_patterns.clone(); let running = Arc::clone(&self.running); // Read under the registry's own lock, before the async block: a @@ -636,6 +741,7 @@ impl RunDriver for AdapterRegistry { mcp, broker, tui, + protocol, org_security_patterns, ) .await @@ -827,6 +933,7 @@ fn build_placeholder_adapter() -> Arc { )) } +#[allow(clippy::too_many_arguments)] async fn run_one( ctx: &RunDriverContext, authorization: &Arc, @@ -834,6 +941,7 @@ async fn run_one( mcp: Option, broker: Option>, tui: Option>, + protocol: Option>, org_security_patterns: Vec, ) -> Result< ( @@ -873,6 +981,7 @@ async fn run_one( mcp, broker, tui, + protocol, Arc::clone(&ctx.db), ctx.project_id, ctx.events_tx.clone(), @@ -892,7 +1001,7 @@ async fn run_one( ctx.project_id, ctx.events_tx.clone(), org_security_patterns, - effective_capabilities.nested != NestedCapability::Managed, + effective_capabilities.capabilities().nested != NestedCapability::Managed, Arc::clone(&ctx.violation_service), ctx.workspace_path.is_some(), ) { @@ -1005,13 +1114,20 @@ async fn gate_profile( profile: &WorkerProfile, policy: Option<&crate::config::RuntimePolicy>, mode: AdapterMode, -) -> Result { +) -> Result { // Handle TerminalDegraded specially (it has no adapter kind) let effective_capabilities = if profile.adapter_kind().is_none() { // TerminalDegraded uses the terminal adapter with degraded capabilities // We need to extract the backend from the startup options if let StartupOptions::TerminalDegraded(opts) = &profile.startup_options() { - super::terminal::TerminalAdapter::new(opts.backend.clone()).capabilities() + // A `TerminalAdapter`'s capabilities are a pure function of + // its own static profile, with no vendor CLI (and therefore + // no conformance question) involved at all -- `Proven` here + // is not a shortcut, it is the literal truth: there is + // nothing this could disagree with. + GatedCapabilities::Proven( + super::terminal::TerminalAdapter::new(opts.backend.clone()).capabilities(), + ) } else { return Err("TerminalDegraded profile has no startup options".to_string()); } @@ -1034,6 +1150,33 @@ async fn gate_profile( RegistryError::HeadlessControlPlaneRetired(kind.wire_name().to_string()).into(), ); } + // The protocol-first control plane (claude only for now: + // `crate::adapter::claude_protocol`) has no conformance + // suite to run at all yet -- unlike `Headless` above, this is + // not a permanent rejection, just an as-yet-unproven one. Reject + // OTHER kinds under this mode (nothing implements them), and for + // claude, proceed with the adapter's own DECLARED capabilities, + // wrapped `Unproven` so `authorize` below (and any future + // capability check inside it) cannot mistake this for a suite + // that actually ran. Checked, and short-circuited, before the + // conformance dispatch below for the identical reason the + // `Headless` check above is. + if mode == AdapterMode::Protocol { + if kind != AdapterKind::Claude { + return Err(format!( + "adapter {} has no protocol-mode implementation yet; mode: \"protocol\" is \ + unavailable for it", + kind.wire_name() + )); + } + let effective_capabilities = + GatedCapabilities::Unproven(super::claude_protocol::declared_capabilities()); + authorization + .authorize(profile, &effective_capabilities, policy) + .map_err(RegistryError::AuthorizationDenied) + .map_err(String::from)?; + return Ok(effective_capabilities); + } // The full suite is memoized per `(kind, mode)`, stamped // with the vendor-CLI version the availability probe observed; a // changed version (upgrade, downgrade, install) is the @@ -1082,7 +1225,7 @@ async fn gate_profile( .and_then(|(stamped_version, cached)| { (*stamped_version == probed_version).then_some(*cached) }); - match suite_hit { + GatedCapabilities::Proven(match suite_hit { Some(cached) => cached, None => { let effective = conformance::run_fixture_conformance(kind, mode) @@ -1093,7 +1236,7 @@ async fn gate_profile( .insert(cache_key, (probed_version, effective)); effective } - } + }) }; // Policy decision: exactly once, for cache hits and misses alike -- @@ -1251,6 +1394,7 @@ fn build_adapter( mcp: Option, broker: Option>, tui: Option>, + protocol: Option>, db: Arc, project_id: crew_protocol::ProjectId, events_tx: tokio::sync::broadcast::Sender, @@ -1366,6 +1510,65 @@ fn build_adapter( )); } + // `mode: "protocol"`: `gate_profile` already refuses every + // kind but Claude before this function is ever reached under this + // mode, so the only real branch here is Claude's own. `protocol` + // absent is the same defense-in-depth shape `tui` absent is above -- + // reachable only if Claude protocol mode is requested before + // `AdapterRegistry::set_protocol_support` is ever called. + if requested_mode(&profile.startup_options) == Some(super::profile::AdapterMode::Protocol) { + let kind = profile + .startup_options + .adapter_kind() + .expect("Protocol mode only applies to a startup_options variant with an AdapterKind"); + let Some(protocol) = protocol else { + return Err(RegistryError::ProtocolModeUnavailable( + kind.wire_name().to_string(), + )); + }; + 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. + // + // `profile.model` passed straight through, trimmed -- + // see `super::claude_protocol::adapter`'s own module doc + // comment on why the boot-config-precedence + // `build_tui_adapter` needs has nothing to apply to + // here: no boot-loaded per-vendor config exists for + // protocol mode to override. Trimmed defensively per + // `WorkerProfile::validate`'s own non-empty guarantee at + // `ProfileStore::insert` -- belt and suspenders, not a + // guess at emptiness this adapter has any reason to + // expect. + let model = Some(profile.model.trim().to_string()).filter(|m| !m.is_empty()); + Ok(Arc::new( + super::claude_protocol::adapter::ClaudeProtocolAdapter::new( + repo_root.to_path_buf(), + profile.environment_allowlist.clone(), + model, + super::claude_protocol::adapter::ProtocolBundle { + approval_service: Arc::clone(&protocol.approval_service), + callback: Arc::clone(&protocol.callback), + }, + ), + )) + } + _ => Err(RegistryError::ProtocolModeUnavailable( + kind.wire_name().to_string(), + )), + }; + } + // Every reserved kind's Headless fallback (three headless vendor // adapters, plus ompRpc's) was retired along with the adapter code // itself (`docs/adr/0026-headless-retirement.md`) -- `gate_profile` already refuses a @@ -1660,6 +1863,7 @@ mod build_adapter_tests { None, None, None, // no TuiSupport supplied + None, db, crew_protocol::ProjectId::new(), events_tx, @@ -1697,6 +1901,7 @@ mod build_adapter_tests { None, None, Some(test_tui_support()), + None, db, crew_protocol::ProjectId::new(), events_tx, @@ -1731,6 +1936,7 @@ mod build_adapter_tests { None, None, Some(test_tui_support()), + None, db, crew_protocol::ProjectId::new(), events_tx, @@ -1771,6 +1977,7 @@ mod build_adapter_tests { None, None, Some(test_tui_support()), + None, db, crew_protocol::ProjectId::new(), events_tx, @@ -1806,6 +2013,7 @@ mod build_adapter_tests { None, None, Some(test_tui_support()), + None, db, crew_protocol::ProjectId::new(), events_tx, @@ -1860,6 +2068,7 @@ mod build_adapter_tests { None, None, None, + None, Arc::clone(&db), crew_protocol::ProjectId::new(), events_tx.clone(), @@ -2044,7 +2253,7 @@ mod settlement_tests { fn authorize( &self, _profile: &WorkerProfile, - _effective_capabilities: &AdapterCapabilities, + _effective_capabilities: &GatedCapabilities, _policy: Option<&crate::config::RuntimePolicy>, ) -> Result<(), String> { Ok(()) @@ -2180,7 +2389,7 @@ mod settlement_tests { // all (see `AdapterAuthorization::authorize`'s doc comment), so // any value proves the concurrency-ceiling point -- deliberately // not read from a specific adapter's `declared_capabilities()`. - let capabilities = AdapterCapabilities { + let capabilities = GatedCapabilities::Proven(AdapterCapabilities { protocol: crate::adapter::capability::ProtocolKind::Terminal, resume: crate::adapter::capability::ResumeCapability::Session, steering: crate::adapter::capability::SteeringCapability::ActiveTurn, @@ -2191,7 +2400,7 @@ mod settlement_tests { native_view: crate::adapter::capability::NativeViewCapability::IndependentTui, workspace_control: crate::adapter::capability::WorkspaceControlCapability::Write, durability: crate::adapter::capability::DurabilityCapability::VendorResumable, - }; + }); // Book the one slot, then prove the ceiling is exhausted. authorization @@ -2391,8 +2600,12 @@ mod conformance_cache_tests { .await .expect("tui gate must pass"); + assert!( + effective.is_proven(), + "a TUI-mode submit's gate must be Proven, not merely declared: {effective:?}" + ); assert_eq!( - effective.protocol, + effective.capabilities().protocol, crate::adapter::capability::ProtocolKind::Terminal, "a TUI-mode submit must be gated on the TUI suite's effective capabilities: \ {effective:?}" @@ -2446,6 +2659,157 @@ mod conformance_cache_tests { ); } + fn claude_profile_with_mode(mode: AdapterMode) -> WorkerProfile { + WorkerProfile { + id: crate::adapter::profile::ProfileId::new(), + adapter: "claude".to_string(), + model: String::new(), + permission_envelope: serde_json::Value::Object(serde_json::Map::new()), + startup_options: StartupOptions::Claude(ClaudeStartupOptions { + mode, + ..Default::default() + }), + environment_allowlist: Vec::new(), + source: "test".to_string(), + } + } + + /// `AdapterMode::Protocol` has no conformance suite to run at all + /// yet -- unlike `Headless`, this is + /// a temporary, not permanent, gap, so `gate_profile` proceeds rather + /// than refusing outright, but must not let the result look like a + /// suite that actually ran. Pins the whole shape of that: no suite + /// runs, no cache entry is written (both exactly like the Headless + /// rejection above), the capabilities returned are the module's own + /// `declared_capabilities()`, and -- the part that matters -- + /// `is_proven()` is `false`. + #[tokio::test] + async fn a_protocol_mode_claude_submit_is_gated_on_declared_capabilities_marked_unproven() { + let _serial = SERIAL.lock().await; + CONFORMANCE_CACHE.lock().clear(); + FIXTURE_SUITE_RUNS.store(0, Ordering::Relaxed); + let authorization: Arc = + Arc::new(FixtureAuthorization { allow: true }); + + let gated = gate_profile( + &authorization, + &claude_profile_with_mode(AdapterMode::Protocol), + None, + AdapterMode::Protocol, + ) + .await + .expect("a protocol-mode claude submit must be gated, not refused"); + + assert!( + !gated.is_proven(), + "protocol-mode capabilities must never look Proven -- no suite has run for this \ + mode yet: {gated:?}" + ); + assert_eq!( + *gated.capabilities(), + crate::adapter::claude_protocol::declared_capabilities(), + "must carry the module's own declared capabilities, not some other value" + ); + assert_eq!( + FIXTURE_SUITE_RUNS.load(Ordering::Relaxed), + 0, + "protocol mode must never dispatch to the fixture conformance suite" + ); + assert!( + CONFORMANCE_CACHE.lock().is_empty(), + "an unproven protocol-mode submit must never write a cache entry meant for proven \ + results" + ); + } + + /// Only claude implements `AdapterMode::Protocol` in this spike -- + /// every other kind gets a typed rejection, the same shape as + /// `TuiModeUnavailable` for a `TuiVendor`-less kind under `Tui` mode, + /// rather than silently proceeding with a declared-capabilities set + /// for an adapter that does not exist. + #[tokio::test] + async fn a_protocol_mode_submit_for_an_unimplemented_kind_is_rejected() { + let _serial = SERIAL.lock().await; + let authorization: Arc = + Arc::new(FixtureAuthorization { allow: true }); + + let err = gate_profile( + &authorization, + &omp_rpc_profile_with_mode(AdapterMode::Protocol), + None, + AdapterMode::Protocol, + ) + .await + .expect_err("ompRpc has no protocol-mode implementation"); + + assert!( + err.contains("ompRpc") && err.contains("protocol"), + "the rejection must name the kind and the mode: {err}" + ); + } + + /// The distinguishability the `Unproven` marker exists for, proven + /// rather than merely documented: an `AdapterAuthorization` that + /// actually reads the marker denies a protocol-mode (unproven) + /// profile and allows a TUI-mode (proven) one, using the SAME + /// authorizer for both -- so the difference in outcome can only come + /// from `GatedCapabilities` itself. This is the test that fails the + /// day a real capability check forgets the deny-on-unproven + /// constraint from `AdapterAuthorization::authorize`'s own doc + /// comment. + struct DenyOnUnprovenAuthorization; + + impl AdapterAuthorization for DenyOnUnprovenAuthorization { + fn authorize( + &self, + _profile: &WorkerProfile, + effective_capabilities: &GatedCapabilities, + _policy: Option<&crate::config::RuntimePolicy>, + ) -> Result<(), String> { + if effective_capabilities.is_proven() { + Ok(()) + } else { + Err("denied: capabilities are unproven for this mode".to_string()) + } + } + + fn release(&self) {} + } + + #[tokio::test] + async fn a_real_capability_check_denies_unproven_protocol_capabilities_and_allows_proven_ones() + { + let _serial = SERIAL.lock().await; + CONFORMANCE_CACHE.lock().clear(); + let authorization: Arc = Arc::new(DenyOnUnprovenAuthorization); + + let denied = gate_profile( + &authorization, + &claude_profile_with_mode(AdapterMode::Protocol), + None, + AdapterMode::Protocol, + ) + .await + .expect_err("a checker that reads the marker must deny an unproven protocol submit"); + assert!( + denied.contains("unproven"), + "must be this authorizer's own denial, not some other failure: {denied}" + ); + + gate_profile( + &authorization, + &omp_rpc_profile_with_mode(AdapterMode::Tui), + None, + AdapterMode::Tui, + ) + .await + .expect( + "the SAME checker must allow a Proven (TUI, conformance-suite-backed) submit -- the \ + denial above is about the marker, not about this authorizer refusing everything", + ); + CONFORMANCE_CACHE.lock().clear(); + } + /// Seeds a task/worker/run with a resolved Claude profile snapshot in /// the given `mode`, for driving the REAL `resume_run`/`start` call /// sites end to end (not `gate_profile` in isolation). diff --git a/crates/runtime/src/adapter/tui/mod.rs b/crates/runtime/src/adapter/tui/mod.rs index 998986f6..2234028b 100644 --- a/crates/runtime/src/adapter/tui/mod.rs +++ b/crates/runtime/src/adapter/tui/mod.rs @@ -79,6 +79,13 @@ use crate::conformance::ConformanceReport; /// the resolved `crewd` binary path, the state root, and the display /// registry/config -- and `AdapterRegistry` builds a fresh /// `PaneCoordinator` per run from this plus that run's own context. +/// +/// Not inherently TUI-specific despite the name: every field here is a +/// display/pane-coordination concern, not a PTY-specific one, so +/// `AdapterMode::Protocol`'s own pane renderer (not yet built) is +/// expected to reuse this bundle rather than invent a second one. The +/// name stays `TuiSupport` for now, to avoid churning a shared struct +/// ahead of that renderer actually existing. #[derive(Clone)] pub struct TuiSupport { pub display_registry: Arc, diff --git a/crates/runtime/src/cli.rs b/crates/runtime/src/cli.rs index ba843f10..94087816 100644 --- a/crates/runtime/src/cli.rs +++ b/crates/runtime/src/cli.rs @@ -9,7 +9,7 @@ use std::path::{Path, PathBuf}; use std::process::ExitCode; -use clap::{Parser, Subcommand}; +use clap::{Parser, Subcommand, ValueEnum}; use crew_runtime::{StateRoot, VERSION}; /// Selector for `crewd conformance --live`: which control plane to exercise @@ -1253,17 +1253,29 @@ async fn run_conformance( } }; - // `--mode headless` is a typed rejection now, for both `--fixture` - // and `--live` -- never silently accepted-and-discarded. Before this, - // `--fixture` ignored `--mode` entirely (always headless-sourced) and `--live - // --mode headless` silently reached each adapter's own headless - // `live_report`; both dispatch targets are deleted along with the - // headless control plane itself (`docs/adr/0026-headless-retirement.md`). - if matches!(mode, ConformanceModeArg::Headless) { - return fail( - &"mode: \"headless\" is retired in crew v2 -- the headless control \ - plane has no adapter implementation to dispatch to; use --mode tui (the default)", - ); + // Only `Tui` dispatches anywhere below -- both `--fixture` and + // `--live` always run the TUI-sourced report regardless of `mode`, + // so anything else must be refused here, named, rather than let a + // dispatch built for one mode silently run under a caller's belief + // it asked for another. Deliberately an ALLOWLIST (`!= Tui`), not a + // `Headless`-specific denylist: a denylist only refuses the variant + // someone thought to write down, so `ConformanceModeArg` gaining a + // `Protocol` value later would sail straight past a `Headless`-only + // check and reach `run_fixture_conformance`'s own `unreachable!()` + // arm for it -- a panic from a supported command line, not a typed + // rejection. `Headless` itself is retired + // (`docs/adr/0026-headless-retirement.md`); today it is the only + // other variant, but the guarantee here must hold by construction, + // not by nobody having added a second one yet. + if !matches!(mode, ConformanceModeArg::Tui) { + let mode_name = mode + .to_possible_value() + .map(|value| value.get_name().to_string()) + .unwrap_or_else(|| format!("{mode:?}")); + return fail(&format!( + "mode: \"{mode_name}\" has no conformance dispatch to run -- only \"tui\" does; use \ + --mode tui (the default)" + )); } let mut reports: Vec = Vec::with_capacity(kinds.len()); diff --git a/crates/runtime/src/config/crew.rs b/crates/runtime/src/config/crew.rs index ad814fee..77539d60 100644 --- a/crates/runtime/src/config/crew.rs +++ b/crates/runtime/src/config/crew.rs @@ -194,11 +194,21 @@ impl Default for DisplayConfig { /// `load_layers` refuses any reserved adapter kind whose `mode` is /// `Headless`, naming the retirement and directing the operator to /// `"tui"`. See `docs/adr/0026-headless-retirement.md`. +/// +/// `Protocol` drives a vendor over its own structured protocol -- +/// newline-delimited JSON over a subprocess's stdio, not a PTY -- with +/// the terminal reduced to a self-rendered view rather than the control +/// surface. A second, independent control path alongside `Tui`, not a +/// replacement for it: worth saying here so a future reader does not +/// assume the TUI path itself has no protocol underneath it (it does -- +/// the vendor's own screen-rendering protocol -- `Protocol` mode is +/// about which one crew treats as authoritative). #[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] pub enum AdapterMode { Tui, Headless, + Protocol, } /// Abstract permission posture, mapped to each vendor's own flags. diff --git a/crates/runtime/src/conformance/mod.rs b/crates/runtime/src/conformance/mod.rs index a6d75cc9..5bba1992 100644 --- a/crates/runtime/src/conformance/mod.rs +++ b/crates/runtime/src/conformance/mod.rs @@ -204,6 +204,16 @@ pub async fn run_fixture_conformance(kind: AdapterKind, mode: AdapterMode) -> Co "run_fixture_conformance called with the retired Headless mode for {kind} -- the \ caller must reject Headless before calling this function" ), + // The protocol-first control plane (currently just + // `crate::adapter::claude_protocol`) has no fixture conformance + // suite of its own yet -- unlike `Headless`, this + // is not a permanent rejection, just an as-yet-unbuilt one, but + // the honest behavior is the same: fail loudly here rather than + // fabricate a report for a suite that does not exist. + AdapterMode::Protocol => unreachable!( + "run_fixture_conformance called with Protocol mode for {kind} -- no fixture \ + conformance suite exists for the protocol-first control plane yet" + ), AdapterMode::Tui => { use crate::adapter::tui::{ claude_conformance, codex_conformance, copilot_conformance, omp_conformance, diff --git a/crates/runtime/src/domain/repository.rs b/crates/runtime/src/domain/repository.rs index 7e25f088..dcf11816 100644 --- a/crates/runtime/src/domain/repository.rs +++ b/crates/runtime/src/domain/repository.rs @@ -3075,6 +3075,24 @@ impl<'c> DomainRepository<'c> { RuntimeEvent::WorkerQuestion { question, .. } => question.clone(), _ => None, }; + // A journaled `EscalationRaised` reaching this generic path -- + // today, only `claude_protocol::adapter::ClaudeProtocolAdapter`'s + // `AdapterEventPayload::WorkspaceTrustPending` constructs one + // here, rather than through the dedicated + // `record_escalation_raised` the other two production sites use + // -- gets the identical escalations-row projection that method + // inserts, on the same "commit together" basis (I1) this + // function's own `WorkerQuestion` case above already follows. + // No open-row dedup here (unlike `WorkerQuestion`'s "one open + // question per run"): every current caller of this path emits + // at most once per run before failing it outright, so there is + // nothing yet to deduplicate against. + let escalation = match event { + RuntimeEvent::EscalationRaised { + reason, question, .. + } => Some((reason.clone(), question.clone())), + _ => None, + }; self.append_and_apply( event, Some(task_id), @@ -3114,6 +3132,19 @@ impl<'c> DomainRepository<'c> { )?; } } + if let Some((reason, question)) = escalation { + tx.execute( + "INSERT INTO escalations (escalation_id, run_id, kind, question, created_at) \ + VALUES (?1, ?2, ?3, ?4, ?5)", + rusqlite::params![ + crew_protocol::EscalationId::new().to_string(), + run_id.to_string(), + reason, + question.as_ref().map(crew_protocol::Redacted::as_str), + Timestamp::now().as_str(), + ], + )?; + } Ok(()) }, ) diff --git a/crates/runtime/src/lifecycle.rs b/crates/runtime/src/lifecycle.rs index 8a524c99..8dc36c45 100644 --- a/crates/runtime/src/lifecycle.rs +++ b/crates/runtime/src/lifecycle.rs @@ -323,12 +323,27 @@ pub async fn serve(opts: &ServeOptions) -> Result<(), ServeError> { // it). Created before `ServerConfig` so both holders get the same Arc. let activity_clock = Arc::new(crate::adapter::ActivityClock::new()); + // The one `ApprovalCallback` this daemon ever constructs -- see + // `crate::adapter::registry::ProtocolSupport`'s own doc comment for + // why one shared instance (keyed by approval id, never by adapter + // instance) is correct here rather than merely convenient. + // Behaviorally identical to `NoopApprovalCallback` for any approval + // no protocol-mode adapter ever registered (proven directly by + // `claude_protocol::approval_bridge`'s own + // `acknowledging_an_unregistered_or_dropped_approval_never_errors`), + // so replacing the default here is safe for every mode, not just + // Claude's protocol one. + let protocol_callback = + Arc::new(crate::adapter::claude_protocol::approval_bridge::ProtocolApprovalCallback::new()); + let config = ServerConfig { binary_source: opts.binary_source, run_driver: Some(Arc::clone(®istry) as Arc), artifact_store: Some(artifact_store), repository: repo_root.clone(), worker_verifier: Arc::new(ScopeTokenVerifier::new(Arc::clone(&scope_tokens))), + approval_callback: Arc::clone(&protocol_callback) + as Arc, nested_violation_action, policy: Some((config_paths.clone(), Arc::clone(&policy))), turn_budget_default: crew_config.limits.turn_budget_per_subtask, @@ -395,6 +410,18 @@ pub async fn serve(opts: &ServeOptions) -> Result<(), ServeError> { events_tx: server.events_sender(), })); + // Protocol-mode support: the server-owned `ApprovalService` + // only exists once `Server::bind` has constructed its own + // `OrchestrationService` -- a post-construction setter for the same + // reason `set_resume_support` above is. `protocol_callback` is the + // SAME instance already wired as `ServerConfig::approval_callback`, + // so a protocol adapter's own bridge registers into the identical + // map `ApprovalService::decide` calls back into. + registry.set_protocol_support(Arc::new(crate::adapter::registry::ProtocolSupport { + approval_service: server.orchestration_service().approval_service(), + callback: Arc::clone(&protocol_callback), + })); + // Crash recovery: the sweep is now RESUME FIRST. It runs here -- // not earlier -- because it must follow both post-construction registry // supports: `set_tui_support` above (a `mode: "tui"` run cannot even diff --git a/crates/runtime/src/policy/evaluate.rs b/crates/runtime/src/policy/evaluate.rs index f2624e8b..ee247d4f 100644 --- a/crates/runtime/src/policy/evaluate.rs +++ b/crates/runtime/src/policy/evaluate.rs @@ -299,11 +299,22 @@ impl AdapterAuthorization for PolicyEvaluator { fn authorize( &self, profile: &WorkerProfile, - effective_capabilities: &AdapterCapabilities, + effective_capabilities: &crate::adapter::registry::GatedCapabilities, policy: Option<&RuntimePolicy>, ) -> Result<(), String> { - self.evaluate(profile, effective_capabilities, false, policy) - .map_err(|e| e.to_string()) + // Reads zero capability fields today (this trait method's own + // doc comment), so unwrapping the proven/unproven distinction + // here rather than branching on it is not a shortcut yet -- it + // is the literal truth of what this evaluation looks at. The + // binding constraint is on whoever adds the first real + // capability check, not on this call. + self.evaluate( + profile, + effective_capabilities.capabilities(), + false, + policy, + ) + .map_err(|e| e.to_string()) } fn release(&self) { @@ -540,10 +551,16 @@ mod tests { let stripped_evaluator = PolicyEvaluator::new(test_policy()); let profile = test_profile("any-model-at-all"); - let declared_result = - declared_evaluator.authorize(&profile, &fully_declared_capabilities(), None); - let stripped_result = - stripped_evaluator.authorize(&profile, &fully_stripped_capabilities(), None); + let declared_result = declared_evaluator.authorize( + &profile, + &crate::adapter::registry::GatedCapabilities::Proven(fully_declared_capabilities()), + None, + ); + let stripped_result = stripped_evaluator.authorize( + &profile, + &crate::adapter::registry::GatedCapabilities::Proven(fully_stripped_capabilities()), + None, + ); assert!(declared_result.is_ok(), "{declared_result:?}"); assert_eq!( diff --git a/crates/runtime/src/recovery.rs b/crates/runtime/src/recovery.rs index e6605b0d..bb14fcf8 100644 --- a/crates/runtime/src/recovery.rs +++ b/crates/runtime/src/recovery.rs @@ -758,6 +758,18 @@ impl RecoveryCoordinator { implementation to dispatch to; use mode: \"tui\"" )); } + Some(AdapterMode::Protocol) => { + // Protocol-first adapters (`crate::adapter::claude_protocol`) + // do not implement resume yet -- explicitly out of scope + // for the first one. + // Reject here, at the same point the `Headless` rejection + // above fires, rather than let a resume attempt reach a + // downstream failure shaped like something else. + return Err(format!( + "adapter {kind} was requested with mode: \"protocol\", which does not \ + implement resume yet" + )); + } None => { return Err( "terminal-degraded runs declare no resumable vendor session".to_string() diff --git a/crates/runtime/src/service/orchestration.rs b/crates/runtime/src/service/orchestration.rs index 6a2c7766..e8c7a332 100644 --- a/crates/runtime/src/service/orchestration.rs +++ b/crates/runtime/src/service/orchestration.rs @@ -288,6 +288,21 @@ impl OrchestrationService { } } + /// The `ApprovalService` this service dispatches `approval/*` RPCs + /// through -- exposed (like `crate::ipc::Server::violation_service`, + /// for the identical construction-order reason) so a caller + /// constructed after this service's own `new` can share the SAME + /// instance rather than build a second one with independent state. + /// `crate::adapter::registry::AdapterRegistry::set_protocol_support` + /// is the one caller today: a protocol-mode Claude adapter's + /// `respond_to_approval` bridge must submit approvals to the exact + /// service `approval/decide` itself dispatches through, or a real + /// decision would never reach it. + #[must_use] + pub fn approval_service(&self) -> Arc { + Arc::clone(&self.approval) + } + /// Enables the configured, on-demand retention maintenance RPC. #[must_use] pub fn with_retention(mut self, retention: crate::audit::Retention) -> Self { diff --git a/crates/runtime/tests/adapter_registry.rs b/crates/runtime/tests/adapter_registry.rs index 198de882..b8e5d0c6 100644 --- a/crates/runtime/tests/adapter_registry.rs +++ b/crates/runtime/tests/adapter_registry.rs @@ -14,9 +14,10 @@ use std::sync::Arc; use crew_protocol::{ProjectId, RunId, TaskId, WorkerId}; use crew_runtime::adapter::{ AdapterAuthorization, AdapterCapabilities, AdapterMode, AdapterRegistry, ApprovalsCapability, - DurabilityCapability, FixtureAuthorization, NativeViewCapability, NestedCapability, - OmpRpcStartupOptions, ProtocolKind, ResumeCapability, StartupOptions, SteeringCapability, - TerminalDegradedStartupOptions, UsageCapability, WorkerProfile, WorkspaceControlCapability, + DurabilityCapability, FixtureAuthorization, GatedCapabilities, NativeViewCapability, + NestedCapability, OmpRpcStartupOptions, ProtocolKind, ResumeCapability, StartupOptions, + SteeringCapability, TerminalDegradedStartupOptions, UsageCapability, WorkerProfile, + WorkspaceControlCapability, }; use crew_runtime::config::{NestedViolationAction, RuntimePolicy, crew::DisplayBackend}; use crew_runtime::db::DatabaseHandle; @@ -190,7 +191,7 @@ impl AdapterAuthorization for BlockingAuthorization { fn authorize( &self, _profile: &WorkerProfile, - _effective_capabilities: &AdapterCapabilities, + _effective_capabilities: &GatedCapabilities, _policy: Option<&crew_runtime::config::RuntimePolicy>, ) -> Result<(), String> { let _ = self.entered_tx.send(()); @@ -408,7 +409,11 @@ async fn releasing_a_policy_evaluator_slot_frees_the_registry_ceiling() { // Book the one slot directly, as a real prior `run_one` authorize // call would have. authorization - .authorize(&gpt4_profile, &test_capabilities(), None) + .authorize( + &gpt4_profile, + &GatedCapabilities::Proven(test_capabilities()), + None, + ) .expect("the first slot is within the ceiling of 1"); let (db, _dir, project_id) = harness().await; diff --git a/crates/runtime/tests/claude_protocol_reconcile.rs b/crates/runtime/tests/claude_protocol_reconcile.rs new file mode 100644 index 00000000..5651401e --- /dev/null +++ b/crates/runtime/tests/claude_protocol_reconcile.rs @@ -0,0 +1,103 @@ +//! Exercises the claude-protocol reconciliation reader's REAL (never +//! injected) parse path in a build shaped like the one that ships -- +//! not an approximation of it. +//! +//! **What guarantees the seam's absence from a release build is the +//! `#[cfg(test)]` attribute itself, a compile-time property -- this file +//! does not, and structurally cannot, prove that.** If `#[cfg(test)]` +//! were ever removed from `reconcile::arm_drop_for_test`, leaving the +//! seam compiled into release builds but simply never armed, every test +//! in this file would still compile and pass unchanged: an unarmed seam +//! and an absent one are behaviourally identical from here. What this +//! file DOES prove is narrower and still worth having: that +//! `reconcile::find_gaps` works correctly through the real parse path in +//! a compilation where `cfg(test)` is off for this crate's own code -- +//! which no test living inside `crates/runtime/src/` can ever do, +//! because `cfg!(test)` is true throughout `cargo test`'s compilation of +//! this crate's OWN unit-test binary. An integration test under +//! `crates/runtime/tests/` is different: it links `crew_runtime` as an +//! ordinary external dependency, and the `test` cfg is not applied to +//! that compilation -- so this file is, structurally, the one place able +//! to exercise the shipping parse path at all. +//! +//! `reconcile::find_gaps`/`TranscriptEntry`/`TranscriptEntryKind` are +//! `pub` for exactly this reason -- see that module's own doc comment. +//! The seam itself (`reconcile::arm_drop_for_test`) stays private to the +//! crate: this file cannot reach it, cannot import it, and cannot even +//! name it -- which is consistent with the attribute doing its job, not +//! a demonstration that it is. + +use std::collections::HashSet; + +use crew_runtime::adapter::claude_protocol::reconcile::{ + TranscriptEntry, TranscriptEntryKind, find_gaps, +}; + +fn entry(uuid: &str, kind: &str) -> String { + format!(r#"{{"uuid":"{uuid}","type":"{kind}","message":{{}}}}"#) +} + +/// The real parse path, exercised in a build where `cfg(test)` is off +/// for this crate's own code: every entry in the transcript, present in +/// `journaled_entry_ids` or not, is examined and correctly classified as +/// present or as a gap, with no unexplained loss. This does not by +/// itself prove the seam is absent here (see this file's own module doc +/// comment for why it structurally cannot) -- it proves the thing the +/// seam's absence is supposed to leave behind: the real parse, working. +#[test] +fn the_real_parse_drops_nothing_and_finds_exactly_the_true_gaps() { + let transcript = [ + entry("u1", "user"), + entry("a1", "assistant"), + entry("a2", "assistant"), + entry("a3", "assistant"), + ] + .join("\n"); + let journaled: HashSet = ["u1".to_string(), "a1".to_string()].into_iter().collect(); + + let (examined, gaps) = find_gaps(transcript.as_bytes(), &journaled); + + assert_eq!( + examined, 4, + "a production build must examine every entry in the transcript, not silently drop any \ + of them" + ); + let mut gap_ids: Vec<&str> = gaps.iter().map(|g| g.entry_id.as_str()).collect(); + gap_ids.sort_unstable(); + assert_eq!( + gap_ids, + vec!["a2", "a3"], + "a production build must find exactly the entries genuinely missing from the journal -- \ + no more, no fewer" + ); + assert_eq!( + gaps, + vec![ + TranscriptEntry { + entry_id: "a2".to_string(), + kind: TranscriptEntryKind::Assistant, + }, + TranscriptEntry { + entry_id: "a3".to_string(), + kind: TranscriptEntryKind::Assistant, + }, + ] + ); +} + +/// The other half: when every transcript entry genuinely was journaled, +/// a production build must find nothing at all -- proving the real +/// parse path is not itself a source of spurious gaps. +#[test] +fn the_real_parse_finds_no_gaps_when_the_journal_is_already_complete() { + let transcript = [entry("u1", "user"), entry("a1", "assistant")].join("\n"); + let journaled: HashSet = ["u1".to_string(), "a1".to_string()].into_iter().collect(); + + let (examined, gaps) = find_gaps(transcript.as_bytes(), &journaled); + + assert_eq!(examined, 2); + assert!( + gaps.is_empty(), + "a production build with a complete journal must find no gaps: {gaps:?}" + ); +} diff --git a/crates/runtime/tests/tui_adapter.rs b/crates/runtime/tests/tui_adapter.rs index d2e94bf0..21442353 100644 --- a/crates/runtime/tests/tui_adapter.rs +++ b/crates/runtime/tests/tui_adapter.rs @@ -112,6 +112,8 @@ fn kind_of(payload: &AdapterEventPayload) -> &'static str { AdapterEventPayload::OutOfBandInput { .. } => "OutOfBandInput", AdapterEventPayload::TurnEnded { .. } => "TurnEnded", AdapterEventPayload::FirstRunGateDetected { .. } => "FirstRunGateDetected", + AdapterEventPayload::ReconciliationCompleted { .. } => "ReconciliationCompleted", + AdapterEventPayload::WorkspaceTrustPending => "WorkspaceTrustPending", } } diff --git a/crates/runtime/tests/tui_claude_registry.rs b/crates/runtime/tests/tui_claude_registry.rs index 7ff39752..aaa91195 100644 --- a/crates/runtime/tests/tui_claude_registry.rs +++ b/crates/runtime/tests/tui_claude_registry.rs @@ -614,6 +614,178 @@ async fn submitting_a_tui_mode_claude_run_reaches_the_tui_path_and_emits_lifecyc db.shutdown().await.expect("shutdown database"); } +/// Installing crew's REAL daemon-shaped approval wiring -- an +/// `ApprovalService` backed by `ProtocolApprovalCallback`, exactly as +/// `lifecycle.rs` wires it for the whole daemon -- ahead of a TUI run +/// must not change that run's own behavior at all. This is provable +/// directly, not merely inferable from reading the code: `TuiAdapter` +/// never touches `ApprovalService`/`ApprovalCallback` at all (no field +/// carries either, and `build_tui_adapter`'s own call sites never pass +/// one), but "was asserted, not proven" is exactly the failure class +/// this spike exists to catch -- so this drives a full TUI run through +/// a real tool call (`write_fake_claude_script`'s own `tool_use` entry, +/// the kind of event that WOULD be approval-shaped on a permission- +/// aware adapter) with the real wiring installed, and asserts three +/// things: `respond_to_approval` still refuses with +/// `capability_unsupported`, no row was ever inserted into `approvals` +/// for this run, and no `EscalationRaised` reached the journal as a +/// side effect of the tool call either. +#[tokio::test] +async fn the_real_approval_callback_wiring_never_touches_a_tui_run() { + let _guard = SERIAL_PTY.lock().await; + let (db, dir, project_id) = harness().await; + let profile = claude_tui_profile(); + let (run_id, task_id, worker_id) = seed_worker_and_run(&db, project_id, &profile).await; + + let session_dir = dir.path().join("session"); + std::fs::create_dir_all(&session_dir).expect("create session dir"); + let script_path = write_fake_claude_script(dir.path(), &session_dir); + + let mut adapters = BTreeMap::new(); + adapters.insert( + "claude".to_string(), + AdapterConfig { + enabled: true, + bin: script_path.to_string_lossy().into_owned(), + mode: CrewAdapterMode::Tui, + permission_mode: PermissionMode::Default, + model: None, + profile: "test".to_string(), + session_dir: Some(session_dir.to_string_lossy().into_owned()), + extra_args: Vec::new(), + }, + ); + + let mut display_registry = DisplayRegistry::new(); + display_registry.register(Box::new(HiddenDisplay::new( + crew_protocol::DisplayConfig::default(), + ))); + + let panes_dir = dir.path().join("panes"); + std::fs::create_dir_all(&panes_dir).expect("create panes dir"); + + let registry = AdapterRegistry::new( + Arc::new(FixtureAuthorization { allow: true }), + dir.path().to_path_buf(), + None, + vec![], + ); + registry.set_tui_support(Arc::new(TuiSupport { + display_registry: Arc::new(display_registry), + panes_dir, + crewd_path: PathBuf::from("/opt/crew/bin/crewd"), + state_dir: dir.path().to_path_buf(), + close_on_exit: CloseOnExit::Always, + forced_backend: None, + force_hidden_displays: false, + adapters, + timings: fast_timings(), + org_security_patterns: Vec::new(), + })); + + // The real daemon wiring: ONE `ApprovalService` backed by + // `ProtocolApprovalCallback`, sharing this test's own `db` -- + // exactly `lifecycle.rs`'s own shape, not a second, disconnected + // instance that would prove nothing about the real path. + let (approval_events_tx, _approval_events_rx) = tokio::sync::broadcast::channel(100); + let callback = Arc::new( + crew_runtime::adapter::claude_protocol::approval_bridge::ProtocolApprovalCallback::new(), + ); + let approval_service = Arc::new(crew_runtime::ApprovalService::new( + Arc::clone(&db), + project_id, + Arc::clone(&callback) as Arc, + approval_events_tx, + )); + registry.set_protocol_support(Arc::new(crew_runtime::adapter::ProtocolSupport { + approval_service, + callback, + })); + + let result = registry + .start(ctx( + Arc::clone(&db), + project_id, + run_id, + task_id, + worker_id, + "say hi", + )) + .await; + assert!( + result.is_ok(), + "starting a TUI-mode Claude run must succeed: {result:?}" + ); + + let adapter = registry + .running_adapter(run_id) + .expect("the started adapter must be tracked as running"); + + // Wait for the fake script's own tool_use entry to actually be + // tailed and journaled as an `adapterToolStarted` event -- proof the + // approval-shaped moment this test cares about really happened, not + // just that the run started. The vendor's own `toolu_1` id is + // reassigned to a crew-generated one by the TUI tailer's own + // normalization, so the journaled event kind is the marker to look + // for, not the vendor's own id string. + let saw_tool_call = wait_until( + || { + let db = Arc::clone(&db); + async move { journal_count(&db, run_id, "adapterToolStarted").await } + }, + |count: &usize| *count > 0, + Duration::from_secs(5), + ) + .await; + assert!( + saw_tool_call, + "expected the fake script's tool_use entry to be journaled as adapterToolStarted" + ); + + // 1. `respond_to_approval` still refuses -- unconditionally, exactly + // as it did before this wiring existed. + let respond_result = adapter + .respond_to_approval("fake-approval-id", "approve") + .await; + assert!( + matches!(respond_result, Err(ref err) if err.code() == "capability_unsupported"), + "TuiAdapter::respond_to_approval must still refuse: {respond_result:?}" + ); + + // 2. No row was ever inserted into `approvals` for this run. + let approval_rows: i64 = { + let run_id_string = run_id.to_string(); + db.run_domain_op(Box::new(move |conn| { + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM approvals WHERE run_id = ?1", + rusqlite::params![run_id_string], + |row| row.get(0), + )?; + Ok(serde_json::json!(count)) + })) + .await + .expect("approvals count query") + .as_i64() + .expect("count is an integer") + }; + assert_eq!( + approval_rows, 0, + "a TUI run's tool call must never raise a real approval" + ); + + // 3. No `EscalationRaised` was journaled as a side effect of the + // tool call either -- the approval wiring being present did not + // fabricate one. + assert_eq!( + journal_count(&db, run_id, "escalationRaised").await, + 0, + "a TUI run's tool call must never raise an escalation either" + ); + + let _ = adapter.dispose().await; + db.shutdown().await.expect("shutdown database"); +} + /// The model chosen for THIS run -- the resolved `WorkerProfile.model`, /// what a per-run model picker in crew's own UI dialog would produce -- /// must reach the vendor process `crewd` actually execs. Asserted from diff --git a/crew-config.schema.json b/crew-config.schema.json index edc4d9f0..7db6d6b3 100644 --- a/crew-config.schema.json +++ b/crew-config.schema.json @@ -181,11 +181,12 @@ ] }, "AdapterMode": { - "description": "Whether a vendor adapter runs its worker attached to a real TUI pane\nor drives a headless protocol adapter. `\"headless\"` is retired:\nit still parses here (an old\nconfig file naming it must not fail to load with a schema-shaped\nerror) but is rejected with a typed error at validation time --\n`load_layers` refuses any reserved adapter kind whose `mode` is\n`Headless`, naming the retirement and directing the operator to\n`\"tui\"`. See `docs/adr/0026-headless-retirement.md`.", + "description": "Whether a vendor adapter runs its worker attached to a real TUI pane\nor drives a headless protocol adapter. `\"headless\"` is retired:\nit still parses here (an old\nconfig file naming it must not fail to load with a schema-shaped\nerror) but is rejected with a typed error at validation time --\n`load_layers` refuses any reserved adapter kind whose `mode` is\n`Headless`, naming the retirement and directing the operator to\n`\"tui\"`. See `docs/adr/0026-headless-retirement.md`.\n\n`Protocol` drives a vendor over its own structured protocol --\nnewline-delimited JSON over a subprocess's stdio, not a PTY -- with\nthe terminal reduced to a self-rendered view rather than the control\nsurface. A second, independent control path alongside `Tui`, not a\nreplacement for it: worth saying here so a future reader does not\nassume the TUI path itself has no protocol underneath it (it does --\nthe vendor's own screen-rendering protocol -- `Protocol` mode is\nabout which one crew treats as authoritative).", "type": "string", "enum": [ "tui", - "headless" + "headless", + "protocol" ] }, "PermissionMode": { diff --git a/packages/extension/dist/index.js b/packages/extension/dist/index.js index 4977e32a..3198465e 100644 --- a/packages/extension/dist/index.js +++ b/packages/extension/dist/index.js @@ -8439,6 +8439,59 @@ only *that* the turn ended, and how.`, ], additionalProperties: false }, + { + description: "A protocol-first adapter reconciled its journal against the\nvendor's own durable transcript for this run. Carries no free\ntext, only counts. `examined == 0` is itself a finding, not a\nclean pass: a reconciliation that never looked at anything is\nindistinguishable from one that never ran at all, and this is the\nfield that lets a reader (or a future check) tell those two\napart. No repair mechanism exists yet, so `gapsRepaired` is\nalways `0` today; `gapsFound` and `gapsRepaired` being unequal is\nthe current, expected state, not a defect, until an actual repair\nstep is built.", + type: "object", + properties: { + type: { + type: "string", + const: "adapterReconciliationEvent" + }, + payload: { + type: "object", + properties: { + runId: { + $ref: "#/$defs/RunId" + }, + taskId: { + $ref: "#/$defs/TaskId" + }, + workerId: { + $ref: "#/$defs/WorkerId" + }, + examined: { + type: "integer", + format: "uint64", + minimum: 0 + }, + gapsFound: { + type: "integer", + format: "uint64", + minimum: 0 + }, + gapsRepaired: { + type: "integer", + format: "uint64", + minimum: 0 + } + }, + additionalProperties: false, + required: [ + "runId", + "taskId", + "workerId", + "examined", + "gapsFound", + "gapsRepaired" + ] + } + }, + required: [ + "type", + "payload" + ], + additionalProperties: false + }, { description: "An artifact produced by a worker adapter.", type: "object", diff --git a/packages/protocol-ts/schema/crew.schema.json b/packages/protocol-ts/schema/crew.schema.json index 5a95ef87..963097cd 100644 --- a/packages/protocol-ts/schema/crew.schema.json +++ b/packages/protocol-ts/schema/crew.schema.json @@ -1464,6 +1464,59 @@ ], "additionalProperties": false }, + { + "description": "A protocol-first adapter reconciled its journal against the\nvendor's own durable transcript for this run. Carries no free\ntext, only counts. `examined == 0` is itself a finding, not a\nclean pass: a reconciliation that never looked at anything is\nindistinguishable from one that never ran at all, and this is the\nfield that lets a reader (or a future check) tell those two\napart. No repair mechanism exists yet, so `gapsRepaired` is\nalways `0` today; `gapsFound` and `gapsRepaired` being unequal is\nthe current, expected state, not a defect, until an actual repair\nstep is built.", + "type": "object", + "properties": { + "type": { + "type": "string", + "const": "adapterReconciliationEvent" + }, + "payload": { + "type": "object", + "properties": { + "runId": { + "$ref": "#/$defs/RunId" + }, + "taskId": { + "$ref": "#/$defs/TaskId" + }, + "workerId": { + "$ref": "#/$defs/WorkerId" + }, + "examined": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "gapsFound": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "gapsRepaired": { + "type": "integer", + "format": "uint64", + "minimum": 0 + } + }, + "additionalProperties": false, + "required": [ + "runId", + "taskId", + "workerId", + "examined", + "gapsFound", + "gapsRepaired" + ] + } + }, + "required": [ + "type", + "payload" + ], + "additionalProperties": false + }, { "description": "An artifact produced by a worker adapter.", "type": "object", diff --git a/packages/protocol-ts/src/generated/RuntimeEvent.ts b/packages/protocol-ts/src/generated/RuntimeEvent.ts index afbfaf21..81b70d04 100644 --- a/packages/protocol-ts/src/generated/RuntimeEvent.ts +++ b/packages/protocol-ts/src/generated/RuntimeEvent.ts @@ -35,7 +35,7 @@ message: Redacted, } } | { "type": "taskEvent", "payload": { kind: RuntimeEventK * The decision's rationale, when one was supplied. Absent on * `approvalDecided` events written before this field existed. */ -reason?: Redacted, } } | { "type": "childEvent", "payload": { kind: RuntimeEventKind, parentRunId: RunId, childTaskId: TaskId | null, childWorkerId: WorkerId | null, childRunId: RunId | null, reason: Redacted | null, } } | { "type": "reconcileEvent", "payload": { taskId: TaskId, oldOwnerClientInstanceId: string, newOwnerClientInstanceId: string, revision: number, } } | { "type": "adapterProcessEvent", "payload": { kind: RuntimeEventKind, runId: RunId, taskId: TaskId, workerId: WorkerId, pid: number | null, exitCode: number | null, signal: string | null, } } | { "type": "adapterVendorSessionEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, vendorSessionId: string, } } | { "type": "adapterTurnEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, outcome: TurnOutcome, } } | { "type": "adapterMessageEvent", "payload": { kind: RuntimeEventKind, runId: RunId, taskId: TaskId, workerId: WorkerId, role: string, text: Redacted | null, } } | { "type": "adapterToolEvent", "payload": { kind: RuntimeEventKind, runId: RunId, taskId: TaskId, workerId: WorkerId, toolCallId: string, name: string, ok: boolean | null, detail: Redacted | null, } } | { "type": "adapterUsageEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, inputTokens: number, outputTokens: number, costUsd: number | null, } } | { "type": "adapterArtifactEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, artifactId: ArtifactId, artifactKind: string, } } | { "type": "adapterProtocolHealthEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, healthy: boolean, detail: Redacted | null, } } | { "type": "workspaceEvent", "payload": { kind: WorkspaceEvent, runId: RunId, leaseId: string, } } | { "type": "adapterNestedWorkerEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, +reason?: Redacted, } } | { "type": "childEvent", "payload": { kind: RuntimeEventKind, parentRunId: RunId, childTaskId: TaskId | null, childWorkerId: WorkerId | null, childRunId: RunId | null, reason: Redacted | null, } } | { "type": "reconcileEvent", "payload": { taskId: TaskId, oldOwnerClientInstanceId: string, newOwnerClientInstanceId: string, revision: number, } } | { "type": "adapterProcessEvent", "payload": { kind: RuntimeEventKind, runId: RunId, taskId: TaskId, workerId: WorkerId, pid: number | null, exitCode: number | null, signal: string | null, } } | { "type": "adapterVendorSessionEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, vendorSessionId: string, } } | { "type": "adapterTurnEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, outcome: TurnOutcome, } } | { "type": "adapterMessageEvent", "payload": { kind: RuntimeEventKind, runId: RunId, taskId: TaskId, workerId: WorkerId, role: string, text: Redacted | null, } } | { "type": "adapterToolEvent", "payload": { kind: RuntimeEventKind, runId: RunId, taskId: TaskId, workerId: WorkerId, toolCallId: string, name: string, ok: boolean | null, detail: Redacted | null, } } | { "type": "adapterUsageEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, inputTokens: number, outputTokens: number, costUsd: number | null, } } | { "type": "adapterReconciliationEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, examined: number, gapsFound: number, gapsRepaired: number, } } | { "type": "adapterArtifactEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, artifactId: ArtifactId, artifactKind: string, } } | { "type": "adapterProtocolHealthEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, healthy: boolean, detail: Redacted | null, } } | { "type": "workspaceEvent", "payload": { kind: WorkspaceEvent, runId: RunId, leaseId: string, } } | { "type": "adapterNestedWorkerEvent", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, /** * The vendor-reported child worker reference. Already redacted: * secret-shaped substrings are masked before this event is built.