shepherd-cli 6.6.1

//! Deterministic real-process provider fixture, never live LLM evidence.
//! Every child reads the compiled carrier and uses Native prepare/claim/activate.
//! Nested launches originate in the actual parent process, not a root proxy.
#![allow(dead_code)] // Shared test binaries exercise different protocol branches.

use std::{
    collections::BTreeMap,
    fs,
    os::unix::fs::MetadataExt,
    path::{Path, PathBuf},
    process::{Child, Command, Stdio},
    thread,
    time::{Duration, Instant},
};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use shepherd_cli::{
    BrokerClient, BrokerLaunchId, LaunchHandle, PreparePendingDispatchRequest,
    shepherd::{
        Harness,
        compiler::{HarnessProfile, compile},
        dispatch::{
            AgentId, AttachmentKind, DispatchId, DispatchRecord, LoadedCarrierAttestationV1, Role,
        },
    },
};

type Result<T> = std::result::Result<T, String>;

#[derive(Clone, Deserialize, Serialize)]
struct ProviderInput {
    endpoint: PathBuf,
    installed: PathBuf,
    scratch: PathBuf,
    launch: String,
    nonce: [u8; 32],
    request: PreparePendingDispatchRequest,
}

#[derive(Deserialize, Serialize)]
enum Operation {
    Spawn(Box<PreparePendingDispatchRequest>),
    CompleteChild(String),
    Complete,
    Exit,
}

fn error(value: impl std::fmt::Display) -> String {
    value.to_string()
}

fn atomic_json(path: &Path, value: &impl Serialize) -> Result<()> {
    let temporary = path.with_extension("writing");
    fs::write(&temporary, serde_json::to_vec(value).map_err(error)?).map_err(error)?;
    fs::rename(temporary, path).map_err(error)
}

fn read_json<T: serde::de::DeserializeOwned>(path: &Path) -> Result<T> {
    serde_json::from_slice(&fs::read(path).map_err(error)?).map_err(error)
}

fn hex_digest(value: &str) -> [u8; 32] {
    assert_eq!(value.len(), 64);
    let mut output = [0; 32];
    for (index, pair) in value.as_bytes().as_chunks::<2>().0.iter().enumerate() {
        output[index] = u8::from_str_radix(std::str::from_utf8(pair).unwrap(), 16).unwrap();
    }
    output
}

fn loaded(input: &ProviderInput) -> LoadedCarrierAttestationV1 {
    let authored = shepherd_cli::content_compiler::embedded_compile_input().unwrap();
    let tree = compile(&authored, &HarnessProfile::pi()).unwrap();
    let manifest: serde_json::Value =
        read_json(&input.installed.join(".shepherd-generated.json")).unwrap();
    assert_eq!(manifest["schema"], "shepherd.compiled-tree/4");
    assert_eq!(manifest["target"], "pi");
    assert_eq!(manifest["tree_digest"], tree.digest);
    assert_eq!(
        manifest["files"].as_array().unwrap().len(),
        tree.files.len()
    );
    for file in &tree.files {
        let path = input.installed.join(&file.path);
        assert_eq!(fs::read(&path).unwrap(), file.content.as_bytes());
        assert_eq!(fs::metadata(path).unwrap().mode() & 0o777, file.mode);
    }
    let role = Role::from_carrier(&input.request.role)
        .or_else(|_| Role::from_name(&input.request.role))
        .unwrap();
    let entry = manifest["roles"]
        .as_array()
        .unwrap()
        .iter()
        .find(|entry| entry["role"] == role.as_str())
        .unwrap();
    let compiled = tree
        .roles
        .iter()
        .find(|entry| entry.role == role.as_str())
        .unwrap();
    assert_eq!(
        entry["startup_skill_sha256"],
        compiled.startup_skill_sha256.as_deref().unwrap()
    );
    let carrier = input
        .installed
        .join(entry["carrier_path"].as_str().unwrap());
    let bytes = fs::read(&carrier).unwrap();
    let metadata = fs::metadata(&carrier).unwrap();
    let mut content = Sha256::new();
    content.update(b"regular-file\0");
    content.update(0o100644_u32.to_be_bytes());
    content.update(i64::try_from(bytes.len()).unwrap().to_be_bytes());
    content.update(bytes);
    let mut identity = Sha256::new();
    identity.update(b"unix-identity\0");
    identity.update(metadata.dev().to_be_bytes());
    identity.update(metadata.ino().to_be_bytes());
    identity.update(metadata.nlink().to_be_bytes());
    identity.update(metadata.mode().to_be_bytes());
    identity.update(metadata.len().to_be_bytes());
    let mut carrier_hash = Sha256::new();
    carrier_hash.update(b"carrier-identity/1\0");
    carrier_hash.update(content.finalize());
    carrier_hash.update(identity.finalize());
    LoadedCarrierAttestationV1 {
        schema: "shepherd.loaded-carrier/1".into(),
        nonce_sha256: input.nonce,
        target: Harness::Pi,
        role,
        agent_id: AgentId::new(&input.request.expected_attachment.agent_id).unwrap(),
        installed_carrier_path: carrier.display().to_string(),
        candidate_sha256: Sha256::digest(fs::read(std::env::current_exe().unwrap()).unwrap())
            .into(),
        carrier_sha256: carrier_hash.finalize().into(),
        compiler_tree_sha256: hex_digest(manifest["tree_digest"].as_str().unwrap()),
        startup_skill: entry["startup_skill"].as_str().unwrap().into(),
        skill_bundle_sha256: hex_digest(entry["startup_skill_sha256"].as_str().unwrap()),
        attachment_kind: AttachmentKind::PiSkillPath,
    }
}

