//! The `BuildMeta` node: build one agent from `internal/meta` on the swarm
//! builder before its deploy is sent, and record the closure in
//! [`swarm_wire::agent_closures`].
//!
//! The node reads `internal/meta`'s head when it runs, because in most DAGs it
//! is declared before the `UpdateMeta` that commits that head. It sends the
//! builder the agent and the head's commit, and on a `Built` reply writes the
//! key itself: the builder holds no grant on the bucket. A `Built` reply and
//! nothing else writes, so a red build leaves the previous record.
//!
//! The node ends one of two ways. `Done` is a green build. Every `Failed` skips
//! the deploy behind it, and the Jobs page shows why:
//! - `nix` failed, its timeout included: `Failed` with `nix`'s tail, or `nix
//!   timed out`. The deploy stays held and counts as a failed build.
//! - The build got no verdict (no builder, a forge error, no `internal/meta`
//!   head, a missing read grant, a failed closure write): `Failed` starting with
//!   [`NO_VERDICT_PREFIX`], which the held-deploys gauge reads to label the
//!   reason.
//!
//! Both are built again on every round of the meta timer
//! (`jobq::queue_meta_update`) until a build succeeds.
//!
//! A `nix` failure of a merged config PR's build (`pr` set) also leaves one
//! comment on that PR, deduped by a marker in the comment per agent: the retry
//! rounds stay quiet whatever rev they build, until a green build of the agent
//! clears the marker. A no-verdict failure says nothing about the config and
//! leaves none.

use std::fmt::Display;
use std::future::Future;

use anyhow::{Context as _, Result};
use hive_jobq::scheduler::Outcome;
use swarm_wire::agent_closures::AgentClosure;
use swarm_wire::builder::{BuildFromMeta, FromMetaReply, META_SUBJECT, REPLY_DEADLINE};

use crate::agent_renewal::unix_now;
use crate::builder::runs::{BuildOutcome, BuildRecord, Builds};
use crate::forge::{self, objects::SHARED_ORG};
use crate::meta_flake::{META_REPO, MetaForge};

/// What a `Failed` outcome starts with when the build got no verdict.
pub(crate) const NO_VERDICT_PREFIX: &str = "no verdict: ";

/// Whether a `Failed` node's `error` says the build got no verdict.
pub(crate) fn is_no_verdict(error: &str) -> bool {
    error.starts_with(NO_VERDICT_PREFIX)
}

fn no_verdict(why: impl Display) -> Outcome {
    Outcome::Failed(format!("{NO_VERDICT_PREFIX}{why}"))
}

/// Where `internal/meta` is read and shared with the builder: the forge in
/// production, a fake in the tests.
pub(crate) trait MetaRepo {
    /// The head commit; `None` when the repo has none.
    fn head(&self) -> impl Future<Output = Result<Option<String>>> + Send;
    /// Make sure the builder's forge account can read the repo.
    fn grant_builder_read(&self) -> impl Future<Output = Result<()>> + Send;
    /// Comment `body` on `pr` of `agent`'s config repo unless a comment there
    /// already contains `marker`. Whether a comment was posted.
    fn comment_once(
        &self,
        agent: &str,
        pr: u64,
        marker: &str,
        body: &str,
    ) -> impl Future<Output = Result<bool>> + Send;
    /// Rewrite the comments on `pr` of `agent`'s config repo that contain
    /// `marker` so they contain `cleared` and end with `resolved`, which lets
    /// `marker` be posted again. Only the controller's own comments count.
    fn clear_marker(
        &self,
        agent: &str,
        pr: u64,
        marker: &str,
        cleared: &str,
        resolved: &str,
    ) -> impl Future<Output = Result<usize>> + Send;
}

impl MetaRepo for forge::Client {
    async fn head(&self) -> Result<Option<String>> {
        let read = self.head_files(SHARED_ORG, META_REPO, None, &[]);
        let head = crate::jobq::forge_node("build_meta", read).await?;
        Ok(head.map(|h| h.commit))
    }

    async fn grant_builder_read(&self) -> Result<()> {
        crate::jobq::forge_node(
            "build_meta",
            self.grant_meta_read(forge::builder_account::LOGIN),
        )
        .await
    }

