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_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();

    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) = rollout_status(&codex, locator.storage.path()) {
                    return codex_activity(locator, status, observed_at_ms);
                }
            }
            persisted_activity(locator, observed_at_ms)
        })
        .collect()
}

#[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::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('/'));
        }
    }
}