diff --git a/flake.nix b/flake.nix index 01d6a133..46dd2c17 100644 --- a/flake.nix +++ b/flake.nix @@ -287,6 +287,15 @@ grep -q '@earendil-works/pi-coding-agent' hooks/pi-channel.ts tsc --noEmit -p hooks/typecheck/tsconfig.json + + # Runtime smoke: the type gate is provably blind to execution-order defects (a TDZ + # use-before-declaration shipped green through it), so the asset is transpiled and + # actually driven through its open path. + ${pkgs.esbuild}/bin/esbuild hooks/pi-channel.ts \ + --format=esm --platform=node --target=es2022 \ + --outfile=hooks/typecheck/smoke-out/pi-channel.mjs + SMOKE_TRUE_BIN=${pkgs.coreutils}/bin/true \ + ${pkgs.nodejs}/bin/node hooks/typecheck/smoke.mjs touch $out ''; diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index 60530123..d6911d62 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -25,6 +25,9 @@ const PROTOCOL = 1; const BIN = "ST2_PI_CHANNEL_BIN"; const CATALOG = "ST2_PI_CHANNEL_CATALOG"; const IDENTITY = "ST2_PI_CHANNEL_IDENTITY"; +const RUNTIME_ID = "ST2_PI_CHANNEL_RUNTIME_ID"; +const SESSION = "ST2_PI_CHANNEL_SESSION"; +const SEQ = "ST2_PI_CHANNEL_SEQ"; // pi starts the session even if st2 is slow to answer. Restored context is worth a short wait and // never worth a hung agent. @@ -51,6 +54,9 @@ type Stash = { bin?: string; catalog?: string; identity?: string; + runtimeId?: string; + session?: string; + seq?: string; child?: childProcess.ChildProcess; }; @@ -67,26 +73,46 @@ type Stash = { const stash = (): Stash => { const globals = globalThis as { __st2PiChannel?: Stash }; if (!globals.__st2PiChannel) { + // EVERY ST2_PI_CHANNEL_* value is stashed and unexported — the ownership pair included: a + // leaked runtime id or session token would hand a nested pi (or any tool child) this seat's + // registry key and record ownership. The channel subprocess receives them explicitly below. globals.__st2PiChannel = { bin: process.env[BIN], catalog: process.env[CATALOG], identity: process.env[IDENTITY], + runtimeId: process.env[RUNTIME_ID], + session: process.env[SESSION], + seq: process.env[SEQ], }; delete process.env[BIN]; delete process.env[CATALOG]; delete process.env[IDENTITY]; + delete process.env[RUNTIME_ID]; + delete process.env[SESSION]; + delete process.env[SEQ]; } return globals.__st2PiChannel; }; export default function (pi: ExtensionAPI) { const state = stash(); - const { bin, catalog, identity } = state; + const { bin, catalog, identity, runtimeId, session, seq } = state; // Always close a NAMED channel, never "whatever is current". A session replacement (/new, // /resume, /fork) tears the old session down around the new one's start, so a teardown handler // that closed `current` would reap the successor it just opened — measured, and it silently // stopped all delivery after `/new`. + const awaitExit = (child: childProcess.ChildProcess | undefined, ms: number) => + new Promise((resolve) => { + if (!child || child.exitCode !== null || child.signalCode !== null) return resolve(); + const timer = setTimeout(resolve, ms); + timer.unref?.(); + child.once("exit", () => { + clearTimeout(timer); + resolve(); + }); + }); + const closeChild = (child: childProcess.ChildProcess | undefined) => { if (!child) return; if (state.child === child) state.child = undefined; @@ -96,7 +122,7 @@ export default function (pi: ExtensionAPI) { }; /** Open a channel and resolve with the hello's restored context (empty if none, or on timeout). */ - const open = (ctx: ExtensionContext): Promise => { + const open = async (ctx: ExtensionContext): Promise => { if (!bin || !catalog || !identity) return Promise.resolve(""); if (typeof ctx.isIdle !== "function") { // Refuse rather than degrade. Without a positive idle proof this extension cannot choose @@ -109,13 +135,22 @@ export default function (pi: ExtensionAPI) { ); return Promise.resolve(""); } - // Closes the channel opened by the PREVIOUS session, whichever extension instance opened it. - closeChild(state.child); + // Closes the channel opened by the PREVIOUS session, whichever extension instance opened + // it — and WAITS (bounded) for it to exit before the replacement spawns: the successor + // shares the seat's record, and a predecessor draining its queued frames after the new + // session's seed would land stale state into fresh records. + const previous = state.child; + closeChild(previous); + await awaitExit(previous, 2000); + const channelEnv: NodeJS.ProcessEnv = { ...process.env }; + if (runtimeId) channelEnv[RUNTIME_ID] = runtimeId; + if (session) channelEnv[SESSION] = session; + if (seq) channelEnv[SEQ] = seq; const child = childProcess.spawn( bin, ["--catalog", catalog, "driver", "pi-channel", "--identity", identity], - { stdio: ["pipe", "pipe", "inherit"] }, + { stdio: ["pipe", "pipe", "inherit"], env: channelEnv }, ); state.child = child; @@ -132,6 +167,12 @@ export default function (pi: ExtensionAPI) { if (state.child === child) state.child = undefined; settle(""); }); + // An observability pipe must never take pi down: a channel that closed its stdin mid-write + // surfaces EPIPE on the stream, which without a listener is an uncaught exception in the + // host process. Retire the channel instead — frames simply stop, fail-open. + child.stdin.on("error", () => { + if (state.child === child) state.child = undefined; + }); child.on("exit", () => settle("")); const send = (frame: Record) => { @@ -197,11 +238,32 @@ export default function (pi: ExtensionAPI) { }); }; + // Observed harness state, extension side. pi's own turn boundaries are the positive signal, + // and the idle edge is `agent_settled`, not `agent_end`: measured against the repo's own pi + // captures, `ctx.isIdle()` is still false through `agent_end`, and a queued follow-up turn + // starts exactly at that boundary — an `agent_end` emit would blip a spurious idle before it. + // `agent_settled` is the first point pi is provably idle. The frame is observational — st2 + // decides what becomes of it — and a closed channel drops it silently, matching the fail-open + // rule this file already follows. pi 0.84.2 exposes no typed waiting-on-a-human event, so no + // frame here ever claims one. + const sendState = (word: "active" | "idle") => { + const child = state.child; + if (!child || !child.stdin || child.stdin.destroyed) return; + child.stdin.write(JSON.stringify({ type: "state", state: word }) + "\n"); + }; + pi.on("agent_start", async () => sendState("active")); + pi.on("agent_settled", async () => sendState("idle")); + pi.on("session_start", async (_event, ctx) => { // Awaited before the session's first turn, which is what makes restored context reach the boot // prompt rather than the turn after it. const restored = await open(ctx); const opened = state.child; + // Seed the observed state with the idle proof's answer at open time, so the record does not + // wait for the first turn boundary to exist. + if (opened && typeof ctx.isIdle === "function") { + sendState(ctx.isIdle() ? "idle" : "active"); + } if (restored.trim()) { // A custom message participates in LLM context without triggering a turn of its own — the // closest pi equivalent to the other harnesses' `additionalContext` hook output. diff --git a/hooks/typecheck/smoke.mjs b/hooks/typecheck/smoke.mjs new file mode 100644 index 00000000..bfa76367 --- /dev/null +++ b/hooks/typecheck/smoke.mjs @@ -0,0 +1,38 @@ +// Runtime smoke of the shipped pi extension: drives the channel-open path far enough that a +// use-before-declaration (TDZ), a broken import, or a top-level throw fails the check — the +// classes a type-only gate is provably blind to. The channel binary is `true`, so the open +// times out its hello and resolves empty; any thrown error fails the smoke. +import assert from "node:assert"; + +process.env.ST2_PI_CHANNEL_BIN = process.env.SMOKE_TRUE_BIN ?? "/bin/true"; +process.env.ST2_PI_CHANNEL_CATALOG = "/tmp/st2-smoke-catalog"; +process.env.ST2_PI_CHANNEL_IDENTITY = "smoke.worker"; +process.env.ST2_PI_CHANNEL_RUNTIME_ID = "smoke.worker"; +process.env.ST2_PI_CHANNEL_SESSION = "smoke-session"; +process.env.ST2_PI_CHANNEL_SEQ = "1"; + +const mod = await import("./smoke-out/pi-channel.mjs"); +assert.strictEqual(typeof mod.default, "function", "extension exports its entry point"); + +const handlers = new Map(); +const pi = { + on: (name, handler) => handlers.set(name, handler), +}; +mod.default(pi); +for (const name of ["session_start", "session_shutdown", "agent_start", "agent_settled"]) { + assert.ok(handlers.has(name), `extension registers ${name}`); +} + +const ctx = { + isIdle: () => true, + ui: { notify: () => {} }, +}; +// Two session starts in a row: the second exercises the predecessor close-and-await path — the +// exact region the TDZ regression lived in. +await handlers.get("session_start")({}, ctx); +await handlers.get("session_start")({}, ctx); +await handlers.get("agent_start")({}, ctx); +await handlers.get("agent_settled")({}, ctx); +await handlers.get("session_shutdown")({ reason: "smoke" }, ctx); +console.log("pi extension smoke: ok"); +process.exit(0); diff --git a/src/pi_channel.rs b/src/pi_channel.rs index 7f31aae9..e0eedd4c 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -12,14 +12,14 @@ use std::collections::HashSet; use std::io::{self, BufRead, Write}; use std::path::Path; -use std::sync::mpsc::{self, RecvTimeoutError}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::{Context as _, Result}; use serde_json::{Value, json}; -use crate::{context, message}; +use crate::{context, harness_state, message}; const POLL: Duration = Duration::from_millis(250); @@ -88,34 +88,126 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { }), )?; stdout.flush()?; + // The channel owns the live half of observed harness state: it is the one process that sees + // pi's own turn events, and its stdio connection to the extension is the evidence that those + // events are still being watched. The terminal half belongs to the outer session wrapper, + // which alone sees the provider die. + // The pty session vouching for the record is the wrapper's task: its runtime ID arrives in + // the channel environment, and only aliases the identity on driver-expanded seats. + let pty_session = std::env::var(crate::pi_session::CHANNEL_RUNTIME_ID) + .ok() + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| identity.to_string()); + // The wrapper mints the session token; adopting it makes the wrapper's terminal record own + // this channel's live records (so a queued frame after `ended` is suppressed) while a + // predecessor incarnation's records are foreign: the first frame opens a fresh transition + // and a predecessor's terminal record never silences this session. + let mut writer = harness_state::Writer::new(&agent_dir, identity, "pi", Some(pty_session)); + if let Ok(session) = std::env::var(crate::pi_session::CHANNEL_SESSION) + && !session.is_empty() + { + // Full adopted ownership when the wrapper exported it: the claimed sequence gives the + // token a direction, so a straggler channel from a superseded session is refused. + writer = match std::env::var(crate::pi_session::CHANNEL_SEQ) + .ok() + .and_then(|seq| seq.parse::().ok()) + { + Some(seq) => writer.with_ownership(session, seq), + None => writer.with_session(session), + }; + } + writer.interrupt(); + channel_loop( + &input_rx, + &mut stdout, + &inbox, + &mut writer, + identity, + POLL, + harness_state::HARNESS_STATE_REFRESH, + ) +} + +/// The channel's steady state: forward inbox entries out, fold extension frames in, and keep the +/// observed-state heartbeat exactly as fresh as the stdio connection that justifies it. EOF is the +/// session-lifetime boundary — the loop then returns without writing anything, so the record ages +/// to `unknown` rather than asserting a state nobody is watching. +fn channel_loop( + input: &Receiver>, + out: &mut impl Write, + inbox: &Path, + writer: &mut harness_state::Writer, + identity: &str, + poll: Duration, + heartbeat_every: Duration, +) -> Result<()> { let mut delivered = HashSet::new(); + let mut next_heartbeat = Instant::now() + heartbeat_every; loop { - match input_rx.recv_timeout(POLL) { + match input.recv_timeout(poll) { Ok(line) => { let line = line.context("reading pi channel input")?; if line.trim().is_empty() { continue; } - // Frames from the extension are observational today. An unknown *or malformed* + // Frames from the extension are otherwise observational. An unknown *or malformed* // frame is dropped rather than fatal: a newer asset, or one line of stray output, - // must not be able to take the channel down and stall an inbox. - let _ = serde_json::from_str::(&line); + // must not be able to take the channel down and stall an inbox. A failed record + // write degrades the same way — delivery never depends on observability. + if let Some(observation) = serde_json::from_str::(&line) + .ok() + .as_ref() + .and_then(state_observation) + // A queued live frame must never overwrite the wrapper's terminal record: + // the channel and the wrapper are separate processes, so the flock alone + // serializes but does not order their writes. + && let Err(error) = writer.observe_unless_ended(observation) + { + eprintln!("st2 pi channel: recording observed state failed: {error}"); + } } Err(RecvTimeoutError::Timeout) => {} // pi's extension owns this child over stdio. EOF is the session-lifetime boundary, so // do not leave a detached watcher behind. Err(RecvTimeoutError::Disconnected) => return Ok(()), } - for msg in message::list_inbox(&inbox)? { + let now = Instant::now(); + if now >= next_heartbeat { + if let Err(error) = writer.heartbeat() { + eprintln!("st2 pi channel: refreshing observed state failed: {error}"); + } + next_heartbeat = now + heartbeat_every; + } + for msg in message::list_inbox(inbox)? { if delivered.insert(msg.filename.clone()) { - write_json(&mut stdout, &message_frame(msg, identity))?; + write_json(out, &message_frame(msg, identity))?; } } - stdout.flush()?; - thread::sleep(POLL); + out.flush()?; + thread::sleep(poll); } } +/// The observed-state frame the shipped extension emits on pi's own turn boundaries. Only +/// positively recognized words become observations: an unrecognized state word is dropped like any +/// other unknown frame, so a newer asset cannot make this channel record something it cannot vouch +/// for. pi offers no waiting-on-a-human signal, so no frame sets `blockedOn` here. +fn state_observation(frame: &Value) -> Option { + if frame.get("type").and_then(Value::as_str) != Some("state") { + return None; + } + let state = match frame.get("state").and_then(Value::as_str)? { + "active" => harness_state::Activity::Active, + "idle" => harness_state::Activity::Idle, + _ => return None, + }; + Some(harness_state::Observation::new( + state, + harness_state::BlockedOn::None, + harness_state::InputBuffer::Unknown, + )) +} + /// What a starting or restarting pi session is told about its own durable state. /// /// pi has no session-start hook, so this is the payload that stands in for @@ -171,6 +263,120 @@ fn write_json(out: &mut impl Write, value: &Value) -> Result<()> { mod tests { use super::*; + /// Only the two words pi's own turn boundaries can vouch for become observations. Everything + /// else — other frame types, unknown state words, missing fields — is dropped, so a newer + /// extension asset cannot push this channel into recording something it cannot prove. + #[test] + fn only_recognized_state_frames_become_observations() { + let active = state_observation(&json!({"type":"state","state":"active"})).unwrap(); + assert_eq!(active.state, harness_state::Activity::Active); + assert_eq!(active.blocked_on, harness_state::BlockedOn::None); + assert_eq!(active.input_buffer, harness_state::InputBuffer::Unknown); + assert_eq!( + state_observation(&json!({"type":"state","state":"idle"})) + .unwrap() + .state, + harness_state::Activity::Idle + ); + + for frame in [ + json!({"type":"state","state":"child"}), + json!({"type":"state","state":"unknown"}), + json!({"type":"state"}), + json!({"type":"delivered","state":"active"}), + json!({"state":"active"}), + ] { + assert_eq!(state_observation(&frame), None, "frame: {frame}"); + } + } + + /// The stdio connection is the evidence. While it lives, the record's heartbeat advances + /// without a new observation; when it ends, the loop returns having written nothing more, so + /// the last state is left to age to `unknown` instead of being asserted or terminated — the + /// terminal record belongs to the session wrapper, which sees the provider die. + #[test] + fn heartbeats_while_connected_then_leaves_the_record_to_age_on_eof() { + let tmp = tempfile::tempdir().unwrap(); + let agent_dir = tmp.path(); + std::fs::create_dir_all(message::inbox_dir(agent_dir)).unwrap(); + let record = harness_state::harness_state_path(agent_dir); + let mut writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); + let (tx, rx) = mpsc::channel(); + + tx.send(Ok(r#"{"type":"state","state":"active"}"#.to_string())) + .unwrap(); + let disconnect = thread::spawn(move || { + thread::sleep(Duration::from_millis(40)); + drop(tx); + }); + let mut out = Vec::new(); + channel_loop( + &rx, + &mut out, + &message::inbox_dir(agent_dir), + &mut writer, + "h.worker", + Duration::from_millis(2), + Duration::from_millis(5), + ) + .unwrap(); + disconnect.join().unwrap(); + + let raw: Value = serde_json::from_slice(&std::fs::read(&record).unwrap()).unwrap(); + assert_eq!(raw["state"], "active", "EOF must not rewrite the state"); + assert!( + raw["writtenAtMs"].as_u64().unwrap() > raw["sinceMs"].as_u64().unwrap(), + "a heartbeat re-stamped the record while the connection lived: {raw}" + ); + let after_eof = std::fs::read(&record).unwrap(); + thread::sleep(Duration::from_millis(15)); + assert_eq!( + std::fs::read(&record).unwrap(), + after_eof, + "nothing may write after the connection is gone" + ); + } + + /// The wrapper's terminal record is the incarnation's last word: a live frame the extension + /// queued before dying must not resurrect the session after the wrapper reaped it. + #[test] + fn a_queued_live_frame_never_overwrites_the_wrappers_terminal_record() { + let tmp = tempfile::tempdir().unwrap(); + let agent_dir = tmp.path(); + std::fs::create_dir_all(message::inbox_dir(agent_dir)).unwrap(); + let record = harness_state::harness_state_path(agent_dir); + // The wrapper mints the session token and the channel adopts it — that sharing is what + // makes the wrapper's terminal record this session's last word. + let session = harness_state::session_token(); + let mut channel_writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())) + .with_session(session.clone()); + let mut wrapper_writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())) + .with_session(session); + wrapper_writer.ended("signal 9").unwrap(); + let terminal = std::fs::read(&record).unwrap(); + + let (tx, rx) = mpsc::channel(); + tx.send(Ok(r#"{"type":"state","state":"idle"}"#.to_string())) + .unwrap(); + drop(tx); + let mut out = Vec::new(); + channel_loop( + &rx, + &mut out, + &message::inbox_dir(agent_dir), + &mut channel_writer, + "h.worker", + Duration::from_millis(2), + Duration::from_millis(5), + ) + .unwrap(); + + assert_eq!(std::fs::read(&record).unwrap(), terminal); + } + #[test] fn channel_content_reuses_the_claude_channel_envelope() { assert_eq!( diff --git a/src/pi_session.rs b/src/pi_session.rs index 05cfae8e..551d253c 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -13,11 +13,14 @@ //! staleness, exactly as for the other harnesses. use std::path::Path; +use std::process::ExitStatus; use anyhow::{Context as _, Result}; -use crate::provider_session::{PROVIDER_POLL, STOP, install_signal_handler, run_provider}; -use crate::{hooks, message, status}; +use crate::provider_session::{ + PROVIDER_POLL, ProviderOutcome, STOP, install_signal_handler, run_provider_observed, +}; +use crate::{harness_state, hooks, message, status}; /// The extension file inside this binary's immutable hook set. const EXTENSION: &str = "pi-channel.ts"; @@ -28,6 +31,14 @@ pub const CHANNEL_BIN: &str = "ST2_PI_CHANNEL_BIN"; pub const CHANNEL_CATALOG: &str = "ST2_PI_CHANNEL_CATALOG"; /// The host-qualified bus identity the channel binds. pub const CHANNEL_IDENTITY: &str = "ST2_PI_CHANNEL_IDENTITY"; +/// The wrapper's runtime/task ID — the pty session whose liveness vouches for observed state. +pub const CHANNEL_RUNTIME_ID: &str = "ST2_PI_CHANNEL_RUNTIME_ID"; +/// The session incarnation token the wrapper mints. The channel adopts it so the wrapper's +/// terminal record owns — and thereby fences — the live records the channel writes. +pub const CHANNEL_SESSION: &str = "ST2_PI_CHANNEL_SESSION"; +/// The ownership sequence the wrapper claimed at startup — exported beside the token so the +/// 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. /// @@ -52,16 +63,54 @@ pub fn run( ); let executable = std::env::current_exe().context("resolving st2 executable for the pi channel")?; - let mut env = channel_env(&executable, catalog_root, &identity)?; - 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`" - ) - })?; - let pi_argv = with_channel_extension(pi_argv, &set)?; + 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(); - run_provider( + let outcome = run_provider_observed( "pi", &status::status_path(&agent_dir), &pi_argv, @@ -70,7 +119,50 @@ pub fn run( PROVIDER_POLL, &STOP, ) - .with_context(|| format!("running pi driver '{runtime_id}'")) + .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) { + eprintln!("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. @@ -103,6 +195,9 @@ fn channel_env( executable: &Path, catalog_root: &Path, identity: &str, + runtime_id: &str, + session: &str, + seq: u64, ) -> Result> { let executable = executable .to_str() @@ -112,6 +207,9 @@ fn channel_env( (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()), ]) } @@ -123,6 +221,7 @@ mod tests { use std::time::Duration; use super::*; + use crate::provider_session::run_provider; #[test] fn idle_pi_provider_refreshes_presence_without_channel_input() { @@ -151,6 +250,79 @@ 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] + fn run_provider_observed_reports_the_child_exit_status() { + let tmp = tempfile::tempdir().unwrap(); + let stop = AtomicBool::new(false); + let outcome = crate::provider_session::run_provider_observed( + "pi", + &status::status_path(tmp.path()), + &["sh".into(), "-c".into(), "exit 3".into()], + &[], + Duration::from_secs(60), + Duration::from_millis(5), + &stop, + ) + .unwrap(); + match outcome { + ProviderOutcome::Exited(exit) => assert_eq!(exit.code(), Some(3)), + other => panic!("expected an exit outcome, got {other:?}"), + } + } + #[test] fn the_channel_extension_is_injected_from_the_verified_set_not_the_declaration() { let argv = with_channel_extension( @@ -203,6 +375,9 @@ mod tests { &PathBuf::from("/opt/st2/bin/st2"), &PathBuf::from("/catalog"), "host.worker", + "host.worker-task", + "session-test", + 7, ) .unwrap(); @@ -212,6 +387,12 @@ mod tests { (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()), ] ); } diff --git a/src/provider_session.rs b/src/provider_session.rs index 8d6c064e..44933278 100644 --- a/src/provider_session.rs +++ b/src/provider_session.rs @@ -41,8 +41,20 @@ pub(crate) fn install_signal_handler() { } } +/// How one provider session ended, as the wrapper saw it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ProviderOutcome { + /// The child exited on its own, with this status. + Exited(ExitStatus), + /// The wrapper stopped its process group; the reaped status when the group yielded inside the + /// grace window. `None` means the SIGKILL escalation ran — and since that kill targets the + /// wrapper's own group, code after it usually never runs at all. + Stopped(Option), +} + /// Run one interactive provider in this wrapper's terminal process group, refreshing presence on -/// `refresh_interval` for exactly as long as the spawned child lives. +/// `refresh_interval` for exactly as long as the spawned child lives. Fails on a nonzero exit; +/// wrappers that need the exit itself use [`run_provider_observed`]. pub(crate) fn run_provider( provider: &str, status_path: &Path, @@ -52,6 +64,31 @@ pub(crate) fn run_provider( poll: Duration, stop: &AtomicBool, ) -> Result<()> { + match run_provider_observed( + provider, + status_path, + argv, + env, + refresh_interval, + poll, + stop, + )? { + ProviderOutcome::Exited(exit) => completed_provider(provider, exit), + ProviderOutcome::Stopped(_) => Ok(()), + } +} + +/// [`run_provider`], but reporting how the session ended instead of judging it, so a wrapper can +/// record a terminal observation before deciding what the exit means. +pub(crate) fn run_provider_observed( + provider: &str, + status_path: &Path, + argv: &[String], + env: &[(String, String)], + refresh_interval: Duration, + poll: Duration, + stop: &AtomicBool, +) -> Result { let (program, args) = argv .split_first() .with_context(|| format!("{provider} provider argv is empty"))?; @@ -77,13 +114,13 @@ pub(crate) fn run_provider( let mut next_refresh = Instant::now(); loop { if stop.load(Ordering::SeqCst) { - return stop_provider_group(&mut child); + return stop_provider_group(&mut child).map(ProviderOutcome::Stopped); } if let Some(exit) = child .try_wait() .with_context(|| format!("checking {provider} provider"))? { - return completed_provider(provider, exit); + return Ok(ProviderOutcome::Exited(exit)); } let now = Instant::now(); if now >= next_refresh { @@ -99,7 +136,7 @@ fn completed_provider(provider: &str, exit: ExitStatus) -> Result<()> { Ok(()) } -fn stop_provider_group(child: &mut Child) -> Result<()> { +fn stop_provider_group(child: &mut Child) -> Result> { let process_group = unsafe { libc::getpgrp() }; anyhow::ensure!( process_group > 1, @@ -110,14 +147,13 @@ fn stop_provider_group(child: &mut Child) -> Result<()> { } let deadline = Instant::now() + STOP_GRACE; while Instant::now() < deadline { - if child.try_wait()?.is_some() { - return Ok(()); + if let Some(exit) = child.try_wait()? { + return Ok(Some(exit)); } thread::sleep(Duration::from_millis(25)); } unsafe { libc::kill(-process_group, libc::SIGKILL); } - let _ = child.wait(); - Ok(()) + Ok(child.wait().ok()) }