supercode_harness/

runtime_registry.rs

//! Authenticated inventory and attachment for live and persisted sessions.
//!
//! The registry joins durable harness discovery with private live-runtime
//! receipts. Receipts remain local routing hints: canonical/native sessions,
//! sidecars, and exports are never deleted during stale reconciliation.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::PathBuf;
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};

use crate::{
    find_live_runtime, forget_live_runtime, list_live_runtimes, resolve_live_runtime,
    DiscoveryQuery, FrontendActions, FrontendConnectionState, FrontendRuntimeDescriptor,
    FrontendTurnState, HarnessCatalog, HttpFrontendRuntime, LiveRuntimeEndpoint,
    RuntimeAuthorization, RuntimeClientId, RuntimeControllerLease, RuntimeObserverLease,
    RuntimePermission, SdkError, SdkOperation, Session,
};
use supercode_interchange::catalog::StorageLocator;

/// Filters controlling one joined live/persisted inventory read.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct RuntimeRegistryQuery {
    /// Persisted harness discovery filters and roots.
    pub persisted: DiscoveryQuery,
    /// Include active SDK runtimes.
    pub include_live: bool,
    /// Include durable sessions which are not necessarily live.
    pub include_persisted: bool,
}

impl Default for RuntimeRegistryQuery {
    fn default() -> Self {
        Self {
            persisted: DiscoveryQuery::default(),
            include_live: true,
            include_persisted: true,
        }
    }
}

/// Reconciled lifecycle state reported by list/describe/watch.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RuntimeRegistryState {
    /// Durable session exists but has no registered live runtime.
    Persisted,
    /// Live runtime is ready for a turn.
    Idle,
    /// Live runtime owns an active turn.
    Busy,
    /// Live runtime is shutting down.
    ShuttingDown,
}

impl RuntimeRegistryState {
    /// Stable wire token, identical to this value's serde representation.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Persisted => "persisted",
            Self::Idle => "idle",
            Self::Busy => "busy",
            Self::ShuttingDown => "shutting_down",
        }
    }
}

/// Process and controller ownership for one live entry.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeRegistryOwner {
    /// Local process that owns execution and persistence.
    pub pid: u32,
    /// Current frontend controller lease, if any.
    pub controller: Option<RuntimeControllerLease>,
}

/// Stable joined descriptor returned by the registry.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeRegistryEntry {
    /// Stable selector. Live entries use the SDK runtime id; persisted entries
    /// use `<harness>:<native-session-id>`.
    pub id: String,
    /// Stable SDK runtime id when live.
    pub runtime_id: Option<String>,
    /// Harness-native source session id.
    pub source_session_id: String,
    /// Workspace owned by the runtime/source identity.
    pub source_workspace: Option<PathBuf>,
    /// Source harness.
    pub source_harness: String,
    /// Resolved emulation profile.
    pub profile: Option<String>,
    /// Current durable/live state.
    pub state: RuntimeRegistryState,
    /// Current model label when live, otherwise lightweight persisted metadata.
    pub model: Option<String>,
    /// Runtime process/controller ownership.
    pub owner: Option<RuntimeRegistryOwner>,
    /// Attached observer leases in stable client-id order.
    pub observers: Vec<RuntimeObserverLease>,
    /// Live registration time.
    pub started_at_ms: Option<u128>,
    /// Last persisted update time.
    pub updated_at_ms: Option<u64>,
    /// Opaque live endpoint safe to display.
    pub endpoint: Option<LiveRuntimeEndpoint>,
    /// Available endpoint transports such as HTTP and ACP.
    pub endpoint_capabilities: Vec<String>,
    /// Actions permitted by the credential used for this registry read.
    pub actions: Option<FrontendActions>,
    /// Canonical or native durable location; never inferred from a tmux pane.
    pub persistence_location: Option<PathBuf>,
    /// Optional local process supervisor. Never used as session authority.
    pub supervisor: Option<crate::LiveRuntimeSupervisor>,
    /// Optional persisted title.
    pub title: Option<String>,
}

