supercode-core 0.2.1

//! A live-runtime receipt is a routing hint that nothing re-issues: it is
//! written once, when the runtime registers itself, and no loop re-announces
//! it. Deleting one is therefore permanent for the rest of that runtime's
//! life — the frontend loses both the reported state and the route it would
//! attach through. Registry reads run continuously (the followed-session
//! projection probes every `harness serve` tick), so a single failed probe
//! must be treated as a hiccup rather than as evidence, while a runtime that
//! is genuinely gone must still reconcile away.

use std::fs;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use supercode::server::{run_http, RpcEngine};
use supercode::{
    find_live_runtime, register_live_runtime, Agent, ChatMessage, ChatRequest, Config,
    LiveRuntimeSource, LocalRuntimeRegistry, Provider, RuntimeAuthorization, RuntimeClientId,
    RuntimeRegistryState, Usage,
};
use tokio::io::AsyncWriteExt;
use tokio::sync::watch;

struct SaysProvider;

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

/// `SUPERCODE_HOME` is process-global, so the receipt directory is too.
fn environment_lock() -> &'static tokio::sync::Mutex<()> {
    static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}

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

/// A loopback TCP relay in front of a real runtime server. While it is
/// unhealthy it accepts and immediately closes, and it drops connections that
/// were already open — which is what a transient loopback hiccup looks like to
/// the probe's HTTP client, without pretending to know the hiccup's cause.
struct Relay {
    address: SocketAddr,
    health: watch::Sender<bool>,
    /// TCP connections the relay has accepted, i.e. what one registry read
    /// actually costs on the wire.
    connections: Arc<AtomicUsize>,
    /// Describe requests seen on the wire, counted from the bytes the code
    /// actually sends rather than from an assumed call graph.
    describes: Arc<AtomicUsize>,
}

impl Relay {
    async fn start(upstream: SocketAddr, healthy: bool) -> Self {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let (health, receiver) = watch::channel(healthy);
        let connections = Arc::new(AtomicUsize::new(0));
        let describes = Arc::new(AtomicUsize::new(0));
        let counter = connections.clone();
        let described = describes.clone();
        tokio::spawn(async move {
            loop {
                let Ok((mut inbound, _)) = listener.accept().await else {
                    return;
                };
                counter.fetch_add(1, Ordering::SeqCst);
                let healthy_now = *receiver.borrow();
                if !healthy_now {
                    let _ = inbound.shutdown().await;
                    continue;
                }
                let receiver = receiver.clone();
                let described = described.clone();
                tokio::spawn(async move {
                    let Ok(outbound) = tokio::net::TcpStream::connect(upstream).await else {
                        return;
                    };
                    let (client_read, mut client_write) = inbound.into_split();
                    let (mut server_read, server_write) = outbound.into_split();
                    tokio::select! {
                        _ = count_and_forward(client_read, server_write, described) => {}
                        _ = tokio::io::copy(&mut server_read, &mut client_write) => {}
                        _ = wait_unhealthy(receiver) => {}
                    }
                });
            }
        });
        Self {
            address,
            health,
            connections,
            describes,
        }
    }

    fn set_healthy(&self, healthy: bool) {
        self.health.send_replace(healthy);
    }

    fn connections(&self) -> usize {
        self.connections.load(Ordering::SeqCst)
    }

    fn describes(&self) -> usize {
        self.describes.load(Ordering::SeqCst)
    }
}

/// Forward one direction, counting Describe requests as they pass. A small
/// loopback POST arrives in one read, so this counts requests, not fragments.
async fn count_and_forward(
    mut reader: tokio::net::tcp::OwnedReadHalf,
    mut writer: tokio::net::tcp::OwnedWriteHalf,
    describes: Arc<AtomicUsize>,
) {
    const NEEDLE: &[u8] = b"frontend.v2.describe";
    let mut buffer = vec![0_u8; 16 * 1024];
    loop {
        let Ok(read) = tokio::io::AsyncReadExt::read(&mut reader, &mut buffer).await else {
            return;
        };
        if read == 0 {
            let _ = writer.shutdown().await;
            return;
        }
        let chunk = &buffer[..read];
        describes.fetch_add(
            chunk
                .windows(NEEDLE.len())
                .filter(|window| *window == NEEDLE)
                .count(),
            Ordering::SeqCst,
        );
        if writer.write_all(chunk).await.is_err() {
            return;
        }
    }
}

