trusty-mpm 1.8.2

//! Activity-inspection route for managed sessions.
//!
//! Why: extracted from `managed_routes/mod.rs` to keep that file under the 500-SLOC
//! production cap while the route stays cohesive in its own submodule.
//! What: defines [`ActivityResponse`] and the [`get_session_activity`] handler
//! (`GET /api/v1/sessions/managed/{id}/activity`).
//! Test: `activity_no_key_returns_raw_pane` in tests/session_manager_mvp.rs;
//! `handler_activity_cache_hit`.

use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};

use axum::{
    extract::{Path as AxumPath, State},
    response::IntoResponse,
};
use serde::Serialize;
use tracing::warn;
use trusty_common::credentials::SecretResolveError;

use crate::daemon::rpc::managed::outcome::RouteOutcome;
use crate::daemon::state::DaemonState;

use super::parse_id;

/// Return the last `n` lines of `text` (for capping the scrollback snapshot).
///
/// Why: the scrollback file may be arbitrarily large; the activity response
/// contract caps `raw_pane` at 60 lines, matching the live capture limit.
/// What: splits on newlines, takes the trailing min(n, len) lines, re-joins.
/// Test: covered by the activity handler returning capped content.
fn tail_lines(text: &str, n: usize) -> String {
    let lines: Vec<&str> = text.lines().collect();
    let start = lines.len().saturating_sub(n);
    lines[start..].join("\n")
}

/// How long one answer to "is the classifier key present?" is reused.
///
/// Why (#8563): the TUI polls `/activity` every second. A key added to the
/// store shows up within this window; nothing asks the store more often.
const KEY_PRESENCE_TTL: Duration = Duration::from_secs(30);

/// The daemon-wide presence cache behind `classification`.
static KEY_PRESENCE: KeyPresence = KeyPresence::new(KEY_PRESENCE_TTL);

/// A TTL-cached answer to "does the resolver hold an `OPENROUTER_API_KEY`?".
///
/// Why (#8236, #8563): the route read plain `std::env::var`, so a key moved out
/// of the LaunchAgent plist into the store hid `classification`. Resolving on
/// every request instead cost one store read per poll and, on a host with no
/// key, one ERROR line per poll. Cached presence was chosen over deriving the
/// field from the classifier: the monitor caches a missing-key outcome as a
/// plain `Unknown` verdict, and carrying the signal would add a field to the
/// public `ActivityCheckResult` — a semver break in a security patch.
/// What: an answer younger than `ttl` is reused; otherwise [`Self::present`]
/// runs the probe on a blocking thread and records it.
/// Test: `an_absent_key_logs_nothing_on_repeated_requests`,
/// `classifier_key_is_found_in_the_store_when_the_env_lacks_it`.
struct KeyPresence {
    /// How long an answer stays fresh.
    ttl: Duration,
    /// When the last answer was taken, and what it was.
    slot: Mutex<Option<(Instant, bool)>>,
}

impl KeyPresence {
    const fn new(ttl: Duration) -> Self {
        Self {
            ttl,
            slot: Mutex::new(None),
        }
    }

    /// The recorded answer, when it is younger than `ttl` at `now`.
    fn cached(&self, now: Instant) -> Option<bool> {
        let slot = self.slot.lock().unwrap_or_else(PoisonError::into_inner);
        slot.filter(|(at, _)| now.saturating_duration_since(*at) < self.ttl)
            .map(|(_, present)| present)
    }

    /// The cached answer, or `probe`'s, run on a blocking thread so a slow
    /// store holds no runtime worker. A panicked probe counts as absent.
    async fn present<F>(&self, probe: F) -> bool
    where
        F: FnOnce() -> bool + Send + 'static,
    {
        if let Some(present) = self.cached(Instant::now()) {
            return present;
        }
        // The request's subscriber follows the probe onto the blocking thread.
        let dispatch = tracing::dispatcher::get_default(Clone::clone);
        let present = tokio::task::spawn_blocking(move || {
            tracing::dispatcher::with_default(&dispatch, probe)
        })
        .await
        .unwrap_or(false);
        *self.slot.lock().unwrap_or_else(PoisonError::into_inner) = Some((Instant::now(), present));
        present
    }
}

