liminal-cli 0.3.30

//! ACCESS-015 R4: an open exchange (no `[auth]`) with `[open.construction]`,
//! `[agents]` and a template policy table runs the whole construction layer
//! through the liminal binary. `open:tom` creates a site and binds the seat
//! `builder`, which has listened once; the builder writes and finishes its
//! plan; `open:tom` proposes the program under the root, completes the site
//! and starts it; a call to the site's subject is answered by the program.
//! Every act is answered `executed`, approved by its declared holder.

use std::error::Error;
use std::io::{Read, Write};
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::Path;

use liminal::protocol::{WorkerActivityDescriptor, WorkerRegistration};
use liminal_sdk::{PushClient, PushReceiveOutcome};
use serde_json::{Value, json};

use super::network::OpenServer;
use super::open_services::seat_init;
use super::{Host, Ticket};

const OBJECT: &str = "{\"type\":\"object\"}";

/// The registry, its channel bound, the construction bounds and the root
/// policy over `root`.
fn config(root: &Path) -> Result<String, Box<dyn Error>> {
    let root = root.to_str().ok_or("root is not Unicode")?;
    Ok(format!(
        "[agents]\n\n[limits]\nmax_channels = 64\n\n[open.construction]\nsites = 8\nsubjects = 8\n\
         schema_bytes = 16384\ntext_bytes = 1024\nsteps = 32\ntransitions = 128\ndrafts = 128\n\
         approvers = 2\nsnapshot_bytes = 1048576\nrequest_bytes = 65536\n\n[templates]\n\
         roots = ['{root}']\ncommand_bytes = 1048576\nrecords = 8\narguments = 8\n\
         environment_names = 8\ntext_bytes = 1024\nstop_timeout_ms = 2000\n"
    ))
}

/// The builder's program under the root: this test binary's
/// [`site_process`], told the ready socket by its one argument. A rebuilt
/// program marks its answers `rebuilt`.
fn program(root: &Path, rebuilt: bool) -> Result<std::path::PathBuf, Box<dyn Error>> {
    let exe = std::env::current_exe()?;
    let path = root.join("forge");
    let mark = if rebuilt {
        "export LIMINAL_REBUILT=1\n"
    } else {
        ""
    };
    std::fs::write(
        &path,
        format!(
            "#!/bin/sh\n{mark}export LIMINAL_LIFECYCLE_READY=\"$1\"\nexec '{}' --exact \
             open_construction::site_process --nocapture\n",
            exe.to_str().ok_or("executable path is not Unicode")?
        ),
    )?;
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
    Ok(path)
}

/// The site's program: serves the site's subject with the site's contract on
/// the credential the server gave it, answering `{"echoed": <input>}` (with
/// `"rebuilt": true` when it is the rebuilt program), until it is stopped.
/// In an ordinary run it has no token and returns at once.
#[test]
fn site_process() -> Result<(), Box<dyn Error>> {
    let (Ok(server), Ok(subjects), Ok(token)) = (
        std::env::var("LIMINAL_SERVER"),
        std::env::var("LIMINAL_SUBJECTS"),
        std::env::var("LIMINAL_TOKEN"),
    ) else {
        return Ok(());
    };
    let subject = subjects.split(',').next().unwrap_or_default().to_owned();
    let registration = WorkerRegistration {
        namespaces: vec![subject.clone()],
        task_queue: subject.clone(),
        node: None,
        activity_types: vec![subject.clone()],
        identity: std::env::var("LIMINAL_SERVICE").unwrap_or_default(),
        activities: vec![WorkerActivityDescriptor {
            name: subject,
            input_schema_json: OBJECT.into(),
            output_schema_json: OBJECT.into(),
        }],
    };
    let worker =
        PushClient::connect_with_registration_and_auth(&server, registration, token.as_bytes())?;
    let mut ready = UnixStream::connect(std::env::var("LIMINAL_LIFECYCLE_READY")?)?;
    ready.write_all(b"R")?;
    drop(ready);
    let receiver = worker.receiver_handle()?;
    let writer = worker.writer_handle();
    let rebuilt = std::env::var_os("LIMINAL_REBUILT").is_some();
    while let Ok(PushReceiveOutcome::Frame(push)) = receiver.recv() {
        let input: Value = serde_json::from_slice(push.payload()).unwrap_or(Value::Null);
        let answer = if rebuilt {
            json!({"echoed": input, "rebuilt": true})
        } else {
            json!({"echoed": input})
        };
        writer.reply(push.correlation_id(), serde_json::to_vec(&answer)?)?;
    }
    Ok(())
}

