fix(pi): settle-edge idle, terminal-record precedence, and runtime-id sessions

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 <noreply@anthropic.com>

Author: schickling

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

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!(

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<Vec<(String, String)>> {
     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()
+                ),
             ]
         );
     }