supercode-harness 0.4.13

//! Live acceptance (opt-in): message a real running Claude Code session.
//!
//! Run with:
//!
//! ```text
//! SUPERCODE_LIVE_CLAUDE_PEER=1 cargo test -p supercode-core --test claude_peer_live -- --ignored --nocapture
//! ```
//!
//! It starts its OWN claude-code runtime through Supercode's managed-runtime
//! path, proves that runtime registered itself as a live session, messages
//! exactly that session through `harness.v1.sessions.message`, and follows the
//! session's own store until the text shows up in its transcript. It never
//! addresses a session it did not start: the target is resolved from the
//! runtime id Supercode minted, and the test asserts the registry record's cwd
//! is the temporary workspace it created.
//!
//! It spawns real processes and spends real tokens (one small turn on the
//! runtime, one haiku-class courier), which is why it is `#[ignore]`d behind an
//! explicit env gate rather than run by default.

use std::time::{Duration, Instant};

use serde_json::{json, Value};
use supercode_harness::claude_peer::{read_registry, registry_dir, ClaudePeerEndpoint};
use supercode_harness::harness_service::HarnessSessionService;
use supercode_harness::{HarnessCatalog, HarnessHomes, SessionLocator};

const READY_TIMEOUT: Duration = Duration::from_secs(90);
const ARRIVAL_TIMEOUT: Duration = Duration::from_secs(60);

