diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b63191..1226bf6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,13 @@ Versioning and Keep a Changelog conventions. ### Added +- Contain observer panic-payload cleanup failures and disable failed observers + across runtime clones. Report event-delivery wait separately from observed + first-text latency, preserving bounded backpressure. +- Opt-in payload-free startup timing observers for validation, concurrency admission, + sandbox preparation, process spawn, first output/text and terminal outcomes. + Existing process lifetime and wire events are unchanged. + - Bidirectional OpenCode support through `opencode serve`. `OpenCodeTurnMode` selects the transport; `OpenCode::serve()` starts the server on a reserved loopback port and drives it over HTTP and Server-Sent Events, adding live diff --git a/Cargo.toml b/Cargo.toml index a774545..6b5bfba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -127,3 +127,7 @@ needless_pass_by_value = "allow" return_self_not_must_use = "allow" struct_excessive_bools = "allow" too_many_lines = "allow" + +[[example]] +name = "startup_timings" +required-features = ["claude", "codex"] diff --git a/README.md b/README.md index cbd3d75..6520a6e 100644 --- a/README.md +++ b/README.md @@ -255,3 +255,5 @@ to separate local verification from publishing and hosting decisions. Licensed under either the Apache License, Version 2.0 or the MIT License, at your option. + +- [Measure session startup](docs/how-to/measure-session-startup.md) diff --git a/docs/adr/0004-prepared-provider-processes.md b/docs/adr/0004-prepared-provider-processes.md new file mode 100644 index 0000000..71f2370 --- /dev/null +++ b/docs/adr/0004-prepared-provider-processes.md @@ -0,0 +1,110 @@ +# ADR 0004: Prepared provider processes + +- Status: Proposed +- Date: 2026-09-23 + +## Problem + +`InProcessRuntimeClient` retains logical session identity, not the provider process. +The built-in executor reports `retained_process: false` and calls `AgentRuntime::run` +for each invocation. Both Claude and Codex therefore pay process startup again on +subsequent turns. Calling `acquire` earlier cannot remove this cost. + +The first implementation step is stage timing so an embedder can distinguish +validation, concurrency wait, sandbox setup, process spawn, protocol output and +first assistant text. A first output frame is not evidence of provider readiness +or model request submission. Provider-specific readiness requires an explicit +handshake acknowledgement, not a sleep or an empty model turn. + +## Proposed ownership + +The SDK owns a bounded process supervisor and provider protocol state. Fleet owns +when a user has selected enough configuration to prepare, durable conversation +records, authorization, feature rollout, and presentation. Listing projects or +conversations must never spawn provider processes. + +Existing `AgentRuntime::run` and retained-client constructors preserve lazy, +one-process-per-turn behavior. A separate opt-in client configuration enables +retained processes. Unsupported adapters and remote protocol versions report a +typed capability error rather than silently claiming preparation succeeded. + +## Proposed lifecycle + +1. Acquire a logical runtime with project, provider, sandbox and launch settings. +2. Explicitly prepare it, without a prompt or invocation identifier. This starts + the process and completes the provider handshake; it does not call a model, + create a synthetic transcript message, or grant tool execution. +3. Send a turn. Lazy sending performs preparation automatically. Sending during + preparation joins the same bounded operation and submits exactly once. +4. On a successful turn, keep the provider connection and continue draining its + bounded event stream. Idle tool/background events belong to the runtime and + must not be attached to the next invocation. +5. Dispose or expire the idle process, confirming process-tree teardown. Preserve + session identity so later work can explicitly resume from provider persistence. + +The process lifecycle distinguishes unprepared, queued, preparing, ready, busy, +failed, stopping and stopped. A logical runtime being acquired is not provider +readiness. Preparation errors expose delivery=not_sent. Failure after submission +preserves the existing accepted/possibly_sent semantics and must not replay a +prompt automatically. + +Preparation has a configurable deadline, bounded concurrency, global process +capacity and idle expiration. Active turns cannot be evicted to admit speculative +preparation. Abandoned preparations release their capacity. Disposing while +preparing prevents late successful readiness from resurrecting the runtime. + +## Configuration and credentials + +Use one launch-configuration representation for preparation and sending. Fleet +currently provides MCP definitions, capability credentials, system instructions, +allowed tools and tailnet environment at turn time. Preparing without these and +then spawning again on send would provide no benefit and could misconfigure the +sandbox. Fleet must resolve that context before requesting preparation. + +Never share prepared processes across users, projects, sandboxes or credential +contexts. Working-directory, environment, MCP, sandbox revision and credential +changes require disposal/repreparation unless the driver has a tested live-update +operation. Compare full launch configuration without logging secret values. +Runtime and per-turn settings must have explicit precedence. + +Fleet's MCP capability lifetime must be reconciled with idle process retention: +retaining a process must neither keep a revoked turn token valid nor silently +broaden it to an unbounded credential. Use scoped runtime credentials with explicit +revocation, or safely replace the child when credentials rotate. This contract is +a prerequisite to Fleet enabling retention. + +## Provider implementation + +Claude keeps a streaming input channel open and separates initialization from user +messages. Codex keeps one app-server connection and thread alive, issues one +initialize handshake per process, and starts later turns on that connection. Both +need bounded background draining, crash detection, cancellation, late-event +isolation and cleanup. One-shot Codex exec and unsupported providers remain lazy. + +## Fleet adoption + +1. Upgrade to the verified Windows baseline. +2. Collect startup stage timings without changing process behavior. +3. Add and verify retained-process drivers and preparation in the SDK. +4. Add a feature-gated Fleet preparation endpoint using the generated API client. +5. Trigger preparation only after explicit project/provider selection; display + queued/running/ready/failed feedback while the composer stays usable. +6. Measure cold/prepared first-turn and subsequent-turn latency, process count, + idle memory, abandoned preparation cost and failure rate before wider rollout. + +The dependency upgrade alone must not start extra processes. Existing sessions, +legacy adapters, remote execution and unattended work retain their behavior until +that path explicitly opts in and has equivalent lifecycle coverage. + +## Acceptance evidence + +- No prompt or model request during preparation; one process for prepare+send. +- Concurrent prepare/send is deduplicated; exactly one submitted invocation. +- Two turns reuse the same native PID and preserve provider session identity. +- Sandbox/MCP/environment/credential changes never reuse a stale process. +- Deadline, cancellation, disposal, crash and failed initialization release slots. +- Capacity remains bounded; idle expiration does not interrupt active work. +- Background events are drained and never misattributed to a later turn. +- Existing non-opt-in callers and all provider permission tests continue to pass. +- Native Windows/macOS/Linux lifecycle tests, plus authenticated Claude/Codex + smoke measurements with an isolated workspace before enabling Fleet by default. diff --git a/docs/how-to/measure-session-startup.md b/docs/how-to/measure-session-startup.md new file mode 100644 index 0000000..56e8626 --- /dev/null +++ b/docs/how-to/measure-session-startup.md @@ -0,0 +1,132 @@ +# Measure session startup + +Enable an optional observer when constructing the runtime to find where a turn +spends time. This does not change process lifetime, add model requests, or change +provider events or the remote protocol. + +```rust +use std::sync::Arc; +use temps_agent_runtime::{AgentRuntime, StartupObserver, StartupTiming}; + +struct Timings; +impl StartupObserver for Timings { + fn observe(&self, timing: StartupTiming) { + // For production exporters, use a bounded channel's try_send here. + // Keep this callback quick: it runs on the execution task. + eprintln!("{} {:?} {:?} elapsed={:?} event_delivery={:?}", + timing.observation_id, timing.provider, timing.stage, + timing.elapsed, timing.event_delivery_elapsed); + } +} + +let runtime = AgentRuntime::builder() + .startup_observer(Arc::new(Timings)) + .build()?; +# Ok::<(), temps_agent_runtime::RuntimeError>(()) +``` + +Samples contain only a process-local observation ID, provider, stage and monotonic +elapsed duration plus cumulative event-delivery wait. No prompt, path, provider session ID, account, environment value +or error diagnostic is included. IDs distinguish concurrent runs, but do not +survive a host restart and are not durable invocation identifiers. Treat observers +as trusted: timing metadata can still reveal provider choice and activity. Do not +expose a shared observer stream across tenants. + +For `run`, elapsed time begins before validation. Subtract consecutive timestamps +to locate waits: + +- `WaitingForPermit` to `PermitAcquired`: process concurrency queue. +- `CommandPrepared` to `SandboxPrepared`: sandbox preparation, when configured. +- The last preparation boundary to `ProcessSpawned`: transport process launch. +- `ProcessSpawned` to `InitialInputWritten`: initial stdin flush, when present. +- `StreamsAttached` to `FirstOutput`: waiting for the first provider output line. +- `FirstOutput` to `FirstText`: time until the runtime parses its first nonempty + assistant text. This includes waits while delivering earlier events to the + application's consumer, not just protocol initialization and provider/model work. + +Each sample includes `event_delivery_elapsed`: cumulative time spent awaiting +`EventSink::emit` before that boundary, including sink persistence, queue waits +and scheduler delay during delivery. Subtract the change in this field from the +change in `elapsed` between two samples to separate direct delivery waits. A +consumer that spends five seconds persisting a session event can otherwise make +already-buffered text appear five seconds late. Interrupted delivery is counted +when its future is dropped, including on timeout. + +The remainder is still not pure model latency: parsing, provider protocol, +observer and scheduler overhead remain. Backpressure may also indirectly delay +the provider itself. The SDK deliberately keeps its existing bounded read and +event-delivery loop; it does not add an unbounded background reader to manufacture +an arrival timestamp. + +`FirstOutput` may be a handshake, warning or error. `InitialInputWritten` means +bytes were flushed, not that a prompt was accepted. `StreamsAttached` is not +universal provider readiness. Claude and Codex may perform initialization after +these boundaries. Only the first nonempty normalized text delta triggers +`FirstText`, and a tool-only or failed turn may never reach it. These observations +do not claim to separate MCP readiness from model latency yet. + +A healthy observer receives one terminal observation for each polled run: `Succeeded`, `Failed`, +`Cancelled`, `TimedOut`, or `Abandoned` when its future is dropped. Observation +callbacks are synchronous and must not block. Use bounded export with dropped +samples under overload; observer panics are contained under unwinding builds +(the standard panic hook can still report them). A panicking observer is disabled +for that runtime and its clones, including later turns; callbacks already in +flight may finish. A callback is never invoked during stack unwinding, so an +abandoned run may have no terminal observation when its caller panics. + +Panic-payload cleanup runs inside a second containment boundary. If cleanup also +panics, its secondary payload is intentionally retained rather than dropped again: +this avoids recursively unwinding the host. Further calls are disabled, limiting +this exceptional leak to callbacks already in flight at failure. With a disabled +observer, telemetry can be incomplete while provider execution continues. As with +other Rust panics, `panic=abort` terminates the process. + +For `run_with_sandbox_recovery`, each provider attempt receives a separate +observation ID after sandbox policy resolution and concurrency admission. Its +elapsed time excludes those outer steps. Attempts can therefore be compared for +process startup, but should not be used as whole-operation latency. + +No observer is installed by default. Normalized provider events and invocation +journals stay unchanged. Built-in retained clients still report +`retained_process: false`; keeping a logical session does not avoid process +startup. See [the preparation proposal](../adr/0004-prepared-provider-processes.md) +for the opt-in process-retention work needed before early preparation can help. + +## Optional live baseline + +With an authenticated CLI installed, run: + +```sh +cargo run --example startup_timings -- claude +cargo run --example startup_timings -- codex +``` + +These commands send two small real prompts and can incur model usage. They use a +temporary working directory, deny interactive approvals through the default +handler, and print timing metadata only. Provider-native session persistence can +still write to the CLI's normal home directory. The second turn resumes the first +session; the current driver still spawns a second process. Neither command +measures a prepared process, and the result is specific to the configured model, +provider account, network and host load. + +## Initial live sample (2026-09-23) + +One pair per provider on macOS, with normal installed CLI configuration, an +isolated temporary workspace and the prompt from the example. Provider pairs ran +concurrently on the same host, so these are diagnostic samples, not percentiles or +a controlled performance comparison. Compilation time is outside these samples. + +| Provider / turn | Process spawned | First output | First assistant text | +| --- | ---: | ---: | ---: | +| Claude, new session | 4 ms | 666 ms | 5,274 ms | +| Claude, resumed session | 3 ms | 689 ms | 3,548 ms | +| Codex, new session | 6 ms | 122 ms | 5,236 ms | +| Codex, resumed session | 1 ms | 67 ms | 4,719 ms | + +Both resumed turns launched a fresh process. Most observed first-text latency was +after process creation, but these samples cannot separate CLI initialization, MCP +setup, authentication, provider network latency and model computation. They do not +include Fleet's own context resolution, sandbox preparation or UI rendering. A +prepared process is not a promise of an immediate model response. Measure explicit +provider-ready and turn-accepted acknowledgements before assigning the remaining +latency to the model or predicting a speedup. diff --git a/examples/startup_timings.rs b/examples/startup_timings.rs new file mode 100644 index 0000000..e293fad --- /dev/null +++ b/examples/startup_timings.rs @@ -0,0 +1,54 @@ +//! Opt-in live baseline: two turns with provider-native session continuity. +//! Uses the selected CLI's normal authentication and can incur model usage. + +use std::sync::Arc; + +use temps_agent_runtime::{ + AgentRuntime, NoopEventSink, Provider, StartupObserver, StartupTiming, TurnRequest, +}; + +struct PrintTimings; + +impl StartupObserver for PrintTimings { + fn observe(&self, sample: StartupTiming) { + println!( + "run={} provider={:?} stage={:?} elapsed_ms={} event_delivery_ms={}", + sample.observation_id, + sample.provider, + sample.stage, + sample.elapsed.as_millis(), + sample.event_delivery_elapsed.as_millis(), + ); + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let provider = match std::env::args().nth(1).as_deref() { + Some("claude") => Provider::Claude, + Some("codex") => Provider::Codex, + _ => return Err("usage: cargo run --example startup_timings -- ".into()), + }; + let workspace = tempfile::tempdir()?; + let mut builder = AgentRuntime::builder(); + builder.register(temps_agent_runtime::providers::Codex::app_server()); + let runtime = builder.startup_observer(Arc::new(PrintTimings)).build()?; + let mut session = None; + for _ in 0..2 { + let mut request = TurnRequest::new( + provider, + workspace.path(), + "Reply with exactly READY. Do not use tools or modify files.", + ); + request.session_id = session; + request.timeout = std::time::Duration::from_secs(60); + let result = runtime.run(request, &NoopEventSink, None).await?; + session = result.session_id; + if session.is_none() { + return Err( + "provider did not report a resumable session; cannot measure continuation".into(), + ); + } + } + Ok(()) +} diff --git a/src/lib.rs b/src/lib.rs index 67b53f7..44aadaf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,6 +30,7 @@ pub mod retained; mod runtime; mod sandbox; pub mod services; +pub mod startup; pub mod transport; mod types; mod url_security; @@ -104,3 +105,5 @@ pub use types::{ QuestionPrompt, QuestionRequest, RunStatus, SecretString, ToolCallStatus, ToolProcessPolicy, TurnCapabilities, TurnEvent, TurnProvenance, TurnRequest, TurnResult, Usage, }; + +pub use startup::{StartupObserver, StartupStage, StartupTiming}; diff --git a/src/runtime.rs b/src/runtime.rs index 9146764..5a1cadb 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -8,6 +8,7 @@ use tokio::sync::Semaphore; use crate::adapter::{AdapterState, AgentAdapter, CommandSpec, InteractionRequest}; use crate::error::classify_provider_failure; +use crate::startup::{StartupObserver, StartupObserverState, StartupStage, StartupTrace}; use crate::{ AccountUsageReport, DenyAll, EventSink, ExecutionTransport, HarnessAuthentication, HarnessCatalogError, HarnessCatalogErrorKind, HarnessCatalogStatus, HarnessControlGroup, @@ -329,6 +330,7 @@ pub struct AgentRuntimeBuilder { concurrency_limit: usize, max_prompt_bytes: usize, max_event_line_bytes: usize, + startup_observer: Option>, } impl AgentRuntimeBuilder { @@ -341,6 +343,7 @@ impl AgentRuntimeBuilder { concurrency_limit: 2, max_prompt_bytes: DEFAULT_MAX_PROMPT_BYTES, max_event_line_bytes: DEFAULT_MAX_EVENT_LINE_BYTES, + startup_observer: None, }; #[cfg(feature = "claude")] builder.register(crate::providers::Claude::default()); @@ -387,6 +390,16 @@ impl AgentRuntimeBuilder { self } + /// Observe payload-free startup boundaries. Disabled by default. + /// + /// Callbacks must return promptly; see [`StartupObserver`]. Observations + /// do not add provider events, change the remote protocol, or start extra + /// processes. A resumed session still starts a fresh process in this driver. + pub fn startup_observer(mut self, observer: Arc) -> Self { + self.startup_observer = Some(Arc::new(StartupObserverState::new(observer))); + self + } + /// Validate limits and construct the runtime. pub fn build(self) -> Result { if self.concurrency_limit == 0 { @@ -407,6 +420,7 @@ impl AgentRuntimeBuilder { permits: Arc::new(Semaphore::new(self.concurrency_limit)), max_prompt_bytes: self.max_prompt_bytes, max_event_line_bytes: self.max_event_line_bytes, + startup_observer: self.startup_observer, }) } } @@ -723,6 +737,7 @@ pub struct AgentRuntime { permits: Arc, max_prompt_bytes: usize, max_event_line_bytes: usize, + startup_observer: Option>, } impl AgentRuntime { @@ -1896,6 +1911,19 @@ impl AgentRuntime { request: TurnRequest, events: &dyn EventSink, interactions: Option<&dyn InteractionHandler>, + ) -> Result { + let mut trace = StartupTrace::new(request.provider, self.startup_observer.clone()); + let result = self.run_inner(request, events, interactions, &trace).await; + trace.finish(&result); + result + } + + async fn run_inner( + &self, + request: TurnRequest, + events: &dyn EventSink, + interactions: Option<&dyn InteractionHandler>, + trace: &StartupTrace, ) -> Result { self.validate(&request)?; self.validate_working_directory(&request).await?; @@ -1941,21 +1969,30 @@ impl AgentRuntime { }); } } + trace.record(StartupStage::Validated); // A pre-cancelled turn must never race an immediately available permit // into spawning a provider executable. if request.cancellation.is_cancelled() { return Err(RuntimeError::Cancelled { provider }); } + trace.record(StartupStage::WaitingForPermit); let permit = tokio::select! { _ = request.cancellation.cancelled() => { return Err(RuntimeError::Cancelled { provider }); } permit = self.permits.clone().acquire_owned() => permit.map_err(|_| RuntimeError::Cancelled { provider })?, }; + trace.record(StartupStage::PermitAcquired); let timeout = request.timeout; let result = tokio::time::timeout( timeout, - self.run_process(adapter, &request, events, interactions.unwrap_or(&DenyAll)), + self.run_process( + adapter, + &request, + events, + interactions.unwrap_or(&DenyAll), + trace, + ), ) .await; drop(permit); @@ -2078,6 +2115,7 @@ impl AgentRuntime { attempt.sandbox = Some(resolved.as_request()); let tracking_events = SandboxEventSink::new(events, &resolved); let timeout = attempt.timeout; + let mut trace = StartupTrace::new(provider, self.startup_observer.clone()); let result = tokio::time::timeout( timeout, self.run_process( @@ -2085,13 +2123,17 @@ impl AgentRuntime { &attempt, &tracking_events, interactions.unwrap_or(&DenyAll), + &trace, ), ) .await .map_err(|_| RuntimeError::Timeout { provider, seconds: timeout.as_secs(), - })??; + }) + .and_then(std::convert::identity); + trace.finish(&result); + let result = result?; let Some(violation) = tracking_events.violation() else { return Ok(result); }; @@ -2265,6 +2307,7 @@ impl AgentRuntime { request: &TurnRequest, events: &dyn EventSink, interactions: &dyn InteractionHandler, + trace: &StartupTrace, ) -> Result { let provider = request.provider; let mut state = AdapterState::default(); @@ -2282,6 +2325,7 @@ impl AgentRuntime { for (name, value) in &request.environment { spec.environment.insert(name.into(), value.expose().into()); } + trace.record(StartupStage::CommandPrepared); if let Some(sandbox) = &request.sandbox { let context = SandboxContext { provider, @@ -2293,6 +2337,7 @@ impl AgentRuntime { } prepared = sandbox.prepare(context, spec) => prepared?, }; + trace.record(StartupStage::SandboxPrepared); } let capabilities = self.transport.capabilities(); if spec.interactive_stdin && !capabilities.interactive_stdin { @@ -2333,6 +2378,7 @@ impl AgentRuntime { } } })?; + trace.record(StartupStage::ProcessSpawned); let stdin = process .take_stdin() .ok_or_else(|| RuntimeError::ProcessIo { @@ -2368,6 +2414,9 @@ impl AgentRuntime { source, })?; } + if spec.initial_stdin.is_some() { + trace.record(StartupStage::InitialInputWritten); + } if !spec.interactive_stdin { stdin.take(); } @@ -2406,6 +2455,9 @@ impl AgentRuntime { return Err(error); } }; + trace.record(StartupStage::StreamsAttached); + let mut first_output = true; + let mut first_text = true; let mut lines = BufReader::new(reader).lines(); let mut protocol_completed = false; loop { @@ -2429,6 +2481,10 @@ impl AgentRuntime { })?, }; let Some(line) = line else { break }; + if first_output { + first_output = false; + trace.record(StartupStage::FirstOutput); + } if line.len() > self.max_event_line_bytes { let _ = process.terminate().await; stderr_task.abort(); @@ -2439,7 +2495,15 @@ impl AgentRuntime { } let output = adapter.parse_line(&line, &mut state)?; for event in output.events { - events.emit(event).await?; + if first_text && matches!(&event, TurnEvent::TextDelta { text } if !text.is_empty()) + { + first_text = false; + trace.record(StartupStage::FirstText); + } + { + let _delivery = trace.event_delivery(); + events.emit(event).await?; + } } write_provider_frames(provider, stdin.as_mut(), &output.writes, "provider write") .await?; @@ -2643,6 +2707,7 @@ impl AgentRuntime { }); } if state.result.text.is_empty() { + let _delivery = trace.event_delivery(); events .emit(TurnEvent::Warning { message: format!("{provider} completed without a text response"), @@ -4048,6 +4113,292 @@ mod tests { } } + #[derive(Default)] + struct TimingCollector { + samples: Mutex>, + waiting: tokio::sync::Notify, + } + + impl crate::StartupObserver for TimingCollector { + fn observe(&self, sample: crate::StartupTiming) { + self.samples.lock().unwrap().push(sample); + if sample.stage == StartupStage::WaitingForPermit { + self.waiting.notify_one(); + } + } + } + + impl TimingCollector { + fn stages(&self) -> Vec { + self.samples + .lock() + .unwrap() + .iter() + .map(|sample| sample.stage) + .collect() + } + } + + fn observed_runtime( + script: &str, + observer: Arc, + ) -> AgentRuntime { + let mut builder = AgentRuntime::builder() + .startup_observer(observer) + .concurrency_limit(1); + builder.register(ShellAdapter { + script: script.into(), + args: Vec::new(), + }); + builder.build().unwrap() + } + + #[tokio::test] + async fn startup_timings_separate_output_from_text_without_changing_events() { + let samples = Arc::new(TimingCollector::default()); + let runtime = observed_runtime( + r#"printf '%s\n' '{"session":"private-session"}' '{"text":""}' '{"text":"hello"}' '{"text":"world"}' '{"terminal":true}'"#, + samples.clone(), + ); + let directory = tempfile::tempdir().unwrap(); + let request = TurnRequest::new(Provider::Claude, directory.path(), "private prompt"); + let events = CollectEvents::default(); + let result = runtime.run(request, &events, None).await.unwrap(); + assert_eq!(result.text, "helloworld"); + assert_eq!(events.0.lock().unwrap().len(), 4); + assert_eq!( + samples.stages(), + vec![ + StartupStage::Started, + StartupStage::Validated, + StartupStage::WaitingForPermit, + StartupStage::PermitAcquired, + StartupStage::CommandPrepared, + StartupStage::ProcessSpawned, + StartupStage::StreamsAttached, + StartupStage::FirstOutput, + StartupStage::FirstText, + StartupStage::Succeeded, + ] + ); + let recorded = samples.samples.lock().unwrap(); + assert!(recorded + .windows(2) + .all(|pair| pair[0].elapsed <= pair[1].elapsed)); + assert!(recorded + .iter() + .all(|sample| sample.observation_id == recorded[0].observation_id)); + let diagnostic = format!("{recorded:?}"); + for private in [ + "private prompt", + "private-session", + "helloworld", + directory.path().to_str().unwrap(), + ] { + assert!(!diagnostic.contains(private)); + } + } + + #[tokio::test] + async fn startup_timings_separate_event_delivery_from_first_text_latency() { + #[derive(Default)] + struct SlowEvents { + waited: Mutex, + } + #[async_trait] + impl EventSink for SlowEvents { + async fn emit(&self, event: TurnEvent) -> Result<()> { + if matches!(event, TurnEvent::SessionStarted { .. }) { + let start = std::time::Instant::now(); + tokio::time::sleep(Duration::from_millis(40)).await; + *self.waited.lock().unwrap() = start.elapsed(); + } + Ok(()) + } + } + let samples = Arc::new(TimingCollector::default()); + let runtime = observed_runtime( + r#"printf '%s\n' '{"session":"test"}' '{"text":"done"}' '{"terminal":true}'"#, + samples.clone(), + ); + let directory = tempfile::tempdir().unwrap(); + let events = SlowEvents::default(); + let result = runtime + .run( + TurnRequest::new(Provider::Claude, directory.path(), "test"), + &events, + None, + ) + .await + .unwrap(); + assert_eq!(result.text, "done"); + let recorded = samples.samples.lock().unwrap(); + let output = recorded + .iter() + .find(|s| s.stage == StartupStage::FirstOutput) + .unwrap(); + let text = recorded + .iter() + .find(|s| s.stage == StartupStage::FirstText) + .unwrap(); + let actual_wait = *events.waited.lock().unwrap(); + assert!(actual_wait >= Duration::from_millis(40)); + assert!(text.elapsed.checked_sub(output.elapsed).unwrap() >= actual_wait); + assert!( + text.event_delivery_elapsed + .checked_sub(output.event_delivery_elapsed) + .unwrap() + >= actual_wait, + "first-text latency must separately report time spent delivering earlier events" + ); + assert!(text.event_delivery_elapsed <= text.elapsed); + } + + #[tokio::test] + async fn startup_timings_include_terminal_warning_delivery() { + #[derive(Default)] + struct SlowWarning { + waited: Mutex, + } + #[async_trait] + impl EventSink for SlowWarning { + async fn emit(&self, event: TurnEvent) -> Result<()> { + if matches!(event, TurnEvent::Warning { .. }) { + let start = std::time::Instant::now(); + tokio::time::sleep(Duration::from_millis(40)).await; + *self.waited.lock().unwrap() = start.elapsed(); + } + Ok(()) + } + } + let samples = Arc::new(TimingCollector::default()); + let runtime = observed_runtime(r#"printf '%s\n' '{"terminal":true}'"#, samples.clone()); + let directory = tempfile::tempdir().unwrap(); + let events = SlowWarning::default(); + let result = runtime + .run( + TurnRequest::new(Provider::Claude, directory.path(), "test"), + &events, + None, + ) + .await + .unwrap(); + assert!(result.text.is_empty()); + let recorded = samples.samples.lock().unwrap(); + let terminal = recorded.last().unwrap(); + assert_eq!(terminal.stage, StartupStage::Succeeded); + let actual_wait = *events.waited.lock().unwrap(); + assert!(actual_wait >= Duration::from_millis(40)); + assert!( + terminal.event_delivery_elapsed >= actual_wait, + "terminal warning delivery must be included in cumulative sink wait" + ); + assert!(terminal.event_delivery_elapsed <= terminal.elapsed); + } + + #[tokio::test] + async fn startup_timings_distinguish_queue_wait_and_abandoned_execution() { + let samples = Arc::new(TimingCollector::default()); + let runtime = observed_runtime("exit 0", samples.clone()); + let permit = runtime.permits.clone().acquire_owned().await.unwrap(); + let directory = tempfile::tempdir().unwrap(); + let request = TurnRequest::new(Provider::Claude, directory.path(), "test"); + let task = + tokio::spawn( + async move { runtime.run(request, &crate::NoopEventSink, None).await }, + ); + tokio::time::timeout(Duration::from_secs(5), samples.waiting.notified()) + .await + .unwrap(); + assert_eq!( + samples.stages(), + vec![ + StartupStage::Started, + StartupStage::Validated, + StartupStage::WaitingForPermit + ] + ); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + drop(permit); + assert_eq!(samples.stages().last(), Some(&StartupStage::Abandoned)); + assert!(!samples.stages().contains(&StartupStage::ProcessSpawned)); + } + + #[tokio::test] + async fn startup_timings_report_failure_cancellation_and_timeout_once() { + let samples = Arc::new(TimingCollector::default()); + let runtime = observed_runtime("exec sleep 60", samples.clone()); + let directory = tempfile::tempdir().unwrap(); + let cancelled = TurnRequest::new(Provider::Claude, directory.path(), "cancel"); + cancelled.cancellation.cancel(); + assert!(matches!( + runtime.run(cancelled, &crate::NoopEventSink, None).await, + Err(RuntimeError::Cancelled { .. }) + )); + let mut timeout = TurnRequest::new(Provider::Claude, directory.path(), "timeout"); + timeout.timeout = Duration::from_millis(25); + assert!(matches!( + runtime.run(timeout, &crate::NoopEventSink, None).await, + Err(RuntimeError::Timeout { .. }) + )); + let mut invalid = TurnRequest::new(Provider::Claude, directory.path(), "invalid"); + invalid.timeout = Duration::ZERO; + assert!(runtime + .run(invalid, &crate::NoopEventSink, None) + .await + .is_err()); + let samples = samples.samples.lock().unwrap(); + let mut observations = std::collections::BTreeMap::new(); + for sample in samples.iter() { + observations + .entry(sample.observation_id) + .or_insert_with(Vec::new) + .push(sample.stage); + } + assert_eq!(observations.len(), 3); + let terminal: Vec<_> = observations + .values() + .map(|stages| *stages.last().unwrap()) + .collect(); + assert_eq!( + terminal, + vec![ + StartupStage::Cancelled, + StartupStage::TimedOut, + StartupStage::Failed + ] + ); + assert!(!samples + .iter() + .any(|sample| sample.stage == StartupStage::Abandoned + || sample.stage == StartupStage::FirstText)); + } + + #[tokio::test] + async fn startup_observer_failure_cannot_fail_a_provider_turn() { + struct Panics; + impl crate::StartupObserver for Panics { + fn observe(&self, _: crate::StartupTiming) { + panic!("observer unavailable"); + } + } + let runtime = observed_runtime( + r#"printf '%s\n' '{"text":"done"}' '{"terminal":true}'"#, + Arc::new(Panics), + ); + let directory = tempfile::tempdir().unwrap(); + let result = runtime + .run( + TurnRequest::new(Provider::Claude, directory.path(), "test"), + &crate::NoopEventSink, + None, + ) + .await + .unwrap(); + assert_eq!(result.text, "done"); + } + #[tokio::test] async fn streams_normalized_events_and_returns_result() { let mut builder = AgentRuntime::builder(); diff --git a/src/startup.rs b/src/startup.rs new file mode 100644 index 0000000..9ec4e60 --- /dev/null +++ b/src/startup.rs @@ -0,0 +1,302 @@ +//! Opt-in, payload-free timing observations for provider startup. + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use crate::{Provider, RuntimeError}; + +/// A measured boundary, not an inferred provider readiness signal. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum StartupStage { + /// The runtime received a turn request, before validation. + Started, + /// Request, working-directory and provider controls passed validation. + Validated, + /// Waiting for the runtime's process concurrency permit. + WaitingForPermit, + /// The runtime acquired a process concurrency permit. + PermitAcquired, + /// The adapter built its command and explicit environment. + CommandPrepared, + /// The requested sandbox finished preparing the command. + SandboxPrepared, + /// The execution transport returned a spawned process. + ProcessSpawned, + /// Initial protocol bytes were flushed; this does not prove acceptance. + InitialInputWritten, + /// Provider streams were attached, including HTTP readiness where required. + StreamsAttached, + /// The runtime read its first provider output line; it may be a handshake or error. + FirstOutput, + /// The runtime parsed its first nonempty assistant text delta, before delivery. + /// Earlier event-sink waits are included in `elapsed`; inspect + /// [`StartupTiming::event_delivery_elapsed`] to distinguish them. + FirstText, + /// The turn returned successfully. + Succeeded, + /// The turn returned an error other than cancellation or timeout. + Failed, + /// The turn returned cooperative cancellation. + Cancelled, + /// The turn deadline expired. + TimedOut, + /// The caller dropped the execution future before it returned. + Abandoned, +} + +/// A single timing sample containing no prompts, paths, session IDs or secrets. +#[derive(Debug, Clone, Copy)] +pub struct StartupTiming { + /// Process-local observation ID; shared by all samples for one run call. + pub observation_id: u64, + /// Provider whose execution is being observed. + pub provider: Provider, + /// Boundary just reached. + pub stage: StartupStage, + /// Monotonic elapsed time since the run call began. + pub elapsed: Duration, + /// Cumulative time awaiting the application's event sink before this sample. + /// Subtract interval differences to identify delivery backpressure; the + /// remainder still includes protocol, observer and scheduling overhead. + pub event_delivery_elapsed: Duration, +} + +/// Receives startup boundaries without modifying provider or wire events. +/// +/// Implementations must return promptly: use a bounded `try_send` to export +/// samples, or record them locally. Do not perform network or filesystem work +/// in this callback. Under unwinding builds, observer panics are contained and +/// cannot fail a turn; `panic=abort` still terminates the host process. +/// A panicking observer is disabled for the runtime and its clones. Already +/// in-flight callbacks may finish. Callbacks are skipped during stack unwinding. +pub trait StartupObserver: Send + Sync { + /// Record one payload-free sample. + fn observe(&self, timing: StartupTiming); +} + +pub(crate) struct StartupObserverState { + observer: Arc, + disabled: AtomicBool, +} + +impl StartupObserverState { + pub(crate) fn new(observer: Arc) -> Self { + Self { + observer, + disabled: AtomicBool::new(false), + } + } + + fn enabled(&self) -> bool { + !self.disabled.load(Ordering::Acquire) && !std::thread::panicking() + } + + fn observe(&self, timing: StartupTiming) { + if !self.enabled() { + return; + } + if let Err(payload) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + self.observer.observe(timing); + })) { + // Disable before cleanup, which may itself call user-defined Drop. + self.disabled.store(true, Ordering::Release); + if let Err(cleanup_payload) = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + drop(payload); + })) + { + // The second panic payload can also have a panicking Drop. + // Quarantine it rather than recursively unwinding the host. + // Further callbacks are disabled, limiting this exceptional + // leak to callbacks already in flight when the observer failed. + std::mem::forget(cleanup_payload); + } + } + } +} + +static NEXT_OBSERVATION: AtomicU64 = AtomicU64::new(1); + +pub(crate) struct StartupTrace { + observer: Option>, + observation_id: u64, + provider: Provider, + started: Instant, + finished: bool, + event_delivery_nanos: AtomicU64, +} + +impl StartupTrace { + pub(crate) fn new(provider: Provider, observer: Option>) -> Self { + let trace = Self { + observation_id: if observer.is_some() { + NEXT_OBSERVATION.fetch_add(1, Ordering::Relaxed) + } else { + 0 + }, + observer, + provider, + started: Instant::now(), + finished: false, + event_delivery_nanos: AtomicU64::new(0), + }; + trace.record(StartupStage::Started); + trace + } + + pub(crate) fn record(&self, stage: StartupStage) { + if let Some(observer) = &self.observer { + let timing = StartupTiming { + observation_id: self.observation_id, + provider: self.provider, + stage, + elapsed: self.started.elapsed(), + event_delivery_elapsed: Duration::from_nanos( + self.event_delivery_nanos.load(Ordering::Relaxed), + ), + }; + observer.observe(timing); + } + } + + pub(crate) fn event_delivery(&self) -> EventDeliveryTimer<'_> { + EventDeliveryTimer { + trace: self, + started: self + .observer + .as_ref() + .filter(|observer| observer.enabled()) + .map(|_| Instant::now()), + } + } + + pub(crate) fn finish(&mut self, result: &crate::Result) { + let stage = match result { + Ok(_) => StartupStage::Succeeded, + Err(RuntimeError::Cancelled { .. }) => StartupStage::Cancelled, + Err(RuntimeError::Timeout { .. }) => StartupStage::TimedOut, + Err(_) => StartupStage::Failed, + }; + self.finished = true; + self.record(stage); + } +} + +pub(crate) struct EventDeliveryTimer<'a> { + trace: &'a StartupTrace, + started: Option, +} + +impl Drop for EventDeliveryTimer<'_> { + fn drop(&mut self) { + if let Some(started) = self.started { + let nanos = u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX); + let _ = self.trace.event_delivery_nanos.fetch_update( + Ordering::Relaxed, + Ordering::Relaxed, + |total| Some(total.saturating_add(nanos)), + ); + } + } +} + +impl Drop for StartupTrace { + fn drop(&mut self) { + if !self.finished { + self.record(StartupStage::Abandoned); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + + struct SecondaryCleanupPanic; + impl Drop for SecondaryCleanupPanic { + fn drop(&mut self) { + panic!("secondary panic payload must not be dropped"); + } + } + + struct CleanupPanic; + impl Drop for CleanupPanic { + fn drop(&mut self) { + std::panic::panic_any(SecondaryCleanupPanic); + } + } + + struct FailingObserver { + calls: AtomicUsize, + } + impl StartupObserver for FailingObserver { + fn observe(&self, _: StartupTiming) { + if self.calls.fetch_add(1, Ordering::Relaxed) == 0 { + std::panic::panic_any(CleanupPanic); + } + } + } + + #[test] + fn observer_payload_cleanup_cannot_escape_or_repeat_after_failure() { + let observer = Arc::new(FailingObserver { + calls: AtomicUsize::new(0), + }); + let state = Arc::new(StartupObserverState::new(observer.clone())); + let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let mut trace = StartupTrace::new(Provider::Claude, Some(state.clone())); + trace.record(StartupStage::Validated); + trace.finish(&Ok::<_, RuntimeError>(())); + let mut next_turn = StartupTrace::new(Provider::Claude, Some(state.clone())); + next_turn.finish(&Ok::<_, RuntimeError>(())); + })); + assert!( + outcome.is_ok(), + "observer cleanup escaped the isolation boundary" + ); + assert_eq!(observer.calls.load(Ordering::Relaxed), 1); + } + #[derive(Default)] + struct CountingObserver { + calls: AtomicUsize, + } + impl StartupObserver for CountingObserver { + fn observe(&self, _: StartupTiming) { + self.calls.fetch_add(1, Ordering::Relaxed); + } + } + + #[test] + fn observer_is_not_called_while_the_caller_unwinds() { + let observer = Arc::new(CountingObserver::default()); + let state = Arc::new(StartupObserverState::new(observer.clone())); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _trace = StartupTrace::new(Provider::Claude, Some(state)); + panic!("caller failed"); + })); + assert!(result.is_err()); + assert_eq!(observer.calls.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn interrupted_event_delivery_is_included_in_elapsed_accounting() { + let state = Arc::new(StartupObserverState::new(Arc::new( + CountingObserver::default(), + ))); + let mut trace = StartupTrace::new(Provider::Claude, Some(state)); + let deadline = Duration::from_millis(25); + let result = tokio::time::timeout(deadline, async { + let _delivery = trace.event_delivery(); + std::future::pending::<()>().await; + }) + .await; + assert!(result.is_err()); + assert!( + Duration::from_nanos(trace.event_delivery_nanos.load(Ordering::Relaxed)) >= deadline + ); + trace.finish(&Ok::<_, RuntimeError>(())); + } +}