supercode_harness/

session_activity.rs

//! Protocol-neutral activity for persisted and live harness sessions.
//!
//! Activity is deliberately separate from transcript freshness and UI
//! attention. A process receipt proves presence; only a harness lifecycle
//! boundary or runtime state proves whether a turn is working.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};

use crate::claude_peer::{process_is_live, ClaudePeerSession, ClaudePeerStatus};
#[cfg(feature = "adapter-api")]
use crate::codex_peer::CodexPeerTracker;
use crate::codex_peer::{live_rollouts, rollout_lineage, rollout_status, CodexPeerStatus};
use crate::{HarnessHomes, HarnessId, SessionLocator};

/// Whether a durable session currently has a proven live owner.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionPresence {
    /// The session is durable, but no live owner is proven.
    Persisted,
    /// A harness or Supercode runtime currently owns the session.
    Running,
    /// A Supercode-owned runtime is shutting down.
    ShuttingDown,
}

/// Turn activity, independent of presence and frontend attention.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionTurnState {
    /// Presence is known but the harness exposes no trustworthy turn state.
    Unknown,
    /// The runtime is ready for user input.
    Idle,
    /// A model, tool, or scheduler turn is active.
    Working,
    /// The runtime has issued a structured request that needs a response.
    NeedsInput,
}

/// Provenance for one normalized activity observation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionActivityEvidence {
    /// Stable, non-sensitive evidence source.
    pub source: String,
    /// Harness-native state token, when one was published.
    pub native_state: Option<String>,
    /// Wall-clock time at which Supercode sampled the evidence.
    pub observed_at_ms: u64,
    /// Harness version attached to the evidence, when available.
    pub harness_version: Option<String>,
}

/// Normalized lifecycle state for one harness-native session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionActivity {
    /// Owning harness.
    pub harness: HarnessId,
    /// Harness-native durable session id.
    pub session_id: String,
    /// Live ownership, independent of turn state.
    pub presence: SessionPresence,
    /// Current turn state, independent of unread/attention state.
    pub turn: SessionTurnState,
    /// Why this state is trustworthy. Never contains a pid, path, socket, or token.
    pub evidence: SessionActivityEvidence,
}

impl SessionActivity {
    /// Compare transition-bearing state while ignoring the observation clock.
    pub(crate) fn same_state(&self, other: &Self) -> bool {
        self.harness == other.harness
            && self.session_id == other.session_id
            && self.presence == other.presence
            && self.turn == other.turn
            && self.evidence.source == other.evidence.source
            && self.evidence.native_state == other.evidence.native_state
            && self.evidence.harness_version == other.evidence.harness_version
    }

    /// Stable subscription identity without exposing a persistence path.
    pub(crate) fn key(&self) -> (String, String) {
        (self.harness.as_str().to_string(), self.session_id.clone())
    }
}

/// The harness session running beneath each of `roots` (a terminal pane's
/// shell, say) with its activity: the nearest process below the root that
/// owns a Claude Code session (its registry pid) or a Codex conversation (the
/// rollout it holds open). Roots with no harness session beneath them are
/// left out. This is how a pane's liveness comes from the harness's own
/// lifecycle instead of a guess from the pane's foreground command.
#[cfg(feature = "adapter-api")]
pub(crate) async fn activity_under(
    roots: &[u32],
    homes: &HarnessHomes,
) -> Result<Vec<(u32, SessionActivity)>, crate::SdkError> {
    let table = process_table();
    let mut children: HashMap<u32, Vec<u32>> = HashMap::new();
    for (&pid, (parent, _)) in &table {
        children.entry(*parent).or_default().push(pid);
    }
    let registry = crate::claude_peer::read_registry(&crate::claude_peer::registry_dir(homes));
    let mut found: Vec<(u32, SessionLocator)> = Vec::new();
    for &root in roots {
        let mut queue = std::collections::VecDeque::from([root]);
        let mut visited = 0;
        while let Some(pid) = queue.pop_front() {
            visited += 1;
            if visited > 256 {
                break;
            }
            if let Some(session) = registry.iter().find(|session| session.pid == pid) {
                found.push((
                    root,
                    SessionLocator {
                        harness: HarnessId::new(HarnessId::CLAUDE_CODE),
                        session_id: session.session_id.clone(),
                        storage: crate::StorageLocator::File {
                            path: PathBuf::new(),
                        },
                    },
                ));
                break;
            }
            let is_codex = table
                .get(&pid)
                .is_some_and(|(_, command)| command.rsplit('/').next() == Some("codex"));
            if is_codex {
                // A conversation resumed and idle since holds no rollout open: its own `resume <id>` names it.
                let resolved = crate::codex_peer::session_of_process(pid).or_else(|| {
                    let session_id = crate::codex_peer::resumed_session_of_process(pid)?;
                    let path = crate::codex_peer::rollout_of_session(&homes.codex, &session_id)?;
                    Some((session_id, path))
                });
                if let Some((session_id, path)) = resolved {
                    found.push((
                        root,
                        SessionLocator {
                            harness: HarnessId::new(HarnessId::CODEX),
                            session_id,
                            storage: crate::StorageLocator::File { path },
                        },
                    ));
                    break;
                }
            }
            queue.extend(children.get(&pid).into_iter().flatten().copied());
        }
    }
    if found.is_empty() {
        return Ok(Vec::new());
    }
    let locators: Vec<SessionLocator> = found.iter().map(|(_, locator)| locator.clone()).collect();
    let activities = SessionActivityMonitor::default()
        .resolve(&locators, homes)
        .await?;
    Ok(found
        .into_iter()
        .filter_map(|(root, locator)| {
            activities
                .iter()
                .find(|activity| {
                    activity.harness == locator.harness && activity.session_id == locator.session_id
                })
                .cloned()
                .map(|activity| (root, activity))
        })
        .collect())
}