/// Change emitted by [`RuntimeRegistryWatch`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RuntimeRegistryEvent {
    /// New stable entry.
    Added {
        /// Complete current descriptor.
        entry: RuntimeRegistryEntry,
    },
    /// Existing entry changed state, ownership, metadata, or capabilities.
    Updated {
        /// Complete replacement descriptor.
        entry: RuntimeRegistryEntry,
    },
    /// Entry disappeared after close or stale reconciliation.
    Removed {
        /// Stable id that disappeared.
        id: String,
    },
    /// A polling iteration failed without terminating the watch.
    Error {
        /// Stable human-readable failure detail.
        message: String,
    },
}

/// Bounded watch subscription. Dropping it stops the polling task.
pub struct RuntimeRegistryWatch {
    receiver: tokio::sync::mpsc::Receiver<RuntimeRegistryEvent>,
    task: tokio::task::JoinHandle<()>,
}

impl RuntimeRegistryWatch {
    /// Receive the next registry change.
    pub async fn next(&mut self) -> Option<RuntimeRegistryEvent> {
        self.receiver.recv().await
    }
}

impl Drop for RuntimeRegistryWatch {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Local authenticated registry backed by harness discovery and private
/// live-runtime receipts.
#[derive(Debug, Clone, Copy, Default)]
pub struct LocalRuntimeRegistry;

impl LocalRuntimeRegistry {
    /// Construct a stateless registry facade.
    pub fn new() -> Self {
        Self
    }

    /// List live and/or persisted sessions after enforcing observe authority.
    pub async fn list(
        &self,
        query: &RuntimeRegistryQuery,
        authorization: &RuntimeAuthorization,
    ) -> Result<Vec<RuntimeRegistryEntry>, SdkError> {
        require_permission(authorization, RuntimePermission::Observe)?;
        let mut entries = BTreeMap::<String, RuntimeRegistryEntry>::new();
        if query.include_persisted {
            let persisted = HarnessCatalog::new()
                .discover(&query.persisted)
                .map_err(|error| SdkError::Execution {
                    operation: SdkOperation::Discover,
                    message: error.to_string(),
                })?;
            for descriptor in persisted {
                let id = format!(
                    "{}:{}",
                    descriptor.locator.harness.as_str(),
                    descriptor.locator.session_id
                );
                let persistence_location = Some(match &descriptor.locator.storage {
                    StorageLocator::File { path } | StorageLocator::Sqlite { path, .. } => {
                        path.clone()
                    }
                });
                entries.insert(
                    id.clone(),
                    RuntimeRegistryEntry {
                        id,
                        runtime_id: None,
                        source_session_id: descriptor.locator.session_id,
                        source_workspace: None,
                        source_harness: descriptor.locator.harness.0,
                        profile: None,
                        state: RuntimeRegistryState::Persisted,
                        model: descriptor.model,
                        owner: None,
                        observers: Vec::new(),
                        started_at_ms: None,
                        updated_at_ms: descriptor.updated_at_ms,
                        endpoint: None,
                        endpoint_capabilities: Vec::new(),
                        actions: None,
                        persistence_location,
                        supervisor: None,
                        title: descriptor.title,
                    },
                );
            }
        }
        if query.include_live {
            for record in list_live_runtimes().map_err(registry_receipt_error)? {
                let Some((probe, descriptor)) = probe_receipt(&record).await? else {
                    continue;
                };
                let leases = probe.lease_snapshot().await?;
                let entry = live_entry(record, descriptor, leases);
                if entries.insert(entry.id.clone(), entry).is_some() {
                    return Err(SdkError::Execution {
                        operation: SdkOperation::Discover,
                        message: "duplicate stable runtime id in live registry".into(),
                    });
                }
            }
        }
        Ok(entries.into_values().collect())
    }