pub(crate) struct LiveProvider {
    process: Child,
    scratch: PathBuf,
    sequence: usize,
    record: Option<DispatchRecord>,
}

impl LiveProvider {
    pub(crate) fn launch(
        parent: &mut BrokerClient,
        endpoint: &Path,
        installed: &Path,
        scratch: &Path,
        request: PreparePendingDispatchRequest,
    ) -> Result<Self> {
        let handle = parent.prepare(request.clone()).map_err(error)?;
        Self::launch_prepared(parent, endpoint, installed, scratch, request, &handle)
    }

    /// Allows Native review-replace between preparation and actual activation.
    pub(crate) fn launch_prepared(
        parent: &mut BrokerClient,
        endpoint: &Path,
        installed: &Path,
        scratch: &Path,
        request: PreparePendingDispatchRequest,
        handle: &LaunchHandle,
    ) -> Result<Self> {
        fs::create_dir_all(scratch).map_err(error)?;
        let directory = scratch.join(&request.expected_attachment.agent_id);
        fs::create_dir(&directory).map_err(error)?;
        let input = ProviderInput {
            endpoint: endpoint.into(),
            installed: installed.into(),
            scratch: directory.clone(),
            launch: handle.launch_id().opaque_string(),
            nonce: handle.nonce_sha256(),
            request,
        };
        let input_path = directory.join("input.json");
        atomic_json(&input_path, &input)?;
        let output = fs::File::create(directory.join("process.log")).map_err(error)?;
        let process = Command::new(std::env::current_exe().map_err(error)?)
            .args(["--exact", "broker_fixture_child", "--nocapture"])
            .env("SHEPHERD_TEST_BROKER_PROVIDER", &input_path)
            .stdin(Stdio::null())
            .stdout(output.try_clone().map_err(error)?)
            .stderr(output)
            .spawn()
            .map_err(error)?;
        let mut provider = Self {
            process,
            scratch: directory,
            sequence: 0,
            record: None,
        };
        provider.wait_file("ready")?;
        parent
            .register_child(handle, provider.process.id())
            .map_err(error)?;
        fs::write(provider.scratch.join("registered"), b"registered").map_err(error)?;
        provider.wait_file("activated.json")?;
        provider.record = Some(read_json::<Result<DispatchRecord>>(
            &provider.scratch.join("activated.json"),
        )??);
        Ok(provider)
    }

    fn wait_file(&mut self, name: &str) -> Result<()> {
        let deadline = Instant::now() + Duration::from_secs(30);
        loop {
            if self.scratch.join(name).is_file() {
                return Ok(());
            }
            if let Some(status) = self.process.try_wait().map_err(error)? {
                return Err(format!(
                    "provider exited {status} before {name}: {}",
                    fs::read_to_string(self.scratch.join("process.log")).unwrap_or_default()
                ));
            }
            if Instant::now() >= deadline {
                return Err(format!(
                    "provider timed out at {}",
                    self.scratch.join(name).display()
                ));
            }
            thread::sleep(Duration::from_millis(5));
        }
    }

    fn operation(&mut self, operation: Operation) -> Result<DispatchRecord> {
        let sequence = self.sequence;
        self.sequence += 1;
        atomic_json(
            &self.scratch.join(format!("command-{sequence}.json")),
            &operation,
        )?;
        let name = format!("response-{sequence}.json");
        self.wait_file(&name)?;
        read_json::<Result<DispatchRecord>>(&self.scratch.join(name))?
    }

    pub(crate) fn record(&self) -> &DispatchRecord {
        self.record.as_ref().expect("activated provider")
    }

    pub(crate) fn spawn(
        &mut self,
        request: PreparePendingDispatchRequest,
    ) -> Result<DispatchRecord> {
        self.operation(Operation::Spawn(Box::new(request)))
    }