/// Every process: pid → (parent pid, command), from `ps`.
#[cfg(feature = "adapter-api")]
fn process_table() -> HashMap<u32, (u32, String)> {
    let Ok(output) = std::process::Command::new("ps")
        .args(["-axo", "pid=,ppid=,comm="])
        .output()
    else {
        return HashMap::new();
    };
    String::from_utf8_lossy(&output.stdout)
        .lines()
        .filter_map(|line| {
            let mut fields = line.split_whitespace();
            let pid = fields.next()?.parse().ok()?;
            let parent = fields.next()?.parse().ok()?;
            let command = fields.collect::<Vec<_>>().join(" ");
            Some((pid, (parent, command)))
        })
        .collect()
}

/// Stateful activity resolver. It caches only expensive process-ownership
/// discovery; every lifecycle boundary is still sampled on each poll.
#[cfg(feature = "adapter-api")]
#[derive(Debug, Default)]
pub(crate) struct SessionActivityMonitor {
    codex: CodexPeerTracker,
}

#[cfg(feature = "adapter-api")]
impl SessionActivityMonitor {
    pub(crate) async fn resolve(
        &mut self,
        locators: &[SessionLocator],
        homes: &HarnessHomes,
    ) -> Result<Vec<SessionActivity>, crate::SdkError> {
        let authorization = crate::RuntimeAuthorization::observer();
        let sources = locators
            .iter()
            .map(|locator| (locator.harness.0.clone(), locator.session_id.clone()))
            .collect::<BTreeSet<_>>();
        let owned = crate::LocalRuntimeRegistry::new()
            .source_states(&sources, &authorization)
            .await?;
        let claude = read_claude(locators, homes);
        let codex = if locators
            .iter()
            .any(|locator| locator.harness.as_str() == HarnessId::CODEX)
        {
            self.codex.sample(&homes.codex)
        } else {
            HashMap::new()
        };
        let hermes = read_hermes(locators, homes);
        let mut activities = resolve_stock_with_evidence(locators, &claude, &codex, &hermes)
            .into_iter()
            .map(|activity| (activity.key(), activity))
            .collect::<BTreeMap<_, _>>();
        let observed_at_ms = now_ms();
        for locator in locators {
            let key = (
                locator.harness.as_str().to_string(),
                locator.session_id.clone(),
            );
            if let Some(state) = owned.get(&key).copied() {
                activities.insert(key, owned_activity(locator, state, observed_at_ms));
            }
        }
        Ok(locators
            .iter()
            .filter_map(|locator| {
                activities.remove(&(
                    locator.harness.as_str().to_string(),
                    locator.session_id.clone(),
                ))
            })
            .collect())
    }
}

/// Resolve stock-harness activity without consulting Supercode-owned runtime
/// receipts. Discovery and the subscription lane share this exact mapping.
pub(crate) fn resolve_stock_session_activities(
    locators: &[SessionLocator],
    homes: &HarnessHomes,
) -> Vec<SessionActivity> {
    let claude = read_claude(locators, homes);
    let codex = if locators
        .iter()
        .any(|locator| locator.harness.as_str() == HarnessId::CODEX)
    {
        live_rollouts(&homes.codex)
    } else {
        HashMap::new()
    };
    let hermes = read_hermes(locators, homes);
    resolve_stock_with_evidence(locators, &claude, &codex, &hermes)
}

/// What Hermes's own execution ledger says about a session: the fire that
/// opened it (joined by the orchestration codec, which also resolves
/// compression chains) and the pid Hermes recorded for the attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct HermesFire {
    status: supercode_interchange::orchestration::FireStatus,
    pid: Option<u32>,
}