/// `tool` on `host` with `arguments`, its structured answer when it is not
/// an error.
fn acted(host: &mut Host, tool: &str, arguments: &Value) -> Result<Value, Box<dyn Error>> {
    let answer = host.tool(tool, arguments)?;
    if answer["isError"] != false {
        return Err(format!("{tool} failed: {answer}").into());
    }
    Ok(answer["structuredContent"].clone())
}

/// An act's answer: executed now, approved by and answerable to `open:tom`.
fn executed_by_tom(answer: &Value, operation: &str) {
    assert_eq!(
        answer,
        &json!({"state": "executed", "grant": "open", "approved_by": "open:tom",
            "responsible": "open:tom", "operation": operation}),
    );
}

#[test]
fn access015_an_open_holder_builds_a_service_from_create_site_to_a_call()
-> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let root = dir.path().join("root");
    std::fs::create_dir(&root)?;
    let command = program(&root, false)?;
    let ready_path = dir.path().join("ready.sock");
    let ready = UnixListener::bind(&ready_path)?;
    let server = OpenServer::start_on_with(dir.path(), "127.0.0.1", &config(&root)?)?;
    let address = server.address.to_string();
    let tom_seat = dir.path().join("tom.toml");
    seat_init(&address, Some("tom"), &tom_seat)?;
    let builder_seat = dir.path().join("builder.toml");
    seat_init(&address, Some("builder"), &builder_seat)?;
    let mut tom = Host::open(Ticket::spawn(&tom_seat)?, true)?;
    // The builder's first listen places its station, so it can be bound.
    let mut builder = Host::open(Ticket::spawn(&builder_seat)?, true)?;

    let created = acted(
        &mut tom,
        "create_site",
        &json!({"operation": "op-create", "name": "forge", "purpose": "Echo what it is given.",
            "subjects": ["forge.echo"],
            "contracts": {"forge.echo": {"input_schema_json": OBJECT, "output_schema_json": OBJECT}}}),
    )?;
    executed_by_tom(&created, "op-create");
    let bound = acted(
        &mut tom,
        "bind_builder",
        &json!({"operation": "op-bind", "service": "forge", "agent": "builder"}),
    )?;
    executed_by_tom(&bound, "op-bind");

    let plan = acted(
        &mut builder,
        "plan_set",
        &json!({"service": "forge", "steps": [{"title": "write the echo"}]}),
    )?;
    assert_eq!(plan["steps"][0]["title"], "write the echo", "{plan}");
    acted(
        &mut builder,
        "plan_step",
        &json!({"service": "forge", "step": 1, "status": "done"}),
    )?;
    // Another open seat is not the builder.
    let refused = tom.tool(
        "plan_step",
        &json!({"service": "forge", "step": 1, "status": "todo"}),
    )?;
    assert_eq!(refused["isError"], true, "{refused}");
    assert!(
        refused
            .to_string()
            .contains("pass holder tom is not the builder of forge"),
        "{refused}"
    );

    let ready_arg = ready_path.to_str().ok_or("ready path is not Unicode")?;
    let proposed = acted(
        &mut tom,
        "propose_template",
        &json!({"operation": "op-template", "service": "forge",
            "command": command.to_str().ok_or("command is not Unicode")?,
            "args": [ready_arg], "env": [], "stop_timeout_ms": 2000}),
    )?;
    executed_by_tom(&proposed, "op-template");
    let completed = acted(
        &mut tom,
        "complete",
        &json!({"operation": "op-complete", "service": "forge"}),
    )?;
    executed_by_tom(&completed, "op-complete");

    let listed = acted(&mut tom, "services", &json!({}))?;
    let site = &listed["services"][0];
    assert_eq!(site["name"], "forge", "{listed}");
    assert_eq!(site["state"], "asleep", "{listed}");
    assert_eq!(site["approved_by"], json!(["open:tom"]), "{listed}");
    assert_eq!(site["responsible"], "open:tom", "{listed}");
    assert_eq!(site["placed"], false, "{listed}");

    acted(&mut tom, "start", &json!({"id": "forge"}))?;
    registered(&ready)?;
    let reply = acted(
        &mut tom,
        "call",
        &json!({"subject": "forge.echo", "input": {"text": "hi"}}),
    )?;
    assert_eq!(reply, json!({"echoed": {"text": "hi"}}));

    let held = acted(&mut tom, "stop", &json!({"id": "forge"}))?;
    assert_eq!(held["service"]["state"], "held", "{held}");

    rebuild(&mut tom, &root, ready_arg, &ready)?;
    drop(ready);
    let (stderr, code) = builder.ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    let (stderr, code) = tom.ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    Ok(())
}