    /// Describe one stable entry without attaching an observer.
    pub async fn describe(
        &self,
        id: &str,
        query: &RuntimeRegistryQuery,
        authorization: &RuntimeAuthorization,
    ) -> Result<RuntimeRegistryEntry, SdkError> {
        self.list(query, authorization)
            .await?
            .into_iter()
            .find(|entry| entry.id == id)
            .ok_or_else(|| SdkError::NotFound {
                operation: SdkOperation::Discover,
                message: format!("runtime or persisted session `{id}`"),
            })
    }

    /// Reconciled lifecycle state of the live runtime registered for one
    /// persisted source session.
    ///
    /// `None` means no live Supercode runtime is registered for that identity —
    /// a harness running outside Supercode leaves no receipt, so its activity
    /// is unknowable and is never guessed at. An endpoint that does not answer
    /// is reported as `None` for that read, and its receipt is reconciled away
    /// under exactly the policy [`Self::list`] uses.
    pub async fn source_state(
        &self,
        harness: &str,
        session_id: &str,
        authorization: &RuntimeAuthorization,
    ) -> Result<Option<RuntimeRegistryState>, SdkError> {
        require_permission(authorization, RuntimePermission::Observe)?;
        for record in list_live_runtimes().map_err(registry_receipt_error)? {
            if record.source.harness != harness || record.source.session_id != session_id {
                continue;
            }
            let Some((_probe, descriptor)) = probe_receipt(&record).await? else {
                continue;
            };
            return Ok(Some(reconciled_state(&descriptor)));
        }
        Ok(None)
    }

    /// Sample only the requested source identities in one receipt scan. Activity
    /// polls need fresh lifecycle evidence, but neither unrelated runtime probes
    /// nor the controller/observer leases fetched by a full inventory read.
    pub(crate) async fn source_states(
        &self,
        sources: &BTreeSet<(String, String)>,
        authorization: &RuntimeAuthorization,
    ) -> Result<BTreeMap<(String, String), RuntimeRegistryState>, SdkError> {
        require_permission(authorization, RuntimePermission::Observe)?;
        if sources.is_empty() {
            return Ok(BTreeMap::new());
        }
        // Retain list's stable-runtime ordering and duplicate-id validation
        // before collapsing multiple runtimes for the same source identity.
        let mut runtimes = BTreeMap::new();
        for record in list_live_runtimes().map_err(registry_receipt_error)? {
            let key = (
                record.source.harness.clone(),
                record.source.session_id.clone(),
            );
            if !sources.contains(&key) {
                continue;
            }
            let Some((_probe, descriptor)) = probe_receipt(&record).await? else {
                continue;
            };
            if runtimes
                .insert(
                    record.runtime_session_id,
                    (key, reconciled_state(&descriptor)),
                )
                .is_some()
            {
                return Err(SdkError::Execution {
                    operation: SdkOperation::Discover,
                    message: "duplicate stable runtime id in live registry".into(),
                });
            }
        }
        Ok(runtimes.into_values().collect())
    }

    /// Attach an authenticated SDK client to one live runtime.
    pub async fn attach(
        &self,
        runtime_id: &str,
        client_id: RuntimeClientId,
        authorization: RuntimeAuthorization,
    ) -> Result<Arc<HttpFrontendRuntime>, SdkError> {
        require_permission(&authorization, RuntimePermission::Observe)?;
        let record = find_live_runtime(runtime_id)
            .map_err(registry_receipt_error)?
            .ok_or_else(|| SdkError::NotFound {
                operation: SdkOperation::Resume,
                message: format!("live runtime `{runtime_id}`"),
            })?;
        let resolved = resolve_live_runtime(&record.endpoint, &record.source)
            .map_err(registry_receipt_error)?;
        let attached = HttpFrontendRuntime::connect_with_authorization(
            resolved.base_url,
            resolved.token,
            client_id,
            authorization,
        )
        .await?;
        // A completed attachment is the strongest liveness evidence this
        // module can have — the receipt just did the job it exists for — so it
        // ends any outage a passing probe failure had opened.
        note_reachable(&record.endpoint);
        Ok(attached)
    }