    async fn comment_once(&self, agent: &str, pr: u64, marker: &str, body: &str) -> Result<bool> {
        crate::jobq::forge_node(
            "build_meta",
            forge::Client::comment_once(self, forge::CONFIG_ORG, agent, pr, marker, body),
        )
        .await
    }

    async fn clear_marker(
        &self,
        agent: &str,
        pr: u64,
        marker: &str,
        cleared: &str,
        resolved: &str,
    ) -> Result<usize> {
        crate::jobq::forge_node(
            "build_meta",
            forge::Client::clear_marker(
                self,
                forge::CONFIG_ORG,
                agent,
                pr,
                marker,
                cleared,
                resolved,
            ),
        )
        .await
    }
}

/// Where a build is sent and its closure recorded: the queue in production, a
/// fake in the tests.
pub(crate) trait MetaBuilds {
    /// `Err` is a request that never reached a builder or never came back.
    fn build(&self, request: Vec<u8>)
    -> impl Future<Output = Result<FromMetaReply, String>> + Send;
    fn record(
        &self,
        agent: &str,
        closure: &AgentClosure,
    ) -> impl Future<Output = Result<()>> + Send;
}

impl MetaBuilds for async_nats::Client {
    async fn build(&self, request: Vec<u8>) -> Result<FromMetaReply, String> {
        use async_nats::RequestErrorKind;

        let sent = async_nats::Request::new()
            .payload(request.into())
            .timeout(Some(REPLY_DEADLINE));
        match self.send_request(META_SUBJECT, sent).await {
            Ok(msg) => serde_json::from_slice(&msg.payload)
                .map_err(|e| format!("the builder's meta build reply did not decode: {e}")),
            Err(e) => Err(match e.kind() {
                RequestErrorKind::NoResponders => "no swarm builder is listening for meta \
                     builds: deploy.builder.enable is set on no host, or its builder predates \
                     the request"
                    .to_owned(),
                RequestErrorKind::TimedOut => format!(
                    "the swarm builder sent no answer within {} minutes",
                    REPLY_DEADLINE.as_secs() / 60
                ),
                _ => format!("sending the meta build to the swarm builder failed: {e}"),
            }),
        }
    }

    async fn record(&self, agent: &str, closure: &AgentClosure) -> Result<()> {
        let store = swarm_queue_client::agent_closures::open_or_create(self)
            .await
            .context("opening the agent-closures bucket")?;
        let body = serde_json::to_vec(closure).context("encoding the closure record")?;
        store
            .put(agent, body.into())
            .await
            .map(|_| ())
            .with_context(|| format!("writing agent-closures/{agent}"))
    }
}

/// The most of `nix`'s tail a PR comment carries, in bytes: the builder's own cap
/// on the tail it sends.
const COMMENT_TAIL_BYTES: usize = 4000;

/// The marker a PR's failure comment carries for `agent`: one per PR and agent,
/// whatever `internal/meta` rev a retry builds, until a green build clears it.
fn failure_marker(agent: &str) -> String {
    format!("<!-- swarm-build-failed agent={agent} -->")
}

/// What a green build leaves in place of [`failure_marker`], so the next red
/// build comments again. It does not contain the failure marker.
fn cleared_marker(agent: &str) -> String {
    format!("<!-- swarm-build-cleared agent={agent} -->")
}

/// The comment a red build leaves on its merged config PR, first line the
/// failure marker.
fn failure_comment(agent: &str, meta_rev: &str, error_tail: &str) -> String {
    let marker = failure_marker(agent);
    let mut start = error_tail.len().saturating_sub(COMMENT_TAIL_BYTES);
    while !error_tail.is_char_boundary(start) {
        start += 1;
    }
    let tail = error_tail[start..].replace("```", "'''");
    format!(
        "{marker}\nThe swarm could not build `{agent}` from `internal/meta` at `{meta_rev}` \
         after this PR merged. The deploy is held and built again on every round until a \
         build succeeds. This is the only comment while it stays red; the latest error is on \
         the Jobs page.\n\nLast lines of `nix`:\n\n```\n{}\n```",
        tail.trim_end()
    )
}

