refactor(driver): unify the pi/omp session scaffolding and the exit-label map

src/claude_mcp.rs

@@ -5,7 +5,7 @@
 //! Claude session wrapper owns presence because Claude can close this child before the session ends.
 
 use std::collections::HashSet;
-use std::io::{self, BufRead, Write};
+use std::io::{self, BufRead, Write as _};
 use std::path::Path;
 use std::sync::mpsc::{self, RecvTimeoutError};
 use std::thread;
@@ -15,16 +15,10 @@ use anyhow::{Context as _, Result};
 use serde_json::{Value, json};
 
 use crate::message;
+use crate::native_channel::{channel_content, write_json};
 
 const POLL: Duration = Duration::from_millis(250);
 
-fn channel_content(subject: Option<&str>, body: &str) -> String {
-    match subject.filter(|value| !value.is_empty()) {
-        Some(subject) => format!("Subject: {subject}\n\n{body}"),
-        None => body.to_owned(),
-    }
-}
-
 pub fn run(catalog_root: &Path, identity: &str) -> Result<()> {
     let agent_dir = message::resolve_declared_dir(catalog_root, identity, &crate::run::detect_host())?
         .with_context(|| format!("Claude MCP agent '{identity}' is not declared"))?;
@@ -110,24 +104,3 @@ pub fn run(catalog_root: &Path, identity: &str) -> Result<()> {
         thread::sleep(POLL);
     }
 }
-
-fn write_json(out: &mut impl Write, value: &Value) -> Result<()> {
-    serde_json::to_writer(&mut *out, value)?;
-    out.write_all(b"\n")?;
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::channel_content;
-
-    #[test]
-    fn channel_content_reuses_subject_and_body_envelope() {
-        assert_eq!(
-            channel_content(Some("subject"), "body"),
-            "Subject: subject\n\nbody"
-        );
-        assert_eq!(channel_content(None, "body"), "body");
-        assert_eq!(channel_content(Some(""), "body"), "body");
-    }
-}

src/claude_session.rs

@@ -15,6 +15,7 @@ use std::path::Path;
 
 use anyhow::{Context as _, Result};
 