    /// Load one persisted descriptor through the same catalog used by list.
    pub fn load_persisted(
        &self,
        id: &str,
        query: &RuntimeRegistryQuery,
        authorization: &RuntimeAuthorization,
    ) -> Result<Session, SdkError> {
        require_permission(authorization, RuntimePermission::Observe)?;
        let descriptor = HarnessCatalog::new()
            .discover(&query.persisted)
            .map_err(|error| SdkError::Execution {
                operation: SdkOperation::Discover,
                message: error.to_string(),
            })?
            .into_iter()
            .find(|descriptor| {
                format!(
                    "{}:{}",
                    descriptor.locator.harness.as_str(),
                    descriptor.locator.session_id
                ) == id
            })
            .ok_or_else(|| SdkError::NotFound {
                operation: SdkOperation::Load,
                message: format!("persisted session `{id}`"),
            })?;
        HarnessCatalog::new()
            .load(&descriptor.locator)
            .map_err(|error| SdkError::Execution {
                operation: SdkOperation::Load,
                message: error.to_string(),
            })
    }

    /// Watch joined registry state through a bounded change stream.
    pub fn watch(
        &self,
        query: RuntimeRegistryQuery,
        authorization: RuntimeAuthorization,
        poll_interval: Duration,
    ) -> Result<RuntimeRegistryWatch, SdkError> {
        require_permission(&authorization, RuntimePermission::Observe)?;
        let (sender, receiver) = tokio::sync::mpsc::channel(128);
        let registry = *self;
        let interval = poll_interval.max(Duration::from_millis(25));
        let task = tokio::spawn(async move {
            let mut previous = BTreeMap::<String, RuntimeRegistryEntry>::new();
            let mut ticker = tokio::time::interval(interval);
            loop {
                ticker.tick().await;
                let current = match registry.list(&query, &authorization).await {
                    Ok(entries) => entries
                        .into_iter()
                        .map(|entry| (entry.id.clone(), entry))
                        .collect::<BTreeMap<_, _>>(),
                    Err(error) => {
                        if sender
                            .send(RuntimeRegistryEvent::Error {
                                message: error.to_string(),
                            })
                            .await
                            .is_err()
                        {
                            return;
                        }
                        continue;
                    }
                };
                for (id, entry) in &current {
                    let event = match previous.get(id) {
                        None => Some(RuntimeRegistryEvent::Added {
                            entry: entry.clone(),
                        }),
                        Some(prior) if prior != entry => Some(RuntimeRegistryEvent::Updated {
                            entry: entry.clone(),
                        }),
                        Some(_) => None,
                    };
                    if let Some(event) = event {
                        if sender.send(event).await.is_err() {
                            return;
                        }
                    }
                }
                for id in previous.keys().filter(|id| !current.contains_key(*id)) {
                    if sender
                        .send(RuntimeRegistryEvent::Removed { id: id.clone() })
                        .await
                        .is_err()
                    {
                        return;
                    }
                }
                previous = current;
            }
        });
        Ok(RuntimeRegistryWatch { receiver, task })
    }
}

/// Failed probes within one outage before a receipt is forgotten.
const FORGET_AFTER_FAILED_PROBES: u32 = 3;
/// How long one outage must last before its receipt is forgotten, and — the
/// same bound, deliberately — how far apart two failures may be and still
/// belong to the same outage. A gap wider than this is a stretch the runtime
/// was not observed to be down for, so it ends the outage rather than
/// extending it.
const FORGET_AFTER_UNREACHABLE_FOR: Duration = Duration::from_secs(2);

/// Reach one live receipt, and the ONE place a receipt is ever forgotten.
/// Every registry read — list, describe, watch, and the followed-session
/// projection — goes through this, so the reaping rule cannot fork.
///
/// `Ok(None)` means the runtime did not answer this read: the caller reports
/// it exactly as it reports a session with no receipt at all, which keeps a
/// genuinely gone runtime reconciling to `persisted` immediately.
///
/// Forgetting is the destructive half, and it is deliberately slower. Nothing
/// re-announces a runtime — the receipt is written once at registration — so a
/// forgotten receipt costs the frontend its route to attach for the rest of
/// that runtime's life. A single failed probe is therefore treated as a
/// hiccup, not as evidence: a receipt goes only after
/// `FORGET_AFTER_FAILED_PROBES` failures within ONE outage spanning at least
/// `FORGET_AFTER_UNREACHABLE_FOR`. Both bounds are needed, because a count
/// alone means whatever the caller's poll rate makes it mean (4 Hz on a
/// `harness serve` tick, seconds apart in a watch), and a duration alone would
/// still act on one unlucky probe.
///
/// "One outage" is the load-bearing word, and it is bounded from both ends.
/// Any successful contact through the receipt ends it — a probe here, and an
/// [`LocalRuntimeRegistry::attach`], which is the strongest liveness evidence
/// there is because it is the operation the receipt exists to serve. So does a
/// gap wider than `FORGET_AFTER_UNREACHABLE_FOR` between two failures, which
/// is a stretch nothing observed the runtime to be down for. Without both, the
/// count degenerates into "three unlucky hiccups, however far apart", which
/// destroys the receipt of a runtime that was up — and serving attaches —
/// between them.
///
/// The tally is per process, so a one-shot read — a single `runtime list` from
/// the CLI — leaves a stale receipt behind instead of reaping it, and so does
/// a reader that samples more slowly than the outage window, which can never
/// see two failures close enough together to corroborate. That is the intended
/// trade: such a reader still omits the runtime from its output, a receipt
/// whose owning process is gone is already removed when it is read, and the
/// file costs nothing until a reader that does sample fast enough — the serve
/// tick — corroborates the failure and removes it.
async fn probe_receipt(
    record: &crate::LiveRuntimeRecord,
) -> Result<Option<(Arc<HttpFrontendRuntime>, FrontendRuntimeDescriptor)>, SdkError> {
    let Ok(resolved) = resolve_live_runtime(&record.endpoint, &record.source) else {
        note_unreachable(&record.endpoint);
        return Ok(None);
    };
    let probe_id = registry_probe_id(&record.endpoint)?;
    match HttpFrontendRuntime::probe_described(resolved.base_url, resolved.token, probe_id).await {
        Ok(probed) => {
            note_reachable(&record.endpoint);
            Ok(Some(probed))
        }
        // A live PID whose loopback endpoint has stopped answering for good is
        // a stale routing record. Removing it never addresses durable session
        // or sidecar paths.
        Err(_) => {
            note_unreachable(&record.endpoint);
            Ok(None)
        }
    }
}

/// One in-progress outage: when it started, when it was last confirmed, and
/// how many probes have failed inside it.
struct Outage {
    started: Instant,
    latest: Instant,
    failures: u32,
}

/// Outages in progress, keyed by opaque endpoint. Callers lock it for the
/// duration of one update and never hold the guard across an await or an
/// unlink.
fn outages() -> &'static Mutex<HashMap<String, Outage>> {
    static OUTAGES: OnceLock<Mutex<HashMap<String, Outage>>> = OnceLock::new();
    OUTAGES.get_or_init(|| Mutex::new(HashMap::new()))
}

