//! 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::path::PathBuf;
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 {
    spawn_agent_with_pull(allow, grant, None).await
}

/// A build legitimately runs for minutes, so `/run` must have no whole-request
/// timeout — only `/health` does. This is the trap in the timeout work
/// (2026-07-16): `reqwest`'s request timeout runs until the response *body* has
/// finished, so a client-level timeout would sever a real build mid-flight and
/// look exactly like a build failure. Sleeps past `HEALTH_TIMEOUT` (10s) to
/// prove the cap is not applied here. Slow on purpose — it is the only thing
/// standing between a future "just set a client timeout" and a severed release.
#[tokio::test]
async fn a_step_outliving_the_health_timeout_is_not_severed() {
    let base = spawn_agent(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec!["build".into()],
            observe: vec![],
        }],
        builder_grant(),
    )
    .await;
    let rpc = AgentRpc::new(
        base,
        "mbp",
        CapabilitySet::from_tokens(["build"], Vec::<&str>::new()),
    );

    let mut sink = VecSink::default();
    let step = Step::shell(Action::Build, "sleep 12; printf 'built'");
    let out = rpc
        .run_streaming(&step, &mut sink)
        .await
        .expect("a 12s build must not be cut off");
    assert!(out.success());
    assert_eq!(out.stdout, b"built");
}

/// As [`spawn_agent`], but with a configurable `pull_root` for the file-read path.
async fn spawn_agent_with_pull(
    allow: Vec<CallerGrant>,
    grant: GrantConfig,
    pull_root: Option<PathBuf>,
) -> String {
    let config = AgentConfig {
        listen: "127.0.0.1:0".parse().unwrap(),
        grant,
        allow,
        pull_root,
        pin: Vec::new(),
    };
    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 preflight_accepts_a_matching_protocol_version() {
    let base = spawn_agent(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec!["build".into()],
            observe: vec![],
        }],
        builder_grant(),
    )
    .await;
    let rpc = AgentRpc::new(
        base,
        "mbp",
        CapabilitySet::from_tokens(["build"], Vec::<&str>::new()),
    );
    // The agent advertises PROTOCOL_VERSION; preflight must accept it.
    let health = rpc.health().await.unwrap();
    assert_eq!(health.version, ops_exec::wire::PROTOCOL_VERSION);
    rpc.preflight()
        .await
        .expect("preflight accepts a same-version agent");
}

#[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"
    );
}

/// The caller-side grant a driver needs to pull an artifact.
fn puller() -> CapabilitySet {
    CapabilitySet::from_tokens(Vec::<&str>::new(), ["artifact"])
}

/// An allow-listed caller with an `artifact` observe grant can pull a file that
/// lives under the configured `pull_root`.
#[tokio::test]
async fn agent_pull_serves_an_in_root_file() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("dist");
    tokio::fs::create_dir_all(&root).await.unwrap();
    let artifact = root.join("GoingsOn.dmg");
    tokio::fs::write(&artifact, b"DMGBYTES").await.unwrap();

    let grant = GrantConfig {
        actuate: vec!["build".into()],
        observe: vec!["artifact".into()],
    };
    let base = spawn_agent_with_pull(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec![],
            observe: vec!["artifact".into()],
        }],
        grant,
        Some(root.clone()),
    )
    .await;
    let rpc = AgentRpc::new(base, "mbp", puller());
    let local = dir.path().join("pulled.dmg");
    rpc.pull_file(&artifact, &local, &Default::default())
        .await
        .unwrap();
    assert_eq!(tokio::fs::read(&local).await.unwrap(), b"DMGBYTES");
}

/// `artifact` and `build-log` are independent grants: a caller allowed to read
/// build logs is NOT thereby allowed to retrieve artifacts. This is the split
/// the old `build-log`-gated /pull could not express.
#[tokio::test]
async fn agent_pull_denied_with_only_build_log_observe() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("dist");
    tokio::fs::create_dir_all(&root).await.unwrap();
    let artifact = root.join("a.bin");
    tokio::fs::write(&artifact, b"x").await.unwrap();

    let grant = GrantConfig {
        actuate: vec![],
        observe: vec!["build-log".into(), "artifact".into()],
    };
    let base = spawn_agent_with_pull(
        // The caller may read logs, but was never granted artifacts.
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec![],
            observe: vec!["build-log".into()],
        }],
        grant,
        Some(root.clone()),
    )
    .await;
    // Client-side grant is deliberately wide so the *agent* is what refuses.
    let rpc = AgentRpc::new(base, "mbp", puller());
    let local = dir.path().join("out.bin");
    let err = rpc
        .pull_file(&artifact, &local, &Default::default())
        .await
        .unwrap_err();
    assert!(
        err.to_string().contains("artifact"),
        "denial names the grant: {err}"
    );
    assert!(!local.exists(), "denied pull must not write a file");
}

