supercode-harness 0.4.2

//! A frontend mirroring a persisted session must be able to see that the
//! agent writing it is working. The transcript alone cannot say that, so the
//! live-runtime registry's reconciled state is projected onto the same
//! subscription the transcript follower already uses.

use std::fs;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use serde_json::{json, Value};
use supercode_harness::server::{run_http, RpcEngine};
use supercode_harness::{
    register_live_runtime, Agent, ChatMessage, ChatRequest, Config, FrontendRuntime,
    HarnessSessionService, LiveRuntimeSource, Provider, Usage,
};
use tokio::sync::Semaphore;

/// Holds one turn open until the test releases it, so `busy` is observable
/// rather than a race against a same-tick completion.
struct GatedProvider {
    gate: Arc<Semaphore>,
}

#[async_trait]
impl Provider for GatedProvider {
    async fn complete(
        &self,
        _request: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        let permit = self.gate.acquire().await.expect("gate stays open");
        permit.forget();
        Ok((ChatMessage::assistant("gated reply"), Usage::default()))
    }
}

fn subscribe(service: &mut HarnessSessionService, id: i64, locator: &Value) -> String {
    let opened = service.handle(json!({
        "jsonrpc": "2.0",
        "id": id,
        "method": "harness.v1.sessions.follow",
        "params": {"locator": locator},
    }));
    assert!(opened.get("error").is_none(), "{opened}");
    opened["result"]["subscription"]
        .as_str()
        .expect("follow returns a subscription")
        .to_string()
}

/// Latest reported state per subscription after draining one poll.
async fn poll_states(service: &mut HarnessSessionService) -> Vec<(String, String)> {
    service
        .poll_session_runtime_states()
        .await
        .into_iter()
        .map(|notification| {
            assert_eq!(
                notification["method"], "harness.v1.sessions.event",
                "lifecycle rides the session subscription, not a second channel"
            );
            let event = &notification["params"]["event"];
            assert_eq!(event["type"], "runtime_state");
            assert!(
                event.get("sequence").is_none(),
                "an unsequenced lifecycle event must not disturb transcript sequencing"
            );
            (
                notification["params"]["subscription"]
                    .as_str()
                    .unwrap()
                    .to_string(),
                event["state"].as_str().unwrap().to_string(),
            )
        })
        .collect()
}

async fn wait_for_state(
    service: &mut HarnessSessionService,
    subscription: &str,
    expected: &str,
) -> Vec<(String, String)> {
    let mut seen = Vec::new();
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let states = poll_states(service).await;
            let hit = states
                .iter()
                .any(|(sub, state)| sub == subscription && state == expected);
            seen.extend(states);
            if hit {
                return;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    })
    .await
    .unwrap_or_else(|_| panic!("subscription never reported `{expected}`; saw {seen:?}"));
    seen
}

#[tokio::test]
async fn a_followed_session_reports_its_live_runtime_working_and_idle_again() {
    let nonce = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let root = std::env::temp_dir().join(format!(
        "supercode-session-runtime-state-{}-{nonce}",
        std::process::id()
    ));
    let workspace = root.join("project");
    let codex_home = root.join("codex-sessions");
    fs::create_dir_all(&workspace).unwrap();
    fs::create_dir_all(&codex_home).unwrap();
    std::env::set_var("SUPERCODE_HOME", root.join("supercode-home"));
    for id in ["hosted-source", "terminal-only"] {
        fs::write(
            codex_home.join(format!("{id}.jsonl")),
            json!({
                "timestamp": "2026-01-01T00:00:00Z",
                "type": "session_meta",
                "payload": {"id": id, "cwd": workspace, "model": "test"}
            })
            .to_string()
                + "\n",
        )
        .unwrap();
    }

    let gate = Arc::new(Semaphore::new(0));
    let engine = RpcEngine::new_named(
        Agent::with_provider(
            Config::builder().cwd(workspace.clone()).build(),
            Box::new(GatedProvider { gate: gate.clone() }),
        ),
        "hosted-runtime",
        None,
    );
    let token: Arc<str> = "private-live-token".into();
    let address = run_http(engine.clone(), "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let registration = register_live_runtime(
        engine.session_id(),
        LiveRuntimeSource {
            harness: "codex".into(),
            session_id: "hosted-source".into(),
            workspace: workspace.clone(),
        },
        format!("http://{address}"),
        token.to_string(),
    )
    .unwrap();

    let mut service = HarnessSessionService::new();
    let discovered = service.handle(json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "harness.v1.sessions.discover",
        "params": {
            "workspace": workspace,
            "harnesses": ["codex"],
            "homes": {"codex": codex_home}
        }
    }));
    let sessions = discovered["result"]["sessions"].as_array().unwrap();
    let locator_of = |id: &str| {
        sessions
            .iter()
            .find(|session| session["locator"]["session_id"] == id)
            .unwrap_or_else(|| panic!("discovery should list `{id}`"))["locator"]
            .clone()
    };
    let hosted = locator_of("hosted-source");
    let terminal = locator_of("terminal-only");

    let hosted_sub = subscribe(&mut service, 2, &hosted);
    let terminal_sub = subscribe(&mut service, 3, &terminal);

    // Baseline: a registered runtime that is not running a turn is idle, and a
    // session with no Supercode receipt is reported as merely persisted —
    // never inferred to be working from its transcript.
    let baseline = poll_states(&mut service).await;
    assert!(
        baseline.contains(&(hosted_sub.clone(), "idle".into())),
        "{baseline:?}"
    );
    assert!(
        baseline.contains(&(terminal_sub.clone(), "persisted".into())),
        "{baseline:?}"
    );
    assert!(
        poll_states(&mut service).await.is_empty(),
        "an unchanged state must not re-notify"
    );

    FrontendRuntime::send_input(engine.clone(), "start working".into())
        .await
        .unwrap();
    let working = wait_for_state(&mut service, &hosted_sub, "busy").await;
    assert!(
        !working.iter().any(|(sub, _)| sub == &terminal_sub),
        "an unregistered session must not change state because a neighbour got busy: {working:?}"
    );

    gate.add_permits(1);
    wait_for_state(&mut service, &hosted_sub, "idle").await;

    // Dropping the subscription stops the lifecycle projection with it.
    let unfollowed = service.handle(json!({
        "jsonrpc": "2.0",
        "id": 4,
        "method": "harness.v1.sessions.unfollow",
        "params": {"subscription": hosted_sub},
    }));
    assert_eq!(unfollowed["result"]["removed"], true);
    assert!(poll_states(&mut service).await.is_empty());

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