supercode-harness 0.4.8

//! 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, HashMap};
use std::path::PathBuf;
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};

use crate::catalog::StorageLocator;
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,
};

/// 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)
    }

    /// 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(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::server::{run_http, RpcEngine};
    use crate::{
        register_live_runtime_with_metadata, Agent, ChatMessage, ChatRequest, Config, HarnessHomes,
        HarnessId, LiveRuntimeMetadata, LiveRuntimeSource, Provider, SdkRuntime, Usage,
    };
    use async_trait::async_trait;

    struct SaysProvider;

    #[async_trait]
    impl Provider for SaysProvider {
        async fn complete(
            &self,
            _request: &ChatRequest,
            _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
        ) -> crate::Result<(ChatMessage, Usage)> {
            Ok((ChatMessage::assistant("registry reply"), Usage::default()))
        }
    }

    fn root(label: &str) -> PathBuf {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "supercode-runtime-registry-{label}-{}-{}",
            std::process::id(),
            nonce
        ));
        std::fs::create_dir_all(&path).unwrap();
        path
    }

    #[tokio::test]
    #[allow(clippy::await_holding_lock)]
    async fn joined_registry_lists_watches_attaches_and_reconciles_without_data_loss() {
        let _guard = crate::live_runtime::test_environment_lock();
        let home = root("live");
        let workspace = home.join("workspace");
        std::fs::create_dir_all(&workspace).unwrap();
        let persisted = home.join("canonical.jsonl");
        std::fs::write(&persisted, "SOURCE_BYTES_MUST_SURVIVE\n").unwrap();
        std::env::set_var("SUPERCODE_HOME", &home);

        let agent = Agent::with_provider(
            Config::builder().cwd(workspace.clone()).build(),
            Box::new(SaysProvider),
        );
        let engine = RpcEngine::new_named(agent, "live-registry-1", None);
        let token: Arc<str> = "registry-owner-token".into();
        let address = run_http(engine.clone(), "127.0.0.1:0", token.clone())
            .await
            .unwrap();
        let registry = LocalRuntimeRegistry::new();
        let query = RuntimeRegistryQuery {
            include_live: true,
            include_persisted: false,
            ..RuntimeRegistryQuery::default()
        };
        let mut watch = registry
            .watch(
                query.clone(),
                RuntimeAuthorization::observer(),
                Duration::from_millis(25),
            )
            .unwrap();
        let registration = register_live_runtime_with_metadata(
            "live-registry-1",
            LiveRuntimeSource {
                harness: "claude-code".into(),
                session_id: "source-1".into(),
                workspace: workspace.clone(),
            },
            format!("http://{address}"),
            token.to_string(),
            LiveRuntimeMetadata {
                profile: Some("cc-parity".into()),
                persistence_location: Some(persisted.clone()),
                endpoint_capabilities: vec!["http".into(), "acp".into()],
                supervisor: None,
            },
        )
        .unwrap();

        let added = tokio::time::timeout(Duration::from_secs(2), watch.next())
            .await
            .unwrap()
            .unwrap();
        assert!(matches!(
            added,
            RuntimeRegistryEvent::Added { ref entry }
                if entry.id == "live-registry-1"
                    && entry.profile.as_deref() == Some("cc-parity")
                    && entry.state == RuntimeRegistryState::Idle
                    && entry.persistence_location.as_ref() == Some(&persisted)
                    && entry.owner.as_ref().unwrap().pid == std::process::id()
                    && entry.observers.is_empty()
                    && !entry.actions.as_ref().unwrap().submit
        ));

        let observer = registry
            .attach(
                "live-registry-1",
                RuntimeClientId::parse("registry-observer").unwrap(),
                RuntimeAuthorization::observer(),
            )
            .await
            .unwrap();
        assert!(!observer.describe().await.unwrap().actions.submit);
        assert!(matches!(
            observer.submit("denied".into()).await,
            Err(SdkError::Unauthorized { ref permission }) if permission == "interact"
        ));
        let owner = registry
            .attach(
                "live-registry-1",
                RuntimeClientId::parse("registry-owner").unwrap(),
                RuntimeAuthorization::owner(),
            )
            .await
            .unwrap();
        assert_eq!(
            owner.submit("continue".into()).await.unwrap(),
            "registry reply"
        );
        let listed = registry
            .list(&query, &RuntimeAuthorization::owner())
            .await
            .unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].observers.len(), 2);
        assert_eq!(
            listed[0]
                .owner
                .as_ref()
                .and_then(|owner| owner.controller.as_ref())
                .map(|lease| lease.client_id.as_str()),
            Some("registry-owner")
        );

        owner.close().await.unwrap();
        engine.wait_for_shutdown().await;
        drop(registration);
        let removed = tokio::time::timeout(Duration::from_secs(2), async {
            loop {
                let event = watch.next().await.unwrap();
                if matches!(event, RuntimeRegistryEvent::Removed { .. }) {
                    break event;
                }
            }
        })
        .await
        .unwrap();
        assert_eq!(
            removed,
            RuntimeRegistryEvent::Removed {
                id: "live-registry-1".into()
            }
        );
        assert_eq!(
            std::fs::read_to_string(&persisted).unwrap(),
            "SOURCE_BYTES_MUST_SURVIVE\n"
        );
        std::env::remove_var("SUPERCODE_HOME");
        std::fs::remove_dir_all(home).ok();
    }

    #[test]
    fn persisted_registry_entries_load_through_the_canonical_catalog() {
        let root = root("persisted");
        let workspace = root.join("workspace");
        let claude = root.join("claude");
        std::fs::create_dir_all(&workspace).unwrap();
        std::fs::create_dir_all(&claude).unwrap();
        let session_path = claude.join("session.jsonl");
        std::fs::write(
            &session_path,
            format!(
                "{{\"type\":\"user\",\"sessionId\":\"cc-registry\",\"cwd\":{},\"message\":{{\"role\":\"user\",\"content\":\"persisted fact\"}}}}\n",
                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
            ),
        )
        .unwrap();
        let empty = root.join("empty");
        std::fs::create_dir_all(&empty).unwrap();
        let query = RuntimeRegistryQuery {
            persisted: DiscoveryQuery {
                workspace: Some(workspace),
                harnesses: vec![HarnessId::from(HarnessId::CLAUDE_CODE)],
                homes: HarnessHomes {
                    claude_code: claude,
                    codex: empty.clone(),
                    pi: empty.clone(),
                    opencode: empty.clone(),
                    grok: empty.clone(),
                    gemini: empty.clone(),
                    goose: empty.clone(),
                    supercode: empty,
                },
                cursor: None,
                limit: None,
                query: None,
                include_topic_candidates: false,
                include_child_sessions: false,
            },
            include_live: false,
            include_persisted: true,
        };
        let registry = LocalRuntimeRegistry::new();
        let entries =
            futures::executor::block_on(registry.list(&query, &RuntimeAuthorization::observer()))
                .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].id, "claude-code:cc-registry");
        assert_eq!(entries[0].state, RuntimeRegistryState::Persisted);
        assert_eq!(
            entries[0].persistence_location.as_ref(),
            Some(&session_path)
        );
        let loaded = registry
            .load_persisted(
                "claude-code:cc-registry",
                &query,
                &RuntimeAuthorization::observer(),
            )
            .unwrap();
        assert_eq!(loaded.messages.len(), 1);
        assert_eq!(
            loaded.messages[0].content.as_deref(),
            Some("persisted fact")
        );
        std::fs::remove_dir_all(root).ok();
    }
}