/// Tell `pr` that its merge's build failed, once per agent while the build
/// stays red. A comment that cannot be posted is logged and changes nothing
/// about the node.
async fn report_failure(
    head: &impl MetaRepo,
    pr: u64,
    agent: &str,
    meta_rev: &str,
    error_tail: &str,
) {
    let body = failure_comment(agent, meta_rev, error_tail);
    match head
        .comment_once(agent, pr, &failure_marker(agent), &body)
        .await
    {
        Ok(posted) => {
            tracing::info!(%agent, pr, %meta_rev, posted, "build_meta: reported the failed build on the PR");
        }
        Err(e) => tracing::warn!(
            %agent, pr, %meta_rev, error = %format!("{e:#}"),
            "build_meta: could not comment on the PR"
        ),
    }
}

/// A green build ends the red spell: clear `pr`'s failure marker for `agent`,
/// so a later red comments again. A failure is logged and changes nothing
/// about the node.
async fn clear_failure(head: &impl MetaRepo, pr: u64, agent: &str, meta_rev: &str) {
    let resolved = format!(
        "Resolved: built at `{}`.",
        meta_rev.get(..12).unwrap_or(meta_rev)
    );
    match head
        .clear_marker(
            agent,
            pr,
            &failure_marker(agent),
            &cleared_marker(agent),
            &resolved,
        )
        .await
    {
        Ok(cleared) => {
            tracing::debug!(%agent, pr, cleared, "build_meta: cleared the failure marker");
        }
        Err(e) => tracing::warn!(
            %agent, pr, error = %format!("{e:#}"),
            "build_meta: could not clear the failure marker on the PR"
        ),
    }
}

/// Store the build of `agent` from `meta_rev` that ended as `outcome`, `text`
/// being its store path, `nix`'s tail or the reason, in the list the builds
/// page shows beside the config PRs' builds. A failed write is a warning in
/// `Builds::record` and changes nothing about the node.
async fn record_run(
    builds: &Builds,
    agent: &str,
    meta_rev: &str,
    started_unix: i64,
    outcome: BuildOutcome,
    text: String,
) {
    let mut build = BuildRecord {
        id: format!(
            "meta-{agent}-{}-{started_unix}",
            meta_rev.get(..12).unwrap_or(meta_rev)
        ),
        repo: format!("{SHARED_ORG}/{META_REPO}"),
        pr: None,
        agent: Some(agent.to_owned()),
        head_sha: meta_rev.to_owned(),
        started_unix,
        finished_unix: unix_now(),
        outcome,
        out_path: None,
        error_tail: None,
        detail: None,
    };
    *match outcome {
        BuildOutcome::Built => &mut build.out_path,
        BuildOutcome::Failed => &mut build.error_tail,
        BuildOutcome::Error => &mut build.detail,
    } = Some(text);
    builds.record(&build).await;
}

/// Keep the builder's `reply` to a build of `agent` from `meta_rev` in `builds`.
async fn record_reply(
    builds: &Builds,
    agent: &str,
    meta_rev: &str,
    started_unix: i64,
    reply: &Result<FromMetaReply, String>,
) {
    let (outcome, text) = match reply {
        Err(why) => (BuildOutcome::Error, why),
        Ok(FromMetaReply::Unavailable { reason }) => (BuildOutcome::Error, reason),
        Ok(FromMetaReply::Failed { error_tail }) => (BuildOutcome::Failed, error_tail),
        Ok(FromMetaReply::Built { out_path, .. }) => (BuildOutcome::Built, out_path),
    };
    record_run(builds, agent, meta_rev, started_unix, outcome, text.clone()).await;
}