#[tokio::test]
#[ignore = "starts a real Claude Code runtime and spends courier tokens; set SUPERCODE_LIVE_CLAUDE_PEER=1"]
async fn a_runtime_supercode_started_receives_a_message_through_sessions_message() {
    if std::env::var("SUPERCODE_LIVE_CLAUDE_PEER").is_err() {
        panic!(
            "SUPERCODE_LIVE_CLAUDE_PEER not set — this acceptance test starts a real \
             Claude Code process and pays for a courier turn; set it to 1 to run it."
        );
    }
    let run = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let workspace =
        std::env::temp_dir().join(format!("supercode-live-peer-{}-{run}", std::process::id()));
    std::fs::create_dir_all(&workspace).unwrap();
    // Unique per run, so a transcript match cannot be a leftover from an
    // earlier one.
    let nonce = format!("supercode-live-peer-nonce-{run}");
    let mut service = HarnessSessionService::new();

    let opened = service
        .handle_async(json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "harness.v1.runtimes.start",
            "params": {"harness": "claude-code", "cwd": workspace},
        }))
        .await;
    let connection = opened["result"]["connection"]
        .as_str()
        .unwrap_or_else(|| panic!("could not start a claude-code runtime: {opened}"))
        .to_string();
    let session_id = opened["result"]["handle"]["runtime_id"]
        .as_str()
        .unwrap()
        .to_string();
    eprintln!(
        "started claude-code runtime {session_id} in {}",
        workspace.display()
    );

    // One small turn, because a session's transcript file appears with its
    // first turn and both discovery and following address that file.
    let sent = service
        .handle_async(json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "harness.v1.runtimes.send_input",
            "params": {"connection": connection, "text": "Reply with READY and nothing else."},
        }))
        .await;
    assert!(sent.get("error").is_none(), "send_input failed: {sent}");

    // The runtime is only "live" once IT registered itself; that record is the
    // whole basis for messaging it.
    let homes = HarnessHomes::default();
    let deadline = Instant::now() + READY_TIMEOUT;
    let mut descriptor = None;
    let mut record = None;
    while Instant::now() < deadline && (descriptor.is_none() || record.is_none()) {
        drain(&mut service).await;
        let discovered = service
            .handle_async(json!({
                "jsonrpc": "2.0",
                "id": 3,
                "method": "harness.v1.sessions.discover",
                "params": {"harnesses": ["claude-code"], "workspace": workspace},
            }))
            .await;
        descriptor = discovered["result"]["sessions"]
            .as_array()
            .unwrap()
            .iter()
            .find(|session| session["locator"]["session_id"] == session_id.as_str())
            .cloned();
        record = read_registry(&registry_dir(&homes))
            .into_iter()
            .find(|peer| peer.session_id == session_id);
        if descriptor.is_none() || record.is_none() {
            tokio::time::sleep(Duration::from_millis(500)).await;
        }
    }
    let descriptor =
        descriptor.unwrap_or_else(|| panic!("runtime {session_id} never appeared in discovery"));
    let record = record
        .unwrap_or_else(|| panic!("runtime {session_id} never published a live-session record"));
    // The peer projection is the messageable identity, whatever else discovery
    // chose to publish. A runtime Supercode HOSTS also carries a
    // `supercode-live://` receipt, and that receipt keeps the single
    // `live_endpoint` slot because attaching is strictly richer than sending
    // one message; the peer endpoint fills that slot for the ordinary case,
    // a Claude session running in somebody's own terminal.
    let endpoint = record.endpoint();
    assert_eq!(
        ClaudePeerEndpoint::parse(endpoint.as_str()).unwrap(),
        endpoint
    );
    assert_eq!(endpoint.pid(), record.pid);
    assert_eq!(endpoint.name(), record.name);
    eprintln!(
        "live record: pid {} name {} socket {} (discovery published {:?})",
        record.pid,
        record.name,
        record.socket_path.display(),
        descriptor["live_endpoint"],
    );
    // Never message a session this test did not start.
    assert_eq!(
        std::fs::canonicalize(record.cwd.clone().unwrap()).unwrap(),
        std::fs::canonicalize(&workspace).unwrap(),
        "the target must be the runtime this test started, not a pre-existing session"
    );

    let locator: SessionLocator = serde_json::from_value(descriptor["locator"].clone()).unwrap();
    let started = Instant::now();
    let delivered = service
        .handle_async(json!({
            "jsonrpc": "2.0",
            "id": 4,
            "method": "harness.v1.sessions.message",
            "params": {"locator": locator, "text": nonce},
        }))
        .await;
    let result = &delivered["result"];
    assert_eq!(
        result["delivered_to_bus"], true,
        "delivery refused: {delivered}"
    );
    assert_eq!(result["target"]["name"], record.name.as_str());
    eprintln!(
        "courier delivered in {:?} (report {:?})",
        started.elapsed(),
        result["courier"]["report"]
    );

    // Follow the receiving session's OWN store: the message is real when its
    // transcript says so, not when the courier says so.
    let mut follower = HarnessCatalog::new().follow(&locator).unwrap();
    let deadline = Instant::now() + ARRIVAL_TIMEOUT;
    let mut arrived = false;
    while Instant::now() < deadline && !arrived {
        while let Some(event) = follower.poll().unwrap() {
            if event.to_json().to_string().contains(&nonce) {
                arrived = true;
                break;
            }
        }
        if !arrived {
            tokio::time::sleep(Duration::from_millis(500)).await;
        }
    }
    eprintln!(
        "nonce {} in the followed transcript after {:?}",
        if arrived { "arrived" } else { "NEVER arrived" },
        started.elapsed()
    );

    let closed = service
        .handle_async(json!({
            "jsonrpc": "2.0",
            "id": 5,
            "method": "harness.v1.runtimes.close",
            "params": {"connection": connection},
        }))
        .await;
    assert_eq!(closed["result"]["closed"], true);
    std::fs::remove_dir_all(&workspace).ok();
    // A killed session never gets to clean up after itself, so its record
    // outlives it — harmless, since every read re-checks the pid, but this
    // test should not litter. Only OUR record, and only once its pid is gone.
    let stale = registry_dir(&homes).join(format!("{}.json", record.pid));
    for _ in 0..20 {
        if process_is_gone(record.pid) {
            if std::fs::read_to_string(&stale)
                .map(|text| text.contains(&session_id))
                .unwrap_or(false)
            {
                std::fs::remove_file(&stale).ok();
            }
            break;
        }
        tokio::time::sleep(Duration::from_millis(250)).await;
    }
    assert!(
        arrived,
        "`{nonce}` never reached the live session's transcript within {ARRIVAL_TIMEOUT:?}"
    );
}

/// Keep the hosted runtime's event pump moving while the test waits.
async fn drain(service: &mut HarnessSessionService) {
    let _: Vec<Value> = service.poll_runtimes().await;
}

fn process_is_gone(pid: u32) -> bool {
    !std::process::Command::new("kill")
        .args(["-0", &pid.to_string()])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}