supercode-harness 0.4.21

//! Activity subscriptions must not probe runtimes outside their requested identities.
#![cfg(feature = "adapter-api")]

use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde_json::{json, Value};
use supercode_harness::harness_service::HarnessSessionService;
use supercode_harness::{
    find_live_runtime, register_live_runtime, HarnessId, LiveRuntimeRegistration,
    LiveRuntimeSource, SessionLocator, StorageLocator,
};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};

struct Environment {
    root: PathBuf,
    previous_home: Option<OsString>,
}

impl Environment {
    fn new() -> Self {
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let root = std::env::temp_dir().join(format!(
            "supercode-activity-scope-{}-{nonce}",
            std::process::id()
        ));
        std::fs::create_dir_all(&root).unwrap();
        let previous_home = std::env::var_os("SUPERCODE_HOME");
        std::env::set_var("SUPERCODE_HOME", &root);
        Self {
            root,
            previous_home,
        }
    }
}

impl Drop for Environment {
    fn drop(&mut self) {
        match &self.previous_home {
            Some(home) => std::env::set_var("SUPERCODE_HOME", home),
            None => std::env::remove_var("SUPERCODE_HOME"),
        }
        let _ = std::fs::remove_dir_all(&self.root);
    }
}

/// A protocol fixture that records complete HTTP requests, including unexpected
/// methods. Reading the declared body avoids assuming a request fits one TCP read.
struct Runtime {
    descriptor: Arc<Mutex<Option<Value>>>,
    methods: Arc<Mutex<Vec<String>>>,
    server: tokio::task::JoinHandle<()>,
    _registration: LiveRuntimeRegistration,
}

impl Runtime {
    async fn start(root: &Path, name: &str) -> Self {
        let fixture: Value = serde_json::from_str(include_str!(
            "../../../sdk/frontend/test/fixtures/conformance.json"
        ))
        .unwrap();
        let mut descriptor = fixture["descriptor"].clone();
        descriptor["session_id"] = json!(name);
        let descriptor = Arc::new(Mutex::new(Some(descriptor)));
        let methods = Arc::new(Mutex::new(Vec::new()));
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server_descriptor = descriptor.clone();
        let server_methods = methods.clone();
        let server = tokio::spawn(async move {
            loop {
                let (stream, _) = listener.accept().await.unwrap();
                let mut stream = BufReader::new(stream);
                let mut length = None;
                loop {
                    let mut line = String::new();
                    assert_ne!(stream.read_line(&mut line).await.unwrap(), 0);
                    if line == "\r\n" {
                        break;
                    }
                    if let Some((key, value)) = line.split_once(':') {
                        if key.eq_ignore_ascii_case("content-length") {
                            length = Some(value.trim().parse::<usize>().unwrap());
                        }
                    }
                }
                let mut body = vec![0; length.expect("RPC body length")];
                stream.read_exact(&mut body).await.unwrap();
                let request: Value = serde_json::from_slice(&body).unwrap();
                let method = request["method"].as_str().unwrap();
                server_methods.lock().unwrap().push(method.to_string());
                let descriptor = server_descriptor.lock().unwrap().clone();
                // A closed response simulates a transient runtime outage.
                let Some(descriptor) = descriptor else {
                    continue;
                };
                let result = if method == "frontend.v2.describe" {
                    descriptor
                } else {
                    json!({"controller": null, "observers": [], "lease_ttl_ms": 30000})
                };
                let body =
                    json!({"jsonrpc": "2.0", "id": request["id"], "result": result}).to_string();
                let response = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    body.len(), body
                );
                stream.write_all(response.as_bytes()).await.unwrap();
                stream.shutdown().await.unwrap();
            }
        });
        let registration = register_live_runtime(
            name,
            LiveRuntimeSource {
                harness: "hermes".into(),
                session_id: name.into(),
                workspace: root.to_path_buf(),
            },
            format!("http://{address}"),
            "activity-fixture-token",
        )
        .unwrap();
        Self {
            descriptor,
            methods,
            server,
            _registration: registration,
        }
    }

    fn assert_describes(&self, count: usize) {
        assert_eq!(
            *self.methods.lock().unwrap(),
            vec!["frontend.v2.describe"; count]
        );
    }
}