+use crate::driver_diagnostic::ProviderAuthEdge;
 use crate::harness_context::{self, Compaction, CompactionTrigger, Harness, RateLimits, Reading};
 use crate::harness_state::{Activity, Ask, BlockedOn, InputBuffer, Observation};
 use crate::provider_session::{
@@ -178,7 +179,11 @@ pub fn run_observe(
     // applied before the observation guard below for the same reason the compaction write is:
     // an edge that carries no top-level state change must still reach its own record.
     if let Some(edge) = provider_auth_edge(event, &payload) {
-        publish_provider_auth(&agent_dir, edge);
+        driver_diagnostic::publish_provider_auth(
+            &agent_dir,
+            driver_diagnostic::Driver::Claude,
+            edge,
+        );
     }
     let Some(observation) = observe_hook_event(event, &payload) else {
         return Ok(());
@@ -619,13 +624,6 @@ fn stop_failure_error(payload: &serde_json::Value) -> Option<&str> {
     payload.get("error").and_then(serde_json::Value::as_str)
 }
 
-/// What one hook event proves about the seat's provider credential.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum ProviderAuthEdge {
-    Rejected,
-    Accepted,
-}
-
 /// Read the credential edge out of one hook event, or `None` when the event proves nothing about
 /// it — which must leave a standing rejection alone rather than clearing it.
 fn provider_auth_edge(event: &str, payload: &serde_json::Value) -> Option<ProviderAuthEdge> {
@@ -639,33 +637,6 @@ fn provider_auth_edge(event: &str, payload: &serde_json::Value) -> Option<Provid
     }
 }
 
-/// Record one credential edge on the seat's native-driver diagnostic.
-///
-/// Each hook invocation is its own short-lived writer, so the publisher's stage set starts empty
-/// and its on-disk fallback is what lets a later `Stop` clear a rejection an earlier `StopFailure`
-/// wrote from a different process. Fail-open like every other observation here: the publisher only
-/// warns on a write it cannot land.
-fn publish_provider_auth(agent_dir: &Path, edge: ProviderAuthEdge) {
-    let mut publisher = driver_diagnostic::Publisher::new(
-        agent_dir,
-        driver_diagnostic::Driver::Claude,
-        // A hook payload carries no Claude version — the common hook input is session id,
-        // transcript path, cwd, prompt id, permission mode, agent identity and effort, and
-        // nothing else (2.1.259) — and st2 gates no Claude version, so neither the producer
-        // version nor its support status is knowable from here.
-        None,
-        driver_diagnostic::Support::Unknown,
-    );
-    match edge {
-        ProviderAuthEdge::Rejected => publisher.publish(
-            driver_diagnostic::Stage::ProviderAuth,
-            driver_diagnostic::Reason::ProviderAuthRejected,
-            driver_diagnostic::Source::TurnResult,
-        ),
-        ProviderAuthEdge::Accepted => publisher.clear(driver_diagnostic::Stage::ProviderAuth),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::fs;
@@ -862,8 +833,9 @@ mod tests {
             "error": "authentication_failed",
         });
 
-        publish_provider_auth(
+        driver_diagnostic::publish_provider_auth(
             tmp.path(),
+            driver_diagnostic::Driver::Claude,
             provider_auth_edge("StopFailure", &rejected).unwrap(),
         );
         let driver_diagnostic::Observed::Failure(failure) = driver_diagnostic::read(&record) else {
@@ -897,7 +869,11 @@ mod tests {
             driver_diagnostic::Observed::Failure(_)
         ));
 
-        publish_provider_auth(tmp.path(), ProviderAuthEdge::Accepted);
+        driver_diagnostic::publish_provider_auth(
+            tmp.path(),
+            driver_diagnostic::Driver::Claude,
+            ProviderAuthEdge::Accepted,
+        );
         assert_eq!(
             driver_diagnostic::read(&record),
             driver_diagnostic::Observed::Absent,

src/codex_app_server.rs

@@ -17,7 +17,6 @@ use std::os::unix::ffi::OsStrExt as _;
 use std::os::unix::fs::{FileTypeExt as _, OpenOptionsExt as _, PermissionsExt as _};
 use std::os::unix::io::AsRawFd as _;
 use std::os::unix::net::UnixStream;
-use std::os::unix::process::ExitStatusExt as _;
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, ExitStatus, Stdio};
 use std::sync::mpsc::{self, Receiver, Sender};
@@ -1820,12 +1819,13 @@ enum TuiEnd {
     Stopped(Option<ExitStatus>),
 }
 
+/// The label for a TUI end whose status may not have been observable at all. A status that WAS
+/// reaped is spelled by the one shared exit-label map; no status is the same "unknown" the map's
+/// own unanswerable arm reports.
 fn describe_tui_exit(status: Option<ExitStatus>) -> String {
-    match status.map(|status| (status.code(), status.signal())) {
-        Some((Some(code), _)) => format!("exit {code}"),
-        Some((None, Some(signal))) => format!("signal {signal}"),
-        _ => "exit unknown".to_string(),
-    }
+    status
+        .map(crate::provider_session::describe_exit)
+        .unwrap_or_else(|| "exit unknown".to_string())
 }
 
 /// Start app-server with the authored global configuration inputs that its CLI supports.

src/driver_diagnostic.rs

@@ -588,6 +588,46 @@ impl Publisher {
     }
 }
 
+/// What one observation — a Claude hook event, a pi-family typed turn result — proves about the
+/// seat's provider credential.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ProviderAuthEdge {
+    Rejected,
+    Accepted,
+}
+
+/// Record one credential edge on the seat's native-driver diagnostic.
+///
+/// A fresh publisher per edge on purpose, and every producer of these edges is short-lived: each
+/// Claude hook invocation is its own process, so the publisher's stage set starts empty and its
+/// on-disk fallback is what lets a later `Stop` clear a rejection an earlier `StopFailure` wrote
+/// from a different process; a channel that restarted mid-session inherits the predecessor's
+/// record the same way rather than silently starting clean. Fail-open like every other
+/// observation: the publisher only warns on a write it cannot land, and neither delivery nor
+/// launch depends on it.
+pub(crate) fn publish_provider_auth(agent_dir: &Path, driver: Driver, edge: ProviderAuthEdge) {
+    let mut publisher = Publisher::new(
+        agent_dir,
+        driver,
+        // No producer version and no support verdict is knowable at either edge. A Claude hook
+        // payload carries no version — the common hook input is session id, transcript path, cwd,
+        // prompt id, permission mode, agent identity and effort, and nothing else (2.1.259) — and
+        // st2 gates no Claude version at all. On the pi family the WRAPPER, not the channel, owns
+        // the version gate and refuses the launch on an unadmitted MINOR (OMP-R05), so a running
+        // channel has no version fact of its own to publish and no verdict to restate.
+        None,
+        Support::Unknown,
+    );
+    match edge {
+        ProviderAuthEdge::Rejected => publisher.publish(
+            Stage::ProviderAuth,
+            Reason::ProviderAuthRejected,
+            Source::TurnResult,
+        ),
+        ProviderAuthEdge::Accepted => publisher.clear(Stage::ProviderAuth),
+    }
+}
+
 fn emit(
     driver: Driver,
     stage: Stage,

src/lib.rs

@@ -39,12 +39,17 @@ pub mod identity;
 pub mod isolate;
 pub mod materialize;
 pub mod message;
+/// The stdio framing shared by the native channels; crate-internal.
+mod native_channel;
 pub mod omp_session;
 pub mod metrics;
 pub mod migrations;
 pub mod opencode_session;
 pub mod park;
 pub mod pi_channel;
+/// The launch body shared by the pi-family wrappers; crate-internal, reached through
+/// `pi_session::run` and `omp_session::run`.
+mod pi_family_session;
 pub mod pi_session;
 pub mod pretrust;
 pub mod provider_session;

src/native_channel.rs

@@ -0,0 +1,62 @@
+//! The two framing helpers every stdio native channel shares.
+//!
+//! A native channel is a child of the interactive harness, speaking newline-delimited JSON over
+//! stdio: the Claude MCP watcher ([`crate::claude_mcp`]) and the pi-family channel
+//! ([`crate::pi_channel`]). What a delivered inbox message looks like to the model, and how a
+//! frame is terminated on the wire, are st2's decisions rather than each channel's — so they are
+//! decided once, here.
+
+use std::io::Write;
+
+use anyhow::Result;
+use serde_json::Value;
+
+/// The envelope a delivered inbox message is handed to the model in. A message with no subject is
+/// its body verbatim: an empty `Subject:` line would be noise the model has to read past.
+pub(crate) fn channel_content(subject: Option<&str>, body: &str) -> String {
+    match subject.filter(|value| !value.is_empty()) {
+        Some(subject) => format!("Subject: {subject}\n\n{body}"),
+        None => body.to_owned(),
+    }
+}
+
+/// One frame on the wire: compact JSON followed by the newline that terminates it.
+pub(crate) fn write_json(out: &mut impl Write, value: &Value) -> Result<()> {
+    serde_json::to_writer(&mut *out, value)?;
+    out.write_all(b"\n")?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn channel_content_reuses_one_subject_and_body_envelope() {
+        assert_eq!(
+            channel_content(Some("subject"), "body"),
+            "Subject: subject\n\nbody"
+        );
+        assert_eq!(channel_content(None, "body"), "body");
+        assert_eq!(channel_content(Some(""), "body"), "body");
+    }
+
+    /// The wire is newline-delimited: a reader splitting on newlines must see exactly one frame
+    /// per value, and a body carrying its own newline must not be able to forge a second one.
+    #[test]
+    fn each_frame_is_one_newline_terminated_line() {
+        let mut out = Vec::new();
+        write_json(&mut out, &serde_json::json!({"a": 1})).unwrap();
+        write_json(
+            &mut out,
+            &serde_json::json!({"content": "line one\nline two"}),
+        )
+        .unwrap();
+        let framed = String::from_utf8(out).unwrap();
+        assert_eq!(
+            framed,
+            "{\"a\":1}\n{\"content\":\"line one\\nline two\"}\n"
+        );
+        assert_eq!(framed.lines().count(), 2);
+    }
+}