/// Waits for the site's program to say it registered.
fn registered(ready: &UnixListener) -> Result<(), Box<dyn Error>> {
    let (mut registered, _) = ready.accept()?;
    let mut signal = [0];
    registered.read_exact(&mut signal)?;
    assert_eq!(&signal, b"R");
    Ok(())
}

/// The program is rebuilt while the site is held: a complete site's program
/// changes through `templates_update`, and start then runs the new digest.
fn rebuild(
    tom: &mut Host,
    root: &Path,
    ready_arg: &str,
    ready: &UnixListener,
) -> Result<(), Box<dyn Error>> {
    let command = program(root, true)?;
    let invocation = json!({"service": "forge",
        "command": command.to_str().ok_or("command is not Unicode")?,
        "args": [ready_arg], "env": [], "stop_timeout_ms": 2000});
    let refused = tom.tool("propose_template", &invocation)?;
    assert_eq!(refused["isError"], true, "{refused}");
    assert!(
        refused.to_string().contains("templates_update"),
        "{refused}"
    );
    let mut update = invocation.clone();
    update["operation"] = json!("op-update");
    let updated = acted(tom, "templates_update", &update)?;
    executed_by_tom(&updated, "op-update");
    acted(tom, "start", &json!({"id": "forge"}))?;
    registered(ready)?;
    let reply = acted(
        tom,
        "call",
        &json!({"subject": "forge.echo", "input": {"text": "again"}}),
    )?;
    assert_eq!(reply, json!({"echoed": {"text": "again"}, "rebuilt": true}));
    acted(tom, "stop", &json!({"id": "forge"}))?;
    Ok(())
}

/// Waffles' brief of 25 Oct (Tom: "can we have a generic one and set id at
/// runtime?"): a builder given only `--server` and `--as` has no seat file,
/// declares its name, and its first listen places its station, so it is
/// bound and builds as that name.
#[test]
fn a_seat_named_on_the_command_line_is_placed_on_its_first_listen_and_bound()
-> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let root = dir.path().join("root");
    std::fs::create_dir(&root)?;
    let server = OpenServer::start_on_with(dir.path(), "127.0.0.1", &config(&root)?)?;
    let address = server.address.to_string();
    let tom_seat = dir.path().join("tom.toml");
    seat_init(&address, Some("tom"), &tom_seat)?;
    let mut tom = Host::open(Ticket::spawn(&tom_seat)?, true)?;
    let builder = Ticket::spawn_with(&[
        "--server".as_ref(),
        address.as_ref(),
        "--as".as_ref(),
        "builder".as_ref(),
    ])?;
    let mut builder = Host::open(builder, true)?;
    let created = acted(
        &mut tom,
        "create_site",
        &json!({"operation": "op-create", "name": "forge", "purpose": "Echo what it is given.",
            "subjects": ["forge.echo"],
            "contracts": {"forge.echo": {"input_schema_json": OBJECT, "output_schema_json": OBJECT}}}),
    )?;
    executed_by_tom(&created, "op-create");
    let bound = acted(
        &mut tom,
        "bind_builder",
        &json!({"operation": "op-bind", "service": "forge", "agent": "builder"}),
    )?;
    executed_by_tom(&bound, "op-bind");
    // The builder acts as the name it was given on the command line.
    acted(
        &mut builder,
        "plan_set",
        &json!({"service": "forge", "steps": [{"title": "write the echo"}]}),
    )?;
    let (stderr, code) = builder.ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    let (stderr, code) = tom.ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    Ok(())
}