/// Does `resolve` find the classifier key? Never logs.
///
/// Why (#8563): an absent key is the normal state of a host without
/// OpenRouter, not an error; [`crate::secret_source::resolve_secret`] logs it
/// at ERROR, so this reads through the non-logging resolver instead.
/// Test: `an_absent_key_logs_nothing_on_repeated_requests`.
fn classifier_key_resolves(
    resolve: impl FnOnce(&str) -> Result<String, SecretResolveError>,
) -> bool {
    resolve(trusty_common::env_vars::ENV_OPENROUTER_API_KEY).is_ok()
}

/// Does the daemon's resolver hold the classifier key? Never logs.
///
/// Why (#9121): the probe walked `.env.local` and the Keychain directly, past
/// the sandbox latch.
/// What: `classifier_key_resolves` over
/// [`crate::secret_source::resolve_bounded_gated`], which answers `Absent` in
/// sandbox mode without reading a tier.
/// Test: `tests/sandbox_latch_9121.rs`.
#[must_use]
pub fn classifier_key_present() -> bool {
    classifier_key_resolves(crate::secret_source::resolve_bounded_gated)
}

/// Response body for GET /api/v1/sessions/managed/{id}/activity.
///
/// Why: the calling agentic process needs the full activity picture without
/// requiring an LLM key — raw pane content and structured lifecycle fields are
/// always available; the LLM classification is an optional overlay when
/// OpenRouter is configured. This lets the calling agentic process do its own
/// inference over the raw pane content.
/// What: always-present fields: `raw_pane` (last 60 lines or last-seen snapshot),
/// `runtime_active` (tmux session alive or not), `pane_stale` (true when
/// `raw_pane` was served from the stop-time scrollback snapshot rather than a
/// live capture), `pending_decision`, `proposed_default`. LLM fields (`state`,
/// `summary`, `confidence`, `classification`) are populated when the classifier
/// ran successfully; `classification` is `null` when the key is absent or the
/// classifier was not invoked.
/// Test: activity route handler test; `activity_no_key_returns_raw_pane` test.
#[derive(Debug, Serialize)]
pub struct ActivityResponse {
    /// Raw pane content (last 60 lines, or the stop-time scrollback snapshot
    /// when the runtime is stopped and the live capture is empty). Always
    /// present so the calling agentic process can reason over terminal output.
    pub raw_pane: String,
    /// Whether the tmux runtime session is currently alive.
    pub runtime_active: bool,
    /// True when `raw_pane` was served from the persisted stop-time scrollback
    /// snapshot (i.e. the runtime is stopped and the live pane capture returned
    /// empty). Callers should treat this as a last-known-good snapshot, not a
    /// real-time view (#1840).
    pub pane_stale: bool,
    /// Activity state from LLM classification: working, idle, blocked_on_permission,
    /// errored, done, unknown. Populated from classifier verdict or "unknown".
    pub state: String,
    /// Human-readable summary of what the session is doing (from LLM or fallback).
    pub summary: String,
    /// Confidence of the classification (0.0–1.0). 0.0 when no classifier ran.
    pub confidence: f32,
    /// True when the verdict was served from the content-hash cache.
    pub cache_hit: bool,
    /// Input token count for this check (0 on cache hit or no classifier).
    pub input_tokens: u32,
    /// Output token count for this check (0 on cache hit or no classifier).
    pub output_tokens: u32,
    /// Latency in milliseconds for this check.
    pub latency_ms: u64,
    /// Cumulative input tokens across all checks for this session.
    pub total_input_tokens: u64,
    /// Cumulative output tokens across all checks for this session.
    pub total_output_tokens: u64,
    /// LLM classification result. `null` when no OpenRouter key or classifier
    /// not configured; string state name when classifier ran.
    pub classification: Option<String>,
    /// A pending decision question, if surfaced by a previous activity check.
    pub pending_decision: Option<String>,
    /// Proposed default answer to the pending decision.
    pub proposed_default: Option<String>,
}

