From f99bc5d16557320235cee12bc619b4e068fef2ee Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:14:06 +0200 Subject: [PATCH 1/4] fix(omp): log the session-end failure instead of painting the agent's terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `provider_session` spawns the provider with `stderr(Stdio::inherit())`, so this `eprintln!` painted "recording session end failed" straight into the interactive harness's own terminal — inside the agent's UI, where a human is reading a session, not a log. pi's twin already used `tracing::warn!`. It lands first because the pi/omp unify that follows deletes this line: with the two copies merged there is exactly one sink, and this is the one they should share. No test asserts the wrapper's stderr, so this is review-pinned only. agent-identity: dev3.direct.omp.43sz6ujq agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.1.7 agent-runtime: OMP 18.1.7 tooling-profile: dotfiles@39a19af --- src/omp_session.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/omp_session.rs b/src/omp_session.rs index 5bd8b0ec..adc2ee3f 100644 --- a/src/omp_session.rs +++ b/src/omp_session.rs @@ -209,7 +209,7 @@ fn record_session_end( 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}"); + tracing::warn!("st2 omp driver: recording session end failed: {error}"); } } From 207795752e4b136625d1fdc22aa2f832a63e2012 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:14:25 +0200 Subject: [PATCH 2/4] refactor(driver): unify the pi and omp session scaffolding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pi and omp are the same harness family launched the same way: an extension loaded into the interactive process that reaches st2 by spawning `st2 driver -channel`. `omp_session.rs` carried a forked copy of the whole launch body, and the two copies had already drifted in doc rationale and in style without any behaviour differing. The fork now lives where the family's channel side already puts it: a `HarnessKind` descriptor plus a shared `run_for`, mirroring `pi_channel`'s `ChannelKind` — the same fork solved the same way twice rather than two different ways. What stays per-harness is what genuinely is: each module's doc, its extension asset, its channel variable NAMES (two disjoint sets so an omp seat can never adopt a stray pi configuration — `pi_channel` also reads them by path), omp's measured-build fixtures, and omp's launch-time version gate. That gate rides on the descriptor as `verify_version` and runs exactly where omp ran it: after the empty-argv check and before `harness_state::claim`, so an unadmitted minor still fails without claiming ownership of the seat. Where the two copies carried different rationales for identical code, the merged item carries both halves: the offline defaults keep pi's "a supervised seat must not update-check at boot" and omp's open DQ-OMP-5 note. No behaviour change: every string, ordering and write is preserved — the exit label still spells `(None, None)` as "exited" here, and the session-end warning already went to the tracing log on both sides as of the preceding commit. agent-identity: dev3.direct.omp.43sz6ujq agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.1.7 agent-runtime: OMP 18.1.7 tooling-profile: dotfiles@39a19af --- src/lib.rs | 3 + src/omp_session.rs | 230 +++--------------- src/pi_family_session.rs | 490 +++++++++++++++++++++++++++++++++++++++ src/pi_session.rs | 342 ++------------------------- 4 files changed, 545 insertions(+), 520 deletions(-) create mode 100644 src/pi_family_session.rs diff --git a/src/lib.rs b/src/lib.rs index c5b9e121..2c82f7a9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,6 +45,9 @@ 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/omp_session.rs b/src/omp_session.rs index adc2ee3f..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) { - tracing::warn!("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/pi_family_session.rs b/src/pi_family_session.rs new file mode 100644 index 00000000..992b515d --- /dev/null +++ b/src/pi_family_session.rs @@ -0,0 +1,490 @@ +//! 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 std::process::ExitStatus; + +use anyhow::{Context as _, Result}; + +use crate::provider_session::{ + PROVIDER_POLL, ProviderOutcome, STOP, 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)) => exit_label(*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 + ); + } +} + +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, + 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] From 5fea6b3a19917b627647d71ec1fbb6072f66acd6 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:14:35 +0200 Subject: [PATCH 3/4] refactor(driver): one exit-label map MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five copies of `(status, signal) -> label` existed; the pi/omp unify folded two of them. The remaining copies disagreed on exactly one arm: pi/omp spelled `(None, None)` as "exited", everyone else as "exit unknown". `provider_session::describe_exit` becomes the one map, and codex's `describe_tui_exit` keeps only what is genuinely its own — flattening a status that may never have been observable at all. Behaviour change, deliberate: the pi/omp record's `exit` field spells `(None, None)` as "exit unknown" instead of "exited". "exited" claimed an ordinary end for a status that answers neither question, which is the one thing that arm cannot know. Two facts de-risk it. The arm 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`; only `ExitStatus::from_raw(0x7f)` constructs it, in a test. And no reader parses the label back — `agents::ObservedJson` and `main.rs` pass it to display, `harness_state` only asks `exit.is_some()`. The pin the report claimed existed did not: `provider_session.rs`'s assertion is fed by `SessionObserver::launch_error()`'s hardcoded literal and never reaches this map. `the_exit_label_map_covers_every_arm` is that missing pin — the whole table, `(None, None)` included. agent-identity: dev3.direct.omp.43sz6ujq agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.1.7 agent-runtime: OMP 18.1.7 tooling-profile: dotfiles@39a19af --- src/codex_app_server.rs | 12 ++++++------ src/opencode_session.rs | 11 +---------- src/pi_family_session.rs | 17 +++++------------ src/provider_session.rs | 41 +++++++++++++++++++++++++++++++++++++++- 4 files changed, 52 insertions(+), 29 deletions(-) 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/opencode_session.rs b/src/opencode_session.rs index 750a5dce..5c9a6f08 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,7 @@ 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, 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 @@ -457,14 +456,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_family_session.rs b/src/pi_family_session.rs index 992b515d..6b76f878 100644 --- a/src/pi_family_session.rs +++ b/src/pi_family_session.rs @@ -20,12 +20,12 @@ //! omp, its version gate. use std::path::Path; -use std::process::ExitStatus; use anyhow::{Context as _, Result}; use crate::provider_session::{ - PROVIDER_POLL, ProviderOutcome, STOP, install_signal_handler, run_provider_observed, + PROVIDER_POLL, ProviderOutcome, STOP, describe_exit, install_signal_handler, + run_provider_observed, }; use crate::{harness_state, hooks, message, status}; @@ -199,7 +199,9 @@ fn record_session_end( kind: &HarnessKind, ) { let label = match outcome { - ProviderOutcome::Exited(exit) | ProviderOutcome::Stopped(Some(exit)) => exit_label(*exit), + ProviderOutcome::Exited(exit) | ProviderOutcome::Stopped(Some(exit)) => { + describe_exit(*exit) + } ProviderOutcome::Stopped(None) => "stopped".to_string(), }; let mut writer = harness_state::Writer::new( @@ -217,15 +219,6 @@ fn record_session_end( } } -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 diff --git a/src/provider_session.rs b/src/provider_session.rs index 09648ff3..cbe0704d 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}"), @@ -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)); + } } From d9c13c693cdb57842f80bb21793553dfb5b1758a Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:14:46 +0200 Subject: [PATCH 4/4] refactor(driver): fold the byte-identical adapter copies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five class-(a) duplications, each verified identical before folding: - `channel_content` and `write_json` (claude_mcp, pi_channel) were byte-identical. Both are stdio-native-channel framing decisions st2 owns rather than each channel, so they move to `native_channel`. The NDJSON framing gains the pin it never had: one newline-terminated line per value, including a body that carries its own newline. - `ProviderAuthEdge` had two byte-identical definitions; one moves into `driver_diagnostic`, which already owns `Driver`/`Stage`/`Reason`/`Source`/ `Support`. - `publish_provider_auth` was identical modulo a hardcoded `Driver`, so it joins the enum and takes the driver as an argument, as pi_channel's copy already did. - opencode's `completed` folds onto `completed_provider("opencode", …)`: "{provider} provider exited with {exit}" reproduces its message byte for byte. Where the two copies carried different rationales, the merged item carries both: `publish_provider_auth`'s `None`/`Support::Unknown` now states both why a Claude hook payload has no version to publish AND why a pi-family channel has none (the wrapper owns the gate, OMP-R05). `provider_auth_edge` stays forked in both modules on purpose: the two decode disjoint inputs — a hook event name plus JSON payload versus a typed `TurnResult` — and share only the output enum this commit unifies. agent-identity: dev3.direct.omp.43sz6ujq agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.1.7 agent-runtime: OMP 18.1.7 tooling-profile: dotfiles@39a19af --- src/claude_mcp.rs | 31 ++----------------- src/claude_session.rs | 50 ++++++++----------------------- src/driver_diagnostic.rs | 40 +++++++++++++++++++++++++ src/lib.rs | 2 ++ src/native_channel.rs | 62 ++++++++++++++++++++++++++++++++++++++ src/opencode_session.rs | 11 +++---- src/pi_channel.rs | 65 ++-------------------------------------- src/provider_session.rs | 2 +- 8 files changed, 127 insertions(+), 136 deletions(-) create mode 100644 src/native_channel.rs 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/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 2c82f7a9..64744c32 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -39,6 +39,8 @@ 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; 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/opencode_session.rs b/src/opencode_session.rs index 5c9a6f08..7f5a3d29 100644 --- a/src/opencode_session.rs +++ b/src/opencode_session.rs @@ -37,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, describe_exit, 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 @@ -251,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) => { @@ -429,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!( 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/provider_session.rs b/src/provider_session.rs index cbe0704d..2c645993 100644 --- a/src/provider_session.rs +++ b/src/provider_session.rs @@ -288,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(()) }