From 3fa0adfff72e6dcfb7107bea019cd77562b355a3 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Sun, 23 Aug 2026 19:00:18 +0200 Subject: [PATCH 1/8] feat(pi): produce observed harness state from pi's own turn events The shipped extension emits observational state frames on agent_start/ agent_end (the exact span ctx.isIdle() covers) plus an idle-proof seed at session start. The channel owns the live record: it maps recognized state frames to observations, heartbeats exactly as long as its stdio connection lives, and on EOF leaves the record to age rather than asserting a state nobody watches. The session wrapper owns the one fact the channel cannot see and writes the terminal record with the provider's real exit status, via a new run_provider_observed that reports the exit instead of judging it. pi 0.84.2 exposes no waiting-on-a-human event, so no pi frame ever sets blockedOn. Co-Authored-By: Claude Fable 5 --- hooks/pi-channel.ts | 19 +++++ src/pi_channel.rs | 163 +++++++++++++++++++++++++++++++++++++--- src/pi_session.rs | 114 +++++++++++++++++++++++++++- src/provider_session.rs | 52 +++++++++++-- 4 files changed, 325 insertions(+), 23 deletions(-) diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index 60530123..a1a3d794 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -197,11 +197,30 @@ export default function (pi: ExtensionAPI) { }); }; + // Observed harness state, extension side. pi's own turn boundaries are the positive signal: + // `ctx.isIdle()` is false for exactly the `agent_start`..`agent_end` span, so these two events + // carry the working/idle edge without inspecting anything. 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_end", 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/src/pi_channel.rs b/src/pi_channel.rs index 7f31aae9..e8595034 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,100 @@ 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. + let mut writer = + harness_state::Writer::new(&agent_dir, identity, "pi", Some(identity.to_string())); + 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) + && let Err(error) = writer.observe(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 +237,81 @@ 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" + ); + } + #[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..51b7555e 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"; @@ -61,7 +64,7 @@ pub fn run( })?; let pi_argv = with_channel_extension(pi_argv, &set)?; install_signal_handler(); - run_provider( + let outcome = run_provider_observed( "pi", &status::status_path(&agent_dir), &pi_argv, @@ -70,7 +73,42 @@ 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, &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, 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(identity.to_string())); + 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. @@ -123,6 +161,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 +190,73 @@ 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", + &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", + &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( 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()) } From 6ae97414d5e7e92c3df71ce8df6310e454312f69 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Sun, 23 Aug 2026 21:05:08 +0200 Subject: [PATCH 2/8] fix(pi): settle-edge idle, terminal-record precedence, and runtime-id sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review-pass fixes: idle emits on agent_settled instead of 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 there, so the old edge blipped a spurious idle before it; the channel drops queued live frames once the wrapper's terminal record is on disk (the flock serializes but does not order two processes); and ptySession records the wrapper's runtime ID, delivered to the channel via ST2_PI_CHANNEL_RUNTIME_ID beside the existing three env vars. checks.pi-extension-types stays green against pinned pi, proving the settle event exists on that surface. Co-Authored-By: Claude Fable 5 --- hooks/pi-channel.ts | 10 ++++++---- src/pi_channel.rs | 48 ++++++++++++++++++++++++++++++++++++++++++--- src/pi_session.rs | 24 +++++++++++++++++++---- 3 files changed, 71 insertions(+), 11 deletions(-) diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index a1a3d794..8c351522 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -197,9 +197,11 @@ export default function (pi: ExtensionAPI) { }); }; - // Observed harness state, extension side. pi's own turn boundaries are the positive signal: - // `ctx.isIdle()` is false for exactly the `agent_start`..`agent_end` span, so these two events - // carry the working/idle edge without inspecting anything. The frame is observational — st2 + // 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. @@ -209,7 +211,7 @@ export default function (pi: ExtensionAPI) { child.stdin.write(JSON.stringify({ type: "state", state: word }) + "\n"); }; pi.on("agent_start", async () => sendState("active")); - pi.on("agent_end", async () => sendState("idle")); + 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 diff --git a/src/pi_channel.rs b/src/pi_channel.rs index e8595034..72286e56 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -92,8 +92,13 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { // 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. - let mut writer = - harness_state::Writer::new(&agent_dir, identity, "pi", Some(identity.to_string())); + // 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()); + let mut writer = harness_state::Writer::new(&agent_dir, identity, "pi", Some(pty_session)); channel_loop( &input_rx, &mut stdout, @@ -135,7 +140,10 @@ fn channel_loop( .ok() .as_ref() .and_then(state_observation) - && let Err(error) = writer.observe(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}"); } @@ -312,6 +320,40 @@ mod tests { ); } + /// 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); + let mut channel_writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); + let mut wrapper_writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); + 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 51b7555e..c2ca103f 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -31,6 +31,8 @@ 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"; /// pi's startup network work, which a supervised seat should not be doing. /// @@ -55,7 +57,7 @@ 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)?; + let mut env = channel_env(&executable, catalog_root, &identity, &runtime_id)?; env.extend(offline_defaults(|key| std::env::var_os(key).is_some())); let set = hooks::verify_required_set().with_context(|| { format!( @@ -74,7 +76,7 @@ pub fn run( &STOP, ) .with_context(|| format!("running pi driver '{runtime_id}'"))?; - record_session_end(&agent_dir, &identity, &outcome); + record_session_end(&agent_dir, &identity, &runtime_id, &outcome); match outcome { ProviderOutcome::Exited(exit) => { anyhow::ensure!(exit.success(), "pi provider exited with {exit}"); @@ -90,13 +92,18 @@ pub fn run( /// 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, outcome: &ProviderOutcome) { +fn record_session_end( + agent_dir: &Path, + identity: &str, + runtime_id: &str, + 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(identity.to_string())); + harness_state::Writer::new(agent_dir, identity, "pi", Some(runtime_id.to_string())); if let Err(error) = writer.ended(label) { eprintln!("st2 pi driver: recording session end failed: {error}"); } @@ -141,6 +148,7 @@ fn channel_env( executable: &Path, catalog_root: &Path, identity: &str, + runtime_id: &str, ) -> Result> { let executable = executable .to_str() @@ -150,6 +158,7 @@ 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()), ]) } @@ -213,6 +222,7 @@ mod tests { record_session_end( agent_dir, "h.worker", + "h.worker", &ProviderOutcome::Exited(ExitStatus::from_raw(3 << 8)), ); @@ -229,6 +239,7 @@ mod tests { record_session_end( agent_dir, "h.worker", + "h.worker", &ProviderOutcome::Stopped(Some(ExitStatus::from_raw(9))), ); let observed = crate::harness_state::read(&record, None).unwrap(); @@ -309,6 +320,7 @@ mod tests { &PathBuf::from("/opt/st2/bin/st2"), &PathBuf::from("/catalog"), "host.worker", + "host.worker-task", ) .unwrap(); @@ -318,6 +330,10 @@ 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() + ), ] ); } From 6a5639aad408427ccadaa0a7cfee173a59592d64 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Sun, 23 Aug 2026 21:53:15 +0200 Subject: [PATCH 3/8] fix(pi): a fresh channel is a new session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The channel writer starts interrupted, so a restarted seat opens a new transition instead of coalescing with — or being suppressed by — its predecessor's record, including a predecessor's terminal ended. Co-Authored-By: Claude Fable 5 --- src/pi_channel.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/pi_channel.rs b/src/pi_channel.rs index 72286e56..69549a8e 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -99,6 +99,10 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { .filter(|value| !value.is_empty()) .unwrap_or_else(|| identity.to_string()); let mut writer = harness_state::Writer::new(&agent_dir, identity, "pi", Some(pty_session)); + // A fresh channel is a new session: its first frame opens a new transition rather than + // claiming continuity with whatever a predecessor left behind — including a predecessor's + // terminal record, which (being pre-session) does not suppress this session's live frames. + writer.interrupt(); channel_loop( &input_rx, &mut stdout, From 55f7d56fd07aeeec64ab4b7354b92cb87c5c4064 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Mon, 24 Aug 2026 00:15:24 +0200 Subject: [PATCH 4/8] fix(pi): fail-open channel stdin and a wrapper-minted session token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The extension attaches a stream-level stdin error listener so an EPIPE from a retiring channel can never take pi down; the wrapper mints the session incarnation token, exports it beside the other channel env, and both its terminal writer and the channel adopt it — the terminal record owns exactly this session's live records. Co-Authored-By: Claude Fable 5 --- hooks/pi-channel.ts | 6 ++++++ src/pi_channel.rs | 21 ++++++++++++++++----- src/pi_session.rs | 18 +++++++++++++++--- 3 files changed, 37 insertions(+), 8 deletions(-) diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index 8c351522..98da27f7 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -132,6 +132,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) => { diff --git a/src/pi_channel.rs b/src/pi_channel.rs index 69549a8e..1028588a 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -98,10 +98,16 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { .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)); - // A fresh channel is a new session: its first frame opens a new transition rather than - // claiming continuity with whatever a predecessor left behind — including a predecessor's - // terminal record, which (being pre-session) does not suppress this session's live frames. + if let Ok(session) = std::env::var(crate::pi_session::CHANNEL_SESSION) + && !session.is_empty() + { + writer = writer.with_session(session); + } writer.interrupt(); channel_loop( &input_rx, @@ -332,10 +338,15 @@ mod tests { 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())); + 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())); + 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(); diff --git a/src/pi_session.rs b/src/pi_session.rs index c2ca103f..3f41133c 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -33,6 +33,9 @@ pub const CHANNEL_CATALOG: &str = "ST2_PI_CHANNEL_CATALOG"; 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"; /// pi's startup network work, which a supervised seat should not be doing. /// @@ -57,7 +60,8 @@ 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, &runtime_id)?; + let session = harness_state::session_token(); + let mut env = channel_env(&executable, catalog_root, &identity, &runtime_id, &session)?; env.extend(offline_defaults(|key| std::env::var_os(key).is_some())); let set = hooks::verify_required_set().with_context(|| { format!( @@ -76,7 +80,7 @@ pub fn run( &STOP, ) .with_context(|| format!("running pi driver '{runtime_id}'"))?; - record_session_end(&agent_dir, &identity, &runtime_id, &outcome); + record_session_end(&agent_dir, &identity, &runtime_id, &session, &outcome); match outcome { ProviderOutcome::Exited(exit) => { anyhow::ensure!(exit.success(), "pi provider exited with {exit}"); @@ -96,6 +100,7 @@ fn record_session_end( agent_dir: &Path, identity: &str, runtime_id: &str, + session: &str, outcome: &ProviderOutcome, ) { let label = match outcome { @@ -103,7 +108,8 @@ fn record_session_end( ProviderOutcome::Stopped(None) => "stopped".to_string(), }; let mut writer = - harness_state::Writer::new(agent_dir, identity, "pi", Some(runtime_id.to_string())); + harness_state::Writer::new(agent_dir, identity, "pi", Some(runtime_id.to_string())) + .with_session(session); if let Err(error) = writer.ended(label) { eprintln!("st2 pi driver: recording session end failed: {error}"); } @@ -149,6 +155,7 @@ fn channel_env( catalog_root: &Path, identity: &str, runtime_id: &str, + session: &str, ) -> Result> { let executable = executable .to_str() @@ -159,6 +166,7 @@ fn channel_env( (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()), ]) } @@ -223,6 +231,7 @@ mod tests { agent_dir, "h.worker", "h.worker", + "session-test", &ProviderOutcome::Exited(ExitStatus::from_raw(3 << 8)), ); @@ -240,6 +249,7 @@ mod tests { agent_dir, "h.worker", "h.worker", + "session-test", &ProviderOutcome::Stopped(Some(ExitStatus::from_raw(9))), ); let observed = crate::harness_state::read(&record, None).unwrap(); @@ -321,6 +331,7 @@ mod tests { &PathBuf::from("/catalog"), "host.worker", "host.worker-task", + "session-test", ) .unwrap(); @@ -334,6 +345,7 @@ mod tests { CHANNEL_RUNTIME_ID.to_string(), "host.worker-task".to_string() ), + (CHANNEL_SESSION.to_string(), "session-test".to_string()), ] ); } From fafa411f841c30c6d4b71a7266abb41d843a9fa1 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Mon, 24 Aug 2026 00:46:23 +0200 Subject: [PATCH 5/8] fix(pi): stash every channel variable and export the ownership claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The extension stashes and unexports ALL ST2_PI_CHANNEL_* values — the runtime id and ownership pair included — handing them only to the channel subprocess, so no pi tool child inherits this seat's registry key or record ownership. The wrapper claims the ownership sequence at startup and exports it beside the token; the channel and the terminal writer act under the same directional claim. Co-Authored-By: Claude Fable 5 --- hooks/pi-channel.ts | 23 +++++++++++++++++++++-- src/pi_channel.rs | 10 +++++++++- src/pi_session.rs | 24 +++++++++++++++++++++--- 3 files changed, 51 insertions(+), 6 deletions(-) diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index 98da27f7..f359fab3 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,21 +73,30 @@ 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 @@ -112,10 +127,14 @@ export default function (pi: ExtensionAPI) { // Closes the channel opened by the PREVIOUS session, whichever extension instance opened it. closeChild(state.child); + 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; diff --git a/src/pi_channel.rs b/src/pi_channel.rs index 1028588a..e0eedd4c 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -106,7 +106,15 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> { if let Ok(session) = std::env::var(crate::pi_session::CHANNEL_SESSION) && !session.is_empty() { - writer = writer.with_session(session); + // 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( diff --git a/src/pi_session.rs b/src/pi_session.rs index 3f41133c..3ba14edc 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -36,6 +36,9 @@ 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. /// @@ -61,7 +64,15 @@ pub fn run( let executable = std::env::current_exe().context("resolving st2 executable for the pi channel")?; let session = harness_state::session_token(); - let mut env = channel_env(&executable, catalog_root, &identity, &runtime_id, &session)?; + let seq = harness_state::claim_seq(&agent_dir); + 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!( @@ -80,7 +91,7 @@ pub fn run( &STOP, ) .with_context(|| format!("running pi driver '{runtime_id}'"))?; - record_session_end(&agent_dir, &identity, &runtime_id, &session, &outcome); + 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}"); @@ -101,6 +112,7 @@ fn record_session_end( identity: &str, runtime_id: &str, session: &str, + seq: u64, outcome: &ProviderOutcome, ) { let label = match outcome { @@ -109,7 +121,7 @@ fn record_session_end( }; let mut writer = harness_state::Writer::new(agent_dir, identity, "pi", Some(runtime_id.to_string())) - .with_session(session); + .with_ownership(session, seq); if let Err(error) = writer.ended(label) { eprintln!("st2 pi driver: recording session end failed: {error}"); } @@ -156,6 +168,7 @@ fn channel_env( identity: &str, runtime_id: &str, session: &str, + seq: u64, ) -> Result> { let executable = executable .to_str() @@ -167,6 +180,7 @@ fn channel_env( (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()), ]) } @@ -232,6 +246,7 @@ mod tests { "h.worker", "h.worker", "session-test", + 1, &ProviderOutcome::Exited(ExitStatus::from_raw(3 << 8)), ); @@ -250,6 +265,7 @@ mod tests { "h.worker", "h.worker", "session-test", + 1, &ProviderOutcome::Stopped(Some(ExitStatus::from_raw(9))), ); let observed = crate::harness_state::read(&record, None).unwrap(); @@ -332,6 +348,7 @@ mod tests { "host.worker", "host.worker-task", "session-test", + 7, ) .unwrap(); @@ -346,6 +363,7 @@ mod tests { "host.worker-task".to_string() ), (CHANNEL_SESSION.to_string(), "session-test".to_string()), + (CHANNEL_SEQ.to_string(), "7".to_string()), ] ); } From 1f34a896d16ebcecf2f3ad4d8b0ba182aaf952b6 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Mon, 24 Aug 2026 01:24:18 +0200 Subject: [PATCH 6/8] fix(pi): written ownership claim, and the successor channel waits out its predecessor The wrapper's claim is a written supersession before the channel acts; the extension awaits (bounded) the previous channel's exit before spawning its replacement, so a predecessor cannot drain queued frames into the new session's records after the seed. Co-Authored-By: Claude Fable 5 --- hooks/pi-channel.ts | 21 ++++++++++++++++++--- src/pi_session.rs | 4 +++- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index f359fab3..ede539a6 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -111,7 +111,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 @@ -124,9 +124,24 @@ 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 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 channelEnv: NodeJS.ProcessEnv = { ...process.env }; if (runtimeId) channelEnv[RUNTIME_ID] = runtimeId; if (session) channelEnv[SESSION] = session; diff --git a/src/pi_session.rs b/src/pi_session.rs index 3ba14edc..296c0fe1 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -64,7 +64,9 @@ pub fn run( let executable = std::env::current_exe().context("resolving st2 executable for the pi channel")?; let session = harness_state::session_token(); - let seq = harness_state::claim_seq(&agent_dir); + // 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)?; let mut env = channel_env( &executable, catalog_root, From 82eb391d9e460401f58c6990306bebd9410c933f Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Mon, 24 Aug 2026 01:51:30 +0200 Subject: [PATCH 7/8] fix(pi): hoist the exit-await above its use, and smoke-execute the shipped extension MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cycle-5 bounded-await landed as a use-before-declaration — a TDZ ReferenceError on every channel open, so no managed pi seat got a channel at all — and the type gate shipped it green. The declaration is hoisted beside closeChild, and the gate now transpiles the real asset and drives it through registration, a double session_start (the exact region the regression lived in), the turn events, and shutdown; the injected-TDZ negative test fails the gate as required. Co-Authored-By: Claude Fable 5 --- flake.nix | 9 +++++++++ hooks/pi-channel.ts | 21 +++++++++++---------- hooks/typecheck/smoke.mjs | 38 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 10 deletions(-) create mode 100644 hooks/typecheck/smoke.mjs 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 ede539a6..d6911d62 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -102,6 +102,17 @@ export default function (pi: ExtensionAPI) { // /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; @@ -132,16 +143,6 @@ export default function (pi: ExtensionAPI) { closeChild(previous); await awaitExit(previous, 2000); - 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 channelEnv: NodeJS.ProcessEnv = { ...process.env }; if (runtimeId) channelEnv[RUNTIME_ID] = runtimeId; if (session) channelEnv[SESSION] = session; 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); From e0d4676325fb440d217100978e658dd005972ea6 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Mon, 24 Aug 2026 02:34:25 +0200 Subject: [PATCH 8/8] fix(pi): post-claim launch failures end the record honestly Found by the cycle-7 self-review's error-arm sweep: channel-env, hook-set verification, and extension-injection failures after the written claim left the placeholder as the last word. They now write a real terminal record (ended, launch-error) before propagating. Co-Authored-By: Claude Fable 5 --- src/pi_session.rs | 57 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 42 insertions(+), 15 deletions(-) diff --git a/src/pi_session.rs b/src/pi_session.rs index 296c0fe1..551d253c 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -67,21 +67,48 @@ pub fn run( // 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)?; - 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`" - ) - })?; - let pi_argv = with_channel_extension(pi_argv, &set)?; + // 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(); let outcome = run_provider_observed( "pi",