/// GET /api/v1/sessions/managed/{id}/activity — inspect session activity.
///
/// Why: the calling agentic process needs to know whether the session is
/// working, idle, blocked, errored, or done WITHOUT requiring an LLM key.
/// The raw pane content is ALWAYS returned so the calling agentic process can
/// perform its own inference. The OpenRouter LLM classifier is invoked ONLY
/// when configured (i.e. when OPENROUTER_API_KEY is set).
/// What: captures the pane via the session's tmux driver (last 60 lines);
/// determines `runtime_active` from tmux presence; when the runtime is stopped
/// and the live capture is empty, falls back to the stop-time scrollback
/// snapshot in `record.scrollback_path` and sets `pane_stale=true` (#1840);
/// calls `ActivityMonitor::check` — which already converts `MissingApiKey` to
/// an Unknown verdict non-erroring — and returns the verdict alongside
/// `raw_pane` and `classification` (null when no classifier ran or key absent).
/// Test: `activity_no_key_returns_raw_pane` in tests/session_manager_mvp.rs;
/// `handler_activity_cache_hit`.
pub async fn get_session_activity(
    State(state): State<Arc<DaemonState>>,
    AxumPath(id_str): AxumPath<String>,
) -> impl IntoResponse {
    activity_core(&state, &id_str).await // #6288
}