/// A caller without any observe grant is refused, even for an in-root file —
/// pull is gated, not open.
#[tokio::test]
async fn agent_pull_denied_without_observe_grant() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("dist");
    tokio::fs::create_dir_all(&root).await.unwrap();
    let artifact = root.join("a.bin");
    tokio::fs::write(&artifact, b"x").await.unwrap();

    let base = spawn_agent_with_pull(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec!["build".into()],
            observe: vec![],
        }],
        builder_grant(),
        Some(root.clone()),
    )
    .await;
    let rpc = AgentRpc::new(base, "mbp", puller());
    let local = dir.path().join("out.bin");
    // The agent returns 403; AgentRpc surfaces the reason and the file never
    // transfers.
    assert!(
        rpc.pull_file(&artifact, &local, &Default::default())
            .await
            .is_err()
    );
    assert!(!local.exists(), "denied pull must not write a file");
}

/// The caller-side half of double enforcement: a driver that never declared
/// `artifact` fails before any request leaves the process. This is the check
/// that saves a full build + notary round trip on a misconfigured driver.
#[tokio::test]
async fn agent_pull_denied_caller_side_without_declaring_artifact() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("dist");
    tokio::fs::create_dir_all(&root).await.unwrap();
    let artifact = root.join("a.bin");
    tokio::fs::write(&artifact, b"x").await.unwrap();

    let grant = GrantConfig {
        actuate: vec![],
        observe: vec!["artifact".into()],
    };
    let base = spawn_agent_with_pull(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec![],
            observe: vec!["artifact".into()],
        }],
        grant,
        Some(root.clone()),
    )
    .await;
    // Agent would happily serve this; the caller's own set is what refuses.
    let rpc = AgentRpc::new(
        base,
        "mbp",
        CapabilitySet::from_tokens(["build"], ["build-log"]),
    );
    let local = dir.path().join("out.bin");
    let err = rpc
        .pull_file(&artifact, &local, &Default::default())
        .await
        .unwrap_err();
    assert!(
        err.to_string().contains("observe:artifact"),
        "caller-side denial: {err}"
    );
    assert!(!local.exists(), "denied pull must not write a file");
}

/// A path outside `pull_root` is refused even for an authorized caller — the
/// confinement boundary holds against traversal.
#[tokio::test]
async fn agent_pull_denied_outside_root() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("dist");
    tokio::fs::create_dir_all(&root).await.unwrap();
    let secret = dir.path().join("secret.key");
    tokio::fs::write(&secret, b"topsecret").await.unwrap();

    let grant = GrantConfig {
        actuate: vec![],
        observe: vec!["artifact".into()],
    };
    let base = spawn_agent_with_pull(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec![],
            observe: vec!["artifact".into()],
        }],
        grant,
        Some(root.clone()),
    )
    .await;
    let rpc = AgentRpc::new(base, "mbp", puller());
    let local = dir.path().join("out.key");
    let err = rpc
        .pull_file(&secret, &local, &Default::default())
        .await
        .unwrap_err();
    assert!(!local.exists(), "out-of-root pull must not write a file");
    let _ = err; // status is non-2xx; the point is the secret never transfers
}

/// The agent serves exactly one file per `/pull`; there is no directory form.
/// `pull_dir` must refuse locally rather than invent one — this is the half of
/// the trait split that keeps a transport swap honest.
#[tokio::test]
async fn agent_pull_dir_is_refused_by_design() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("dist");
    tokio::fs::create_dir_all(&root).await.unwrap();

    let grant = GrantConfig {
        actuate: vec![],
        observe: vec!["artifact".into()],
    };
    let base = spawn_agent_with_pull(
        vec![CallerGrant {
            identity: "fw13".into(),
            actuate: vec![],
            observe: vec!["artifact".into()],
        }],
        grant,
        Some(root.clone()),
    )
    .await;
    let rpc = AgentRpc::new(base, "mbp", puller());
    let err = rpc
        .pull_dir(&root, &dir.path().join("out"), &Default::default())
        .await
        .expect_err("pull_dir must be refused, not attempted");
    assert!(err.to_string().contains("unsupported by design"), "{err}");
}