impl Drop for Runtime {
    fn drop(&mut self) {
        self.server.abort();
    }
}

#[tokio::test]
async fn activity_only_probes_requested_identities_and_preserves_transitions() {
    // This integration-test process owns its isolated SUPERCODE_HOME.
    let environment = Environment::new();
    let requested = Runtime::start(&environment.root, "requested").await;
    let second = Runtime::start(&environment.root, "second").await;
    let unrelated = Runtime::start(&environment.root, "unrelated").await;
    let durable = environment.root.join("session.jsonl");
    std::fs::write(&durable, "durable session sentinel\n").unwrap();
    let locator = SessionLocator {
        harness: HarnessId::new("hermes"),
        session_id: "requested".into(),
        storage: StorageLocator::File {
            path: durable.clone(),
        },
    };
    let mut service = HarnessSessionService::new();
    let second_locator = SessionLocator {
        session_id: "second".into(),
        ..locator.clone()
    };
    let initial = service
        .handle_async(json!({
            "jsonrpc": "2.0", "id": 1, "method": "harness.v1.sessions.activity.subscribe",
        "params": {"locators": [locator, locator, second_locator]}
        }))
        .await;
    assert!(initial.get("error").is_none(), "{initial}");
    assert_eq!(initial["result"]["initial"].as_array().unwrap().len(), 2);
    assert_eq!(initial["result"]["initial"][0]["session_id"], "requested");
    assert_eq!(initial["result"]["initial"][1]["session_id"], "second");
    assert_eq!(initial["result"]["initial"][0]["turn"], "idle");
    unrelated.assert_describes(0);
    requested.assert_describes(1);
    second.assert_describes(1);

    for count in 2..=5 {
        assert!(service.poll_session_activities().await.is_empty());
        requested.assert_describes(count);
        second.assert_describes(count);
        unrelated.assert_describes(0);
    }

    for (turn, connection, expected_turn, expected_presence) in [
        ("busy", "connected", "working", "running"),
        ("idle", "connected", "idle", "running"),
        ("idle", "shutting_down", "unknown", "shutting_down"),
    ] {
        {
            let mut descriptor = requested.descriptor.lock().unwrap();
            let descriptor = descriptor.as_mut().unwrap();
            descriptor["turn_state"] = json!(turn);
            descriptor["connection_state"] = json!(connection);
        }
        let events = service.poll_session_activities().await;
        assert_eq!(events.len(), 1);
        let activity = &events[0]["params"]["activities"][0];
        assert_eq!(activity["turn"], expected_turn);
        assert_eq!(activity["presence"], expected_presence);
    }
    requested.assert_describes(8);
    second.assert_describes(8);

    let saved = requested.descriptor.lock().unwrap().take();
    let events = service.poll_session_activities().await;
    assert_eq!(
        events[0]["params"]["activities"][0]["presence"],
        "persisted"
    );
    assert!(find_live_runtime("requested").unwrap().is_some());
    *requested.descriptor.lock().unwrap() = saved;
    let events = service.poll_session_activities().await;
    assert_eq!(
        events[0]["params"]["activities"][0]["presence"],
        "shutting_down"
    );

    requested.descriptor.lock().unwrap().take();
    tokio::time::timeout(Duration::from_secs(20), async {
        while find_live_runtime("requested").unwrap().is_some() {
            service.poll_session_activities().await;
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("a sustained outage still reaps the requested receipt");
    unrelated.assert_describes(0);
    assert!(find_live_runtime("unrelated").unwrap().is_some());
    assert_eq!(
        std::fs::read_to_string(durable).unwrap(),
        "durable session sentinel\n"
    );
    assert!(requested
        .methods
        .lock()
        .unwrap()
        .iter()
        .all(|method| method == "frontend.v2.describe"));
}