/// Hermes keeps no process registry, but every scheduled or task fire records
/// its attempt in `cron/executions.db` with the pid that ran it. A running fire
/// whose pid is alive is a proven owner; a running fire whose pid is gone is an
/// attempt its owner died under, and the ledger will never close it.
fn read_hermes(locators: &[SessionLocator], homes: &HarnessHomes) -> HashMap<String, HermesFire> {
    if !locators
        .iter()
        .any(|locator| locator.harness.as_str() == HarnessId::HERMES)
    {
        return HashMap::new();
    }
    // `HarnessHomes::hermes` addresses `state.db`; the ledger is its sibling.
    let home = homes
        .hermes
        .parent()
        .map_or_else(|| PathBuf::from("."), std::path::Path::to_path_buf);
    let Ok(loaded) = supercode_interchange::orchestration::codec::from_hermes(&home) else {
        return HashMap::new();
    };
    let mut fires = HashMap::new();
    for profile in loaded.orchestration.profiles.values() {
        for fire in &profile.fires {
            let Some(session) = fire.session_id.clone() else {
                continue;
            };
            let pid = fire.residue.0.get("pid").and_then(|value| {
                value
                    .as_u64()
                    .or_else(|| value.as_str().and_then(|text| text.trim().parse().ok()))
                    .and_then(|pid| u32::try_from(pid).ok())
            });
            fires.insert(
                session,
                HermesFire {
                    status: fire.status,
                    pid,
                },
            );
        }
    }
    fires
}

fn hermes_activity(
    locator: &SessionLocator,
    fire: HermesFire,
    live: impl Fn(u32) -> bool,
    observed_at_ms: u64,
) -> SessionActivity {
    use supercode_interchange::orchestration::FireStatus;
    let (presence, turn, native_state) = match fire.status {
        FireStatus::Claimed | FireStatus::Running => match fire.pid {
            Some(pid) if live(pid) => (
                SessionPresence::Running,
                SessionTurnState::Working,
                fire.status.hermes_word(),
            ),
            // The attempt's owner is gone and the ledger still says running:
            // Hermes will never write this fire's end.
            _ => (
                SessionPresence::Persisted,
                SessionTurnState::Unknown,
                "abandoned",
            ),
        },
        status => (
            SessionPresence::Persisted,
            SessionTurnState::Unknown,
            status.hermes_word(),
        ),
    };
    activity(
        locator,
        presence,
        turn,
        "hermes_executions",
        Some(native_state),
        None,
        observed_at_ms,
    )
}

fn read_claude(
    locators: &[SessionLocator],
    homes: &HarnessHomes,
) -> HashMap<String, ClaudePeerSession> {
    if locators
        .iter()
        .any(|locator| locator.harness.as_str() == HarnessId::CLAUDE_CODE)
    {
        crate::claude_peer::read_registry(&crate::claude_peer::registry_dir(homes))
            .into_iter()
            .map(|peer| (peer.session_id.clone(), peer))
            .collect::<HashMap<_, _>>()
    } else {
        HashMap::new()
    }
}

fn resolve_stock_with_evidence(
    locators: &[SessionLocator],
    claude: &HashMap<String, ClaudePeerSession>,
    codex: &HashMap<PathBuf, CodexPeerStatus>,
    hermes: &HashMap<String, HermesFire>,
) -> Vec<SessionActivity> {
    let observed_at_ms = now_ms();
    let codex_descendants = codex_descendant_statuses(codex);

    locators
        .iter()
        .map(|locator| {
            if locator.harness.as_str() == HarnessId::CLAUDE_CODE {
                if let Some(peer) = claude.get(&locator.session_id) {
                    return claude_activity(locator, peer, observed_at_ms);
                }
            }
            if locator.harness.as_str() == HarnessId::HERMES {
                if let Some(fire) = hermes.get(&locator.session_id) {
                    return hermes_activity(locator, *fire, process_is_live, observed_at_ms);
                }
            }
            if locator.harness.as_str() == HarnessId::CODEX {
                if let Some(status) = merge_codex_status(
                    rollout_status(codex, locator.storage.path()),
                    codex_descendants.get(&locator.session_id).copied(),
                ) {
                    return codex_activity(locator, status, observed_at_ms);
                }
            }
            persisted_activity(locator, observed_at_ms)
        })
        .collect()
}