fn lock_outages() -> MutexGuard<'static, HashMap<String, Outage>> {
    outages()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Record that this endpoint answered. Any successful contact ends whatever
/// outage was in progress, which is why [`LocalRuntimeRegistry::attach`] calls
/// this too: a receipt that just served an attachment is demonstrably a good
/// route, and letting hiccups either side of it accumulate would destroy it.
fn note_reachable(endpoint: &LiveRuntimeEndpoint) {
    lock_outages().remove(endpoint.as_str());
}

/// Record that this endpoint did not answer, and forget its receipt once the
/// outage is corroborated. This is the ONLY place a receipt is ever forgotten.
fn note_unreachable(endpoint: &LiveRuntimeEndpoint) {
    let now = Instant::now();
    let corroborated = {
        let mut outages = lock_outages();
        // A failure further from the previous one than the outage window is
        // not part of that outage: nothing observed the runtime to be down in
        // between, and it may well have been serving. Dropping the entry here
        // is what makes this failure start a fresh outage, and it is also what
        // bounds the map — an entry outlives its last failure by one window.
        outages
            .retain(|_, outage| now.duration_since(outage.latest) <= FORGET_AFTER_UNREACHABLE_FOR);
        let outage = outages
            .entry(endpoint.as_str().to_string())
            .or_insert(Outage {
                started: now,
                latest: now,
                failures: 0,
            });
        outage.failures += 1;
        outage.latest = now;
        let corroborated = outage.failures >= FORGET_AFTER_FAILED_PROBES
            && now.duration_since(outage.started) >= FORGET_AFTER_UNREACHABLE_FOR;
        if corroborated {
            outages.remove(endpoint.as_str());
        }
        corroborated
    };
    // Unlink outside the lock: every other endpoint's update would otherwise
    // queue behind this one's filesystem call.
    if corroborated {
        let _ = forget_live_runtime(endpoint);
    }
}