/// The smallest ACP agent the host can open a session with: it answers
/// `initialize` (version 1, no resume) and `session/new`, and reads the rest.
fn acp_harness(root: &Path) -> Result<std::path::PathBuf, Box<dyn Error>> {
    let path = root.join("acp");
    std::fs::write(
        &path,
        "#!/bin/sh\nenv | grep '^LIMINAL_E2E_' > \"$0.env\"\nwhile IFS= read -r line; do\n\
         id=$(printf '%s' \"$line\" | grep -o '\"id\":[0-9]*' | head -n 1 | cut -d: -f2)\n\
         case \"$line\" in\n\
         *'\"method\":\"initialize\"'*) printf '{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":\
         {\"protocolVersion\":1,\"agentCapabilities\":{\"loadSession\":false},\"authMethods\":[]}}\\n' \"$id\" ;;\n\
         *'\"method\":\"session/new\"'*) printf '{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":\
         {\"sessionId\":\"open-session\"}}\\n' \"$id\" ;;\n\
         esac\ndone\n",
    )?;
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
    Ok(path)
}

/// Tom, 25 Oct: an assistant is hosted on an open exchange with no Lys. A
/// seat with `[open] name = "assistant"` and a `[harness]` runs under
/// `liminal agent run`, holds no pass, and is present on its own station.
/// AGT-007 R1: it runs with stdin /dev/null, as a supervised process does,
/// and SIGTERM ends it with its station gone.
#[test]
fn an_open_seat_hosts_its_harness_under_agent_run() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let root = dir.path().join("root");
    std::fs::create_dir(&root)?;
    let server = OpenServer::start_on_with(dir.path(), "127.0.0.1", &config(&root)?)?;
    let address = server.address.to_string();
    let seat = dir.path().join("assistant.toml");
    std::fs::write(
        &seat,
        format!(
            "server = \"{address}\"\nchannel = \"console.topology\"\n\n[open]\nname = \"assistant\"\n\n\
             [limits]\nframe_bytes = 65536\nrecord_bytes = 1048576\nreply_timeout_ms = 10000\n\
             flow_steps = 8\nrequest_bytes = 65536\n\n[harness]\nid = \"claude-code\"\n\
             program = \"{}\"\nargs = []\ncwd = \"{}\"\nenv_remove = [\"LIMINAL_E2E_GONE\"]\n\n\
             [harness.env]\nLIMINAL_E2E_SET = \"yes\"\n",
            acp_harness(&root)?.display(),
            root.display()
        ),
    )?;
    let mut host = std::process::Command::new(env!("CARGO_BIN_EXE_liminal"))
        .args(["agent", "run", "--config"])
        .arg(&seat)
        .env_remove("LYS_AGENT")
        .env_remove("LYS_AGENT_PASS")
        .env("LIMINAL_E2E_GONE", "present")
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::piped())
        .spawn()?;
    let watched = (|| -> Result<(Value, Value), Box<dyn Error>> {
        // The station is placed by the host's first listen; its presence is
        // retained, so a watch opened after it still reads it.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
        let presence = loop {
            match liminal_sdk::SubscriptionStream::open_with_auth(
                &address,
                "agents.assistant.presence",
                Vec::new(),
                b"open:assistant",
            ) {
                Ok(stream) => break stream,
                Err(error) if std::time::Instant::now() >= deadline => return Err(error.into()),
                Err(_) => std::thread::sleep(std::time::Duration::from_millis(50)),
            }
        };
        let next = |present: bool| -> Result<Value, Box<dyn Error>> {
            loop {
                let record = presence.recv_timeout(std::time::Duration::from_secs(20))?;
                let value: Value = serde_json::from_slice(record.payload())?;
                if value["present"] == present {
                    return Ok(value);
                }
            }
        };
        let present = next(true)?;
        // SIGTERM ends the host, which closes the harness and says gone.
        let pid = rustix::process::Pid::from_raw(i32::try_from(host.id())?)
            .ok_or("the host has no pid")?;
        rustix::process::kill_process(pid, rustix::process::Signal::TERM)?;
        Ok((present, next(false)?))
    })();
    if watched.is_err() {
        host.kill()?;
    }
    let output = host.wait_with_output()?;
    let stderr = String::from_utf8_lossy(&output.stderr);
    let (present, gone) = watched.map_err(|error| {
        format!(
            "{error}; host: {stderr}; server: {}",
            server.log().unwrap_or_default()
        )
    })?;
    assert_eq!(present["present"], true, "{present}");
    assert_eq!(gone["present"], false, "{gone}");
    assert!(!stderr.contains("Lys"), "{stderr}");
    assert!(!stderr.contains("os error"), "{stderr}");
    // AGT-007 R4: the seat's [harness] env is set and env_remove removed.
    // Only these two names are read: the harness inherits the host's own
    // environment, which is never printed.
    let seen = std::fs::read_to_string(root.join("acp.env"))?;
    let named = |name: &str| {
        seen.lines()
            .find(|line| line.starts_with(&format!("{name}=")))
            .map(str::to_owned)
    };
    assert_eq!(
        named("LIMINAL_E2E_SET").as_deref(),
        Some("LIMINAL_E2E_SET=yes")
    );
    assert!(
        named("LIMINAL_E2E_GONE").is_none(),
        "LIMINAL_E2E_GONE reached the harness"
    );
    Ok(())
}