diff --git a/src/claude_mcp.rs b/src/claude_mcp.rs index 8eef902b..efd42c04 100644 --- a/src/claude_mcp.rs +++ b/src/claude_mcp.rs @@ -5,7 +5,7 @@ //! Claude session wrapper owns presence because Claude can close this child before the session ends. use std::collections::HashSet; -use std::io::{self, BufRead, Write}; +use std::io::{self, BufRead, Write as _}; use std::path::Path; use std::sync::mpsc::{self, RecvTimeoutError}; use std::thread; @@ -15,16 +15,10 @@ use anyhow::{Context as _, Result}; use serde_json::{Value, json}; use crate::message; +use crate::native_channel::{channel_content, write_json}; const POLL: Duration = Duration::from_millis(250); -fn channel_content(subject: Option<&str>, body: &str) -> String { - match subject.filter(|value| !value.is_empty()) { - Some(subject) => format!("Subject: {subject}\n\n{body}"), - None => body.to_owned(), - } -} - pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { let agent_dir = message::resolve_declared_dir(catalog_root, identity, &crate::run::detect_host())? .with_context(|| format!("Claude MCP agent '{identity}' is not declared"))?; @@ -110,24 +104,3 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { thread::sleep(POLL); } } - -fn write_json(out: &mut impl Write, value: &Value) -> Result<()> { - serde_json::to_writer(&mut *out, value)?; - out.write_all(b"\n")?; - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::channel_content; - - #[test] - fn channel_content_reuses_subject_and_body_envelope() { - assert_eq!( - channel_content(Some("subject"), "body"), - "Subject: subject\n\nbody" - ); - assert_eq!(channel_content(None, "body"), "body"); - assert_eq!(channel_content(Some(""), "body"), "body"); - } -} diff --git a/src/claude_session.rs b/src/claude_session.rs index 73f9b41d..922cbe58 100644 --- a/src/claude_session.rs +++ b/src/claude_session.rs @@ -15,6 +15,7 @@ use std::path::Path; use anyhow::{Context as _, Result}; +use crate::driver_diagnostic::ProviderAuthEdge; use crate::harness_context::{self, Compaction, CompactionTrigger, Harness, RateLimits, Reading}; use crate::harness_state::{Activity, Ask, BlockedOn, InputBuffer, Observation}; use crate::provider_session::{ @@ -178,7 +179,11 @@ pub fn run_observe( // applied before the observation guard below for the same reason the compaction write is: // an edge that carries no top-level state change must still reach its own record. if let Some(edge) = provider_auth_edge(event, &payload) { - publish_provider_auth(&agent_dir, edge); + driver_diagnostic::publish_provider_auth( + &agent_dir, + driver_diagnostic::Driver::Claude, + edge, + ); } let Some(observation) = observe_hook_event(event, &payload) else { return Ok(()); @@ -619,13 +624,6 @@ fn stop_failure_error(payload: &serde_json::Value) -> Option<&str> { payload.get("error").and_then(serde_json::Value::as_str) } -/// What one hook event proves about the seat's provider credential. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum ProviderAuthEdge { - Rejected, - Accepted, -} - /// Read the credential edge out of one hook event, or `None` when the event proves nothing about /// it — which must leave a standing rejection alone rather than clearing it. fn provider_auth_edge(event: &str, payload: &serde_json::Value) -> Option { @@ -639,33 +637,6 @@ fn provider_auth_edge(event: &str, payload: &serde_json::Value) -> Option publisher.publish( - driver_diagnostic::Stage::ProviderAuth, - driver_diagnostic::Reason::ProviderAuthRejected, - driver_diagnostic::Source::TurnResult, - ), - ProviderAuthEdge::Accepted => publisher.clear(driver_diagnostic::Stage::ProviderAuth), - } -} - #[cfg(test)] mod tests { use std::fs; @@ -862,8 +833,9 @@ mod tests { "error": "authentication_failed", }); - publish_provider_auth( + driver_diagnostic::publish_provider_auth( tmp.path(), + driver_diagnostic::Driver::Claude, provider_auth_edge("StopFailure", &rejected).unwrap(), ); let driver_diagnostic::Observed::Failure(failure) = driver_diagnostic::read(&record) else { @@ -897,7 +869,11 @@ mod tests { driver_diagnostic::Observed::Failure(_) )); - publish_provider_auth(tmp.path(), ProviderAuthEdge::Accepted); + driver_diagnostic::publish_provider_auth( + tmp.path(), + driver_diagnostic::Driver::Claude, + ProviderAuthEdge::Accepted, + ); assert_eq!( driver_diagnostic::read(&record), driver_diagnostic::Observed::Absent, diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs index a474f488..e645aae0 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -17,7 +17,6 @@ use std::os::unix::ffi::OsStrExt as _; use std::os::unix::fs::{FileTypeExt as _, OpenOptionsExt as _, PermissionsExt as _}; use std::os::unix::io::AsRawFd as _; use std::os::unix::net::UnixStream; -use std::os::unix::process::ExitStatusExt as _; use std::path::{Path, PathBuf}; use std::process::{Child, Command, ExitStatus, Stdio}; use std::sync::mpsc::{self, Receiver, Sender}; @@ -1820,12 +1819,13 @@ enum TuiEnd { Stopped(Option), } +/// The label for a TUI end whose status may not have been observable at all. A status that WAS +/// reaped is spelled by the one shared exit-label map; no status is the same "unknown" the map's +/// own unanswerable arm reports. fn describe_tui_exit(status: Option) -> String { - match status.map(|status| (status.code(), status.signal())) { - Some((Some(code), _)) => format!("exit {code}"), - Some((None, Some(signal))) => format!("signal {signal}"), - _ => "exit unknown".to_string(), - } + status + .map(crate::provider_session::describe_exit) + .unwrap_or_else(|| "exit unknown".to_string()) } /// Start app-server with the authored global configuration inputs that its CLI supports. diff --git a/src/driver_diagnostic.rs b/src/driver_diagnostic.rs index 12b98e1c..f470227e 100644 --- a/src/driver_diagnostic.rs +++ b/src/driver_diagnostic.rs @@ -588,6 +588,46 @@ impl Publisher { } } +/// What one observation — a Claude hook event, a pi-family typed turn result — proves about the +/// seat's provider credential. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ProviderAuthEdge { + Rejected, + Accepted, +} + +/// Record one credential edge on the seat's native-driver diagnostic. +/// +/// A fresh publisher per edge on purpose, and every producer of these edges is short-lived: each +/// Claude hook invocation is its own process, so the publisher's stage set starts empty and its +/// on-disk fallback is what lets a later `Stop` clear a rejection an earlier `StopFailure` wrote +/// from a different process; a channel that restarted mid-session inherits the predecessor's +/// record the same way rather than silently starting clean. Fail-open like every other +/// observation: the publisher only warns on a write it cannot land, and neither delivery nor +/// launch depends on it. +pub(crate) fn publish_provider_auth(agent_dir: &Path, driver: Driver, edge: ProviderAuthEdge) { + let mut publisher = Publisher::new( + agent_dir, + driver, + // No producer version and no support verdict is knowable at either edge. A Claude hook + // payload carries no version — the common hook input is session id, transcript path, cwd, + // prompt id, permission mode, agent identity and effort, and nothing else (2.1.259) — and + // st2 gates no Claude version at all. On the pi family the WRAPPER, not the channel, owns + // the version gate and refuses the launch on an unadmitted MINOR (OMP-R05), so a running + // channel has no version fact of its own to publish and no verdict to restate. + None, + Support::Unknown, + ); + match edge { + ProviderAuthEdge::Rejected => publisher.publish( + Stage::ProviderAuth, + Reason::ProviderAuthRejected, + Source::TurnResult, + ), + ProviderAuthEdge::Accepted => publisher.clear(Stage::ProviderAuth), + } +} + fn emit( driver: Driver, stage: Stage, diff --git a/src/lib.rs b/src/lib.rs index c5b9e121..64744c32 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -39,12 +39,17 @@ pub mod identity; pub mod isolate; pub mod materialize; pub mod message; +/// The stdio framing shared by the native channels; crate-internal. +mod native_channel; pub mod omp_session; pub mod metrics; pub mod migrations; pub mod opencode_session; pub mod park; pub mod pi_channel; +/// The launch body shared by the pi-family wrappers; crate-internal, reached through +/// `pi_session::run` and `omp_session::run`. +mod pi_family_session; pub mod pi_session; pub mod pretrust; pub mod provider_session; diff --git a/src/native_channel.rs b/src/native_channel.rs new file mode 100644 index 00000000..a1a9d5fd --- /dev/null +++ b/src/native_channel.rs @@ -0,0 +1,62 @@ +//! The two framing helpers every stdio native channel shares. +//! +//! A native channel is a child of the interactive harness, speaking newline-delimited JSON over +//! stdio: the Claude MCP watcher ([`crate::claude_mcp`]) and the pi-family channel +//! ([`crate::pi_channel`]). What a delivered inbox message looks like to the model, and how a +//! frame is terminated on the wire, are st2's decisions rather than each channel's — so they are +//! decided once, here. + +use std::io::Write; + +use anyhow::Result; +use serde_json::Value; + +/// The envelope a delivered inbox message is handed to the model in. A message with no subject is +/// its body verbatim: an empty `Subject:` line would be noise the model has to read past. +pub(crate) fn channel_content(subject: Option<&str>, body: &str) -> String { + match subject.filter(|value| !value.is_empty()) { + Some(subject) => format!("Subject: {subject}\n\n{body}"), + None => body.to_owned(), + } +} + +/// One frame on the wire: compact JSON followed by the newline that terminates it. +pub(crate) fn write_json(out: &mut impl Write, value: &Value) -> Result<()> { + serde_json::to_writer(&mut *out, value)?; + out.write_all(b"\n")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn channel_content_reuses_one_subject_and_body_envelope() { + assert_eq!( + channel_content(Some("subject"), "body"), + "Subject: subject\n\nbody" + ); + assert_eq!(channel_content(None, "body"), "body"); + assert_eq!(channel_content(Some(""), "body"), "body"); + } + + /// The wire is newline-delimited: a reader splitting on newlines must see exactly one frame + /// per value, and a body carrying its own newline must not be able to forge a second one. + #[test] + fn each_frame_is_one_newline_terminated_line() { + let mut out = Vec::new(); + write_json(&mut out, &serde_json::json!({"a": 1})).unwrap(); + write_json( + &mut out, + &serde_json::json!({"content": "line one\nline two"}), + ) + .unwrap(); + let framed = String::from_utf8(out).unwrap(); + assert_eq!( + framed, + "{\"a\":1}\n{\"content\":\"line one\\nline two\"}\n" + ); + assert_eq!(framed.lines().count(), 2); + } +} diff --git a/src/omp_session.rs b/src/omp_session.rs index 5bd8b0ec..ec02a404 100644 --- a/src/omp_session.rs +++ b/src/omp_session.rs @@ -3,9 +3,7 @@ //! omp is pi-family: its integration point is a pi-style extension loaded into the interactive //! process, which reaches st2 by spawning `st2 driver omp-channel` (`hooks/omp-channel.ts`, //! forked from the pi channel — see `docs/vrs/06-omp-driver/spec.md` for the measured -//! divergences). The wrapper owns presence for the same reason the pi wrapper does: the extension -//! lives only as long as omp's process, and a SIGKILL of omp produces no terminal record at all, -//! so presence decays by staleness exactly as for the other harnesses. +//! divergences). The launch body itself is shared with pi in [`crate::pi_family_session`]. //! //! Unlike pi, the wrapper hard-gates the provider version (OMP-R05): the delivery-critical //! surface — event names, the sampled idle edge, the approval events — is versioned behavior, not @@ -13,14 +11,11 @@ //! Patches inside an admitted minor launch without new evidence (decision 0007-omp-is-a-fifth-native-driver-with-its-own-channel-and-a-hard-version-gate). use std::path::Path; -use std::process::ExitStatus; use anyhow::{Context as _, Result}; -use crate::provider_session::{ - install_signal_handler, run_provider_observed, ProviderOutcome, PROVIDER_POLL, STOP, -}; -use crate::{harness_state, harness_version, hooks, message, status}; +use crate::harness_version; +use crate::pi_family_session::{self, HarnessKind}; /// The extension file inside this binary's immutable hook set. const EXTENSION: &str = "omp-channel.ts"; @@ -39,11 +34,6 @@ pub const CHANNEL_SESSION: &str = "ST2_OMP_CHANNEL_SESSION"; /// The ownership sequence the wrapper claimed at startup. pub const CHANNEL_SEQ: &str = "ST2_OMP_CHANNEL_SEQ"; -/// omp reads its pi ancestor's env fallbacks, so the same offline defaults apply. Whether they -/// suppress the update banner in interactive boots is still open (DQ-OMP-5); shipping them is -/// harmless either way. -const OFFLINE_DEFAULTS: [(&str, &str); 2] = [("PI_OFFLINE", "1"), ("PI_SKIP_VERSION_CHECK", "1")]; - /// The omp MINORS verified against the admission checks in `docs/vrs/06-omp-driver/spec.md`. /// /// 18.0 was measured twice: at 18.0.3 on 2026-08-25 and again at 18.0.9 on 2026-08-28. @@ -66,9 +56,20 @@ const SUPPORTED_OMP_MINORS: [(u32, u32); 2] = [(18, 0), (18, 1)]; /// `the_measured_context_builds_are_admitted_by_this_gate` keeps them from drifting apart. pub const MEASURED_CONTEXT_VERSIONS: [&str; 2] = ["18.0.9", "18.0.3"]; -/// What the wrapper hands the provider process: the channel environment plus the launch argv with -/// the channel extension spliced in. -type PreparedLaunch = (Vec<(String, String)>, Vec); +/// omp's half of the pi-family launch fork. The version gate rides on the descriptor so the shared +/// body runs it where omp has always run it: after the empty-argv check and before the ownership +/// claim, so an unadmitted minor fails without claiming the seat. +pub(crate) const OMP_KIND: HarnessKind = HarnessKind { + label: "omp", + extension: EXTENSION, + bin_env: CHANNEL_BIN, + catalog_env: CHANNEL_CATALOG, + identity_env: CHANNEL_IDENTITY, + runtime_id_env: CHANNEL_RUNTIME_ID, + session_env: CHANNEL_SESSION, + seq_env: CHANNEL_SEQ, + verify_version: Some(verify_supported_version), +}; /// Run one interactive omp provider and maintain its presence until it exits. pub fn run( @@ -77,93 +78,7 @@ pub fn run( runtime_id: String, omp_argv: Vec, ) -> Result<()> { - let agent_dir = - message::resolve_declared_dir(catalog_root, &identity, &crate::run::detect_host())? - .with_context(|| format!("omp driver agent '{identity}' is not declared"))?; - anyhow::ensure!( - !omp_argv.is_empty(), - "omp driver '{runtime_id}' has no provider argv" - ); - verify_supported_version(&omp_argv[0])?; - let executable = - std::env::current_exe().context("resolving st2 executable for the omp channel")?; - let session = harness_state::session_token(); - // The claim is written: it supersedes whatever the predecessor left — including a - // still-fresh live record — before the channel or terminal writer act under it. - let seq = harness_state::claim(&agent_dir, identity.clone(), "omp", &session)?; - // Every fallible step past the claim must end the record honestly on failure — the claim - // placeholder standing as the last word would read as a takeover, not a launch that never - // ran. - let prepared = (|| -> Result { - let mut env = channel_env( - &executable, - catalog_root, - &identity, - &runtime_id, - &session, - seq, - )?; - env.extend(offline_defaults(|key| std::env::var_os(key).is_some())); - let set = hooks::verify_required_set().with_context(|| { - format!( - "omp driver '{runtime_id}' needs this binary's verified hook set for {EXTENSION}; run `st2 hooks install`" - ) - })?; - Ok((env, with_channel_extension(omp_argv, &set)?)) - })(); - let (env, omp_argv) = match prepared { - Ok(prepared) => prepared, - Err(error) => { - let mut writer = harness_state::Writer::new( - &agent_dir, - identity.clone(), - "omp", - Some(runtime_id.clone()), - ) - .with_ownership(session.clone(), seq); - let _ = writer.observe( - harness_state::Observation::new( - harness_state::Activity::Ended, - harness_state::BlockedOn::None, - harness_state::InputBuffer::Unknown, - ) - .with_reason("launch-error") - .with_exit("exit unknown"), - ); - return Err(error); - } - }; - install_signal_handler(); - // Terminal-only: the channel owns the live record and its heartbeat, but only this wrapper - // survives long enough to see the stop path. Same token as the channel, so the terminal - // record fences exactly this session's live records. - let observer = crate::provider_session::SessionObserver::terminal_only( - &agent_dir, - &identity, - "omp", - &runtime_id, - &session, - seq, - ); - let outcome = run_provider_observed( - "omp", - &status::status_path(&agent_dir), - &omp_argv, - &env, - status::STATUS_REFRESH, - PROVIDER_POLL, - &STOP, - Some(&observer), - ) - .with_context(|| format!("running omp driver '{runtime_id}'"))?; - record_session_end(&agent_dir, &identity, &runtime_id, &session, seq, &outcome); - match outcome { - ProviderOutcome::Exited(exit) => { - anyhow::ensure!(exit.success(), "omp provider exited with {exit}"); - Ok(()) - } - ProviderOutcome::Stopped(_) => Ok(()), - } + pi_family_session::run_for(catalog_root, identity, runtime_id, omp_argv, &OMP_KIND) } /// Refuse any provider whose MINOR this binary was not verified against. Failing loudly at launch @@ -190,86 +105,6 @@ fn verify_supported_version(binary: &str) -> Result<()> { Ok(()) } -/// The wrapper's one write into observed harness state: the terminal record. Live states and -/// heartbeats belong to the omp channel; the wrapper sees exactly one fact the channel cannot — -/// that the provider process is gone. -fn record_session_end( - agent_dir: &Path, - identity: &str, - runtime_id: &str, - session: &str, - seq: u64, - outcome: &ProviderOutcome, -) { - let label = match outcome { - ProviderOutcome::Exited(exit) | ProviderOutcome::Stopped(Some(exit)) => exit_label(*exit), - ProviderOutcome::Stopped(None) => "stopped".to_string(), - }; - let mut writer = - harness_state::Writer::new(agent_dir, identity, "omp", Some(runtime_id.to_string())) - .with_ownership(session, seq); - if let Err(error) = writer.ended(label) { - eprintln!("st2 omp driver: recording session end failed: {error}"); - } -} - -fn exit_label(exit: ExitStatus) -> String { - use std::os::unix::process::ExitStatusExt as _; - match (exit.code(), exit.signal()) { - (Some(code), _) => format!("exit {code}"), - (None, Some(signal)) => format!("signal {signal}"), - (None, None) => "exited".to_string(), - } -} - -/// Load the channel extension from the verified set, immediately after the provider program. -/// -/// Resolving it here means a launch uses the exact asset this binary was built with; a rendered -/// machine-local path in a declaration would pin one host's layout into a catalog. -fn with_channel_extension(mut argv: Vec, set: &Path) -> Result> { - let extension = set.join(EXTENSION); - let extension = extension - .to_str() - .context("verified hook set path is not UTF-8")? - .to_owned(); - argv.splice(1..1, ["-e".to_string(), extension]); - Ok(argv) -} - -/// The offline defaults this launch should add, skipping any the operator already declared. -fn offline_defaults(is_set: impl Fn(&str) -> bool) -> Vec<(String, String)> { - OFFLINE_DEFAULTS - .iter() - .filter(|(key, _)| !is_set(key)) - .map(|(key, value)| ((*key).to_string(), (*value).to_string())) - .collect() -} - -/// The environment the shipped omp extension reads to reach this exact control plane. -/// -/// Fresh variable names: an omp seat must never adopt a stray pi channel configuration. -fn channel_env( - executable: &Path, - catalog_root: &Path, - identity: &str, - runtime_id: &str, - session: &str, - seq: u64, -) -> Result> { - let executable = executable - .to_str() - .context("st2 executable path is not UTF-8")?; - let catalog_root = catalog_root.to_str().context("catalog root is not UTF-8")?; - Ok(vec![ - (CHANNEL_BIN.to_string(), executable.to_string()), - (CHANNEL_CATALOG.to_string(), catalog_root.to_string()), - (CHANNEL_IDENTITY.to_string(), identity.to_string()), - (CHANNEL_RUNTIME_ID.to_string(), runtime_id.to_string()), - (CHANNEL_SESSION.to_string(), session.to_string()), - (CHANNEL_SEQ.to_string(), seq.to_string()), - ]) -} - #[cfg(test)] mod tests { use super::*; @@ -486,24 +321,19 @@ mod tests { assert_eq!(paths.iter().collect::>().len(), paths.len()); } + /// The gate must run before the wrapper claims the seat: an unadmitted minor that took + /// ownership would leave the seat's record owned by a session that never launched. #[test] - fn offline_defaults_skip_operator_declared_keys() { - let defaults = offline_defaults(|key| key == "PI_OFFLINE"); - assert_eq!( - defaults, - vec![("PI_SKIP_VERSION_CHECK".to_string(), "1".to_string())] + fn the_version_gate_is_wired_into_the_shared_launch_fork() { + assert!( + OMP_KIND.verify_version.is_some(), + "omp must carry a launch-time version gate" + ); + let fake = FakeExecutable::new("#!/bin/sh\nprintf '18.2.0\\n'\n"); + let gate = OMP_KIND.verify_version.unwrap(); + assert!( + gate(fake.path().to_str().unwrap()).is_err(), + "the descriptor's gate must be the refusing one" ); - } - - #[test] - fn channel_extension_is_spliced_right_after_the_program() { - let dir = tempfile::tempdir().unwrap(); - let argv = - with_channel_extension(vec!["omp".into(), "--model".into(), "x".into()], dir.path()) - .unwrap(); - assert_eq!(argv[0], "omp"); - assert_eq!(argv[1], "-e"); - assert!(argv[2].ends_with("omp-channel.ts")); - assert_eq!(argv[3], "--model"); } } diff --git a/src/opencode_session.rs b/src/opencode_session.rs index 750a5dce..7f5a3d29 100644 --- a/src/opencode_session.rs +++ b/src/opencode_session.rs @@ -21,7 +21,6 @@ use std::collections::BTreeMap; use std::io::{BufRead as _, BufReader, Read as _, Write as _}; use std::net::{TcpListener, TcpStream}; -use std::os::unix::process::ExitStatusExt as _; use std::path::{Path, PathBuf}; use std::process::{Child, ExitStatus}; use std::sync::atomic::{AtomicBool, Ordering}; @@ -38,7 +37,9 @@ use crate::driver_diagnostic::{ Source as DiagnosticSource, Stage as DiagnosticStage, Support as DiagnosticSupport, }; use crate::harness_state::{self, Activity, Ask, BlockedOn, InputBuffer, Observation, Writer}; -use crate::provider_session::{PROVIDER_POLL, STOP, install_signal_handler}; +use crate::provider_session::{ + PROVIDER_POLL, STOP, completed_provider, describe_exit, install_signal_handler, +}; use crate::{delivery_ledger, ding, harness_context, harness_version, message, status}; /// OpenCode MINORS whose `/event`, `/session`, and `prompt_async` surfaces were verified @@ -252,7 +253,7 @@ fn run_session(mut session: Session, child: &mut Child, agent_dir: &Path) -> Res match child.try_wait() { Ok(Some(exit)) => { let _ = session.writer.ended(describe_exit(exit)); - break completed(exit); + break completed_provider("opencode", exit); } Ok(None) => {} Err(error) => { @@ -430,11 +431,6 @@ fn spawn_provider(argv: &[String], password: &str) -> Result { .with_context(|| format!("starting opencode provider {program}")) } -fn completed(exit: ExitStatus) -> Result<()> { - anyhow::ensure!(exit.success(), "opencode provider exited with {exit}"); - Ok(()) -} - fn stop_provider_group(child: &mut Child) -> Result> { let process_group = unsafe { libc::getpgrp() }; anyhow::ensure!( @@ -457,14 +453,6 @@ fn stop_provider_group(child: &mut Child) -> Result> { Ok(child.wait().ok()) } -fn describe_exit(exit: ExitStatus) -> String { - match (exit.code(), exit.signal()) { - (Some(code), _) => format!("exit {code}"), - (None, Some(signal)) => format!("signal {signal}"), - (None, None) => "exit unknown".to_string(), - } -} - fn supported_version(binary: &str) -> Result { let output = std::process::Command::new(binary) .arg("--version") diff --git a/src/pi_channel.rs b/src/pi_channel.rs index 1e1370d1..bea558a2 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -19,6 +19,8 @@ use std::time::{Duration, Instant}; use anyhow::{Context as _, Result}; use serde_json::{Value, json}; +use crate::driver_diagnostic::ProviderAuthEdge; +use crate::native_channel::{channel_content, write_json}; use crate::{context, driver_diagnostic, harness_context, harness_state, message}; const POLL: Duration = Duration::from_millis(250); @@ -64,13 +66,6 @@ const PRE_COMPACT_ERROR_REASON: &str = "pre-compact context recovery failed"; /// guaranteed is that displaced work resumes: the model chose to continue, once, on one model. const DELIVER_AS: &str = "steer"; -fn channel_content(subject: Option<&str>, body: &str) -> String { - match subject.filter(|value| !value.is_empty()) { - Some(subject) => format!("Subject: {subject}\n\n{body}"), - None => body.to_owned(), - } -} - /// The harness-specific facts the shared channel loop needs: which env names carry the wrapper's /// exported ownership triple, what label goes on records and errors, and which native-driver /// diagnostic word — if any — this channel publishes under. @@ -268,7 +263,7 @@ fn channel_loop( if let Some(driver) = kind.diagnostic_driver && let Some(edge) = turn.as_ref().and_then(provider_auth_edge) { - publish_provider_auth(agent_dir, driver, edge); + driver_diagnostic::publish_provider_auth(agent_dir, driver, edge); } // The numeric axis. There is deliberately no cadence here and no heartbeat timer: // a producer holding no fresh reading must write nothing at all, so the record @@ -459,13 +454,6 @@ fn turn_observation(result: &TurnResult<'_>) -> Option) -> Option { } } -/// Record one credential edge on the seat's native-driver diagnostic. -/// -/// A fresh publisher per edge, like the Claude hook's: the on-disk fallback is what lets an -/// ordinary turn end clear a rejection, and a channel that restarted mid-session inherits the -/// predecessor's record rather than silently starting clean. Fail-open like every other -/// observation in this loop — the publisher only warns on a write it cannot land, and delivery -/// never depends on it. -fn publish_provider_auth( - agent_dir: &Path, - driver: driver_diagnostic::Driver, - edge: ProviderAuthEdge, -) { - let mut publisher = driver_diagnostic::Publisher::new( - agent_dir, - driver, - // The wrapper — not the channel — owns the version gate, and it refuses the launch on an - // unadmitted MINOR (OMP-R05), so a running channel has no version fact of its own to - // publish and no support verdict to restate. - None, - driver_diagnostic::Support::Unknown, - ); - match edge { - ProviderAuthEdge::Rejected => publisher.publish( - driver_diagnostic::Stage::ProviderAuth, - driver_diagnostic::Reason::ProviderAuthRejected, - driver_diagnostic::Source::TurnResult, - ), - ProviderAuthEdge::Accepted => publisher.clear(driver_diagnostic::Stage::ProviderAuth), - } -} - /// Write the recovery stub only when durable working state is absent or whitespace-only. /// /// The extension cannot perform this check: it owns neither the resolved agent directory nor the @@ -689,12 +646,6 @@ fn message_frame(msg: message::Message, identity: &str) -> Value { }}) } -fn write_json(out: &mut impl Write, value: &Value) -> Result<()> { - serde_json::to_writer(&mut *out, value)?; - out.write_all(b"\n")?; - Ok(()) -} - #[cfg(test)] mod tests { use super::*; @@ -1237,16 +1188,6 @@ mod tests { ); } - #[test] - fn channel_content_reuses_the_claude_channel_envelope() { - assert_eq!( - channel_content(Some("subject"), "body"), - "Subject: subject\n\nbody" - ); - assert_eq!(channel_content(None, "body"), "body"); - assert_eq!(channel_content(Some(""), "body"), "body"); - } - /// A restarting pi agent has to be told the same three things the Codex and Claude session-start /// hooks tell theirs, in the same order — otherwise "restart" means something different per /// harness. diff --git a/src/pi_family_session.rs b/src/pi_family_session.rs new file mode 100644 index 00000000..6b76f878 --- /dev/null +++ b/src/pi_family_session.rs @@ -0,0 +1,483 @@ +//! The controlled-launch body shared by the pi-family wrappers (pi and omp). +//! +//! Both harnesses integrate the same way: an extension loaded into the interactive process, which +//! reaches st2 by spawning `st2 driver -channel`. Two facts have to be handed to that +//! extension, and neither is discoverable from inside the harness. The first is *which* st2 to run +//! — resolving `st2` from `PATH` would let a replaced control plane and its live agents disagree, +//! which the R11 control-plane replacement guarantee exists to prevent — so the wrapper exports its +//! own executable path. The second is the catalog and identity the channel must bind. +//! +//! The wrapper also owns presence for the same reason the Claude wrapper does: the extension lives +//! only as long as the provider's process, and a measured SIGKILL of pi produces no terminal record +//! at all (`docs/vrs/.experiments/2026-08-18-pi-harness-integration.md`). Presence therefore decays +//! by staleness, exactly as for the other harnesses. +//! +//! The fork between the two harnesses is a [`HarnessKind`] descriptor, mirroring the +//! [`crate::pi_channel::ChannelKind`] fork the family's channel side already uses: the same fork +//! solved the same way twice rather than two different ways. What stays in each harness module is +//! what is genuinely per-harness — its module doc, its extension asset, its channel env NAMES (the +//! whole point of two sets is that an omp seat can never adopt a stray pi configuration), and, for +//! omp, its version gate. + +use std::path::Path; + +use anyhow::{Context as _, Result}; + +use crate::provider_session::{ + PROVIDER_POLL, ProviderOutcome, STOP, describe_exit, install_signal_handler, + run_provider_observed, +}; +use crate::{harness_state, hooks, message, status}; + +/// pi's startup network work, which a supervised seat should not be doing. +/// +/// A managed agent that update-checks or self-updates at boot makes its own launch latency and its +/// own behaviour depend on the network, and lets a release change a running fleet. Each is applied +/// only when the operator has not already set it, so a declaration's `env` still wins. +/// +/// omp reads its pi ancestor's env fallbacks, so the same defaults apply to it. Whether they +/// suppress the update banner in interactive boots is still open (DQ-OMP-5); shipping them is +/// harmless either way. +const OFFLINE_DEFAULTS: [(&str, &str); 2] = [("PI_OFFLINE", "1"), ("PI_SKIP_VERSION_CHECK", "1")]; + +/// The harness-specific facts the shared wrapper body needs: the label that goes on records and +/// errors, the extension asset to inject, the env names the shipped extension reads, and the +/// launch gate — if any — this harness enforces. +pub(crate) struct HarnessKind { + /// The harness word on observed records, claims and launch errors. + pub(crate) label: &'static str, + /// The extension file inside this binary's immutable hook set. + pub(crate) extension: &'static str, + /// The exact st2 executable the extension must spawn for its channel. + pub(crate) bin_env: &'static str, + /// The catalog root that executable must be pointed at. + pub(crate) catalog_env: &'static str, + /// The host-qualified bus identity the channel binds. + pub(crate) identity_env: &'static str, + /// The wrapper's runtime/task ID — the pty session whose liveness vouches for observed state. + pub(crate) runtime_id_env: &'static str, + /// The session incarnation token the wrapper mints. + pub(crate) session_env: &'static str, + /// The ownership sequence the wrapper claimed at startup. + pub(crate) seq_env: &'static str, + /// The launch-time provider version gate, for the harness that has one. omp hard-gates its + /// MINOR (OMP-R05); pi does not gate at runtime at all, so its slot is `None` rather than a + /// function that always succeeds — a gate that cannot refuse is not a gate. + pub(crate) verify_version: Option Result<()>>, +} + +/// What the wrapper hands the provider process: the channel environment plus the launch argv with +/// the channel extension spliced in. +type PreparedLaunch = (Vec<(String, String)>, Vec); + +/// Run one interactive pi-family provider and maintain its presence until it exits. +pub(crate) fn run_for( + catalog_root: &Path, + identity: String, + runtime_id: String, + provider_argv: Vec, + kind: &HarnessKind, +) -> Result<()> { + let label = kind.label; + let agent_dir = + message::resolve_declared_dir(catalog_root, &identity, &crate::run::detect_host())? + .with_context(|| format!("{label} driver agent '{identity}' is not declared"))?; + anyhow::ensure!( + !provider_argv.is_empty(), + "{label} driver '{runtime_id}' has no provider argv" + ); + // Before the claim on purpose: an unadmitted provider must fail without taking ownership of + // the seat's observed record, so a refused launch leaves the predecessor's state alone. + if let Some(verify_version) = kind.verify_version { + verify_version(&provider_argv[0])?; + } + let executable = std::env::current_exe() + .with_context(|| format!("resolving st2 executable for the {label} channel"))?; + let session = harness_state::session_token(); + // The claim is written: it supersedes whatever the predecessor left — including a + // still-fresh live record — before the channel or terminal writer act under it. + let seq = harness_state::claim(&agent_dir, identity.clone(), label, &session)?; + // Every fallible step past the claim must end the record honestly on failure — the claim + // placeholder standing as the last word would read as a takeover, not a launch that never + // ran. + let prepared = (|| -> Result { + let mut env = channel_env( + kind, + &executable, + catalog_root, + &identity, + &runtime_id, + &session, + seq, + )?; + env.extend(offline_defaults(|key| std::env::var_os(key).is_some())); + let set = hooks::verify_required_set().with_context(|| { + format!( + "{label} driver '{runtime_id}' needs this binary's verified hook set for {}; run `st2 hooks install`", + kind.extension + ) + })?; + Ok((env, with_channel_extension(provider_argv, &set, kind.extension)?)) + })(); + let (env, provider_argv) = match prepared { + Ok(prepared) => prepared, + Err(error) => { + let mut writer = harness_state::Writer::new( + &agent_dir, + identity.clone(), + label, + Some(runtime_id.clone()), + ) + .with_ownership(session.clone(), seq); + let _ = writer.observe( + harness_state::Observation::new( + harness_state::Activity::Ended, + harness_state::BlockedOn::None, + harness_state::InputBuffer::Unknown, + ) + .with_reason("launch-error") + .with_exit("exit unknown"), + ); + return Err(error); + } + }; + install_signal_handler(); + // Terminal-only: the channel owns the live record and its heartbeat, but only this wrapper + // survives long enough to see the stop path — its pre-escalation `ended` write is the one + // that makes `Stopped(None)` observable at all. Same token as the channel, so the terminal + // record fences exactly this session's live records. + let observer = crate::provider_session::SessionObserver::terminal_only( + &agent_dir, + &identity, + label, + &runtime_id, + &session, + seq, + ); + let outcome = run_provider_observed( + label, + &status::status_path(&agent_dir), + &provider_argv, + &env, + status::STATUS_REFRESH, + PROVIDER_POLL, + &STOP, + Some(&observer), + ) + .with_context(|| format!("running {label} driver '{runtime_id}'"))?; + record_session_end( + &agent_dir, + &identity, + &runtime_id, + &session, + seq, + &outcome, + kind, + ); + match outcome { + ProviderOutcome::Exited(exit) => { + anyhow::ensure!(exit.success(), "{label} provider exited with {exit}"); + Ok(()) + } + ProviderOutcome::Stopped(_) => Ok(()), + } +} + +/// The wrapper's one write into observed harness state: the terminal record. Live states and +/// heartbeats belong to the harness channel, which sees the provider's own turn events over stdio; +/// the wrapper sees exactly one fact the channel cannot — that the provider process is gone — so +/// that is the one fact it records. The `Writer` is constructed at the terminal edge on purpose: it +/// re-reads whatever the channel last wrote and continues its transition counter, and by the time +/// the wrapper has reaped the provider the extension (and with it the channel) is already gone. +fn record_session_end( + agent_dir: &Path, + identity: &str, + runtime_id: &str, + session: &str, + seq: u64, + outcome: &ProviderOutcome, + kind: &HarnessKind, +) { + let label = match outcome { + ProviderOutcome::Exited(exit) | ProviderOutcome::Stopped(Some(exit)) => { + describe_exit(*exit) + } + ProviderOutcome::Stopped(None) => "stopped".to_string(), + }; + let mut writer = harness_state::Writer::new( + agent_dir, + identity, + kind.label, + Some(runtime_id.to_string()), + ) + .with_ownership(session, seq); + if let Err(error) = writer.ended(label) { + tracing::warn!( + "st2 {} driver: recording session end failed: {error}", + kind.label + ); + } +} + +/// Load the channel extension from the verified set, immediately after the provider program. +/// +/// The declaration deliberately carries no path to it: a rendered machine-local path would pin one +/// host's layout into a catalog, and a `$ST_HOOKS` token in an argv would resolve to the +/// receipt-bearing root rather than the selected set. Resolving it here means a launch uses the +/// exact asset this binary was built with. +fn with_channel_extension( + mut argv: Vec, + set: &Path, + extension: &str, +) -> Result> { + let extension = set.join(extension); + let extension = extension + .to_str() + .context("verified hook set path is not UTF-8")? + .to_owned(); + argv.splice(1..1, ["-e".to_string(), extension]); + Ok(argv) +} + +/// The offline defaults this launch should add, skipping any the operator already declared. +fn offline_defaults(is_set: impl Fn(&str) -> bool) -> Vec<(String, String)> { + OFFLINE_DEFAULTS + .iter() + .filter(|(key, _)| !is_set(key)) + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect() +} + +/// The environment the shipped extension reads to reach this exact control plane. +/// +/// The names come from the descriptor rather than being shared: an omp seat must never adopt a +/// stray pi channel configuration. +fn channel_env( + kind: &HarnessKind, + executable: &Path, + catalog_root: &Path, + identity: &str, + runtime_id: &str, + session: &str, + seq: u64, +) -> Result> { + let executable = executable + .to_str() + .context("st2 executable path is not UTF-8")?; + let catalog_root = catalog_root.to_str().context("catalog root is not UTF-8")?; + Ok(vec![ + (kind.bin_env.to_string(), executable.to_string()), + (kind.catalog_env.to_string(), catalog_root.to_string()), + (kind.identity_env.to_string(), identity.to_string()), + (kind.runtime_id_env.to_string(), runtime_id.to_string()), + (kind.session_env.to_string(), session.to_string()), + (kind.seq_env.to_string(), seq.to_string()), + ]) +} + +#[cfg(test)] +mod tests { + use std::fs; + use std::path::PathBuf; + use std::process::ExitStatus; + + use super::*; + use crate::omp_session::OMP_KIND; + use crate::pi_session::PI_KIND; + + /// The wrapper writes the one observation the channel cannot: the terminal record, carrying + /// the exit. It continues the transition counter of whatever the channel last wrote, so the + /// death of a session is a transition in the same record, not a new history. + #[test] + fn provider_exit_writes_the_terminal_record_with_its_status() { + use std::os::unix::process::ExitStatusExt as _; + + let tmp = tempfile::tempdir().unwrap(); + let agent_dir = tmp.path(); + let mut channel_writer = + crate::harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); + channel_writer + .observe(crate::harness_state::Observation::new( + crate::harness_state::Activity::Active, + crate::harness_state::BlockedOn::None, + crate::harness_state::InputBuffer::Unknown, + )) + .unwrap(); + drop(channel_writer); + + record_session_end( + agent_dir, + "h.worker", + "h.worker", + "session-test", + 1, + &ProviderOutcome::Exited(ExitStatus::from_raw(3 << 8)), + &PI_KIND, + ); + + let record = crate::harness_state::harness_state_path(agent_dir); + let observed = crate::harness_state::read(&record, None).unwrap(); + assert_eq!(observed.state, crate::harness_state::Activity::Ended); + assert_eq!(observed.exit.as_deref(), Some("exit 3")); + let raw: serde_json::Value = serde_json::from_slice(&fs::read(&record).unwrap()).unwrap(); + assert_eq!( + raw["transitions"], 1, + "counter continues the channel's record" + ); + + record_session_end( + agent_dir, + "h.worker", + "h.worker", + "session-test", + 1, + &ProviderOutcome::Stopped(Some(ExitStatus::from_raw(9))), + &PI_KIND, + ); + let observed = crate::harness_state::read(&record, None).unwrap(); + assert_eq!(observed.exit.as_deref(), Some("signal 9")); + } + + /// Each harness injects its OWN extension asset from the verified set, not a path the + /// declaration carried, and it lands immediately after the provider program. + #[test] + fn the_channel_extension_is_injected_from_the_verified_set_not_the_declaration() { + let argv = with_channel_extension( + vec![ + "pi".into(), + "-a".into(), + "--model".into(), + "anthropic/opus".into(), + "Start work.".into(), + ], + &PathBuf::from("/state/st2/hooks/sets/sha256-abc"), + PI_KIND.extension, + ) + .unwrap(); + + assert_eq!( + argv, + vec![ + "pi", + "-e", + "/state/st2/hooks/sets/sha256-abc/pi-channel.ts", + "-a", + "--model", + "anthropic/opus", + "Start work.", + ] + ); + + let argv = with_channel_extension( + vec!["omp".into(), "--model".into(), "x".into()], + &PathBuf::from("/state/st2/hooks/sets/sha256-abc"), + OMP_KIND.extension, + ) + .unwrap(); + + assert_eq!( + argv, + vec![ + "omp", + "-e", + "/state/st2/hooks/sets/sha256-abc/omp-channel.ts", + "--model", + "x", + ] + ); + } + + /// A supervised seat is offline by default, but an operator who declared otherwise keeps their + /// value — otherwise the wrapper would silently overrule the declaration. + #[test] + fn offline_defaults_apply_only_where_the_operator_declared_nothing() { + assert_eq!( + offline_defaults(|_| false), + vec![ + ("PI_OFFLINE".to_string(), "1".to_string()), + ("PI_SKIP_VERSION_CHECK".to_string(), "1".to_string()), + ] + ); + assert_eq!( + offline_defaults(|key| key == "PI_OFFLINE"), + vec![("PI_SKIP_VERSION_CHECK".to_string(), "1".to_string())] + ); + assert!(offline_defaults(|_| true).is_empty()); + } + + #[test] + fn the_extension_receives_this_binary_not_a_path_lookup() { + let env = channel_env( + &PI_KIND, + &PathBuf::from("/opt/st2/bin/st2"), + &PathBuf::from("/catalog"), + "host.worker", + "host.worker-task", + "session-test", + 7, + ) + .unwrap(); + + assert_eq!( + env, + vec![ + ( + crate::pi_session::CHANNEL_BIN.to_string(), + "/opt/st2/bin/st2".to_string() + ), + ( + crate::pi_session::CHANNEL_CATALOG.to_string(), + "/catalog".to_string() + ), + ( + crate::pi_session::CHANNEL_IDENTITY.to_string(), + "host.worker".to_string() + ), + ( + crate::pi_session::CHANNEL_RUNTIME_ID.to_string(), + "host.worker-task".to_string() + ), + ( + crate::pi_session::CHANNEL_SESSION.to_string(), + "session-test".to_string() + ), + (crate::pi_session::CHANNEL_SEQ.to_string(), "7".to_string()), + ] + ); + } + + /// The two harnesses carry DISJOINT env names on purpose: an omp seat that inherited a stray + /// pi channel configuration would point its channel at another harness's control plane. + #[test] + fn the_two_harnesses_export_disjoint_channel_variable_names() { + let names = |kind: &HarnessKind| { + channel_env( + kind, + &PathBuf::from("/opt/st2/bin/st2"), + &PathBuf::from("/catalog"), + "host.worker", + "host.worker-task", + "session-test", + 7, + ) + .unwrap() + .into_iter() + .map(|(key, _)| key) + .collect::>() + }; + let pi = names(&PI_KIND); + let omp = names(&OMP_KIND); + assert!( + pi.iter().all(|name| name.starts_with("ST2_PI_CHANNEL_")), + "{pi:?}" + ); + assert!( + omp.iter().all(|name| name.starts_with("ST2_OMP_CHANNEL_")), + "{omp:?}" + ); + assert!( + pi.iter().all(|name| !omp.contains(name)), + "the two harnesses must share no channel variable name" + ); + } +} diff --git a/src/pi_session.rs b/src/pi_session.rs index 84d26557..d39ae1e0 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -1,26 +1,16 @@ //! Controlled pi launch with a session-owned presence lease. //! //! pi has no MCP and no app-server: its integration point is an extension loaded into the -//! interactive process, and that extension reaches st2 by spawning `st2 driver pi-channel`. Two -//! facts have to be handed to it, and neither is discoverable from inside pi. The first is *which* -//! st2 to run — resolving `st2` from `PATH` would let a replaced control plane and its live agents -//! disagree, which the R11 control-plane replacement guarantee exists to prevent — so the wrapper -//! exports its own executable path. The second is the catalog and identity the channel must bind. -//! -//! The wrapper also owns presence for the same reason the Claude wrapper does: the extension lives -//! only as long as pi's process, and a measured SIGKILL of pi produces no terminal record at all -//! (`docs/vrs/.experiments/2026-08-18-pi-harness-integration.md`). Presence therefore decays by -//! staleness, exactly as for the other harnesses. +//! interactive process, and that extension reaches st2 by spawning `st2 driver pi-channel`. The +//! launch body itself is shared with omp — the family's other extension-driven harness — in +//! [`crate::pi_family_session`]; what lives here is what is genuinely pi's: its extension asset, +//! its channel variable names, and the release its harness-context arithmetic was measured on. use std::path::Path; -use std::process::ExitStatus; -use anyhow::{Context as _, Result}; +use anyhow::Result; -use crate::provider_session::{ - PROVIDER_POLL, ProviderOutcome, STOP, install_signal_handler, run_provider_observed, -}; -use crate::{harness_state, hooks, message, status}; +use crate::pi_family_session::{self, HarnessKind}; /// The extension file inside this binary's immutable hook set. const EXTENSION: &str = "pi-channel.ts"; @@ -49,12 +39,19 @@ pub const CHANNEL_SESSION: &str = "ST2_PI_CHANNEL_SESSION"; /// channel's writes act under the same directional claim. pub const CHANNEL_SEQ: &str = "ST2_PI_CHANNEL_SEQ"; -/// pi's startup network work, which a supervised seat should not be doing. -/// -/// A managed agent that update-checks or self-updates at boot makes its own launch latency and its -/// own behaviour depend on the network, and lets a release change a running fleet. Each is applied -/// only when the operator has not already set it, so a declaration's `env` still wins. -const OFFLINE_DEFAULTS: [(&str, &str); 2] = [("PI_OFFLINE", "1"), ("PI_SKIP_VERSION_CHECK", "1")]; +/// pi's half of the pi-family launch fork. No `verify_version`: pi gates its build in +/// `flake.nix`'s extension check rather than at launch. +pub(crate) const PI_KIND: HarnessKind = HarnessKind { + label: "pi", + extension: EXTENSION, + bin_env: CHANNEL_BIN, + catalog_env: CHANNEL_CATALOG, + identity_env: CHANNEL_IDENTITY, + runtime_id_env: CHANNEL_RUNTIME_ID, + session_env: CHANNEL_SESSION, + seq_env: CHANNEL_SEQ, + verify_version: None, +}; /// Run one interactive pi provider and maintain its presence until it exits. pub fn run( @@ -63,187 +60,17 @@ pub fn run( runtime_id: String, pi_argv: Vec, ) -> Result<()> { - let agent_dir = - message::resolve_declared_dir(catalog_root, &identity, &crate::run::detect_host())? - .with_context(|| format!("pi driver agent '{identity}' is not declared"))?; - anyhow::ensure!( - !pi_argv.is_empty(), - "pi driver '{runtime_id}' has no provider argv" - ); - let executable = - std::env::current_exe().context("resolving st2 executable for the pi channel")?; - let session = harness_state::session_token(); - // The claim is written: it supersedes whatever the predecessor left — including a - // still-fresh live record — before the channel or terminal writer act under it. - let seq = harness_state::claim(&agent_dir, identity.clone(), "pi", &session)?; - // Every fallible step past the claim must end the record honestly on failure — the claim - // placeholder standing as the last word would read as a takeover, not a launch that never - // ran. - let prepared = (|| -> Result<(Vec<(String, String)>, Vec)> { - let mut env = channel_env( - &executable, - catalog_root, - &identity, - &runtime_id, - &session, - seq, - )?; - env.extend(offline_defaults(|key| std::env::var_os(key).is_some())); - let set = hooks::verify_required_set().with_context(|| { - format!( - "pi driver '{runtime_id}' needs this binary's verified hook set for {EXTENSION}; run `st2 hooks install`" - ) - })?; - Ok((env, with_channel_extension(pi_argv, &set)?)) - })(); - let (env, pi_argv) = match prepared { - Ok(prepared) => prepared, - Err(error) => { - let mut writer = harness_state::Writer::new( - &agent_dir, - identity.clone(), - "pi", - Some(runtime_id.clone()), - ) - .with_ownership(session.clone(), seq); - let _ = writer.observe( - harness_state::Observation::new( - harness_state::Activity::Ended, - harness_state::BlockedOn::None, - harness_state::InputBuffer::Unknown, - ) - .with_reason("launch-error") - .with_exit("exit unknown"), - ); - return Err(error); - } - }; - install_signal_handler(); - // Terminal-only: the channel owns the live record and its heartbeat, but only this wrapper - // survives long enough to see the stop path — its pre-escalation `ended` write is the one - // that makes `Stopped(None)` observable at all. Same token as the channel, so the terminal - // record fences exactly this session's live records. - let observer = crate::provider_session::SessionObserver::terminal_only( - &agent_dir, - &identity, - "pi", - &runtime_id, - &session, - seq, - ); - let outcome = run_provider_observed( - "pi", - &status::status_path(&agent_dir), - &pi_argv, - &env, - status::STATUS_REFRESH, - PROVIDER_POLL, - &STOP, - Some(&observer), - ) - .with_context(|| format!("running pi driver '{runtime_id}'"))?; - record_session_end(&agent_dir, &identity, &runtime_id, &session, seq, &outcome); - match outcome { - ProviderOutcome::Exited(exit) => { - anyhow::ensure!(exit.success(), "pi provider exited with {exit}"); - Ok(()) - } - ProviderOutcome::Stopped(_) => Ok(()), - } -} - -/// The wrapper's one write into observed harness state: the terminal record. Live states and -/// heartbeats belong to the pi channel, which sees pi's own turn events over stdio; the wrapper -/// sees exactly one fact the channel cannot — that the provider process is gone — so that is the -/// one fact it records. The `Writer` is constructed at the terminal edge on purpose: it re-reads -/// whatever the channel last wrote and continues its transition counter, and by the time the -/// wrapper has reaped pi the extension (and with it the channel) is already gone. -fn record_session_end( - agent_dir: &Path, - identity: &str, - runtime_id: &str, - session: &str, - seq: u64, - outcome: &ProviderOutcome, -) { - let label = match outcome { - ProviderOutcome::Exited(exit) | ProviderOutcome::Stopped(Some(exit)) => exit_label(*exit), - ProviderOutcome::Stopped(None) => "stopped".to_string(), - }; - let mut writer = - harness_state::Writer::new(agent_dir, identity, "pi", Some(runtime_id.to_string())) - .with_ownership(session, seq); - if let Err(error) = writer.ended(label) { - tracing::warn!("st2 pi driver: recording session end failed: {error}"); - } -} - -fn exit_label(exit: ExitStatus) -> String { - use std::os::unix::process::ExitStatusExt as _; - match (exit.code(), exit.signal()) { - (Some(code), _) => format!("exit {code}"), - (None, Some(signal)) => format!("signal {signal}"), - (None, None) => "exited".to_string(), - } -} - -/// Load the channel extension from the verified set, immediately after the provider program. -/// -/// The declaration deliberately carries no path to it: a rendered machine-local path would pin one -/// host's layout into a catalog, and a `$ST_HOOKS` token in an argv would resolve to the -/// receipt-bearing root rather than the selected set. Resolving it here means a launch uses the -/// exact asset this binary was built with. -fn with_channel_extension(mut argv: Vec, set: &Path) -> Result> { - let extension = set.join(EXTENSION); - let extension = extension - .to_str() - .context("verified hook set path is not UTF-8")? - .to_owned(); - argv.splice(1..1, ["-e".to_string(), extension]); - Ok(argv) -} - -/// The offline defaults this launch should add, skipping any the operator already declared. -fn offline_defaults(is_set: impl Fn(&str) -> bool) -> Vec<(String, String)> { - OFFLINE_DEFAULTS - .iter() - .filter(|(key, _)| !is_set(key)) - .map(|(key, value)| ((*key).to_string(), (*value).to_string())) - .collect() -} - -/// The environment the shipped pi extension reads to reach this exact control plane. -fn channel_env( - executable: &Path, - catalog_root: &Path, - identity: &str, - runtime_id: &str, - session: &str, - seq: u64, -) -> Result> { - let executable = executable - .to_str() - .context("st2 executable path is not UTF-8")?; - let catalog_root = catalog_root.to_str().context("catalog root is not UTF-8")?; - Ok(vec![ - (CHANNEL_BIN.to_string(), executable.to_string()), - (CHANNEL_CATALOG.to_string(), catalog_root.to_string()), - (CHANNEL_IDENTITY.to_string(), identity.to_string()), - (CHANNEL_RUNTIME_ID.to_string(), runtime_id.to_string()), - (CHANNEL_SESSION.to_string(), session.to_string()), - (CHANNEL_SEQ.to_string(), seq.to_string()), - ]) + pi_family_session::run_for(catalog_root, identity, runtime_id, pi_argv, &PI_KIND) } #[cfg(test)] mod tests { use std::fs; - use std::path::PathBuf; use std::sync::atomic::AtomicBool; use std::time::Duration; - use super::*; - use crate::provider_session::run_provider; + use crate::provider_session::{ProviderOutcome, run_provider}; + use crate::status; #[test] fn idle_pi_provider_refreshes_presence_without_channel_input() { @@ -273,57 +100,6 @@ mod tests { assert_eq!(status::read_state(&presence), status::State::Available); } - /// The wrapper writes the one observation the channel cannot: the terminal record, carrying - /// the exit. It continues the transition counter of whatever the channel last wrote, so the - /// death of a session is a transition in the same record, not a new history. - #[test] - fn provider_exit_writes_the_terminal_record_with_its_status() { - use std::os::unix::process::ExitStatusExt as _; - - let tmp = tempfile::tempdir().unwrap(); - let agent_dir = tmp.path(); - let mut channel_writer = - crate::harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); - channel_writer - .observe(crate::harness_state::Observation::new( - crate::harness_state::Activity::Active, - crate::harness_state::BlockedOn::None, - crate::harness_state::InputBuffer::Unknown, - )) - .unwrap(); - drop(channel_writer); - - record_session_end( - agent_dir, - "h.worker", - "h.worker", - "session-test", - 1, - &ProviderOutcome::Exited(ExitStatus::from_raw(3 << 8)), - ); - - let record = crate::harness_state::harness_state_path(agent_dir); - let observed = crate::harness_state::read(&record, None).unwrap(); - assert_eq!(observed.state, crate::harness_state::Activity::Ended); - assert_eq!(observed.exit.as_deref(), Some("exit 3")); - let raw: serde_json::Value = serde_json::from_slice(&fs::read(&record).unwrap()).unwrap(); - assert_eq!( - raw["transitions"], 1, - "counter continues the channel's record" - ); - - record_session_end( - agent_dir, - "h.worker", - "h.worker", - "session-test", - 1, - &ProviderOutcome::Stopped(Some(ExitStatus::from_raw(9))), - ); - let observed = crate::harness_state::read(&record, None).unwrap(); - assert_eq!(observed.exit.as_deref(), Some("signal 9")); - } - /// The observed variant reports a nonzero exit instead of judging it, which is what lets the /// wrapper record the terminal state before failing the launch. #[test] @@ -347,80 +123,6 @@ mod tests { } } - #[test] - fn the_channel_extension_is_injected_from_the_verified_set_not_the_declaration() { - let argv = with_channel_extension( - vec![ - "pi".into(), - "-a".into(), - "--model".into(), - "anthropic/opus".into(), - "Start work.".into(), - ], - &PathBuf::from("/state/st2/hooks/sets/sha256-abc"), - ) - .unwrap(); - - assert_eq!( - argv, - vec![ - "pi", - "-e", - "/state/st2/hooks/sets/sha256-abc/pi-channel.ts", - "-a", - "--model", - "anthropic/opus", - "Start work.", - ] - ); - } - - /// A supervised seat is offline by default, but an operator who declared otherwise keeps their - /// value — otherwise the wrapper would silently overrule the declaration. - #[test] - fn offline_defaults_apply_only_where_the_operator_declared_nothing() { - assert_eq!( - offline_defaults(|_| false), - vec![ - ("PI_OFFLINE".to_string(), "1".to_string()), - ("PI_SKIP_VERSION_CHECK".to_string(), "1".to_string()), - ] - ); - assert_eq!( - offline_defaults(|key| key == "PI_OFFLINE"), - vec![("PI_SKIP_VERSION_CHECK".to_string(), "1".to_string())] - ); - assert!(offline_defaults(|_| true).is_empty()); - } - - #[test] - fn the_extension_receives_this_binary_not_a_path_lookup() { - let env = channel_env( - &PathBuf::from("/opt/st2/bin/st2"), - &PathBuf::from("/catalog"), - "host.worker", - "host.worker-task", - "session-test", - 7, - ) - .unwrap(); - - assert_eq!( - env, - vec![ - (CHANNEL_BIN.to_string(), "/opt/st2/bin/st2".to_string()), - (CHANNEL_CATALOG.to_string(), "/catalog".to_string()), - (CHANNEL_IDENTITY.to_string(), "host.worker".to_string()), - ( - CHANNEL_RUNTIME_ID.to_string(), - "host.worker-task".to_string() - ), - (CHANNEL_SESSION.to_string(), "session-test".to_string()), - (CHANNEL_SEQ.to_string(), "7".to_string()), - ] - ); - } - /// W6: the terminal-only observer records how the session ended but never re-stamps live /// state — the channel owns the heartbeat — and its token makes the write this session's. #[test] diff --git a/src/provider_session.rs b/src/provider_session.rs index 09648ff3..2c645993 100644 --- a/src/provider_session.rs +++ b/src/provider_session.rs @@ -164,7 +164,15 @@ impl SessionObserver { } } -fn describe_exit(exit: ExitStatus) -> String { +/// The one exit-label map: how every wrapper spells a reaped child's outcome into the observed +/// record's `exit` field. +/// +/// `(None, None)` is unreachable for a child this process reaped — `Child::wait`/`try_wait` call +/// `waitpid` with neither `WUNTRACED` nor `WCONTINUED`, so every status they return satisfies +/// `WIFEXITED` or `WIFSIGNALED` — but the label still has to be honest for the arm the type +/// admits, which is why it says "unknown" rather than inventing an ordinary exit. +/// `the_exit_label_map_covers_every_arm` pins the whole table. +pub(crate) fn describe_exit(exit: ExitStatus) -> String { match (exit.code(), exit.signal()) { (Some(code), _) => format!("exit {code}"), (None, Some(signal)) => format!("signal {signal}"), @@ -280,7 +288,7 @@ pub(crate) fn run_provider_observed( } } -fn completed_provider(provider: &str, exit: ExitStatus) -> Result<()> { +pub(crate) fn completed_provider(provider: &str, exit: ExitStatus) -> Result<()> { anyhow::ensure!(exit.success(), "{provider} provider exited with {exit}"); Ok(()) } @@ -348,4 +356,35 @@ mod tests { assert_eq!(record.exit.as_deref(), Some("exit unknown")); assert_eq!(record.reason.as_deref(), Some("launch-error")); } + + /// The whole `(status, signal) -> label` table, including the `(None, None)` arm no reaped + /// child can produce. Every wrapper's terminal record now spells its exit through this map, + /// so the table is the contract: a fifth harness that wants a different word has to change + /// it here, in front of this test, rather than forking a private copy that quietly disagrees. + #[test] + fn the_exit_label_map_covers_every_arm() { + // `wait_status` values as `waitpid` yields them: `code << 8` for an ordinary exit, + // the bare signal number for a killed child, and `0x7f` for the stopped shape only + // `WUNTRACED` could deliver — which is what makes `(None, None)` constructible at all. + for (raw, label) in [ + (0, "exit 0"), + (3 << 8, "exit 3"), + (127 << 8, "exit 127"), + (libc::SIGKILL, "signal 9"), + (libc::SIGTERM, "signal 15"), + (0x7f, "exit unknown"), + ] { + let exit = ExitStatus::from_raw(raw); + assert_eq!( + describe_exit(exit), + label, + "raw wait status {raw:#x} => {:?}/{:?}", + exit.code(), + exit.signal() + ); + } + // The arm the label calls unknown really is the one neither half of the pair answers. + let stopped = ExitStatus::from_raw(0x7f); + assert_eq!((stopped.code(), stopped.signal()), (None, None)); + } }