async fn wait_unhealthy(mut receiver: watch::Receiver<bool>) {
    while receiver.changed().await.is_ok() {
        if !*receiver.borrow_and_update() {
            return;
        }
    }
    std::future::pending::<()>().await
}

struct Fixture {
    root: PathBuf,
    engine: Arc<RpcEngine>,
    relay: Relay,
    source: LiveRuntimeSource,
    registration: supercode::LiveRuntimeRegistration,
}

impl Fixture {
    async fn start(label: &str, healthy: bool) -> Self {
        let root = temp_root(label);
        std::env::set_var("SUPERCODE_HOME", root.join("supercode-home"));
        let workspace = root.join("project");
        fs::create_dir_all(&workspace).unwrap();
        let engine = RpcEngine::new_named(
            Agent::with_provider(
                Config::builder().cwd(workspace.clone()).build(),
                Box::new(SaysProvider),
            ),
            label,
            None,
        );
        let token: Arc<str> = "receipt-reap-token".into();
        let upstream = run_http(engine.clone(), "127.0.0.1:0", token.clone())
            .await
            .unwrap();
        let relay = Relay::start(upstream, healthy).await;
        let source = LiveRuntimeSource {
            harness: "codex".into(),
            session_id: format!("{label}-source"),
            workspace,
        };
        let registration = register_live_runtime(
            label,
            source.clone(),
            format!("http://{}", relay.address),
            token.to_string(),
        )
        .unwrap();
        Self {
            root,
            engine,
            relay,
            source,
            registration,
        }
    }

    async fn state(&self) -> Option<RuntimeRegistryState> {
        LocalRuntimeRegistry::new()
            .source_state(
                &self.source.harness,
                &self.source.session_id,
                &RuntimeAuthorization::observer(),
            )
            .await
            .unwrap()
    }

    async fn finish(self) {
        self.engine.shutdown().await;
        drop(self.registration);
        std::env::remove_var("SUPERCODE_HOME");
        fs::remove_dir_all(self.root).ok();
    }
}