/// The one mapping from a live runtime's own report to the registry's
/// reconciled lifecycle state. Every reader — list, describe, watch, and the
/// followed-session projection — goes through it.
fn reconciled_state(descriptor: &FrontendRuntimeDescriptor) -> RuntimeRegistryState {
    if descriptor.connection_state == FrontendConnectionState::ShuttingDown {
        RuntimeRegistryState::ShuttingDown
    } else if descriptor.turn_state == FrontendTurnState::Busy {
        RuntimeRegistryState::Busy
    } else {
        RuntimeRegistryState::Idle
    }
}

fn registry_probe_id(endpoint: &LiveRuntimeEndpoint) -> Result<RuntimeClientId, SdkError> {
    RuntimeClientId::parse(format!(
        "registry-{}",
        endpoint.as_str().rsplit('/').next().unwrap_or("probe")
    ))
    .map_err(|error| SdkError::InvalidArgument {
        operation: SdkOperation::Discover,
        message: error.to_string(),
    })
}

fn live_entry(
    record: crate::LiveRuntimeRecord,
    descriptor: FrontendRuntimeDescriptor,
    leases: crate::RuntimeLeaseSnapshot,
) -> RuntimeRegistryEntry {
    let state = reconciled_state(&descriptor);
    RuntimeRegistryEntry {
        id: record.runtime_session_id.clone(),
        runtime_id: Some(record.runtime_session_id),
        source_session_id: record.source.session_id,
        source_workspace: Some(record.source.workspace),
        source_harness: record.source.harness,
        profile: descriptor
            .emulation_profile
            .or(record.metadata.profile.clone()),
        state,
        model: Some(descriptor.model),
        owner: Some(RuntimeRegistryOwner {
            pid: record.pid,
            controller: leases.controller,
        }),
        observers: leases.observers,
        started_at_ms: Some(record.created_at_ms),
        updated_at_ms: None,
        endpoint: Some(record.endpoint),
        endpoint_capabilities: record.metadata.endpoint_capabilities,
        actions: Some(descriptor.actions),
        persistence_location: record.metadata.persistence_location,
        supervisor: record.metadata.supervisor,
        title: None,
    }
}

fn require_permission(
    authorization: &RuntimeAuthorization,
    permission: RuntimePermission,
) -> Result<(), SdkError> {
    if authorization.allows(permission) {
        Ok(())
    } else {
        Err(SdkError::Unauthorized {
            permission: permission.as_str().into(),
        })
    }
}

fn registry_receipt_error(error: crate::LiveRuntimeReceiptError) -> SdkError {
    SdkError::Execution {
        operation: SdkOperation::Discover,
        message: error.to_string(),
    }
}