From 2a3676662c37a8303397e9ad1d5c26843e268e11 Mon Sep 17 00:00:00 2001 From: Nikolas Demiridis Date: Wed, 9 Sep 2026 23:29:41 +0300 Subject: [PATCH 1/3] feat(runtime): park and escalate on a recognized first-run gate wait_for_readiness no longer fails a run outright the instant it recognizes a vendor first-run gate (workspace trust, a theme picker, sign-in). It now journals a new AdapterEventPayload::FirstRunGateDetected (mapped to RuntimeEvent::FirstRunGateDetected { kind }, a new closed FirstRunGateKind enum mirroring the TUI adapter's own internal GateKind), raises the paired EscalationRaised { reason: "vendorFirstRunGate" } with a runtime-authored question (Redacted::assert_runtime_authored -- every word is a fixed template plus a closed-enum-derived phrase, never vendor or captured content), and keeps polling classify_surface on no deadline of its own until the surface resolves to PromptReady (resumes the ordinary paste/Enter path, unmodified) or the process exits (the existing exit watcher reports it, naming the gate that was blocking the run). The escalation is journaled exactly once per distinct gate even though the same Gate(kind) classifies on every tick until it resolves. The escalation's DB/journal side effect -- inserting the escalations row and raising EscalationRaised transactionally with the event append -- is wired the same way the existing write-violation escalation already is: a second mutation DomainAdapterEventSink::emit raises after its main event commits, keyed off the just-built event's shape, rather than a new bespoke call path. This is also the first production call site to populate EscalationRaised.question; the redaction-guard allowlist entry for EscalationRaised.reason is updated to name it. packages/extension/src/milestones.ts's escalationRaised digest case rendered only reason, never question -- a bug dormant since no production call site had ever populated the field until now. It now appends question after reason when present, and renders exactly today's sentence when it is absent. Tests: a fixture-driven replay of a real captured claude workspace-trust gate proves wait_for_readiness escalates exactly once (naming the right kind) and then fails naming the gate once the process is forced to exit; a call-counting test vendor proves a gate that clears after a few ticks resumes readiness normally; two new event_sink.rs tests prove the full DB path -- FirstRunGateDetected and the paired EscalationRaised both journal and broadcast, with a question naming the right vendor and gate phrase, for both a claude and a codex gate kind. milestones.test.ts gains both the question-present and question-absent digest cases. bun run generate regenerates the FirstRunGateKind TS binding and the RuntimeEvent union; packages/protocol-ts/src/index.ts's barrel export is updated to match (caught by generate --check, not by hand). Deliberately not yet fixed here: cancel() and dispose() both open on self.run's async lock, which wait_for_readiness's new unbounded gate-park poll now holds for as long as it runs -- a run parked at a gate cannot yet be cancelled. That is a real, separate concern (a lock-ordering fix, not an escalation-shape one) and lands as its own following commit so it is reviewable on its own. cargo fmt --all, cargo clippy --all-targets --all-features -D warnings, CREW_DISABLE_VENDOR_CLI=1 cargo test --workspace (69 test binaries, all passing), bun run typecheck, and bun test all verified clean. --- crates/protocol/src/event.rs | 42 +- crates/protocol/src/lib.rs | 6 +- crates/runtime/src/adapter/event_sink.rs | 312 ++++++++++++++ crates/runtime/src/adapter/tui/adapter.rs | 393 ++++++++++++++++-- crates/runtime/tests/tui_adapter.rs | 1 + packages/extension/src/milestones.test.ts | 33 ++ packages/extension/src/milestones.ts | 4 +- packages/protocol-ts/schema/crew.schema.json | 50 +++ .../src/generated/FirstRunGateKind.ts | 9 + .../protocol-ts/src/generated/RuntimeEvent.ts | 3 +- packages/protocol-ts/src/index.ts | 1 + 11 files changed, 805 insertions(+), 49 deletions(-) create mode 100644 packages/protocol-ts/src/generated/FirstRunGateKind.ts diff --git a/crates/protocol/src/event.rs b/crates/protocol/src/event.rs index c22aef45..057f8337 100644 --- a/crates/protocol/src/event.rs +++ b/crates/protocol/src/event.rs @@ -348,6 +348,25 @@ pub enum TimeoutKind { Total, } +// Mirrored 1:1 from the TUI adapter's own internal `GateKind` +// (crates/runtime/src/adapter/tui/classify.rs) -- a Rust-only name, not +// part of the shipped schema, so it stays in this plain comment rather +// than the doc below. +/// Which recognized vendor first-run gate blocked a run at readiness. +/// Closed set, never derived from captured PTY text, so a +/// `firstRunGateDetected` event carries no redaction obligation of its +/// own (see that event's doc comment). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export)] +pub enum FirstRunGateKind { + ClaudeWorkspaceTrust, + ClaudeThemePicker, + ClaudeSignIn, + CodexDirectoryTrust, + CodexSignIn, +} + /// Why a settled run resumed to `working`. The two causes map exactly /// to `runResumed`'s two journaling call sites, which race each other /// for the same edge -- see that event's own doc comment. @@ -1043,6 +1062,18 @@ pub enum RuntimeEvent { worker_id: WorkerId, question: Option, }, + /// A TUI adapter's readiness poll recognized a vendor first-run gate + /// blocking the run (workspace trust, a theme picker, sign-in, ...). + /// Journaled once, immediately before the paired + /// `escalationRaised { reason: "vendorFirstRunGate" }` -- `kind` is a + /// closed enum, never captured PTY text, so this event carries no + /// redaction obligation of its own. + FirstRunGateDetected { + run_id: RunId, + task_id: TaskId, + worker_id: WorkerId, + kind: FirstRunGateKind, + }, /// A worker escalated a blocking condition to its leader or a human /// operator. `reason` is a plain, machine-assigned code (never raw /// worker content); `question` has already crossed the redaction @@ -1584,14 +1615,15 @@ 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 only two production construction sites pass \ + caller- or vendor-derived: the three production construction sites pass \ the literals `repeated_failure` \ - (crates/runtime/src/adapter/run_lifecycle.rs) and `write_violation` \ - (crates/runtime/src/domain/repository.rs). This reason previously \ + (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 \ 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. What actually secures the \ - field is the literals above.", + `question`, which no production site populates before `vendorFirstRunGate`. \ + What actually secures the field is the literals above.", ), ( "RuntimeEventKind::PolicyViolation.profile_id", diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index f8c421a6..1fe3d4c1 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -39,9 +39,9 @@ pub use display::{ }; pub use event::RunFlags; pub use event::{ - AnsweredBy, Classified, ContentClass, DiagnosticLevel, EventEnvelope, EventSource, PlanSpec, - Redacted, ResumeCause, RuntimeEvent, RuntimeEventKind, SubtaskSpec, TimeoutKind, Timestamp, - TimestampParseError, TurnOutcome, + AnsweredBy, Classified, ContentClass, DiagnosticLevel, EventEnvelope, EventSource, + FirstRunGateKind, PlanSpec, Redacted, ResumeCause, RuntimeEvent, RuntimeEventKind, SubtaskSpec, + TimeoutKind, Timestamp, TimestampParseError, TurnOutcome, }; pub use ids::{ ApprovalId, ArtifactId, EscalationId, MessageId, OperationId, PolicyViolationId, ProjectId, diff --git a/crates/runtime/src/adapter/event_sink.rs b/crates/runtime/src/adapter/event_sink.rs index 1d7e385a..59baa56d 100644 --- a/crates/runtime/src/adapter/event_sink.rs +++ b/crates/runtime/src/adapter/event_sink.rs @@ -141,6 +141,15 @@ pub enum AdapterEventPayload { backend: crew_protocol::DisplayBackend, pane_ref: String, }, + /// A TUI adapter's readiness poll recognized a vendor first-run gate + /// blocking the run. Carries only the closed-set `kind` -- never + /// captured PTY text. Maps to [`RuntimeEvent::FirstRunGateDetected`] + /// and additionally raises the paired `EscalationRaised { reason: + /// "vendorFirstRunGate" }` with a runtime-authored `question` (see + /// [`DomainAdapterEventSink::emit`]). + FirstRunGateDetected { + kind: crew_protocol::FirstRunGateKind, + }, } /// Adapters push ordered normalized events into the runtime journal @@ -413,10 +422,43 @@ impl DomainAdapterEventSink { pane_ref, } } + AdapterEventPayload::FirstRunGateDetected { kind } => { + RuntimeEvent::FirstRunGateDetected { + run_id, + task_id, + worker_id, + kind, + } + } } } } +/// 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, +/// never vendor or captured content, so it may cross the redaction +/// boundary without going through [`Redactor`]. Wording is pinned by +/// `first_run_gate_question_names_the_vendor_and_the_gate` below -- +/// changing it is a deliberate edit, not a refactor side effect. +fn first_run_gate_question(kind: crew_protocol::FirstRunGateKind) -> crew_protocol::Redacted { + use crew_protocol::FirstRunGateKind::{ + ClaudeSignIn, ClaudeThemePicker, ClaudeWorkspaceTrust, CodexDirectoryTrust, CodexSignIn, + }; + let (vendor, phrase) = match kind { + ClaudeWorkspaceTrust => ("Claude", "workspace-trust prompt"), + ClaudeThemePicker => ("Claude", "theme-selection prompt"), + ClaudeSignIn => ("Claude", "sign-in prompt"), + CodexDirectoryTrust => ("Codex", "directory-trust prompt"), + CodexSignIn => ("Codex", "sign-in prompt"), + }; + crew_protocol::Redacted::assert_runtime_authored(format!( + "The {vendor} CLI is waiting on its first-run {phrase} and cannot proceed until it is \ + answered. Answer it in the worker's pane, or cancel the run. Crew will not answer it \ + for you." + )) +} + impl AdapterEventSink for DomainAdapterEventSink { fn emit(&self, event: AdapterEvent) -> AdapterFuture<'_, u64> { let run_id = event.run_id; @@ -431,6 +473,10 @@ impl AdapterEventSink for DomainAdapterEventSink { // block below (this function cannot `?` here -- it returns the // future itself, not a `Result`). let cursor_json = event.cursor.as_ref().map(serde_json::to_string).transpose(); + let first_run_gate_kind = match &event.payload { + AdapterEventPayload::FirstRunGateDetected { kind } => Some(*kind), + _ => None, + }; let write_tool = match &event.payload { AdapterEventPayload::ToolStarted { name, .. } if !self.isolated_workspace @@ -551,6 +597,35 @@ impl AdapterEventSink for DomainAdapterEventSink { } } + if let Some(kind) = first_run_gate_kind { + // The escalation row and its `EscalationRaised` event + // commit together, as a second mutation after the + // `FirstRunGateDetected` event above -- same shape as the + // write-violation escalation, and for the same reason: + // this call site is the first to actually populate + // `EscalationRaised.question` (see that field's own doc). + let question = first_run_gate_question(kind); + let raise_result = db + .run_domain_op(Box::new(move |conn| { + DomainRepository::new(conn, project_id) + .record_escalation_raised(run_id, "vendorFirstRunGate", Some(question)) + .map(|c| embed_envelope(json!({ "sequence": c.sequence }), &c.envelope)) + })) + .await; + match raise_result { + Ok(mut value) => { + let _ = crate::domain::broadcast_committed(&events_tx, &mut value); + } + Err(err) => { + tracing::warn!( + error = %err, + run_id = %run_id, + "failed to record vendor-first-run-gate escalation" + ); + } + } + } + if out_of_band { let mut flag_result = db .run_domain_op(Box::new(move |conn| { @@ -1998,3 +2073,240 @@ mod crash_resume_tests { .collect() } } + +#[cfg(test)] +mod first_run_gate_tests { + //! Requirement (a)'s DB/journal half: emitting + //! `AdapterEventPayload::FirstRunGateDetected` through the real + //! [`DomainAdapterEventSink`] must durably journal + //! `RuntimeEvent::FirstRunGateDetected` AND raise the paired + //! `EscalationRaised { reason: "vendorFirstRunGate" }` with a + //! populated `question`, against a real database -- proving the + //! second-mutation wiring this sink adds (mirrored on the existing + //! write-violation special case), not just the payload shape. The + //! fixture-driven half (that `wait_for_readiness` actually emits this + //! payload for a real replayed gate capture, exactly once) is + //! `adapter.rs`'s own + //! `wait_for_readiness_escalates_and_then_fails_when_the_process_exits_under_a_replayed_gate_capture`. + + use std::sync::Arc; + + use crew_protocol::{ + FirstRunGateKind, ProjectId, RunState, TaskId, Timestamp, Worker, WorkerId, + WorkerProfileRef, + }; + use tempfile::TempDir; + use tokio::sync::broadcast; + + use crate::config::NestedViolationAction; + use crate::db::DatabaseHandle; + use crate::domain::DomainRepository; + use crate::policy::ViolationService; + + use super::*; + + async fn open_db() -> (TempDir, Arc) { + let dir = tempfile::Builder::new() + .prefix("bat-first-run-gate-sink-") + .tempdir_in("/tmp") + .expect("create temp dir"); + let db_path = dir.path().join("state.db"); + let db = Arc::new( + DatabaseHandle::start(db_path) + .await + .expect("start database"), + ); + (dir, db) + } + + /// Mirrors `out_of_band_input_tests::seed_working_run` (duplicated -- + /// private to its own module). + async fn seed_working_run( + db: &DatabaseHandle, + project_id: ProjectId, + ) -> (TaskId, WorkerId, RunId) { + 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, + &crew_protocol::TaskRef { + owner_client_instance_id: "omp-1".to_string(), + revision: 1, + }, + )?; + let worker = Worker { + worker_id, + profile_ref: WorkerProfileRef { + id: worker_id, + fingerprint: "sha256:fake".to_string(), + adapter: "fake".to_string(), + model: "test".to_string(), + permission_envelope: serde_json::json!({}), + }, + parent_worker_id: None, + created_at: Timestamp::now(), + }; + repo.create_worker(&worker)?; + let run = crew_protocol::Run { + run_id, + task_id, + worker_id, + state: RunState::try_from("queued").expect("queued is a valid state"), + flags: crew_protocol::RunFlags::default(), + vendor_session_id: None, + started_at: None, + completed_at: None, + }; + repo.submit_run(&run, None, None)?; + for state in ["starting", "working"] { + repo.transition_run( + run_id, + &RunState::try_from(state).expect("valid state"), + None, + )?; + } + Ok(serde_json::json!({})) + })) + .await + .expect("seed working run"); + (task_id, worker_id, run_id) + } + + fn sink( + db: Arc, + project_id: ProjectId, + events_tx: broadcast::Sender, + ) -> DomainAdapterEventSink { + let violation_service = Arc::new(ViolationService::new( + Arc::clone(&db), + project_id, + events_tx.clone(), + None, + NestedViolationAction::QuarantineAndCancel, + Redactor::new(), + )); + DomainAdapterEventSink::new( + db, + project_id, + events_tx, + Vec::new(), + false, + violation_service, + false, + ) + .expect("built-in redaction rules always compile") + } + + #[tokio::test] + async fn first_run_gate_detected_journals_the_kind_and_raises_a_populated_escalation() { + let (_dir, db) = open_db().await; + let project_id = ProjectId::new(); + let (task_id, worker_id, run_id) = seed_working_run(&db, project_id).await; + let (events_tx, mut events_rx) = broadcast::channel(16); + let sink = sink(Arc::clone(&db), project_id, events_tx); + + sink.emit(AdapterEvent { + run_id, + task_id, + worker_id, + payload: AdapterEventPayload::FirstRunGateDetected { + kind: FirstRunGateKind::ClaudeWorkspaceTrust, + }, + cursor: None, + }) + .await + .expect("emit"); + + let first = events_rx + .try_recv() + .expect("FirstRunGateDetected must broadcast"); + match &first.event { + RuntimeEvent::FirstRunGateDetected { + run_id: got_run_id, + task_id: got_task_id, + worker_id: got_worker_id, + kind, + } => { + assert_eq!(*got_run_id, run_id); + assert_eq!(*got_task_id, task_id); + assert_eq!(*got_worker_id, worker_id); + assert_eq!(*kind, FirstRunGateKind::ClaudeWorkspaceTrust); + } + other => panic!("expected FirstRunGateDetected, got {other:?}"), + } + + let second = events_rx + .try_recv() + .expect("the paired EscalationRaised must also broadcast"); + match &second.event { + RuntimeEvent::EscalationRaised { + run_id: got_run_id, + reason, + question, + .. + } => { + assert_eq!(*got_run_id, run_id); + assert_eq!(reason, "vendorFirstRunGate"); + let question = question + .as_ref() + .expect("a vendorFirstRunGate escalation must carry a question") + .as_str(); + assert!( + question.contains("Claude") && question.contains("workspace-trust prompt"), + "the question must name the vendor and the specific gate: {question}" + ); + assert!( + question.contains("Crew will not answer it for you."), + "the question must close on the approved sentence verbatim: {question}" + ); + } + other => panic!("expected EscalationRaised, got {other:?}"), + } + + assert!( + events_rx.try_recv().is_err(), + "exactly these two events, nothing more" + ); + + db.shutdown().await.expect("shutdown database"); + } + + #[tokio::test] + async fn a_codex_sign_in_gate_names_codex_and_sign_in() { + let (_dir, db) = open_db().await; + let project_id = ProjectId::new(); + let (task_id, worker_id, run_id) = seed_working_run(&db, project_id).await; + let (events_tx, mut events_rx) = broadcast::channel(16); + let sink = sink(Arc::clone(&db), project_id, events_tx); + + sink.emit(AdapterEvent { + run_id, + task_id, + worker_id, + payload: AdapterEventPayload::FirstRunGateDetected { + kind: FirstRunGateKind::CodexSignIn, + }, + cursor: None, + }) + .await + .expect("emit"); + + let _ = events_rx.try_recv().expect("FirstRunGateDetected"); + let escalation = events_rx.try_recv().expect("EscalationRaised"); + match &escalation.event { + RuntimeEvent::EscalationRaised { question, .. } => { + let question = question.as_ref().expect("question").as_str(); + assert!( + question.contains("Codex") && question.contains("sign-in prompt"), + "unexpected question for CodexSignIn: {question}" + ); + } + other => panic!("expected EscalationRaised, got {other:?}"), + } + + db.shutdown().await.expect("shutdown database"); + } +} diff --git a/crates/runtime/src/adapter/tui/adapter.rs b/crates/runtime/src/adapter/tui/adapter.rs index 94d72fff..6da60ab1 100644 --- a/crates/runtime/src/adapter/tui/adapter.rs +++ b/crates/runtime/src/adapter/tui/adapter.rs @@ -52,7 +52,7 @@ use crate::display::{ }; use crate::supervisor::{EscalationTimings, PtyProcess, SupervisorError}; -use super::classify::Surface; +use super::classify::{GateKind, Surface}; use super::discovery::{DiscoveryError, find_transcript_by_nonce}; use super::grid::TerminalGrid; use super::input::{PASTE_CHUNK_BYTES, paste_chunks}; @@ -834,6 +834,7 @@ impl TuiAdapter { /// `ProtocolHealthChanged` diagnostic. `run_slot` is the caller's /// already-held, already-checked-empty `self.run` guard -- see /// `run_pipeline`'s own doc comment for why it stays held throughout. + #[allow(clippy::too_many_arguments)] async fn resume_from( &self, run_slot: &mut Option, @@ -1114,6 +1115,10 @@ impl TuiAdapter { &pty, type_text, spawn_instant + INJECT_MIN_DELAY, + run_id, + task_id, + worker_id, + &sink, ) .await { @@ -1690,8 +1695,19 @@ async fn wait_for_readiness( pty: &Arc, inject: Option>, not_before: tokio::time::Instant, + run_id: RunId, + task_id: TaskId, + worker_id: WorkerId, + sink: &Arc, ) -> Result<(), AdapterError> { let deadline = tokio::time::Instant::now() + cap; + // Set once this poll journals `FirstRunGateDetected` + + // `EscalationRaised` for the gate currently blocking the run -- + // exactly once per distinct gate, even though the loop below keeps + // re-classifying the same `Gate(kind)` on every tick until it + // resolves. A vendor that swaps from one gate to another (observed + // nowhere yet, but not ruled out) re-escalates for the new one. + let mut escalated_gate: Option = None; // Wait for the first output (a single check, not a loop: every // outcome below either proceeds past this point or returns). This @@ -1738,14 +1754,59 @@ async fn wait_for_readiness( match classified { None | Some(Surface::PromptReady) => break, Some(Surface::Gate(gate)) => { - return Err(AdapterError::process( - kind, - "start", - format!( - "a first-run gate is blocking the run ({gate:?}) -- answer it by hand \ - once in this workspace outside crew, then retry" - ), - )); + // A recognized first-run gate does not fail the run: it + // parks it and hands the human the decision, journaling + // exactly once per distinct gate even though this arm + // re-fires on every tick the grid still reads as the same + // `Gate(gate)`. + if escalated_gate != Some(gate) { + emit( + sink, + run_id, + task_id, + worker_id, + AdapterEventPayload::FirstRunGateDetected { + kind: protocol_gate_kind(gate), + }, + None, + ) + .await; + escalated_gate = Some(gate); + } + // Parked on no deadline of its own -- neither `cap` nor + // any other timeout -- until the surface changes (the next + // loop iteration's classify above), the process exits, or + // the run is cancelled. A human answering a first-run gate + // by hand has no bound this adapter may impose; see the + // approval service for the same "no Duration at all" + // judgement made for the same reason. + tokio::select! { + biased; + _ = pty.exit_watcher() => { + let classified = { + let g = grid.lock().expect("terminal-grid mutex never poisoned"); + vendor.classify_surface(&g) + }; + return Err(match classified { + Some(Surface::Gate(gate)) => AdapterError::process( + kind, + "start", + format!( + "the worker process exited while a first-run gate was \ + blocking the run ({gate:?})" + ), + ), + _ => AdapterError::process( + kind, + "start", + "the worker process exited before a recognizable surface \ + appeared", + ), + }); + } + _ = tokio::time::timeout(quiet, rx.recv()) => {} + } + continue; } Some(Surface::Undecided) => { let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); @@ -1852,6 +1913,21 @@ async fn wait_for_readiness( } } +/// Maps this module's internal [`GateKind`] to the durable protocol enum +/// journaled on [`crew_protocol::RuntimeEvent::FirstRunGateDetected`]. +/// Kept as an explicit `match` (not a `From` impl) so a new `GateKind` +/// variant fails this file to compile rather than silently falling +/// through -- the protocol enum is mirrored 1:1 by hand, on purpose. +fn protocol_gate_kind(gate: GateKind) -> crew_protocol::FirstRunGateKind { + match gate { + GateKind::ClaudeWorkspaceTrust => crew_protocol::FirstRunGateKind::ClaudeWorkspaceTrust, + GateKind::ClaudeThemePicker => crew_protocol::FirstRunGateKind::ClaudeThemePicker, + GateKind::ClaudeSignIn => crew_protocol::FirstRunGateKind::ClaudeSignIn, + GateKind::CodexDirectoryTrust => crew_protocol::FirstRunGateKind::CodexDirectoryTrust, + GateKind::CodexSignIn => crew_protocol::FirstRunGateKind::CodexSignIn, + } +} + /// Waits until the PTY output has been silent for `required` (phase 2 of /// prompt delivery -- see the caller's comment). Gives up waiting at `cap` /// and reports it via the returned error so the caller can decide to @@ -2384,7 +2460,121 @@ mod tests { use super::super::classify::GateKind; use super::*; - use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; + + /// Polls `condition` until it is true or `timeout` elapses, returning + /// which. Mirrors `tests/tui_adapter.rs`'s own `wait_until` helper + /// (duplicated here -- that file is a separate integration-test + /// binary, not reachable from this unit-test module). + async fn wait_until(mut condition: impl FnMut() -> bool, timeout: Duration) -> bool { + let deadline = tokio::time::Instant::now() + timeout; + loop { + if condition() { + return true; + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(15)).await; + } + } + + /// A minimal [`AdapterEventSink`] that just records every payload it + /// is given, for the `wait_for_readiness` unit tests below that need + /// to see what the gate-escalation path journaled without a real + /// database (that end-to-end proof, through `DomainAdapterEventSink` + /// into a real `EscalationRaised` row with its `question`, is + /// `event_sink.rs`'s own `first_run_gate_tests` module). + struct GateRecordingSink(StdMutex>); + + impl GateRecordingSink { + fn new() -> Arc { + Arc::new(Self(StdMutex::new(Vec::new()))) + } + + fn payloads(&self) -> Vec { + self.0 + .lock() + .expect("recording sink mutex never poisoned") + .clone() + } + } + + impl AdapterEventSink for GateRecordingSink { + fn emit(&self, event: AdapterEvent) -> AdapterFuture<'_, u64> { + self.0 + .lock() + .expect("recording sink mutex never poisoned") + .push(event.payload); + Box::pin(async { Ok(0) }) + } + + fn note_real_user_turn(&self, _run_id: RunId) -> AdapterFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + } + + /// A test-only vendor whose `classify_surface` deterministically + /// reports a fixed [`GateKind`] for its first `clear_after` calls, + /// then `Surface::PromptReady` forever after -- lets the gate-park + /// loop be tested on TICK COUNTS rather than real elapsed time or + /// fixture-replay timing. A whole captured screen (e.g. + /// `claude-trust-to-composer.raw`) typically arrives on the PTY as a + /// single read, so replaying a fixture that transitions gate -> + /// composer cannot reliably exercise more than one tick of this poll; + /// this double can. + /// + /// `wait_for_readiness` calls no `TuiVendor` method but + /// `classify_surface`, so every other trait method here is + /// unreachable by construction. + struct CountingGateVendor { + gate: GateKind, + clear_after: u32, + calls: AtomicU32, + } + + impl TuiVendor for CountingGateVendor { + fn kind(&self) -> &'static str { + "counting-gate" + } + fn launch(&self, _spec: &StartSpec, _cfg: &AdapterConfig) -> LaunchSpec { + unreachable!("wait_for_readiness never calls TuiVendor::launch") + } + fn resume_launch( + &self, + _session: &VendorSessionRef, + _spec: &StartSpec, + _cfg: &AdapterConfig, + ) -> LaunchSpec { + unreachable!("wait_for_readiness never calls TuiVendor::resume_launch") + } + fn transcript_root(&self, _spec: &StartSpec, _cfg: &AdapterConfig) -> PathBuf { + unreachable!("wait_for_readiness never calls TuiVendor::transcript_root") + } + fn format(&self) -> Arc { + unreachable!("wait_for_readiness never calls TuiVendor::format") + } + fn compose_input(&self, _message: &str) -> Vec { + unreachable!("wait_for_readiness never calls TuiVendor::compose_input") + } + fn interrupt_sequence(&self) -> Vec { + unreachable!("wait_for_readiness never calls TuiVendor::interrupt_sequence") + } + fn permission_args(&self, _mode: crate::config::crew::PermissionMode) -> Vec { + unreachable!("wait_for_readiness never calls TuiVendor::permission_args") + } + fn version_gate(&self, _probed: &str) -> VersionVerdict { + unreachable!("wait_for_readiness never calls TuiVendor::version_gate") + } + fn classify_surface(&self, _grid: &TerminalGrid) -> Option { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + if n < self.clear_after { + Some(Surface::Gate(self.gate)) + } else { + Some(Surface::PromptReady) + } + } + } /// The constant is derived from the bounds it must dominate, /// not chosen. If a production bound is ever raised past it, this fails @@ -2912,19 +3102,24 @@ mod tests { } /// End-to-end proof that `wait_for_readiness` itself, not just the - /// Enter-precondition helper, refuses to paste into a gate: a real - /// double replays a committed capture verbatim onto its own PTY - /// output (the same bytes a live vendor produced), fed into the grid - /// the exact way `run_pipeline` feeds it (a background task on its - /// own subscription), and the readiness poll must recognize the gate - /// and fail closed before ever calling `write_paste`. The double - /// sleeps after replaying so its output channel stays open through - /// the poll -- otherwise the channel closing before - /// `wait_for_readiness` ever reads from it would be misread as "the - /// process exited before producing output", the wrong error for this - /// test to prove. + /// Enter-precondition helper, recognizes a gate from a real replayed + /// capture (the same bytes a live vendor produced, fed into the grid + /// the exact way `run_pipeline` feeds it) and parks on it rather than + /// pasting into it: it journals `FirstRunGateDetected` naming the + /// gate exactly once, then keeps polling -- on no deadline of its + /// own -- until the process exits (forced here via `terminate()` + /// rather than waiting out the double's own sleep), at which point + /// the failure names the gate that was blocking it. This is also + /// this slice's fixture-driven proof of requirement (c) ("a fake + /// vendor that exits under the gate -> failure names the gate"); + /// requirement (a)'s remaining half -- that the paired + /// `EscalationRaised` a real sink raises carries the right `kind` + /// and a populated `question` -- is `event_sink.rs`'s own + /// `first_run_gate_tests` module, which proves the DB/journal side + /// this unit test's stub sink cannot. #[tokio::test] - async fn wait_for_readiness_refuses_to_paste_into_a_replayed_gate_capture() { + async fn wait_for_readiness_escalates_and_then_fails_when_the_process_exits_under_a_replayed_gate_capture() + { use crate::adapter::tui::ClaudeTuiVendor; let fixture_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -2935,7 +3130,7 @@ mod tests { program: PathBuf::from("/bin/sh"), args: vec![ "-c".to_string(), - format!("cat '{}' && sleep 5", fixture_path.display()), + format!("cat '{}' && sleep 30", fixture_path.display()), ], ..crate::supervisor::SpawnSpec::minimal() }, @@ -2964,23 +3159,56 @@ mod tests { } let vendor = ClaudeTuiVendor::new(PathBuf::from("/w"), vec![]); - let result = wait_for_readiness( - &mut readiness_rx, - "claude", - &vendor, - &grid, - Duration::from_millis(50), - Duration::from_secs(3), - &pty, - Some(PromptInjection { - text: "this must never be written", - write_timeout: Duration::from_secs(1), - }), - tokio::time::Instant::now(), - ) - .await; + let sink = GateRecordingSink::new(); + let dyn_sink: Arc = sink.clone(); + let run_id = RunId::new(); + let task_id = TaskId::new(); + let worker_id = WorkerId::new(); + + let pty_for_wait = Arc::clone(&pty); + let wait = tokio::spawn(async move { + wait_for_readiness( + &mut readiness_rx, + "claude", + &vendor, + &grid, + Duration::from_millis(50), + Duration::from_secs(30), + &pty_for_wait, + Some(PromptInjection { + text: "this must never be written", + write_timeout: Duration::from_secs(1), + }), + tokio::time::Instant::now(), + run_id, + task_id, + worker_id, + &dyn_sink, + ) + .await + }); + + // Park, don't fail: wait for the escalation to actually land + // rather than asserting it failed immediately. + assert!( + wait_until(|| !sink.payloads().is_empty(), Duration::from_secs(5)).await, + "a replayed gate capture must escalate, not silently fail closed or hang" + ); + match &sink.payloads()[..] { + [AdapterEventPayload::FirstRunGateDetected { kind }] => { + assert_eq!(*kind, crew_protocol::FirstRunGateKind::ClaudeWorkspaceTrust); + } + other => panic!("expected exactly one FirstRunGateDetected, got {other:?}"), + } + // The gate never clears on its own (the fixture is a static + // replay): force the exit path instead of waiting out the + // double's own sleep, and confirm the failure names the gate. let _ = pty.terminate().await; + let result = tokio::time::timeout(Duration::from_secs(5), wait) + .await + .expect("must return promptly once the process exits") + .expect("wait_for_readiness task must not panic"); match result { Err(err) => { @@ -3043,6 +3271,7 @@ mod tests { } let vendor = ClaudeTuiVendor::new(PathBuf::from("/w"), vec![]); + let sink: Arc = GateRecordingSink::new(); let started = tokio::time::Instant::now(); let result = wait_for_readiness( &mut readiness_rx, @@ -3057,6 +3286,10 @@ mod tests { write_timeout: Duration::from_secs(1), }), tokio::time::Instant::now(), + RunId::new(), + TaskId::new(), + WorkerId::new(), + &sink, ) .await; let elapsed = started.elapsed(); @@ -3120,6 +3353,7 @@ mod tests { } let vendor = ClaudeTuiVendor::new(PathBuf::from("/w"), vec![]); + let sink: Arc = GateRecordingSink::new(); let started = tokio::time::Instant::now(); // A long cap: if the closed-channel case were mishandled as a // pacing tick, this test would either hang here or take @@ -3140,6 +3374,10 @@ mod tests { write_timeout: Duration::from_secs(1), }), tokio::time::Instant::now(), + RunId::new(), + TaskId::new(), + WorkerId::new(), + &sink, ), ) .await @@ -3159,6 +3397,85 @@ mod tests { ); } + /// Requirement (b): a vendor that clears its gate after a few ticks + /// resumes normally. `wait_for_readiness` must escalate exactly once + /// for the gate it first saw, then -- once the identical surface + /// classifies as `PromptReady` on a later tick, nothing else about + /// the call having changed -- return `Ok(())` exactly as an ungated + /// readiness would, so the caller's ordinary phase-2 Enter delivery + /// proceeds unmodified. + #[tokio::test] + async fn wait_for_readiness_resumes_normally_once_a_parked_gate_clears() { + let pty = Arc::new( + PtyProcess::spawn( + &crate::supervisor::SpawnSpec { + program: PathBuf::from("/bin/sh"), + args: vec!["-c".to_string(), "printf hi && sleep 30".to_string()], + ..crate::supervisor::SpawnSpec::minimal() + }, + EscalationTimings::default(), + ) + .expect("spawn the ticking double"), + ); + + let mut readiness_rx = pty.subscribe_output(); + let grid = Arc::new(StdMutex::new(TerminalGrid::new())); + { + let mut grid_rx = pty.subscribe_output(); + let grid = Arc::clone(&grid); + tokio::spawn(async move { + loop { + match grid_rx.recv().await { + Ok(bytes) => grid + .lock() + .expect("terminal-grid mutex never poisoned") + .push(&bytes), + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(_) => break, + } + } + }); + } + + let vendor = CountingGateVendor { + gate: GateKind::ClaudeWorkspaceTrust, + clear_after: 3, + calls: AtomicU32::new(0), + }; + let sink = GateRecordingSink::new(); + let dyn_sink: Arc = sink.clone(); + + let result = wait_for_readiness( + &mut readiness_rx, + "counting-gate", + &vendor, + &grid, + Duration::from_millis(20), + Duration::from_secs(5), + &pty, + None, + tokio::time::Instant::now(), + RunId::new(), + TaskId::new(), + WorkerId::new(), + &dyn_sink, + ) + .await; + + let _ = pty.terminate().await; + + result.expect("a gate that clears must resolve readiness normally, not fail it"); + match &sink.payloads()[..] { + [AdapterEventPayload::FirstRunGateDetected { kind }] => { + assert_eq!(*kind, crew_protocol::FirstRunGateKind::ClaudeWorkspaceTrust); + } + other => panic!( + "the gate must be escalated exactly once even though classify_surface reports \ + it on every tick until it clears: {other:?}" + ), + } + } + // ----------------------------------------------- the bracketed-paste invariant /// Prompt text is always delivered to a vendor TUI as a bracketed diff --git a/crates/runtime/tests/tui_adapter.rs b/crates/runtime/tests/tui_adapter.rs index 7b7f9915..6a8decc5 100644 --- a/crates/runtime/tests/tui_adapter.rs +++ b/crates/runtime/tests/tui_adapter.rs @@ -111,6 +111,7 @@ fn kind_of(payload: &AdapterEventPayload) -> &'static str { AdapterEventPayload::QuestionDetected { .. } => "QuestionDetected", AdapterEventPayload::OutOfBandInput { .. } => "OutOfBandInput", AdapterEventPayload::TurnEnded { .. } => "TurnEnded", + AdapterEventPayload::FirstRunGateDetected { .. } => "FirstRunGateDetected", } } diff --git a/packages/extension/src/milestones.test.ts b/packages/extension/src/milestones.test.ts index 2d8c8c4d..0c37acbf 100644 --- a/packages/extension/src/milestones.test.ts +++ b/packages/extension/src/milestones.test.ts @@ -245,6 +245,39 @@ test("question digest contains the question text and triage instruction", () => expect(digest).toContain("Answer via crew_send"); }); +test("escalation digest with no question renders exactly today's sentence", () => { + const digest = formatDigest( + envelope({ + runId: "run-1", + event: { type: "escalationRaised", payload: { runId: "run-1", taskId: "task-1", workerId: "w1", reason: "write_violation", question: null } }, + }), + ROWS, + ); + expect(digest).toBe("Escalation raised on run run-1 (claude adapter) for task task-1: write_violation."); +}); + +test("escalation digest with a question appends it after the reason", () => { + const digest = formatDigest( + envelope({ + runId: "run-1", + event: { + type: "escalationRaised", + payload: { + runId: "run-1", + taskId: "task-1", + workerId: "w1", + reason: "vendorFirstRunGate", + question: "The Claude CLI is waiting on its first-run workspace-trust prompt and cannot proceed until it is answered. Answer it in the worker's pane, or cancel the run. Crew will not answer it for you.", + }, + }, + }), + ROWS, + ); + expect(digest).toBeDefined(); + expect(digest).toContain("vendorFirstRunGate"); + expect(digest).toContain("Crew will not answer it for you."); +}); + function fakeBridge(): { sent: string[]; dispatch: (e: EventEnvelope, meta?: { replay: boolean }) => void; diff --git a/packages/extension/src/milestones.ts b/packages/extension/src/milestones.ts index ef29c06f..8ea86b8a 100644 --- a/packages/extension/src/milestones.ts +++ b/packages/extension/src/milestones.ts @@ -156,8 +156,8 @@ export function formatDigest(e: EventEnvelope, lookup: RunLookup): string | unde case "budgetExceeded": return `${capitalize(who)} exceeded its turn budget. Escalate to the user or raise the budget via the plan.`; case "escalationRaised": { - const reason = event.payload.reason; - return `Escalation raised on ${who}: ${reason}.`; + const { reason, question } = event.payload; + return question ? `Escalation raised on ${who}: ${reason}. ${question}` : `Escalation raised on ${who}: ${reason}.`; } case "paneDowngraded": { const { requestedBackend, requestedPlacement, actualBackend, reason } = event.payload; diff --git a/packages/protocol-ts/schema/crew.schema.json b/packages/protocol-ts/schema/crew.schema.json index b8dde0df..d0517fea 100644 --- a/packages/protocol-ts/schema/crew.schema.json +++ b/packages/protocol-ts/schema/crew.schema.json @@ -1980,6 +1980,45 @@ ], "additionalProperties": false }, + { + "description": "A TUI adapter's readiness poll recognized a vendor first-run gate\nblocking the run (workspace trust, a theme picker, sign-in, ...).\nJournaled once, immediately before the paired\n`escalationRaised { reason: \"vendorFirstRunGate\" }` -- `kind` is a\nclosed enum, never captured PTY text, so this event carries no\nredaction obligation of its own.", + "type": "object", + "properties": { + "type": { + "type": "string", + "const": "firstRunGateDetected" + }, + "payload": { + "type": "object", + "properties": { + "runId": { + "$ref": "#/$defs/RunId" + }, + "taskId": { + "$ref": "#/$defs/TaskId" + }, + "workerId": { + "$ref": "#/$defs/WorkerId" + }, + "kind": { + "$ref": "#/$defs/FirstRunGateKind" + } + }, + "additionalProperties": false, + "required": [ + "runId", + "taskId", + "workerId", + "kind" + ] + } + }, + "required": [ + "type", + "payload" + ], + "additionalProperties": false + }, { "description": "A worker escalated a blocking condition to its leader or a human\noperator. `reason` is a plain, machine-assigned code (never raw\nworker content); `question` has already crossed the redaction\nboundary the same way `workerQuestion`'s field has.", "type": "object", @@ -3050,6 +3089,17 @@ "writes" ] }, + "FirstRunGateKind": { + "description": "Which recognized vendor first-run gate blocked a run at readiness.\nClosed set, never derived from captured PTY text, so a\n`firstRunGateDetected` event carries no redaction obligation of its\nown (see that event's doc comment).", + "type": "string", + "enum": [ + "claudeWorkspaceTrust", + "claudeThemePicker", + "claudeSignIn", + "codexDirectoryTrust", + "codexSignIn" + ] + }, "AnsweredBy": { "description": "Who answered an `escalationRaised` escalation.", "type": "string", diff --git a/packages/protocol-ts/src/generated/FirstRunGateKind.ts b/packages/protocol-ts/src/generated/FirstRunGateKind.ts new file mode 100644 index 00000000..d17b1da2 --- /dev/null +++ b/packages/protocol-ts/src/generated/FirstRunGateKind.ts @@ -0,0 +1,9 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Which recognized vendor first-run gate blocked a run at readiness. + * Closed set, never derived from captured PTY text, so a + * `firstRunGateDetected` event carries no redaction obligation of its + * own (see that event's doc comment). + */ +export type FirstRunGateKind = "claudeWorkspaceTrust" | "claudeThemePicker" | "claudeSignIn" | "codexDirectoryTrust" | "codexSignIn"; diff --git a/packages/protocol-ts/src/generated/RuntimeEvent.ts b/packages/protocol-ts/src/generated/RuntimeEvent.ts index 0c84d9d9..02004e64 100644 --- a/packages/protocol-ts/src/generated/RuntimeEvent.ts +++ b/packages/protocol-ts/src/generated/RuntimeEvent.ts @@ -6,6 +6,7 @@ import type { DecidedBy } from "./DecidedBy"; import type { DiagnosticLevel } from "./DiagnosticLevel"; import type { DisplayBackend } from "./DisplayBackend"; import type { DisplayPlacement } from "./DisplayPlacement"; +import type { FirstRunGateKind } from "./FirstRunGateKind"; import type { MessageId } from "./MessageId"; import type { PlanSpec } from "./PlanSpec"; import type { Redacted } from "./Redacted"; @@ -76,7 +77,7 @@ reason: Redacted, } } | { "type": "planProposed", "payload": { runId: RunId, tas /** * `null` when no rationale was given for the decision. */ -reason: Redacted | null, } } | { "type": "workerQuestion", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, question: Redacted | null, } } | { "type": "escalationRaised", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, +reason: Redacted | null, } } | { "type": "workerQuestion", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, question: Redacted | null, } } | { "type": "firstRunGateDetected", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, kind: FirstRunGateKind, } } | { "type": "escalationRaised", "payload": { runId: RunId, taskId: TaskId, workerId: WorkerId, /** * Why the run was escalated, as a short machine-assigned code from * a closed set -- never a worker's or a caller's own words. diff --git a/packages/protocol-ts/src/index.ts b/packages/protocol-ts/src/index.ts index f9c827e0..30beb074 100644 --- a/packages/protocol-ts/src/index.ts +++ b/packages/protocol-ts/src/index.ts @@ -36,6 +36,7 @@ export type * from "./generated/DisplayPlacement"; export type * from "./generated/DisplayStatus"; export type * from "./generated/EventEnvelope"; export type * from "./generated/EventSource"; +export type * from "./generated/FirstRunGateKind"; export type * from "./generated/InitializeParams"; export type * from "./generated/InitializeResult"; export type * from "./generated/InspectRequest"; From 54ff6e05c9f7835cba2c690a3aed882caec7ac22 Mon Sep 17 00:00:00 2001 From: Nikolas Demiridis Date: Wed, 9 Sep 2026 23:37:47 +0300 Subject: [PATCH 2/3] fix(runtime): make a run parked at a first-run gate cancellable The previous commit's gate-escalation poll waits on no deadline of its own, holding self.run's async lock for as long as it runs (wait_for_readiness is called with that lock already acquired and held for run_pipeline's whole duration). cancel()'s CancelScope::Worker/Subtree branch, and dispose(), both open by acquiring that same lock before they can send on the run's terminate_tx -- so a run parked at a gate could never actually be cancelled: the poll waits for an exit that only cancel can cause, and cancel waits for the poll. Under the previous bounded readiness_cap wait this was invisible; the new unbounded gate-park wait turns it into a genuine, permanent deadlock. Fixed with a lock-free CancellationToken, not a change to the lock itself: TuiAdapter grows a cancel_token field (a std::sync::Mutex around a tokio_util::sync::CancellationToken, never held across an await). start() and resume() each swap in a fresh, unfired token before entering the pipeline -- never a previous run's, so an earlier run's cancellation can never poison the next one's -- and thread it down through resume_from/run_pipeline into wait_for_readiness. cancel() and dispose() both fire the token first, before either touches self.run's lock; firing needs no lock of its own and always succeeds, and every existing path after that point is unchanged. The gate-park select! (and, for the same reason, the pre-existing bounded Undecided-branch select!) gains one new arm racing cancel_token.cancelled() alongside the existing pty.exit_watcher() arm. Test: a run parked at a gate that never clears on its own, cancelled from outside, ends well within a 2s outer timeout rather than hanging toward its 30s pty sleep or 30s readiness cap. Verified as a genuine regression control, not just a passing test: temporarily removed the new select arm and confirmed the test fails by timeout (not by a wrong assertion) before restoring it. cargo fmt --all, cargo clippy --all-targets --all-features -D warnings, and CREW_DISABLE_VENDOR_CLI=1 cargo test --workspace (69 test binaries, all passing) verified clean. --- crates/runtime/src/adapter/tui/adapter.rs | 194 ++++++++++++++++++++++ 1 file changed, 194 insertions(+) diff --git a/crates/runtime/src/adapter/tui/adapter.rs b/crates/runtime/src/adapter/tui/adapter.rs index 6da60ab1..873ec8d4 100644 --- a/crates/runtime/src/adapter/tui/adapter.rs +++ b/crates/runtime/src/adapter/tui/adapter.rs @@ -29,6 +29,7 @@ use std::time::{Duration, SystemTime}; use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc, oneshot}; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use uuid::Uuid; use crew_protocol::{ @@ -769,6 +770,16 @@ pub struct TuiAdapter { /// through a different seam). resume: ResumeContext, run: AsyncMutex>, + /// Fired by `cancel`/`dispose` to interrupt a run parked in an + /// unbounded wait -- today, `wait_for_readiness`'s gate-escalation + /// poll -- without needing `self.run`'s async lock, which that poll + /// holds for its entire (potentially unbounded) duration. Replaced + /// with a fresh token at the top of every `start`/`resume` (never + /// reused across runs, so a previous run's cancellation can never + /// poison the next one's). Held in a `std::sync::Mutex`, never across + /// an `.await` -- see `cancel`'s own comment for why firing it must + /// never itself block on `self.run`. + cancel_token: StdMutex, } impl TuiAdapter { @@ -804,9 +815,23 @@ impl TuiAdapter { timings, resume, run: AsyncMutex::new(None), + cancel_token: StdMutex::new(CancellationToken::new()), } } + /// Swaps in a fresh, unfired token for a new run and returns a clone + /// to thread down into this run's own pipeline -- never the same + /// token an earlier run may have already cancelled. + fn arm_cancel_token(&self) -> CancellationToken { + let fresh = CancellationToken::new(); + let mut slot = self + .cancel_token + .lock() + .expect("cancel-token mutex never poisoned"); + *slot = fresh.clone(); + fresh + } + fn kind(&self) -> &'static str { self.vendor.kind() } @@ -843,6 +868,7 @@ impl TuiAdapter { worker_id: WorkerId, session: &VendorSessionRef, sink: Arc, + cancel_token: CancellationToken, ) -> Result<(), AdapterError> { let placeholder = StartSpec { run_id, @@ -901,6 +927,7 @@ impl TuiAdapter { Some(transcript_path), tail_from, sink, + cancel_token, ) .await } @@ -939,6 +966,7 @@ impl TuiAdapter { // re-journal) events an earlier run already committed. tail_from: Cursor, sink: Arc, + cancel_token: CancellationToken, ) -> Result<(), AdapterError> { let started_at = SystemTime::now(); let pty = Arc::new( @@ -1119,6 +1147,7 @@ impl TuiAdapter { task_id, worker_id, &sink, + &cancel_token, ) .await { @@ -1699,6 +1728,7 @@ async fn wait_for_readiness( task_id: TaskId, worker_id: WorkerId, sink: &Arc, + cancel_token: &CancellationToken, ) -> Result<(), AdapterError> { let deadline = tokio::time::Instant::now() + cap; // Set once this poll journals `FirstRunGateDetected` + @@ -1804,6 +1834,16 @@ async fn wait_for_readiness( ), }); } + _ = cancel_token.cancelled() => { + return Err(AdapterError::process( + kind, + "start", + format!( + "the run was cancelled while a first-run gate was blocking it \ + ({gate:?})" + ), + )); + } _ = tokio::time::timeout(quiet, rx.recv()) => {} } continue; @@ -1857,6 +1897,13 @@ async fn wait_for_readiness( ), }); } + _ = cancel_token.cancelled() => { + return Err(AdapterError::process( + kind, + "start", + "the run was cancelled while waiting for readiness", + )); + } // A per-tick receive timeout, or any `rx` outcome // other than a still-hypothetical channel close, is a // pacing wake-up, not itself a failure: it means no @@ -2179,6 +2226,10 @@ impl Adapter for TuiAdapter { // and tailing picks up from the stored position. Fresh ids on // the spec are the correlation (the registry binds the same // run/task/worker into this adapter at construction). + // A fresh, unfired token for this run, before either branch + // below -- never the previous run's (already-possibly-fired) + // token. See `arm_cancel_token`'s own comment. + let cancel_token = self.arm_cancel_token(); if let Some(session) = spec.resume { return self .resume_from( @@ -2188,6 +2239,7 @@ impl Adapter for TuiAdapter { spec.worker_id, &session, sink, + cancel_token, ) .await; } @@ -2210,6 +2262,7 @@ impl Adapter for TuiAdapter { None, Cursor::start(), sink, + cancel_token, ) .await }) @@ -2230,6 +2283,7 @@ impl Adapter for TuiAdapter { "adapter already has an active run", )); } + let cancel_token = self.arm_cancel_token(); // No `StartSpec` carries this adapter's real ids across // `Adapter::resume`'s signature; `self.run_id`/`task_id`/ // `worker_id` (bound at construction, see `TuiAdapter`'s own @@ -2243,6 +2297,7 @@ impl Adapter for TuiAdapter { self.worker_id, &session, sink, + cancel_token, ) .await }) @@ -2349,6 +2404,33 @@ impl Adapter for TuiAdapter { fn cancel(&self, scope: CancelScope) -> AdapterFuture<'_, ()> { Box::pin(async move { + // Fired first, before either branch below touches `self.run`'s + // async lock: a run parked in `wait_for_readiness`'s + // gate-escalation poll holds that lock for as long as the + // poll runs (unbounded, by design -- see that function's own + // comment), so a `cancel` that itself waited on the lock + // first could never reach a run stuck there at all. Firing + // the token needs no lock of its own and always succeeds. + // + // Firing unconditionally, for every `scope` including `Turn`, + // is deliberate and widens `Turn`'s ordinary contract + // ("interrupt the current turn, the worker survives") in + // exactly one window: while parked at a first-run gate there + // is no turn in progress to interrupt, so the only coherent + // reading of "stop" at that moment is to stop waiting -- the + // poll's cancel arm fails the start, naming the gate, rather + // than resuming as a no-op `Turn` would everywhere else. The + // alternative -- scoping the fire to `Worker`/`Subtree` only + // -- was considered and rejected: `CancelScope::Turn` would + // then reach the `match` below and block on `self.run`'s lock + // against the very poll this fix exists to make interruptible, + // so a `Turn` cancel during a gate park would simply hang + // forever. A start that fails with an accurate reason is the + // correct trade against a caller that never returns. + self.cancel_token + .lock() + .expect("cancel-token mutex never poisoned") + .cancel(); match scope { CancelScope::Turn => { let guard = self.run.lock().await; @@ -2415,6 +2497,13 @@ impl Adapter for TuiAdapter { fn dispose(&self) -> AdapterFuture<'_, ()> { Box::pin(async move { + // See `cancel`'s own comment: the same lock-ordering hazard + // applies here (`dispose` also opens with `self.run`'s async + // lock), so the token is fired first on the same terms. + self.cancel_token + .lock() + .expect("cancel-token mutex never poisoned") + .cancel(); let run = { let mut guard = self.run.lock().await; guard.take() @@ -3161,6 +3250,7 @@ mod tests { let vendor = ClaudeTuiVendor::new(PathBuf::from("/w"), vec![]); let sink = GateRecordingSink::new(); let dyn_sink: Arc = sink.clone(); + let cancel_token = CancellationToken::new(); let run_id = RunId::new(); let task_id = TaskId::new(); let worker_id = WorkerId::new(); @@ -3184,6 +3274,7 @@ mod tests { task_id, worker_id, &dyn_sink, + &cancel_token, ) .await }); @@ -3272,6 +3363,7 @@ mod tests { let vendor = ClaudeTuiVendor::new(PathBuf::from("/w"), vec![]); let sink: Arc = GateRecordingSink::new(); + let cancel_token = CancellationToken::new(); let started = tokio::time::Instant::now(); let result = wait_for_readiness( &mut readiness_rx, @@ -3290,6 +3382,7 @@ mod tests { TaskId::new(), WorkerId::new(), &sink, + &cancel_token, ) .await; let elapsed = started.elapsed(); @@ -3354,6 +3447,7 @@ mod tests { let vendor = ClaudeTuiVendor::new(PathBuf::from("/w"), vec![]); let sink: Arc = GateRecordingSink::new(); + let cancel_token = CancellationToken::new(); let started = tokio::time::Instant::now(); // A long cap: if the closed-channel case were mishandled as a // pacing tick, this test would either hang here or take @@ -3378,6 +3472,7 @@ mod tests { TaskId::new(), WorkerId::new(), &sink, + &cancel_token, ), ) .await @@ -3444,6 +3539,7 @@ mod tests { }; let sink = GateRecordingSink::new(); let dyn_sink: Arc = sink.clone(); + let cancel_token = CancellationToken::new(); let result = wait_for_readiness( &mut readiness_rx, @@ -3459,6 +3555,7 @@ mod tests { TaskId::new(), WorkerId::new(), &dyn_sink, + &cancel_token, ) .await; @@ -3476,6 +3573,103 @@ mod tests { } } + /// Requirement (d): a run parked at a gate is cancellable. This is + /// the regression control for the deadlock item 4 fixes -- without + /// the `cancel_token.cancelled()` arm in the gate-park `select!`, + /// this vendor's gate never clears on its own (`clear_after` is never + /// reached) and the PTY sleeps far longer than this test's own outer + /// timeout, so removing that arm fails this test by timeout, not by + /// a wrong assertion. + #[tokio::test] + async fn wait_for_readiness_ends_promptly_when_cancelled_while_parked_at_a_gate() { + let pty = Arc::new( + PtyProcess::spawn( + &crate::supervisor::SpawnSpec { + program: PathBuf::from("/bin/sh"), + args: vec!["-c".to_string(), "printf hi && sleep 30".to_string()], + ..crate::supervisor::SpawnSpec::minimal() + }, + EscalationTimings::default(), + ) + .expect("spawn the parked double"), + ); + + let mut readiness_rx = pty.subscribe_output(); + let grid = Arc::new(StdMutex::new(TerminalGrid::new())); + { + let mut grid_rx = pty.subscribe_output(); + let grid = Arc::clone(&grid); + tokio::spawn(async move { + loop { + match grid_rx.recv().await { + Ok(bytes) => grid + .lock() + .expect("terminal-grid mutex never poisoned") + .push(&bytes), + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(_) => break, + } + } + }); + } + + let vendor = CountingGateVendor { + gate: GateKind::ClaudeSignIn, + clear_after: u32::MAX, + calls: AtomicU32::new(0), + }; + let sink = GateRecordingSink::new(); + let dyn_sink: Arc = sink.clone(); + let cancel_token = CancellationToken::new(); + let cancel_token_for_wait = cancel_token.clone(); + let pty_for_wait = Arc::clone(&pty); + + let wait = tokio::spawn(async move { + wait_for_readiness( + &mut readiness_rx, + "counting-gate", + &vendor, + &grid, + Duration::from_millis(15), + Duration::from_secs(30), + &pty_for_wait, + None, + tokio::time::Instant::now(), + RunId::new(), + TaskId::new(), + WorkerId::new(), + &dyn_sink, + &cancel_token_for_wait, + ) + .await + }); + + assert!( + wait_until(|| !sink.payloads().is_empty(), Duration::from_secs(5)).await, + "the gate must be escalated before this test cancels it" + ); + + cancel_token.cancel(); + + let result = tokio::time::timeout(Duration::from_secs(2), wait) + .await + .expect( + "cancelling a parked gate must end the poll well within 2s, not hang toward \ + the 30s pty sleep or the 30s readiness cap", + ) + .expect("wait_for_readiness task must not panic"); + + let _ = pty.terminate().await; + + let message = result + .expect_err("a cancelled gate-park must never resolve as ready") + .to_string(); + assert!( + message.contains("cancelled"), + "the error must say the run was cancelled: {message}" + ); + } + // ----------------------------------------------- the bracketed-paste invariant /// Prompt text is always delivered to a vendor TUI as a bracketed From 17df2919dc9833c9d64c6d48a199c1349fb481e8 Mon Sep 17 00:00:00 2001 From: "crew-bot[bot]" <322634903+crew-bot[bot]@users.noreply.github.com> Date: Wed, 9 Sep 2026 21:03:48 +0000 Subject: [PATCH 3/3] chore(extension): refresh bundle from linux-x64 CI Build: https://github.com/nikolasd/crew/actions/runs/34404712589 Source: ed7e2f4022f63b7fb5f44d47d6183b15cddcf287 --- packages/extension/dist/index.js | 62 ++++++++++++++++++++++++++++++-- 1 file changed, 60 insertions(+), 2 deletions(-) diff --git a/packages/extension/dist/index.js b/packages/extension/dist/index.js index bac51119..c71c5d8a 100644 --- a/packages/extension/dist/index.js +++ b/packages/extension/dist/index.js @@ -8990,6 +8990,50 @@ a listener should act on; this is operator-facing detail only.`, ], additionalProperties: false }, + { + description: `A TUI adapter's readiness poll recognized a vendor first-run gate +blocking the run (workspace trust, a theme picker, sign-in, ...). +Journaled once, immediately before the paired +\`escalationRaised { reason: "vendorFirstRunGate" }\` -- \`kind\` is a +closed enum, never captured PTY text, so this event carries no +redaction obligation of its own.`, + type: "object", + properties: { + type: { + type: "string", + const: "firstRunGateDetected" + }, + payload: { + type: "object", + properties: { + runId: { + $ref: "#/$defs/RunId" + }, + taskId: { + $ref: "#/$defs/TaskId" + }, + workerId: { + $ref: "#/$defs/WorkerId" + }, + kind: { + $ref: "#/$defs/FirstRunGateKind" + } + }, + additionalProperties: false, + required: [ + "runId", + "taskId", + "workerId", + "kind" + ] + } + }, + required: [ + "type", + "payload" + ], + additionalProperties: false + }, { description: "A worker escalated a blocking condition to its leader or a human\noperator. `reason` is a plain, machine-assigned code (never raw\nworker content); `question` has already crossed the redaction\nboundary the same way `workerQuestion`'s field has.", type: "object", @@ -10111,6 +10155,20 @@ new window.`, "writes" ] }, + FirstRunGateKind: { + description: `Which recognized vendor first-run gate blocked a run at readiness. +Closed set, never derived from captured PTY text, so a +\`firstRunGateDetected\` event carries no redaction obligation of its +own (see that event's doc comment).`, + type: "string", + enum: [ + "claudeWorkspaceTrust", + "claudeThemePicker", + "claudeSignIn", + "codexDirectoryTrust", + "codexSignIn" + ] + }, AnsweredBy: { description: "Who answered an `escalationRaised` escalation.", type: "string", @@ -13639,8 +13697,8 @@ function formatDigest(e, lookup) { case "budgetExceeded": return `${capitalize(who)} exceeded its turn budget. Escalate to the user or raise the budget via the plan.`; case "escalationRaised": { - const reason = event.payload.reason; - return `Escalation raised on ${who}: ${reason}.`; + const { reason, question } = event.payload; + return question ? `Escalation raised on ${who}: ${reason}. ${question}` : `Escalation raised on ${who}: ${reason}.`; } case "paneDowngraded": { const { requestedBackend, requestedPlacement, actualBackend, reason } = event.payload;