/// Build `agent` from `internal/meta`'s head and record the closure. A build
/// the builder answers, or fails to be asked, is also kept in `builds`; one that
/// never got as far as the request (no forge, no head) is not.
pub(crate) async fn run(
    head: Option<&impl MetaRepo>,
    queue: Option<&impl MetaBuilds>,
    builds: &Builds,
    agent: String,
    pr: Option<u64>,
) -> Outcome {
    let Some(head) = head else {
        return no_verdict(
            "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token not in the \
             swarm secret store), so internal/meta's head cannot be read",
        );
    };
    let Some(queue) = queue else {
        return no_verdict("no swarm queue is configured on this host, so no build can be sent");
    };
    // The repo is private, and the grant is otherwise made only by a meta update.
    if let Err(e) = head.grant_builder_read().await {
        return no_verdict(format!(
            "the builder's read on internal/meta could not be granted: {e:#}"
        ));
    }
    let meta_rev = match head.head().await {
        Ok(Some(rev)) => rev,
        Ok(None) => return no_verdict("the swarm has not published internal/meta yet"),
        Err(e) => return no_verdict(format!("{e:#}")),
    };
    let request = match serde_json::to_vec(&BuildFromMeta {
        agent: agent.clone(),
        meta_rev: meta_rev.clone(),
    }) {
        Ok(request) => request,
        Err(e) => return no_verdict(format!("encoding the build request: {e}")),
    };
    let started_unix = unix_now();
    let reply = queue.build(request).await;
    record_reply(builds, &agent, &meta_rev, started_unix, &reply).await;
    match reply {
        Err(why) => no_verdict(why),
        Ok(FromMetaReply::Unavailable { reason }) => no_verdict(reason),
        Ok(FromMetaReply::Failed { error_tail }) => {
            if let Some(pr) = pr {
                report_failure(head, pr, &agent, &meta_rev, &error_tail).await;
            }
            Outcome::Failed(error_tail)
        }
        Ok(FromMetaReply::Built {
            out_path,
            system,
            cached,
        }) => {
            let built_rev = meta_rev.clone();
            let closure = AgentClosure {
                meta_rev,
                store_path: out_path.clone(),
                system,
                built_unix: unix_now(),
                cached,
            };
            match queue.record(&agent, &closure).await {
                Ok(()) => {
                    if let Some(pr) = pr {
                        clear_failure(head, pr, &agent, &built_rev).await;
                    }
                    Outcome::Done
                }
                Err(e) => no_verdict(format!("built {out_path}, but recording it failed: {e:#}")),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use crate::builder::runs::tests::FakeDoc;

    use super::*;

    const REV: &str = "0123456789abcdef0123456789abcdef01234567";

    /// `internal/meta` as the forge answers: the head read, and whether the
    /// builder's read grant can be made.
    struct FakeHead {
        head: Result<Option<&'static str>, &'static str>,
        grant_fails: bool,
        /// The forge a PR comment goes to; `None` panics on a comment.
        comments: Option<std::sync::Arc<forge::Client>>,
    }

    impl FakeHead {
        fn reading(head: Result<Option<&'static str>, &'static str>) -> Self {
            Self {
                head,
                grant_fails: false,
                comments: None,
            }
        }
    }

    impl MetaRepo for FakeHead {
        async fn head(&self) -> Result<Option<String>> {
            match self.head {
                Ok(rev) => Ok(rev.map(str::to_owned)),
                Err(e) => Err(anyhow::anyhow!(e)),
            }
        }

        async fn grant_builder_read(&self) -> Result<()> {
            anyhow::ensure!(!self.grant_fails, "the forge refused the collaborator");
            Ok(())
        }

        async fn comment_once(
            &self,
            agent: &str,
            pr: u64,
            marker: &str,
            body: &str,
        ) -> Result<bool> {
            let client = self.comments.as_ref().expect("no comment expected");
            client
                .comment_once(forge::CONFIG_ORG, agent, pr, marker, body)
                .await
        }

        async fn clear_marker(
            &self,
            agent: &str,
            pr: u64,
            marker: &str,
            cleared: &str,
            resolved: &str,
        ) -> Result<usize> {
            let client = self.comments.as_ref().expect("no clear expected");
            client
                .clear_marker(forge::CONFIG_ORG, agent, pr, marker, cleared, resolved)
                .await
        }
    }

    /// A queue that answers every build with `reply`, and keeps what it was
    /// sent and what was recorded.
    struct FakeQueue {
        reply: Result<FromMetaReply, String>,
        record_fails: bool,
        sent: Mutex<Vec<BuildFromMeta>>,
        recorded: Mutex<Vec<(String, AgentClosure)>>,
    }

    impl FakeQueue {
        fn answering(reply: Result<FromMetaReply, String>) -> Self {
            Self {
                reply,
                record_fails: false,
                sent: Mutex::new(Vec::new()),
                recorded: Mutex::new(Vec::new()),
            }
        }

        fn built() -> Self {
            Self::built_cached(false)
        }

        fn built_cached(cached: bool) -> Self {
            Self::answering(Ok(FromMetaReply::Built {
                out_path: "/nix/store/x-toplevel".to_owned(),
                system: "x86_64-linux".to_owned(),
                cached,
            }))
        }
    }

    impl MetaBuilds for FakeQueue {
        async fn build(&self, request: Vec<u8>) -> Result<FromMetaReply, String> {
            self.sent
                .lock()
                .unwrap()
                .push(serde_json::from_slice(&request).expect("a build request"));
            self.reply.clone()
        }

        async fn record(&self, agent: &str, closure: &AgentClosure) -> Result<()> {
            anyhow::ensure!(!self.record_fails, "the bucket refused the write");
            self.recorded
                .lock()
                .unwrap()
                .push((agent.to_owned(), closure.clone()));
            Ok(())
        }
    }

    async fn run_with(head: &FakeHead, queue: &FakeQueue) -> Outcome {
        run_recording(head, queue, &Arc::default()).await
    }

    /// [`run_with`], keeping the builds in `doc`.
    async fn run_recording(head: &FakeHead, queue: &FakeQueue, doc: &Arc<FakeDoc>) -> Outcome {
        let builds = Builds::Fake(Arc::clone(doc));
        run(Some(head), Some(queue), &builds, "atlas".to_owned(), None).await
    }

    async fn run_for_pr(head: &FakeHead, queue: &FakeQueue, pr: u64) -> Outcome {
        let builds = Builds::Fake(Arc::default());
        run(
            Some(head),
            Some(queue),
            &builds,
            "atlas".to_owned(),
            Some(pr),
        )
        .await
    }

    /// The builds page lists these beside the config PRs' builds: which agent,
    /// at which `internal/meta` commit, how it ended, and with what.
    #[tokio::test]
    async fn every_build_the_builder_answers_is_kept_for_the_builds_page() {
        let head = FakeHead::reading(Ok(Some(REV)));
        let green = Arc::new(FakeDoc::default());
        run_recording(&head, &FakeQueue::built(), &green).await;
        let builds = green.builds();
        let [built] = builds.as_slice() else {
            panic!("one build kept: {builds:?}")
        };
        assert_eq!(built.repo, "internal/meta");
        assert_eq!(built.pr, None);
        assert_eq!(built.agent.as_deref(), Some("atlas"));
        assert_eq!(built.head_sha, REV);
        assert_eq!(built.outcome, BuildOutcome::Built);
        assert_eq!(built.out_path.as_deref(), Some("/nix/store/x-toplevel"));
        assert!(
            built.id.starts_with(&format!("meta-atlas-{}-", &REV[..12])),
            "{}",
            built.id
        );

        let red = Arc::new(FakeDoc::default());
        run_recording(&head, &nix_failed("error: boom"), &red).await;
        let builds = red.builds();
        let [failed] = builds.as_slice() else {
            panic!("one build kept: {builds:?}")
        };
        assert_eq!(failed.outcome, BuildOutcome::Failed);
        assert_eq!(failed.error_tail.as_deref(), Some("error: boom"));

        let none = Arc::new(FakeDoc::default());
        let silent = FakeQueue::answering(Err("no builder is listening".to_owned()));
        run_recording(&head, &silent, &none).await;
        let builds = none.builds();
        let [error] = builds.as_slice() else {
            panic!("one build kept: {builds:?}")
        };
        assert_eq!(error.outcome, BuildOutcome::Error);
        assert_eq!(error.detail.as_deref(), Some("no builder is listening"));
    }

    #[tokio::test]
    async fn a_green_build_of_the_head_is_recorded_for_the_agent() {
        let queue = FakeQueue::built();
        let outcome = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await;
        assert_eq!(outcome, Outcome::Done);
        assert_eq!(
            *queue.sent.lock().unwrap(),
            [BuildFromMeta {
                agent: "atlas".to_owned(),
                meta_rev: REV.to_owned(),
            }]
        );
        let recorded = queue.recorded.lock().unwrap();
        let [(agent, closure)] = recorded.as_slice() else {
            panic!("one record: {recorded:?}")
        };
        assert_eq!(agent, "atlas");
        assert_eq!(closure.meta_rev, REV);
        assert_eq!(closure.store_path, "/nix/store/x-toplevel");
        assert_eq!(closure.system, "x86_64-linux");
        assert!(closure.built_unix > 0);
        assert!(!closure.cached);
    }

    /// The record says what the builder said about the cache, both ways.
    #[tokio::test]
    async fn the_record_carries_whether_the_builder_cached_the_closure() {
        for cached in [true, false] {
            let queue = FakeQueue::built_cached(cached);
            let outcome = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await;
            assert_eq!(outcome, Outcome::Done);
            let recorded = queue.recorded.lock().unwrap();
            let [(_, closure)] = recorded.as_slice() else {
                panic!("one record: {recorded:?}")
            };
            assert_eq!(closure.cached, cached);
        }
    }

    /// Any `nix` failure, a timeout included, fails the node at once with what
    /// the builder said: not a no-verdict failure, and nothing is recorded. The
    /// green test above is the control that a record is written when the build
    /// is green.
    #[tokio::test]
    async fn a_nix_failure_fails_the_node_with_its_tail_and_records_nothing() {
        for tail in [
            "error: builder for foo failed",
            "timed out: nix took 60 minutes",
        ] {
            let queue = FakeQueue::answering(Ok(FromMetaReply::Failed {
                error_tail: tail.to_owned(),
            }));
            let outcome = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await;
            assert_eq!(outcome, Outcome::Failed(tail.to_owned()));
            assert!(!is_no_verdict(tail));
            assert!(queue.recorded.lock().unwrap().is_empty(), "{tail}");
        }
    }

    /// A read grant the forge will not make is the controller's precondition,
    /// not the config's: nothing is sent and the failure has no verdict.
    #[tokio::test]
    async fn a_read_grant_that_cannot_be_made_is_a_no_verdict_failure_before_anything_is_sent() {
        let queue = FakeQueue::built();
        let head = FakeHead {
            grant_fails: true,
            ..FakeHead::reading(Ok(Some(REV)))
        };
        let Outcome::Failed(error) = run_with(&head, &queue).await else {
            panic!("not failed")
        };
        assert!(is_no_verdict(&error), "{error}");
        assert!(error.contains("read on internal/meta"), "{error}");
        assert!(queue.sent.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn a_builder_that_gave_no_verdict_is_a_no_verdict_failure_and_records_nothing() {
        for (name, reply) in [
            (
                "no responders",
                Err("no swarm builder is listening for meta builds".to_owned()),
            ),
            (
                "unavailable",
                Ok(FromMetaReply::Unavailable {
                    reason: "busy".to_owned(),
                }),
            ),
        ] {
            let queue = FakeQueue::answering(reply);
            let outcome = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await;
            let Outcome::Failed(error) = outcome else {
                panic!("{name}: not failed")
            };
            assert!(is_no_verdict(&error), "{name}: {error}");
            assert!(queue.recorded.lock().unwrap().is_empty(), "{name}");
        }
    }

    #[tokio::test]
    async fn a_closure_that_cannot_be_recorded_is_a_no_verdict_failure() {
        let queue = FakeQueue {
            record_fails: true,
            ..FakeQueue::built()
        };
        let Outcome::Failed(error) = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await
        else {
            panic!("not failed")
        };
        assert!(is_no_verdict(&error), "{error}");
        assert!(error.contains("/nix/store/x-toplevel"), "{error}");
    }

    #[tokio::test]
    async fn without_a_head_or_a_queue_nothing_is_sent_and_the_failure_has_no_verdict() {
        for (name, head, with_queue) in [
            ("no repo yet", Some(FakeHead::reading(Ok(None))), true),
            (
                "forge error",
                Some(FakeHead::reading(Err("forge down"))),
                true,
            ),
            ("no forge", None, true),
            ("no queue", Some(FakeHead::reading(Ok(Some(REV)))), false),
        ] {
            let queue = FakeQueue::built();
            let doc = Arc::new(FakeDoc::default());
            let outcome = run(
                head.as_ref(),
                with_queue.then_some(&queue),
                &Builds::Fake(Arc::clone(&doc)),
                "atlas".to_owned(),
                None,
            )
            .await;
            assert!(doc.builds().is_empty(), "{name}: nothing was built");
            let Outcome::Failed(error) = outcome else {
                panic!("{name}: not failed")
            };
            assert!(is_no_verdict(&error), "{name}: {error}");
            assert!(queue.sent.lock().unwrap().is_empty(), "{name}");
        }
    }

    const REV2: &str = "fedcba9876543210fedcba9876543210fedcba98";

    fn nix_failed(tail: &str) -> FakeQueue {
        FakeQueue::answering(Ok(FromMetaReply::Failed {
            error_tail: tail.to_owned(),
        }))
    }

    async fn head_with_comments(
        rev: &'static str,
    ) -> (FakeHead, crate::forge::pr_comment::tests::Posted) {
        let (client, posted) = crate::forge::pr_comment::tests::stand_in().await;
        let head = FakeHead {
            comments: Some(std::sync::Arc::new(client)),
            ..FakeHead::reading(Ok(Some(rev)))
        };
        (head, posted)
    }

    /// A `nix` failure of a merged config PR's build is one comment on that PR
    /// with the agent, the `internal/meta` rev, `nix`'s tail and the retry.
    #[tokio::test]
    async fn a_red_build_of_a_merged_pr_comments_once_on_it() {
        let (head, posted) = head_with_comments(REV).await;
        let outcome = run_for_pr(&head, &nix_failed("error: builder for foo failed"), 7).await;
        assert_eq!(
            outcome,
            Outcome::Failed("error: builder for foo failed".to_owned())
        );
        let posted = posted.lock().unwrap();
        let [(path, body)] = posted.as_slice() else {
            panic!("one comment: {posted:?}")
        };
        assert_eq!(
            path,
            &format!("/api/v1/repos/agent-configs/atlas/{}/7/comments", "issues")
        );
        assert!(body.contains("`atlas`"), "{body}");
        assert!(body.contains(REV), "{body}");
        assert!(body.contains("error: builder for foo failed"), "{body}");
        assert!(body.contains("held"), "{body}");
        assert!(body.contains("every round"), "{body}");
    }

    /// The timer builds a failed merge again on every round, and `internal/meta`
    /// moves between rounds: while the agent stays red the PR keeps its one
    /// comment, whatever rev a retry builds.
    #[tokio::test]
    async fn a_build_that_stays_red_comments_once_across_retries_and_meta_revs() {
        let (head, posted) = head_with_comments(REV).await;
        for _ in 0..3 {
            run_for_pr(&head, &nix_failed("error: boom"), 7).await;
        }
        let moved = FakeHead {
            comments: head.comments.clone(),
            ..FakeHead::reading(Ok(Some(REV2)))
        };
        run_for_pr(&moved, &nix_failed("error: other"), 7).await;
        let posted = posted.lock().unwrap();
        assert_eq!(posted.len(), 1, "{posted:?}");
        assert!(posted[0].1.contains(REV), "{:?}", posted[0]);
    }

    /// A green build ends the red spell, so a red after it comments again;
    /// a red on another PR of the same agent is a different spell.
    #[tokio::test]
    async fn a_red_after_a_green_comments_again() {
        let (head, posted) = head_with_comments(REV).await;
        run_for_pr(&head, &nix_failed("error: boom"), 7).await;
        let green = run_for_pr(&head, &FakeQueue::built(), 7).await;
        assert_eq!(green, Outcome::Done);
        let count = posted.lock().unwrap().len();
        assert_eq!(count, 1, "a green build posts nothing");
        run_for_pr(&head, &nix_failed("error: boom"), 7).await;
        run_for_pr(&head, &nix_failed("error: boom"), 7).await;
        let posted = posted.lock().unwrap();
        assert_eq!(posted.len(), 2, "red, green, red: {posted:?}");
        assert!(
            !posted[0].1.contains(&failure_marker("atlas")),
            "the first comment's marker is cleared: {:?}",
            posted[0]
        );
        assert!(posted[0].1.contains(&cleared_marker("atlas")));
        assert!(
            posted[0].1.contains("Resolved: built at `0123456789ab`"),
            "{:?}",
            posted[0]
        );
    }

    /// A build that got no verdict says nothing about the PR's config, so it
    /// leaves no comment. The `nix` failure arm is the control.
    #[tokio::test]
    async fn a_no_verdict_failure_does_not_comment() {
        for (name, reply) in [
            ("no responders", Err("no swarm builder".to_owned())),
            (
                "unavailable",
                Ok(FromMetaReply::Unavailable {
                    reason: "busy".to_owned(),
                }),
            ),
        ] {
            let (head, posted) = head_with_comments(REV).await;
            let outcome = run_for_pr(&head, &FakeQueue::answering(reply), 7).await;
            assert!(
                matches!(outcome, Outcome::Failed(e) if is_no_verdict(&e)),
                "{name}"
            );
            assert!(posted.lock().unwrap().is_empty(), "{name}");
        }
        let (head, posted) = head_with_comments(REV).await;
        run_for_pr(&head, &nix_failed("error: boom"), 7).await;
        assert_eq!(
            posted.lock().unwrap().len(),
            1,
            "control: a nix failure comments"
        );
    }

    /// A build that is not a merged PR's (the pin fan-out, a created agent)
    /// has nobody to comment on.
    #[tokio::test]
    async fn a_failed_build_that_is_not_from_a_pr_does_not_comment() {
        let (head, posted) = head_with_comments(REV).await;
        let outcome = run_with(&head, &nix_failed("error: boom")).await;
        assert_eq!(outcome, Outcome::Failed("error: boom".to_owned()));
        assert!(posted.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn the_comment_carries_the_end_of_a_long_tail_only() {
        let tail = format!("START{}END", "x".repeat(10_000));
        let (head, posted) = head_with_comments(REV).await;
        run_for_pr(&head, &nix_failed(&tail), 7).await;
        let posted = posted.lock().unwrap();
        let body = &posted[0].1;
        assert!(body.contains("END"), "the end of the tail is kept");
        assert!(!body.contains("START"), "the start is cut");
        assert!(body.len() < COMMENT_TAIL_BYTES + 1000, "{}", body.len());
    }

    /// A forge that does not take the comment leaves the node failed as it was.
    #[tokio::test]
    async fn a_comment_that_cannot_be_posted_does_not_change_the_outcome() {
        let head = FakeHead {
            comments: Some(std::sync::Arc::new(crate::forge::Client::for_test_forge(
                url::Url::parse("http://127.0.0.1:1/").unwrap(),
            ))),
            ..FakeHead::reading(Ok(Some(REV)))
        };
        let outcome = run_for_pr(&head, &nix_failed("error: boom"), 7).await;
        assert_eq!(outcome, Outcome::Failed("error: boom".to_owned()));
    }

    #[test]
    fn the_repo_path_the_builder_fetches_is_the_one_the_controller_writes() {
        assert_eq!(
            swarm_wire::builder::META_REPO_PATH,
            format!("{SHARED_ORG}/{META_REPO}")
        );
    }

    /// The bucket on a real server: a record written for an agent reads back
    /// under its key, and a second build replaces it.
    #[tokio::test]
    async fn the_closure_round_trips_through_the_bucket() {
        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        let mut closure = AgentClosure {
            meta_rev: REV.to_owned(),
            store_path: "/nix/store/a".to_owned(),
            system: "x86_64-linux".to_owned(),
            built_unix: 1,
            cached: false,
        };
        client.record("atlas", &closure).await.expect("written");
        closure.store_path = "/nix/store/b".to_owned();
        client.record("atlas", &closure).await.expect("replaced");
        let store = swarm_queue_client::agent_closures::open_or_create(&client)
            .await
            .expect("opens");
        let got = store.get("atlas").await.expect("reads").expect("present");
        let got: AgentClosure = serde_json::from_slice(&got).expect("decodes");
        assert_eq!(got, closure);
        assert!(store.get("argus").await.expect("reads").is_none());
    }
}