    pub(crate) fn complete_child(&mut self, agent: &str) -> Result<DispatchRecord> {
        self.operation(Operation::CompleteChild(agent.into()))
    }

    pub(crate) fn complete(&mut self) -> Result<DispatchRecord> {
        let record = self.operation(Operation::Complete)?;
        let status = self.process.wait().map_err(error)?;
        if !status.success() {
            return Err(format!("provider completion process failed: {status}"));
        }
        self.record = Some(record.clone());
        Ok(record)
    }
}

impl Drop for LiveProvider {
    fn drop(&mut self) {
        if self.process.try_wait().ok().flatten().is_some() {
            return;
        }
        let _ = atomic_json(
            &self.scratch.join(format!("command-{}.json", self.sequence)),
            &Operation::Exit,
        );
        let _ = fs::write(self.scratch.join("exit"), b"exit");
        let deadline = Instant::now() + Duration::from_secs(1);
        while Instant::now() < deadline {
            if self.process.try_wait().ok().flatten().is_some() {
                return;
            }
            thread::sleep(Duration::from_millis(5));
        }
        let _ = self.process.kill();
        let _ = self.process.wait();
    }
}

/// Call from an exactly named `#[test] fn broker_fixture_child()` in each binary.
pub(crate) fn child_main() {
    child_main_with(|_| {});
}

/// Optional host-message fixture behavior, executed by the activated child PID.
pub(crate) fn child_main_with(on_active: impl FnOnce(&DispatchRecord)) {
    let Some(path) = std::env::var_os("SHEPHERD_TEST_BROKER_PROVIDER") else {
        return;
    };
    let input: ProviderInput = read_json(Path::new(&path)).unwrap();
    fs::write(input.scratch.join("ready"), b"ready").unwrap();
    let deadline = Instant::now() + Duration::from_secs(30);
    while !input.scratch.join("registered").is_file() {
        if input.scratch.join("exit").is_file() {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "parent did not register provider"
        );
        thread::sleep(Duration::from_millis(5));
    }
    let launch_id = BrokerLaunchId::from_opaque(&input.launch).unwrap();
    let mut client = BrokerClient::connect_child_by_id(&input.endpoint, launch_id).unwrap();
    let agent = input.request.expected_attachment.agent_id.clone();
    let session = input.request.child_session_id.clone();
    let attestation = loaded(&input);
    let agent_type = format!("pi-subagents:{}", attestation.role.as_str());
    let activated = client
        .claim(
            agent.clone(),
            session.clone(),
            agent_type.clone(),
            attestation.clone(),
        )
        .and_then(|_| {
            client.activate(
                agent.clone(),
                session.clone(),
                agent_type.clone(),
                attestation,
            )
        })
        .map_err(error);
    atomic_json(&input.scratch.join("activated.json"), &activated).unwrap();
    let Ok(record) = activated else {
        return;
    };
    on_active(&record);
    let mut parent = BrokerClient::connect_endpoint(&input.endpoint).unwrap();
    parent
        .register_parent(
            Harness::Pi,
            record.role,
            record.session_id.clone(),
            record.root_session_id.clone(),
            Some(DispatchId::new(record.agent_id.as_str()).unwrap()),
        )
        .unwrap();
    let mut children = BTreeMap::<String, LiveProvider>::new();
    for sequence in 0.. {
        let command = input.scratch.join(format!("command-{sequence}.json"));
        let deadline = Instant::now() + Duration::from_secs(120);
        while !command.is_file() {
            if input.scratch.join("exit").is_file() {
                return;
            }
            assert!(Instant::now() < deadline, "provider fixture was abandoned");
            thread::sleep(Duration::from_millis(5));
        }
        let operation: Operation = read_json(&command).unwrap();
        let mut done = false;
        let response = match operation {
            Operation::Spawn(request) => LiveProvider::launch(
                &mut parent,
                &input.endpoint,
                &input.installed,
                &input.scratch.join("children"),
                *request,
            )
            .map(|child| {
                let record = child.record().clone();
                children.insert(record.agent_id.to_string(), child);
                record
            }),
            Operation::CompleteChild(id) => children
                .get_mut(&id)
                .ok_or_else(|| format!("no live child {id}"))
                .and_then(LiveProvider::complete),
            Operation::Complete => {
                let result = BrokerClient::connect_child_event_by_id(&input.endpoint, launch_id)
                    .and_then(|mut terminal| {
                        terminal.complete(agent.clone(), session.clone(), agent_type.clone())
                    })
                    .map_err(error);
                done = result.is_ok();
                result
            }
            Operation::Exit => return,
        };
        atomic_json(
            &input.scratch.join(format!("response-{sequence}.json")),
            &response,
        )
        .unwrap();
        if done {
            return;
        }
    }
}