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, HashMap};
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};

use crate::claude_peer::{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())
    }
}

/// 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 entries = crate::LocalRuntimeRegistry::new()
            .list(
                &crate::RuntimeRegistryQuery {
                    persisted: Default::default(),
                    include_live: true,
                    include_persisted: false,
                },
                &authorization,
            )
            .await?;
        let owned = entries
            .into_iter()
            .map(|entry| ((entry.source_harness, entry.source_session_id), entry.state))
            .collect::<BTreeMap<_, _>>();
        let claude = read_claude(locators, homes);
        let codex = locators
            .iter()
            .any(|locator| locator.harness.as_str() == HarnessId::CODEX)
            .then(|| self.codex.sample(&homes.codex))
            .unwrap_or_default();
        let mut activities = resolve_stock_with_evidence(locators, &claude, &codex)
            .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 = locators
        .iter()
        .any(|locator| locator.harness.as_str() == HarnessId::CODEX)
        .then(|| live_rollouts(&homes.codex))
        .unwrap_or_default();
    resolve_stock_with_evidence(locators, &claude, &codex)
}

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

fn resolve_stock_with_evidence(
    locators: &[SessionLocator],
    claude: &HashMap<String, ClaudePeerSession>,
    codex: &HashMap<PathBuf, CodexPeerStatus>,
) -> 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::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,
        }
    }
    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,
        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,
    };
    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)
}

/// Internal test fixture for the evidence precedence table.
#[cfg(all(test, feature = "adapter-api"))]
pub(crate) fn resolve_fixture(
    locator: &SessionLocator,
    owned: Option<crate::RuntimeRegistryState>,
) -> SessionActivity {
    owned.map_or_else(
        || persisted_activity(locator, 0),
        |state| owned_activity(locator, state, 0),
    )
}

#[cfg(all(test, feature = "adapter-api"))]
mod tests {
    use std::fs;
    use std::path::PathBuf;

    use super::*;
    use crate::{RuntimeRegistryState, StorageLocator};

    #[test]
    fn normalized_activity_keeps_presence_and_turn_orthogonal() {
        let locator = SessionLocator {
            harness: HarnessId("fixture".into()),
            session_id: "session-1".into(),
            storage: StorageLocator::File {
                path: PathBuf::from("/not-read"),
            },
        };
        let cases = [
            (None, SessionPresence::Persisted, SessionTurnState::Unknown),
            (
                Some(RuntimeRegistryState::Idle),
                SessionPresence::Running,
                SessionTurnState::Idle,
            ),
            (
                Some(RuntimeRegistryState::Busy),
                SessionPresence::Running,
                SessionTurnState::Working,
            ),
            (
                Some(RuntimeRegistryState::ShuttingDown),
                SessionPresence::ShuttingDown,
                SessionTurnState::Unknown,
            ),
        ];
        for (native, presence, turn) in cases {
            let activity = resolve_fixture(&locator, native);
            assert_eq!((activity.presence, activity.turn), (presence, turn));
            assert_eq!(activity.harness, locator.harness);
            assert_eq!(activity.session_id, locator.session_id);
            assert!(!activity.evidence.source.contains('/'));
        }
    }

    #[test]
    fn busy_codex_child_makes_the_root_conversation_working() {
        let child_path = std::env::temp_dir().join(format!(
            "supercode-codex-child-{}-{}.jsonl",
            std::process::id(),
            std::thread::current().name().unwrap_or("test")
        ));
        fs::write(
            &child_path,
            serde_json::json!({
                "timestamp": "2026-01-01T00:00:00Z",
                "type": "session_meta",
                "payload": {
                    "id": "child-session",
                    "cwd": "/project",
                    "parent_thread_id": "root-session",
                    "source": {"subagent":{"thread_spawn":{
                        "parent_thread_id":"root-session",
                        "depth":1,
                        "agent_path":"/root/reviewer"
                    }}}
                }
            })
            .to_string()
                + "\n",
        )
        .unwrap();
        let root = SessionLocator {
            harness: HarnessId::from(HarnessId::CODEX),
            session_id: "root-session".into(),
            storage: StorageLocator::File {
                path: PathBuf::from("/not-open/root.jsonl"),
            },
        };
        let activities = resolve_stock_with_evidence(
            &[root],
            &HashMap::new(),
            &HashMap::from([(child_path.clone(), CodexPeerStatus::Busy)]),
        );
        assert_eq!(activities.len(), 1);
        assert_eq!(activities[0].presence, SessionPresence::Running);
        assert_eq!(activities[0].turn, SessionTurnState::Working);
        assert_eq!(activities[0].session_id, "root-session");
        fs::remove_file(child_path).ok();
    }
}