/// Isolated hiccups are not an outage, however many of them a long-lived
/// reader eventually accumulates. Between each one this runtime proves it is
/// alive through the frontend's own route — a successful attach, the very
/// operation the receipt exists to serve — so no number of them may add up to
/// a reap.
#[tokio::test]
async fn hiccups_separated_by_successful_contact_never_reap() {
    let _environment = environment_lock().lock().await;
    let fixture = Fixture::start("spaced-runtime", true).await;
    assert_eq!(fixture.state().await, Some(RuntimeRegistryState::Idle));

    for round in 0..3 {
        fixture.relay.set_healthy(false);
        assert_eq!(fixture.state().await, None, "round {round}: one hiccup");
        fixture.relay.set_healthy(true);
        let attached = LocalRuntimeRegistry::new()
            .attach(
                "spaced-runtime",
                RuntimeClientId::parse(format!("spaced-observer-{round}")).unwrap(),
                RuntimeAuthorization::observer(),
            )
            .await;
        assert!(
            attached.is_ok(),
            "round {round}: the runtime is alive and serving attaches, but its \
             receipt was destroyed by isolated hiccups around them"
        );
        assert!(
            find_live_runtime("spaced-runtime").unwrap().is_some(),
            "round {round}: a hiccup either side of proven liveness is not an outage"
        );
        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    fixture.finish().await;
}

/// The same defect without the attach: a reader that samples slowly enough can
/// see three failed probes in a row and still never have witnessed a
/// continuous outage. Failures further apart than the outage window each start
/// a new outage rather than extending the first one.
#[tokio::test]
async fn failed_probes_further_apart_than_the_outage_window_never_reap() {
    let _environment = environment_lock().lock().await;
    let fixture = Fixture::start("slow-reader-runtime", true).await;
    assert_eq!(fixture.state().await, Some(RuntimeRegistryState::Idle));

    // No successful probe intervenes — the runtime is simply up while nothing
    // is looking, which is what a slow reader sees.
    for round in 0..3 {
        fixture.relay.set_healthy(false);
        assert_eq!(fixture.state().await, None, "round {round}: one hiccup");
        fixture.relay.set_healthy(true);
        assert!(
            find_live_runtime("slow-reader-runtime").unwrap().is_some(),
            "round {round}: failures minutes apart are not one outage"
        );
        tokio::time::sleep(Duration::from_millis(2_200)).await;
    }
    assert_eq!(
        fixture.state().await,
        Some(RuntimeRegistryState::Idle),
        "the runtime was reachable throughout except for three single probes"
    );

    fixture.finish().await;
}

#[tokio::test]
async fn a_single_failed_probe_keeps_a_receipt_that_answers_again() {
    let _environment = environment_lock().lock().await;
    let fixture = Fixture::start("flaky-runtime", false).await;

    // The runtime is unreachable this instant, so its state is honestly
    // unknown and reconciles to `persisted`. That much is unchanged.
    assert_eq!(
        fixture.state().await,
        None,
        "an unreachable runtime must not have its state guessed at"
    );

    // But one failed probe is not evidence that the runtime is gone, and
    // nothing would ever re-register it.
    assert!(
        find_live_runtime("flaky-runtime").unwrap().is_some(),
        "a single failed probe destroyed a live runtime's only routing record"
    );

    fixture.relay.set_healthy(true);
    assert_eq!(
        fixture.state().await,
        Some(RuntimeRegistryState::Idle),
        "the runtime answered again and must be reported live again"
    );
    assert!(
        LocalRuntimeRegistry::new()
            .attach(
                "flaky-runtime",
                RuntimeClientId::parse("receipt-reap-observer").unwrap(),
                RuntimeAuthorization::observer(),
            )
            .await
            .is_ok(),
        "the frontend must still have a route to attach through"
    );

    fixture.finish().await;
}

#[tokio::test]
async fn a_runtime_that_stays_unreachable_is_still_reconciled_away() {
    let _environment = environment_lock().lock().await;
    let fixture = Fixture::start("dead-runtime", true).await;
    assert_eq!(fixture.state().await, Some(RuntimeRegistryState::Idle));
    let baseline_connections = fixture.relay.connections();

    fixture.engine.shutdown().await;
    fixture.relay.set_healthy(false);

    let started = Instant::now();
    tokio::time::timeout(Duration::from_secs(20), async {
        loop {
            assert_eq!(
                fixture.state().await,
                None,
                "a runtime that is gone reports persisted from the first failed probe"
            );
            if find_live_runtime("dead-runtime").unwrap().is_none() {
                return;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("a genuinely dead runtime's receipt must still be reconciled away");
    println!(
        "reaped a dead receipt after {:?} and {} probes",
        started.elapsed(),
        fixture.relay.connections() - baseline_connections
    );

    fixture.finish().await;
}

/// `harness serve` runs this read on every tick for every followed session, so
/// asking one runtime to describe itself twice for one read doubles that load
/// forever. The connection count is printed rather than asserted: one loopback
/// handshake per read is what the current client does, not a contract worth
/// pinning a pooled client against.
#[tokio::test]
async fn one_registry_read_describes_the_runtime_once() {
    let _environment = environment_lock().lock().await;
    let fixture = Fixture::start("cost-runtime", true).await;
    let connections = fixture.relay.connections();
    let describes = fixture.relay.describes();
    for _ in 0..10 {
        assert_eq!(fixture.state().await, Some(RuntimeRegistryState::Idle));
    }
    println!(
        "10 registry reads: {} TCP connections, {} describe RPCs",
        fixture.relay.connections() - connections,
        fixture.relay.describes() - describes
    );
    assert_eq!(
        fixture.relay.describes() - describes,
        10,
        "each registry read must describe the runtime exactly once"
    );
    fixture.finish().await;
}