/// Strongest proven status among live descendants, keyed by root session id.
/// Codex currently defaults to depth one, but walking the live parent map also
/// handles nested children without turning each rollout into a top-level row.
fn codex_descendant_statuses(
    live: &HashMap<PathBuf, CodexPeerStatus>,
) -> HashMap<String, CodexPeerStatus> {
    let lineage = live
        .iter()
        .filter_map(|(path, status)| {
            rollout_lineage(path).map(|(session_id, parent)| (session_id, parent, *status))
        })
        .collect::<Vec<_>>();
    let parent_by_child = lineage
        .iter()
        .filter_map(|(child, parent, _)| {
            parent
                .as_ref()
                .map(|parent| (child.clone(), parent.clone()))
        })
        .collect::<HashMap<_, _>>();
    let mut statuses = HashMap::new();
    for (_, parent, status) in lineage {
        let Some(mut root) = parent else {
            continue;
        };
        let mut visited = std::collections::HashSet::new();
        while visited.insert(root.clone()) {
            let Some(parent) = parent_by_child.get(&root) else {
                break;
            };
            root = parent.clone();
        }
        statuses
            .entry(root)
            .and_modify(|current| *current = stronger_codex_status(*current, status))
            .or_insert(status);
    }
    statuses
}

fn merge_codex_status(
    own: Option<CodexPeerStatus>,
    descendant: Option<CodexPeerStatus>,
) -> Option<CodexPeerStatus> {
    match (own, descendant) {
        (Some(own), Some(descendant)) => Some(stronger_codex_status(own, descendant)),
        (Some(status), None) | (None, Some(status)) => Some(status),
        (None, None) => None,
    }
}

fn stronger_codex_status(left: CodexPeerStatus, right: CodexPeerStatus) -> CodexPeerStatus {
    fn rank(status: CodexPeerStatus) -> u8 {
        match status {
            CodexPeerStatus::Running => 0,
            CodexPeerStatus::Idle => 1,
            CodexPeerStatus::Busy => 2,
            CodexPeerStatus::Waiting => 3,
        }
    }
    if rank(right) > rank(left) {
        right
    } else {
        left
    }
}

#[cfg(feature = "adapter-api")]
fn owned_activity(
    locator: &SessionLocator,
    state: crate::RuntimeRegistryState,
    observed_at_ms: u64,
) -> SessionActivity {
    use crate::RuntimeRegistryState;
    let (presence, turn) = match state {
        RuntimeRegistryState::Persisted => (SessionPresence::Persisted, SessionTurnState::Unknown),
        RuntimeRegistryState::Idle => (SessionPresence::Running, SessionTurnState::Idle),
        RuntimeRegistryState::Busy => (SessionPresence::Running, SessionTurnState::Working),
        RuntimeRegistryState::ShuttingDown => {
            (SessionPresence::ShuttingDown, SessionTurnState::Unknown)
        }
    };
    activity(
        locator,
        presence,
        turn,
        "supercode_runtime",
        Some(state.as_str()),
        None,
        observed_at_ms,
    )
}

fn claude_activity(
    locator: &SessionLocator,
    peer: &ClaudePeerSession,
    observed_at_ms: u64,
) -> SessionActivity {
    let turn = match peer.status {
        Some(ClaudePeerStatus::Busy) => SessionTurnState::Working,
        Some(ClaudePeerStatus::Idle) => SessionTurnState::Idle,
        Some(ClaudePeerStatus::Waiting) => SessionTurnState::NeedsInput,
        None => SessionTurnState::Unknown,
    };
    activity(
        locator,
        SessionPresence::Running,
        turn,
        "claude_registry",
        peer.status.map(|status| status.as_str()),
        peer.version.as_deref(),
        observed_at_ms,
    )
}

fn codex_activity(
    locator: &SessionLocator,
    status: CodexPeerStatus,
    observed_at_ms: u64,
) -> SessionActivity {
    let turn = match status {
        CodexPeerStatus::Running => SessionTurnState::Unknown,
        CodexPeerStatus::Idle => SessionTurnState::Idle,
        CodexPeerStatus::Busy => SessionTurnState::Working,
        CodexPeerStatus::Waiting => SessionTurnState::NeedsInput,
    };
    activity(
        locator,
        SessionPresence::Running,
        turn,
        "codex_rollout",
        Some(status.as_str()),
        None,
        observed_at_ms,
    )
}

fn persisted_activity(locator: &SessionLocator, observed_at_ms: u64) -> SessionActivity {
    activity(
        locator,
        SessionPresence::Persisted,
        SessionTurnState::Unknown,
        "persisted_store",
        None,
        None,
        observed_at_ms,
    )
}

fn activity(
    locator: &SessionLocator,
    presence: SessionPresence,
    turn: SessionTurnState,
    source: &str,
    native_state: Option<&str>,
    harness_version: Option<&str>,
    observed_at_ms: u64,
) -> SessionActivity {
    SessionActivity {
        harness: locator.harness.clone(),
        session_id: locator.session_id.clone(),
        presence,
        turn,
        evidence: SessionActivityEvidence {
            source: source.to_string(),
            native_state: native_state.map(str::to_string),
            observed_at_ms,
            harness_version: harness_version.map(str::to_string),
        },
    }
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .try_into()
        .unwrap_or(u64::MAX)
}