/// The transport-neutral body of `GET .../{id}/activity` (#6288), served over
/// the socket as `mpm.managed.activity`.
///
/// Test: `managed_activity_parity` in `daemon::rpc::managed_tests`.
pub(crate) async fn activity_core(state: &Arc<DaemonState>, id_str: &str) -> RouteOutcome {
    let id = match parse_id(id_str) {
        Ok(id) => id,
        Err((code, msg)) => return RouteOutcome::text(code.as_u16(), msg),
    };
    let mgr = state.session_manager().await;
    let record = match mgr.get(&id).await {
        Ok(r) => r,
        Err(_) => return RouteOutcome::text(404, format!("session {id_str} not found")),
    };

    // Capture the last 60 pane lines — always available, no LLM key required.
    let pane_text = mgr
        .capture_pane(&id, 60)
        .await
        .unwrap_or_else(|_| String::new());

    // Determine runtime_active from tmux presence (independent of LLM classifier).
    let runtime_active = mgr.tmux_driver().session_exists(&record.tmux_name);

    // #1840: when the runtime is stopped and the live capture is empty, serve
    // the stop-time scrollback snapshot so callers get SOME content rather than
    // an empty string. When no snapshot is available, still mark pane_stale=true
    // so callers can distinguish stopped-quiet from live-quiet (pane_stale=false
    // means the runtime IS active).
    //
    // Three possible pane states:
    // 1. runtime_active=true,  pane_stale=false: live capture (may be empty if session is quiet)
    // 2. runtime_active=false, pane_stale=true,  raw_pane non-empty: serving stop-time snapshot
    // 3. runtime_active=false, pane_stale=true,  raw_pane empty: stopped, no snapshot available
    let (raw_pane, pane_stale) = if !runtime_active && pane_text.is_empty() {
        // Use tokio::fs::read_to_string to avoid blocking the async executor on
        // synchronous file I/O (#1840 E).
        let snapshot = if let Some(p) = record.scrollback_path.as_deref() {
            tokio::fs::read_to_string(p)
                .await
                .ok()
                .map(|s| tail_lines(&s, 60))
                .filter(|s| !s.is_empty())
        } else {
            None
        };
        match snapshot {
            Some(snap) => (snap, true),
            None => (String::new(), true), // stopped + no snapshot: pane_stale=true
        }
    } else {
        (pane_text, false)
    };

    // Run the optional LLM classifier. `ActivityMonitor::check` converts
    // `MissingApiKey` to an `Unknown` verdict non-erroring — the raw pane is
    // always available regardless of whether a key is configured.
    let monitor = state.activity_monitor();
    let result = match monitor.check(id_str, &raw_pane).await {
        Ok(r) => r,
        Err(e) => {
            warn!(session = %id_str, "activity check error (non-key): {e}");
            return RouteOutcome::text(500, format!("activity check failed: {e}"));
        }
    };

    // #8236: resolve through the shared resolver, not `std::env::var`, so a key
    // migrated out of the LaunchAgent plist into the store still surfaces here.
    // #8563: cached for `KEY_PRESENCE_TTL`, and never logged per request.
    // #9121: through the sandbox-gated resolver, never the raw bounded one.
    let api_key_present = KEY_PRESENCE.present(classifier_key_present).await;
    let classification = if api_key_present {
        Some(format!("{:?}", result.verdict.state).to_lowercase())
    } else {
        None
    };

    RouteOutcome::ok(&ActivityResponse {
        raw_pane,
        runtime_active,
        pane_stale,
        state: format!("{:?}", result.verdict.state).to_lowercase(),
        summary: result.verdict.summary,
        confidence: result.verdict.confidence,
        cache_hit: result.cache_hit,
        input_tokens: result.cost.input_tokens,
        output_tokens: result.cost.output_tokens,
        latency_ms: result.cost.latency_ms,
        total_input_tokens: result.tally.total_input_tokens,
        total_output_tokens: result.tally.total_output_tokens,
        classification,
        pending_decision: record.pending_decision,
        proposed_default: record.proposed_default,
    })
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use tracing_subscriber::layer::{Context, Layer};
    use tracing_subscriber::prelude::*;
    use trusty_common::credential_registry::provider_for_env_var;
    use trusty_common::credentials::{KeyStore, MemoryKeyStore};

    use super::*;

    /// A resolver whose env tier is empty and whose store tier is `store`.
    ///
    /// Why: the host running the tests may export `OPENROUTER_API_KEY`, and the
    /// tests must neither read nor mutate the real environment or store. This
    /// stands in for `resolve_env_var_bounded` with its env tier pinned absent.
    fn store_only_resolver(
        store: Arc<MemoryKeyStore>,
    ) -> impl FnOnce(&str) -> Result<String, SecretResolveError> + Send + 'static {
        move |var: &str| {
            provider_for_env_var(var)
                .and_then(|p| store.get(p))
                .ok_or_else(|| SecretResolveError::Absent {
                    var: var.to_string(),
                })
        }
    }

    /// Counts every WARN-or-worse event.
    struct CountWarnings(Arc<AtomicUsize>);

    impl<S: tracing::Subscriber> Layer<S> for CountWarnings {
        fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
            if *event.metadata().level() <= tracing::Level::WARN {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }
    }

    /// Why (#8236): a key migrated from the plist into the store must still
    /// surface `classification` after a restart that dropped it from the env.
    /// Test: this test.
    #[tokio::test]
    async fn classifier_key_is_found_in_the_store_when_the_env_lacks_it() {
        let store = Arc::new(MemoryKeyStore::new());
        store
            .set("openrouter", "sk-test-not-real")
            .expect("memory store accepts a write");
        let presence = KeyPresence::new(KEY_PRESENCE_TTL);

        let resolve = store_only_resolver(store);
        assert!(presence.present(|| classifier_key_resolves(resolve)).await);
    }

    /// Why (#8563): the TUI polls `/activity` every second. On a host without
    /// the key, each poll resolved the key again and logged an ERROR. Repeated
    /// requests must reach the store once per TTL and log nothing.
    /// Test: this test.
    #[tokio::test]
    async fn an_absent_key_logs_nothing_on_repeated_requests() {
        let warnings = Arc::new(AtomicUsize::new(0));
        let _guard = tracing::subscriber::set_default(
            tracing_subscriber::registry().with(CountWarnings(Arc::clone(&warnings))),
        );
        let store = Arc::new(MemoryKeyStore::new());
        let probes = Arc::new(AtomicUsize::new(0));
        let presence = KeyPresence::new(KEY_PRESENCE_TTL);

        for _ in 0..5 {
            let (store, probes) = (Arc::clone(&store), Arc::clone(&probes));
            let present = presence
                .present(move || {
                    probes.fetch_add(1, Ordering::SeqCst);
                    classifier_key_resolves(store_only_resolver(store))
                })
                .await;
            assert!(!present, "an absent key is absent");
        }

        assert_eq!(probes.load(Ordering::SeqCst), 1, "one probe per TTL");
        assert_eq!(warnings.load(Ordering::SeqCst), 0, "no per-request log");
        assert_eq!(
            presence.cached(Instant::now() + KEY_PRESENCE_TTL),
            None,
            "the answer expires, so a key added later is seen"
        );
    }
}