liminal-cli 0.3.27

//! 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(())
}