fix(claude,pi,materialize): shared session tokens, terminal fencing, and hook-set supersession

The wrapper mints and exports its incarnation token (ST2_CLAUDE_SESSION;
wrapperless hooks fall back to Claude's own session_id), late hooks can
no longer overwrite the session's terminal record, pi gains a
terminal-only observer so the pre-escalation ended write is real for its
stop path, union merges supersede st2's own prior hook-set entries while
never touching foreign ones, and the maintained example states its
hooks-only limitation.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: schickling

examples/native/agent-claude.kdl

@@ -18,6 +18,12 @@ agent "<identity>" {
     copy "assets/bus.st2.md" ".st2/bus.md"
     ensure-line ".claude/rules/st2.md" "@../../.st2/PERSONA.md"
     ensure-line ".claude/rules/st2.md" "@../../.st2/bus.md"
+    // Hooks-only observability: this seat launches claude directly (no session wrapper), so the
+    // registrations below give it transitions and blocked-on-you — but no heartbeat owner and no
+    // terminal record. A live-but-idle seat ages to `unknown` after the staleness horizon and an
+    // exit leaves the last state to age out; both read indeterminate, never wrong. The full
+    // producer (heartbeats, terminal exits) comes with the session wrapper, i.e. a typed
+    // `claude {}` driver seat or `deliver "mcp"`.
     json-upsert ".claude/settings.local.json" arrays="union" #"""
 {
   "$schema": "https://json.schemastore.org/claude-code-settings.json",

src/claude_session.rs

@@ -39,7 +39,12 @@ pub fn run(
     let observer = SessionObserver::new(&agent_dir, &identity, "claude", &runtime_id);
     // The runtime ID reaches hook subprocesses through the provider environment, so their
     // transitions carry the same pty session the wrapper's records do.
-    let env = [(RUNTIME_ID_ENV.to_string(), runtime_id.clone())];
+    let env = [
+        (RUNTIME_ID_ENV.to_string(), runtime_id.clone()),
+        // Hook subprocesses adopt the wrapper's incarnation token, so their transitions are this
+        // session's records: the wrapper can re-stamp them, and its terminal record fences them.
+        (SESSION_ENV.to_string(), observer.session().to_string()),
+    ];
     run_provider(
         "Claude",
         &status::status_path(&agent_dir),
@@ -59,6 +64,8 @@ pub fn run(
 /// short-lived writer; the transition counter continues from disk.
 /// The env var carrying the wrapper's runtime/task ID into Claude's hook subprocesses.
 pub const RUNTIME_ID_ENV: &str = "ST2_CLAUDE_RUNTIME_ID";
+/// The env var carrying the wrapper's session incarnation token into Claude's hook subprocesses.
+pub const SESSION_ENV: &str = "ST2_CLAUDE_SESSION";
 
 pub fn run_observe(
     catalog_root: &Path,
@@ -76,12 +83,32 @@ pub fn run_observe(
     };
     let pty_session = runtime_id.unwrap_or(identity).to_string();
     let mut writer = harness_state::Writer::new(&agent_dir, identity, "claude", Some(pty_session));
+    // The wrapper's exported token makes hook writes this session's records. A wrapperless seat
+    // (hooks registered on a plain hand-authored launch) falls back to Claude's own session_id —
+    // stable across one Claude session's hooks, fresh on restart — so restatements still
+    // coalesce and a restart still opens a new transition; what such a seat lacks is a
+    // heartbeat/terminal owner, which is a documented hooks-only limitation.
+    let session = std::env::var(SESSION_ENV)
+        .ok()
+        .filter(|token| !token.is_empty())
+        .or_else(|| {
+            payload
+                .get("session_id")
+                .and_then(serde_json::Value::as_str)
+                .map(|id| format!("claude-session-{id}"))
+        });
+    if let Some(session) = session {
+        writer = writer.with_session(session);
+    }
     if event == "SessionStart" {
         // The one event that names a session boundary: even if the new session's first state
         // matches a fresh predecessor record, continuity must not be claimed across the restart.
         writer.interrupt();
     }
-    writer.observe(observation)
+    // A late hook finishing after the wrapper reaped Claude must not replace the terminal record
+    // with a live state: the wrapper's `ended` carries this same token and is the session's last
+    // word. (`false` = suppressed; the hook has nothing else to do with it.)
+    writer.observe_unless_ended(observation).map(|_wrote| ())
 }
 
 /// Map one Claude hook event to an observation, or `None` when the event says nothing about
@@ -286,13 +313,15 @@ mod tests {
         let record = harness_state_path(tmp.path());
         let observer = SessionObserver::new(tmp.path(), "hetz.worker", "claude", "hetz.worker");
 
-        // A hook process wrote a blocked observation between wrapper ticks.
+        // A hook process wrote a blocked observation between wrapper ticks — carrying the
+        // wrapper's exported token, exactly as the env plumbing arranges in a real seat.
         harness_state::Writer::new(
             tmp.path(),
             "hetz.worker",
             "claude",
             Some("hetz.worker".to_string()),
         )
+        .with_session(observer.session())
         .observe(observe_hook_event("PermissionRequest", &serde_json::Value::Null).unwrap())
         .unwrap();
         let before = fs::read(&record).unwrap();
@@ -387,4 +416,57 @@ mod tests {
         let idle = observe_hook_event("Stop", &serde_json::json!({})).unwrap();
         assert_eq!(idle.ask, Ask::None);
     }
+
+    /// T2: a hook that finishes after the wrapper reaped Claude must not replace the terminal
+    /// record — the wrapper's `ended` carries the shared token and is the session's last word —
+    /// while a NEW session's boundary event still supersedes an old terminal record.
+    #[test]
+    fn a_late_hook_never_overwrites_this_sessions_terminal_record() {
+        use crate::harness_state::{self, Activity};
+        let tmp = tempfile::tempdir().unwrap();
+        let record = harness_state_path(tmp.path());
+        let observer = SessionObserver::new(tmp.path(), "hetz.worker", "claude", "hetz.worker");
+        observer.ended("exit 0");
+
+        // The straggler hook shares the session token (env plumbing) and is suppressed.
+        let mut late = harness_state::Writer::new(
+            tmp.path(),
+            "hetz.worker",
+            "claude",
+            Some("hetz.worker".to_string()),
+        )
+        .with_session(observer.session());
+        assert!(
+            !late
+                .observe_unless_ended(
+                    observe_hook_event("PostToolUse", &serde_json::Value::Null).unwrap()
+                )
+                .unwrap()
+        );
+        assert_eq!(
+            harness_state::read(&record, None).unwrap().state,
+            Activity::Ended
+        );
+
+        // A new Claude session is a new incarnation: its SessionStart replaces the old terminal.
+        let mut fresh = harness_state::Writer::new(
+            tmp.path(),
+            "hetz.worker",
+            "claude",
+            Some("hetz.worker".to_string()),
+        )
+        .with_session("claude-session-fresh");
+        fresh.interrupt();
+        assert!(
+            fresh
+                .observe_unless_ended(
+                    observe_hook_event("SessionStart", &serde_json::Value::Null).unwrap()
+                )
+                .unwrap()
+        );
+        assert_eq!(
+            harness_state::read(&record, None).unwrap().state,
+            Activity::Idle
+        );
+    }
 }

src/materialize.rs

@@ -494,12 +494,17 @@ fn ensure_line(path: &Path, line: &str) -> Result<bool> {
     Ok(true)
 }
 
-fn deep_merge(target: &mut serde_json::Value, patch: serde_json::Value, arrays: ArrayMerge) {
+fn deep_merge(
+    target: &mut serde_json::Value,
+    patch: serde_json::Value,
+    arrays: ArrayMerge,
+    owned_prefixes: &[String],
+) {
     match (target, patch) {
         (serde_json::Value::Object(target), serde_json::Value::Object(patch)) => {
             for (key, value) in patch {
                 match target.get_mut(&key) {
-                    Some(existing) => deep_merge(existing, value, arrays),
+                    Some(existing) => deep_merge(existing, value, arrays, owned_prefixes),
                     None => {
                         target.insert(key, value);
                     }
@@ -509,6 +514,14 @@ fn deep_merge(target: &mut serde_json::Value, patch: serde_json::Value, arrays:
         (serde_json::Value::Array(target), serde_json::Value::Array(patch))
             if arrays == ArrayMerge::Union =>
         {
+            // Exact-equality union alone would accumulate st2's own entries across hook-set
+            // upgrades: `$ST_HOOKS` expands content-addressed, so every upgrade renders each
+            // entry with a new path and the old one would be retained beside it. An element
+            // recognizably st2's — one referencing the hook root — that the patch no longer
+            // states is therefore superseded and dropped; foreign entries are never touched.
+            target.retain(|element| {
+                !contains_owned_string(element, owned_prefixes) || patch.contains(element)
+            });
             for element in patch {
                 if !target.contains(&element) {
                     target.push(element);
@@ -519,6 +532,35 @@ fn deep_merge(target: &mut serde_json::Value, patch: serde_json::Value, arrays:
     }
 }
 
+/// Whether any string inside `value` marks it as an st2-rendered element: a reference to the
+/// installed hook root (any set version) or the unexpanded `$ST_HOOKS` variable.
+fn contains_owned_string(value: &serde_json::Value, owned_prefixes: &[String]) -> bool {
+    match value {
+        serde_json::Value::String(text) => owned_prefixes
+            .iter()
+            .any(|prefix| text.starts_with(prefix.as_str())),
+        serde_json::Value::Array(items) => items
+            .iter()
+            .any(|item| contains_owned_string(item, owned_prefixes)),
+        serde_json::Value::Object(map) => map
+            .values()
+            .any(|item| contains_owned_string(item, owned_prefixes)),
+        _ => false,
+    }
+}
+
+/// The string prefixes that mark a JSON element as st2-rendered for union supersession: the hook
+/// root that contains every installed set version, and the unexpanded variable spellings.
+fn owned_union_prefixes(env: &BTreeMap<String, String>) -> Vec<String> {
+    let mut prefixes = vec!["$ST_HOOKS".to_string(), "${ST_HOOKS}".to_string()];
+    if let Some(hooks) = env.get("ST_HOOKS")
+        && let Some(root) = Path::new(hooks).parent()
+    {
+        prefixes.push(format!("{}/", root.display()));
+    }
+    prefixes
+}
+
 fn git_exclude(workspace: &Path, line: &str) -> Result<bool> {
     let output = Command::new("git")
         .args(["-C"])
@@ -843,7 +885,7 @@ pub fn materialize_agent(root: &Path, spec: &AgentSpec, this_host: &str) -> Resu
                             spec.identity
                         )
                     })?;
-                deep_merge(&mut target, patch, arrays);
+                deep_merge(&mut target, patch, arrays, &owned_union_prefixes(&env));
                 let mut bytes = serde_json::to_vec_pretty(&target)?;
                 bytes.push(b'\n');
                 let note = format!("{}: upserted {}", spec.identity, raw_destination);
@@ -1096,6 +1138,7 @@ mod tests {
                 "array": [2]
             }),
             ArrayMerge::Replace,
+            &[],
         );
         assert_eq!(
             target,
@@ -1117,8 +1160,8 @@ mod tests {
         let ours = serde_json::json!({
             "hooks": {"Stop": [{"hooks": [{"type": "command", "command": "$ST_HOOKS/claude-observe.sh Stop"}]}]}
         });
-        deep_merge(&mut target, ours.clone(), ArrayMerge::Union);
-        deep_merge(&mut target, ours, ArrayMerge::Union);
+        deep_merge(&mut target, ours.clone(), ArrayMerge::Union, &[]);
+        deep_merge(&mut target, ours, ArrayMerge::Union, &[]);
         assert_eq!(
             target,
             serde_json::json!({
@@ -1129,4 +1172,36 @@ mod tests {
             })
         );
     }
+
+    /// A hook-set upgrade renders every entry under a new content-addressed path. Union must
+    /// supersede st2's prior entries — recognizable by the hook root — rather than accumulate
+    /// them, while a user's entry under any other path survives every merge.
+    #[test]
+    fn union_supersedes_prior_hook_set_entries_but_never_foreign_ones() {
+        let owned = vec![
+            "$ST_HOOKS".to_string(),
+            "${ST_HOOKS}".to_string(),
+            "/state/st2/hooks/".to_string(),
+        ];
+        let mut target = serde_json::json!({
+            "hooks": {"Stop": [
+                {"hooks": [{"type": "command", "command": "user-audit.sh"}]},
+                {"hooks": [{"type": "command", "command": "/state/st2/hooks/set-v1/claude-observe.sh Stop"}]}
+            ]}
+        });
+        let upgraded = serde_json::json!({
+            "hooks": {"Stop": [{"hooks": [{"type": "command", "command": "/state/st2/hooks/set-v2/claude-observe.sh Stop"}]}]}
+        });
+        deep_merge(&mut target, upgraded.clone(), ArrayMerge::Union, &owned);
+        deep_merge(&mut target, upgraded, ArrayMerge::Union, &owned);
+        assert_eq!(
+            target,
+            serde_json::json!({
+                "hooks": {"Stop": [
+                    {"hooks": [{"type": "command", "command": "user-audit.sh"}]},
+                    {"hooks": [{"type": "command", "command": "/state/st2/hooks/set-v2/claude-observe.sh Stop"}]}
+                ]}
+            })
+        );
+    }
 }

src/pi_session.rs

@@ -70,6 +70,17 @@ pub fn run(
     })?;
     let pi_argv = with_channel_extension(pi_argv, &set)?;
     install_signal_handler();
+    // Terminal-only: the channel owns the live record and its heartbeat, but only this wrapper
+    // survives long enough to see the stop path — its pre-escalation `ended` write is the one
+    // that makes `Stopped(None)` observable at all. Same token as the channel, so the terminal
+    // record fences exactly this session's live records.
+    let observer = crate::provider_session::SessionObserver::terminal_only(
+        &agent_dir,
+        &identity,
+        "pi",
+        &runtime_id,
+        &session,
+    );
     let outcome = run_provider_observed(
         "pi",
         &status::status_path(&agent_dir),
@@ -78,7 +89,7 @@ pub fn run(
         status::STATUS_REFRESH,
         PROVIDER_POLL,
         &STOP,
-        None,
+        Some(&observer),
     )
     .with_context(|| format!("running pi driver '{runtime_id}'"))?;
     record_session_end(&agent_dir, &identity, &runtime_id, &session, &outcome);
@@ -352,4 +363,40 @@ mod tests {
             ]
         );
     }
+
+    /// W6: the terminal-only observer records how the session ended but never re-stamps live
+    /// state — the channel owns the heartbeat — and its token makes the write this session's.
+    #[test]
+    fn the_terminal_only_observer_ends_but_never_heartbeats() {
+        use crate::harness_state::{self, Activity};
+        let tmp = tempfile::tempdir().unwrap();
+        let record = harness_state::harness_state_path(tmp.path());
+        let session = harness_state::session_token();
+        let mut channel =
+            harness_state::Writer::new(tmp.path(), "h.worker", "pi", Some("h.worker".to_string()))
+                .with_session(session.clone());
+        channel
+            .observe(harness_state::Observation::new(
+                Activity::Active,
+                harness_state::BlockedOn::None,
+                harness_state::InputBuffer::Unknown,
+            ))
+            .unwrap();
+        let live = std::fs::read(&record).unwrap();
+
+        let observer = crate::provider_session::SessionObserver::terminal_only(
+            tmp.path(),
+            "h.worker",
+            "pi",
+            "h.worker",
+            &session,
+        );
+        observer.heartbeat();
+        assert_eq!(std::fs::read(&record).unwrap(), live, "no heartbeat");
+
+        observer.ended("signal 9");
+        let observed = harness_state::read(&record, None).unwrap();
+        assert_eq!(observed.state, Activity::Ended);
+        assert_eq!(observed.exit.as_deref(), Some("signal 9"));
+    }
 }