//! End-to-end: a real `ops-agent` HTTP server (with a stubbed `whois`) driven
//! by the real `AgentRpc` client over a loopback socket. Proves the E2 path —
//! identity → authorization → in-session exec → streamed frames — without a
//! live tailnet.
#![cfg(feature = "agent")]

use ops_exec::agent::{AgentConfig, AgentState, CallerGrant, CallerIdentity, GrantConfig, router};
use ops_exec::{Action, AgentRpc, CapabilityDenied, CapabilitySet, Executor, LogSink, Step};
use std::net::SocketAddr;
use std::sync::Arc;

#[derive(Default)]
struct VecSink(Vec<u8>);
#[async_trait::async_trait]
impl LogSink for VecSink {
    async fn write_chunk(&mut self, bytes: &[u8]) {
        self.0.extend_from_slice(bytes);
    }
}

/// Spin the agent on an ephemeral loopback port; whois always says the caller
/// is `fw13`. Returns the base URL.
async fn spawn_agent(allow: Vec<CallerGrant>, grant: GrantConfig) -> String {
    let config = AgentConfig { listen: "127.0.0.1:0".parse().unwrap(), grant, allow };
    let state = AgentState {
        config: Arc::new(config),
        whois: Arc::new(|_ip| {
            Box::pin(async { Ok(CallerIdentity { node: "fw13".into(), tags: vec![] }) })
        }),
    };
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, router(state).into_make_service_with_connect_info::<SocketAddr>())
            .await
            .unwrap();
    });
    format!("http://{addr}")
}

fn builder_grant() -> GrantConfig {
    GrantConfig {
        actuate: vec!["build".into(), "sign".into(), "notarize".into(), "staple".into()],
        observe: vec![],
    }
}

#[tokio::test]
async fn agent_runs_a_granted_step_and_streams_output() {
    let base = spawn_agent(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec!["build".into(), "sign".into()],
            observe: vec![],
        }],
        builder_grant(),
    )
    .await;

    // The driver's caller-side caps also include sign.
    let rpc = AgentRpc::new(base, "mbp", CapabilitySet::from_tokens(["build", "sign"], Vec::<&str>::new()));
    let health = rpc.health().await.unwrap();
    assert!(health.ok);
    assert!(health.actuate.contains(&"sign".to_string()));

    let mut sink = VecSink::default();
    let step = Step::shell(Action::Sign, "printf 'signed-ok'");
    let out = rpc.run_streaming(&step, &mut sink).await.unwrap();
    assert!(out.success());
    assert_eq!(sink.0, b"signed-ok");
    assert_eq!(out.stdout, b"signed-ok");
}

#[tokio::test]
async fn agent_denies_action_outside_its_grant() {
    // The agent host grants build/sign only; the caller asks to deploy. Even
    // though the client-side caps below include deploy, the agent must refuse.
    let base = spawn_agent(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec!["build".into(), "sign".into(), "deploy".into()],
            observe: vec![],
        }],
        builder_grant(),
    )
    .await;

    let rpc = AgentRpc::new(base, "mbp", CapabilitySet::from_tokens(["deploy"], Vec::<&str>::new()));
    let mut sink = VecSink::default();
    let step = Step::shell(Action::Deploy, "echo should-not-run");
    let err = rpc.run_streaming(&step, &mut sink).await.unwrap_err();
    let msg = format!("{err:#}");
    assert!(msg.contains("denied"), "expected agent denial, got: {msg}");
}

#[tokio::test]
async fn caller_side_gate_rejects_before_round_trip() {
    // The client's own caps omit `sign`, so AgentRpc must refuse before any
    // HTTP call (CapabilityDenied), independent of the agent.
    let rpc = AgentRpc::new(
        "http://127.0.0.1:1", // unreachable; must never be dialed
        "mbp",
        CapabilitySet::from_tokens(["build"], Vec::<&str>::new()),
    );
    let mut sink = VecSink::default();
    let err = rpc
        .run_streaming(&Step::shell(Action::Sign, "true"), &mut sink)
        .await
        .unwrap_err();
    assert!(err.downcast_ref::<CapabilityDenied>().is_some(), "expected caller-side CapabilityDenied");
}

#[tokio::test]
async fn agent_pull_serves_a_file() {
    let dir = tempfile::tempdir().unwrap();
    let artifact = dir.path().join("GoingsOn.dmg");
    tokio::fs::write(&artifact, b"DMGBYTES").await.unwrap();

    let base = spawn_agent(
        vec![CallerGrant { identity: "fw13".into(), actuate: vec![], observe: vec![] }],
        builder_grant(),
    )
    .await;
    let rpc = AgentRpc::new(base, "mbp", CapabilitySet::default());
    let local = dir.path().join("pulled.dmg");
    rpc.pull(&artifact, &local, &Default::default()).await.unwrap();
    assert_eq!(tokio::fs::read(&local).await.unwrap(), b"DMGBYTES");
}