areev 1.12.2

//! `areev trigger` through the real binary.
//!
//! The surface this covers had none: `memory` and `composite` are two of the
//! eight declared kinds, and neither could be declared from the CLI at all —
//! `incoherence()` refuses a memory trigger without a predicate, and there was
//! no flag that could supply one. A unit test on the parser would not have
//! caught that, because the parser was fine; the wiring was missing.

use std::process::Command;
use tempfile::TempDir;

/// A content address shaped like a real one. `trigger add` does not resolve the
/// workflow — that is TRG-E002 at firing time — so a literal is honest here.
const WF: &str = "a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90";
const MEMBER_A: &str = "b1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90";
const MEMBER_B: &str = "c1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90";

fn areev(args: &[&str]) -> (bool, String, String) {
    let out = Command::new(env!("CARGO_BIN_EXE_areev")).args(args).output().expect("spawn areev");
    (
        out.status.success(),
        String::from_utf8_lossy(&out.stdout).to_string(),
        String::from_utf8_lossy(&out.stderr).to_string(),
    )
}

#[test]
fn memory_and_composite_triggers_can_be_declared_from_the_cli() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    // A memory trigger: the predicate is CAL `WHERE` syntax.
    let (ok, out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "memory", "--workflow", WF,
        "--where", r#"grain_type = "fact" AND subject = "acme""#,
        "--because", "start review when an Acme fact lands",
    ]);
    assert!(ok, "memory trigger refused: {err}");
    assert_eq!(out.trim().rsplit(' ').next().unwrap().len(), 64, "{out}");

    // A composite: aliases, a gate over them, and a correlation window.
    let members = format!("invoice={MEMBER_A},purchase_order={MEMBER_B}");
    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "composite", "--workflow", WF,
        "--members", &members,
        "--where", "invoice = true AND purchase_order = true",
        "--correlate", "/thread_id", "--window", "10m",
        "--because", "match an invoice to its purchase order",
    ]);
    assert!(ok, "composite trigger refused: {err}");

    let (ok, out, err) = areev(&["trigger", "list", "--db", db, "--ns", "ops", "--json"]);
    assert!(ok, "list failed: {err}");
    assert!(out.contains("memory"), "{out}");
    assert!(out.contains("composite"), "{out}");
}

#[test]
fn a_declaration_that_could_never_fire_is_refused_rather_than_stored() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    // A memory trigger with no predicate selects nothing, forever.
    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "memory", "--workflow", WF,
        "--because", "no predicate",
    ]);
    assert!(!ok, "a memory trigger with no predicate must be refused");
    assert!(err.contains("predicate"), "{err}");

    // A gate naming a member the declaration does not carry can never be
    // satisfied. Firing already refused this; declaring must too.
    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "composite", "--workflow", WF,
        "--members", &format!("invoice={MEMBER_A},purchase_order={MEMBER_B}"),
        "--where", "invoice = true AND ghost = true",
        "--because", "names a member it does not declare",
    ]);
    assert!(!ok, "a gate over an undeclared member must be refused");
    assert!(err.contains("TRG-E008"), "{err}");
    assert!(err.contains("ghost"), "{err}");

    // A window with no unit is ambiguous between seconds and milliseconds.
    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "composite", "--workflow", WF,
        "--members", &format!("invoice={MEMBER_A},purchase_order={MEMBER_B}"),
        "--where", "invoice = true AND purchase_order = true",
        "--window", "600",
        "--because", "unitless window",
    ]);
    assert!(!ok, "a unitless --window must be refused, not guessed");
    assert!(err.contains("--window"), "{err}");
}

/// The k8s manifest must not carry the authoring machine's binary path (#69).
///
/// Driven through the real binary on purpose: the defect was
/// `std::env::current_exe()` reaching the render, which only has a wrong value
/// when a real process produced it. A unit test supplying `exe: "areev"` cannot
/// see it — that is exactly why this shipped.
#[test]
fn k8s_render_uses_the_image_binary_not_the_authoring_hosts_path() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "interval",
        "--workflow", WF, "--interval", "900", "--because", "heartbeat",
    ]);
    assert!(ok, "{err}");

    let (ok, out, err) =
        areev(&["trigger", "render", "--db", db, "--ns", "ops", "--target", "k8s-cronjob"]);
    assert!(ok, "{err}");

    let exe = env!("CARGO_BIN_EXE_areev");
    assert!(
        !out.contains(exe),
        "a path from this machine cannot be right inside a container:\n{out}"
    );
    assert!(out.contains("\n            - areev\n"), "command[0] must be the PATH name:\n{out}");

    // The host targets run where they were rendered, so for them it IS right.
    for target in ["cron", "launchd", "systemd"] {
        let (ok, out, err) =
            areev(&["trigger", "render", "--db", db, "--ns", "ops", "--target", target]);
        assert!(ok, "{err}");
        assert!(out.contains(exe), "{target} runs locally and must name the local binary:\n{out}");
    }
}

/// A schedule this build cannot evaluate is refused at declaration (#67).
///
/// Since 1.9.0 the CLI carries zone data (#297), so a REAL zone is accepted
/// and reported — an UNKNOWN zone name is what cannot be evaluated, and
/// refusing a typo rather than falling back to UTC is the point: a schedule
/// quietly firing at the wrong hour looks correct.
#[test]
fn an_unknown_timezone_is_refused_when_declared() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "schedule", "--workflow", WF,
        "--cron", "0 9 * * *", "--timezone", "Mars/Olympus", "--because", "probe",
    ]);
    assert!(!ok, "an unknown zone name must be refused");
    assert!(err.contains("TRG-E006"), "{err}");

    // Nothing was stored, so `status` cannot report it as healthy.
    let (ok, out, _err) = areev(&["trigger", "status", "--db", db, "--ns", "ops"]);
    assert!(ok);
    assert!(out.contains("no triggers declared"), "{out}");
}

/// A real IANA zone is accepted, and `trigger status` says which one (#297).
#[test]
fn a_declared_timezone_is_accepted_and_reported() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("tz.db");
    let db = db.to_str().unwrap();

    let (ok, _out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "schedule", "--workflow", WF,
        "--cron", "0 8 * * 1", "--timezone", "America/New_York", "--because", "weekly brief",
    ]);
    assert!(ok, "{err}");

    let (ok, out, err) = areev(&[
        "trigger", "status", "--db", db, "--ns", "ops", "--format", "json",
    ]);
    assert!(ok, "{err}");
    assert!(
        out.contains("America/New_York"),
        "status must name the zone — on a DST boundary the UTC instant moves \
         by an hour while the local wall time does not: {out}"
    );
}

/// The read and lifecycle verbs — `show`, `status`, `pause`, `resume`,
/// `render`, `deliver` — none of which had a test.
///
/// These are the verbs an operator actually types between declaring a trigger
/// and debugging why it has not fired, and the README now points people at
/// them. Each assertion below pins what the operator is told, not merely that
/// the command exited zero.
#[test]
fn the_trigger_read_and_lifecycle_verbs_report_what_an_operator_needs() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    let (ok, out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "interval",
        "--interval", "120", "--workflow", WF, "--because", "poll the mailbox",
    ]);
    assert!(ok, "add refused: {err}");
    let id = out.trim().rsplit(' ').next().unwrap().to_string();
    assert_eq!(id.len(), 64, "add should print the declaration hash: {out}");
    let short = &id[..12];

    // ── show ────────────────────────────────────────────────────────────────
    let (ok, out, err) = areev(&["trigger", "show", short, "--db", db, "--ns", "ops"]);
    assert!(ok, "show refused: {err}");
    assert!(out.contains(&id), "show must name the full trigger: {out}");
    assert!(out.contains("kind         interval"), "{out}");
    assert!(out.contains("enabled      true"), "{out}");
    assert!(out.contains("paused       false"), "{out}");
    // A never-fired trigger must SAY so. "last fired <blank>" would read as a
    // rendering glitch; "never" is the answer to the question being asked.
    assert!(out.contains("last fired   never"), "{out}");

    // A prefix that matches nothing is an error naming the namespace it
    // searched — the usual cause is being in the wrong one.
    let (ok, _out, err) = areev(&["trigger", "show", "deadbeef", "--db", db, "--ns", "ops"]);
    assert!(!ok, "show of an unknown trigger must fail");
    assert!(err.contains("ops"), "the error should name the namespace: {err}");

    let (ok, _out, err) = areev(&["trigger", "show", "--db", db, "--ns", "ops"]);
    assert!(!ok, "show with no argument must fail");
    assert!(err.contains("usage:"), "{err}");

    // ── status ──────────────────────────────────────────────────────────────
    // The staleness warning is the point of this verb: an enabled trigger that
    // has never fired usually means nobody put `trigger run` on a heartbeat,
    // and that failure is otherwise completely silent.
    let (ok, _out, err) = areev(&["trigger", "status", "--db", db, "--ns", "ops"]);
    assert!(ok, "status refused: {err}");
    assert!(
        err.contains("never fired") && err.contains("heartbeat"),
        "status must warn about the never-fired trigger: {err}"
    );

    // ── pause / resume ──────────────────────────────────────────────────────
    // Pausing a standing rule is a governed act: it stops work silently, so
    // the reason is mandatory the same way a loop decision's is.
    let (ok, _out, err) = areev(&["trigger", "pause", short, "--db", db, "--ns", "ops"]);
    assert!(!ok, "pause without a reason must be refused");
    assert!(err.contains("because"), "{err}");

    let (ok, _out, err) = areev(&[
        "trigger", "pause", short, "--db", db, "--ns", "ops",
        "--because", "the vendor mailbox is being migrated",
    ]);
    assert!(ok, "pause refused: {err}");
    let (_ok, out, _err) = areev(&["trigger", "show", short, "--db", db, "--ns", "ops"]);
    assert!(out.contains("paused       true"), "pause must be visible in show: {out}");

    // A paused trigger is not a never-fired-and-forgotten one, so it must NOT
    // raise the staleness warning — otherwise the warning trains operators to
    // ignore it.
    let (_ok, _out, err) = areev(&["trigger", "status", "--db", db, "--ns", "ops"]);
    assert!(
        !err.contains("never fired"),
        "a deliberately paused trigger must not be reported as stale: {err}"
    );

    let (ok, _out, err) = areev(&[
        "trigger", "resume", short, "--db", db, "--ns", "ops",
        "--because", "migration done",
    ]);
    assert!(ok, "resume refused: {err}");
    let (_ok, out, _err) = areev(&["trigger", "show", short, "--db", db, "--ns", "ops"]);
    assert!(out.contains("paused       false"), "resume must be visible in show: {out}");

    // ── show --format json ──────────────────────────────────────────────────
    let (ok, out, err) =
        areev(&["trigger", "show", short, "--db", db, "--ns", "ops", "--format", "json"]);
    assert!(ok, "show --format json refused: {err}");
    let v: serde_json::Value = serde_json::from_str(out.trim()).expect("show --format json");
    assert_eq!(v["trigger"], id.as_str());
    assert_eq!(v["kind"], "interval");
    assert_eq!(v["paused"], false);

    // ── render ──────────────────────────────────────────────────────────────
    // Every declared target must produce something that mentions the verb it
    // is scheduling; a target that renders an empty unit file is worse than an
    // error, because it installs cleanly and never runs.
    for target in ["cron", "launchd", "systemd", "k8s-cronjob"] {
        let (ok, out, err) =
            areev(&["trigger", "render", target, "--db", db, "--ns", "ops"]);
        assert!(ok, "render {target} refused: {err}");
        // launchd renders the argv as separate <string> elements, so match the
        // tokens rather than the joined command line.
        assert!(
            out.contains("trigger") && out.contains("run"),
            "render {target} must schedule the verb: {out}"
        );
        assert!(
            err.contains("heartbeat"),
            "render must explain the cadence it chose: {err}"
        );
    }

    let (ok, _out, _err) = areev(&["trigger", "render", "sysvinit", "--db", db, "--ns", "ops"]);
    assert!(!ok, "an unknown render target must fail");

    // ── unknown subcommand ──────────────────────────────────────────────────
    let (ok, _out, err) = areev(&["trigger", "frobnicate", "--db", db, "--ns", "ops"]);
    assert!(!ok, "an unknown subcommand must fail");
    assert!(
        err.contains("add") && err.contains("deliver"),
        "the error should list the accepted subcommands: {err}"
    );
}

/// `areev trigger deliver` — the push half of the surface. The host owns the
/// listener and hands Areev the payload, so this is the only way a `webhook`
/// or `manual` trigger ever fires.
#[test]
fn deliver_hands_a_payload_to_a_push_trigger() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    let (ok, out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "webhook",
        "--workflow", WF, "--dedup-key", "/invoice_id",
        "--because", "the host receives invoice webhooks",
    ]);
    assert!(ok, "webhook trigger refused: {err}");
    let id = out.trim().rsplit(' ').next().unwrap().to_string();
    let short = &id[..12];

    // A literal payload. Without --tool-cmd nothing executes, which is the
    // documented ingest-only mode rather than a failure.
    let (ok, out, err) = areev(&[
        "trigger", "deliver", "--db", db, "--ns", "ops", "--id", short,
        "--payload", r#"{"invoice_id":"INV-1","amount":420}"#,
    ]);
    assert!(ok, "deliver refused: {err}\n{out}");
    assert!(out.contains("ingested 1"), "the payload should be taken in: {out}");
    assert!(out.contains("runs 0"), "…but nothing executes without --tool-cmd: {out}");

    // Ingesting is not firing. `last fired` tracks runs STARTED, and no tool
    // command was given, so the honest answer is still "never" — the counter
    // does not inflate itself on a payload nobody executed.
    let (_ok, out, _err) = areev(&["trigger", "show", short, "--db", db, "--ns", "ops"]);
    assert!(
        out.contains("last fired   never"),
        "an ingested-but-unexecuted payload must not count as a firing: {out}"
    );

    // Re-delivering the same invoice is a DUPLICATE, not a second run. A
    // webhook that retries must not pay the invoice twice.
    let (ok, out, err) = areev(&[
        "trigger", "deliver", "--db", db, "--ns", "ops", "--id", short,
        "--payload", r#"{"invoice_id":"INV-1","amount":420}"#,
    ]);
    assert!(ok, "re-deliver refused: {err}");
    assert!(out.contains("already delivered"), "a repeat must be reported as such: {out}");

    // A payload with nothing at the declared dedup key cannot be identified,
    // so nothing starts — and the operator is told, rather than left to infer
    // it from a run that never appears.
    let (ok, out, err) = areev(&[
        "trigger", "deliver", "--db", db, "--ns", "ops", "--id", short,
        "--payload", r#"{"no_identity_here":true}"#,
    ]);
    assert!(ok, "deliver refused: {err}");
    assert!(out.contains("dedup key"), "an unidentifiable payload must say so: {out}");

    // Delivering to an unknown id is an error, not a silent no-op — a webhook
    // handler that swallows this drops real traffic.
    let (ok, _out, _err) = areev(&[
        "trigger", "deliver", "--db", db, "--ns", "ops", "--id", "deadbeef",
        "--payload", "{}",
    ]);
    assert!(!ok, "deliver to an unknown trigger must fail");

    // `--id` is required: without it there is no trigger to deliver to.
    let (ok, _out, err) =
        areev(&["trigger", "deliver", "--db", db, "--ns", "ops", "--payload", "{}"]);
    assert!(!ok, "deliver without --id must fail");
    assert!(err.contains("id"), "{err}");
}

// ---- the runner a firing gets (#90) ---------------------------------------

/// Author a one-node plan bound to a code-carrying Definition, plus a trigger
/// pointing at it, and return `(trigger_prefix, pinned_address)`.
///
/// Written in-process rather than through the binary because no CLI verb
/// authors a Workflow or a Tool Definition — `ADD` is limited to fact,
/// observation, goal and skill. The handle is dropped before the binary runs:
/// one memory, one writer.
fn seed_code_plan(db: &str) -> (String, String) {
    use areev_core::types::{Grain, Tool, ToolKind, Trigger, TriggerKind, Workflow};

    let mut m = areev_store::Areev::open(db).unwrap();
    // The blob is a real executable script, so a host that pins it genuinely
    // dispatches rather than only getting past the check.
    let uri = m.put_blob(b"#!/bin/sh\nread -r line\necho '{\"validated\":true}'\n").unwrap();
    let addr = uri.strip_prefix("cas://sha256:").unwrap().to_string();

    let def = Tool::new("validate_rows")
        .kind(ToolKind::Definition)
        .tool_description("a code-carrying tool")
        .executor_uri(&uri)
        .namespace("ops");
    let dh = m.add(&def).unwrap();
    let wf = Workflow::new(vec!["validate_rows".into()])
        .bind("validate_rows", &dh.to_hex())
        .namespace("ops");
    let plan = m.add(&wf).unwrap();
    let t = Trigger::new(TriggerKind::Interval, &plan.to_hex()).interval_secs(60).namespace("ops");
    let trigger = m.add(&t).unwrap().to_hex();
    drop(m);
    (trigger[..12].to_string(), addr)
}

#[cfg(unix)]
#[test]
fn a_firing_runs_a_code_carrying_node_when_the_host_pinned_it() {
    // #90: `trigger run` built its runner with a bare CommandExecutor, so the
    // executor pin and the sandbox command never reached it and passing the
    // flags changed nothing — the same plan ran fine from `run start`. The
    // trigger path silently supported a subset of the plans the host path ran,
    // on the one path nobody is watching.
    //
    // No `--tool-cmd` here on purpose: an all-Tier-C plan needs no subprocess,
    // and gating on one was the same reduction one level down — such a plan
    // was ingested, recorded as fired, and never started.
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();
    let cache = dir.path().join("execache");
    let (_trigger, addr) = seed_code_plan(db);

    let (ok, out, err) = areev(&[
        "trigger", "run", "--db", db, "--ns", "ops",
        "--allow-executor", &addr, "--executor-cache", cache.to_str().unwrap(),
        "--format", "json",
    ]);
    assert!(ok, "a pinned code node must run from a trigger: {out}{err}");
    assert!(out.contains("\"runs_started\":1"), "{out}");
    assert!(!out.contains("RUN-E018"), "{out}");
}

#[cfg(unix)]
#[test]
fn an_unpinned_code_carrying_node_refuses_and_names_the_surface_in_use() {
    // The fix widens what a firing CAN execute, never what it MAY: the
    // authorization to run code still comes from the host. What changed is
    // the message, which used to name `--allow-executor (CLI)`, `Python/Node`
    // and `areev serve` — three surfaces, none of them the one the operator
    // was actually using.
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();
    let (_trigger, _addr) = seed_code_plan(db);

    let (ok, out, err) = areev(&[
        "trigger", "run", "--db", db, "--ns", "ops", "--tool-cmd", "/bin/false",
        "--format", "json",
    ]);
    assert!(!ok, "an unpinned code node must refuse: {out}{err}");
    assert!(out.contains("RUN-E018"), "{out}{err}");
    assert!(out.contains("areev trigger"), "the refusal must name this surface: {out}");
}

#[cfg(unix)]
#[test]
fn the_pin_reaches_a_firing_through_the_environment_too() {
    // A heartbeat is a cron line, a launchd plist or a k8s CronJob, not an
    // interactive command — so the pin has to be settable out of band, which
    // is the second acceptance criterion on #90. `areev-mcp` already read
    // these variables; the CLI did not.
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();
    let cache = dir.path().join("execache");
    let (_trigger, addr) = seed_code_plan(db);

    let out = std::process::Command::new(env!("CARGO_BIN_EXE_areev"))
        .args(["trigger", "run", "--db", db, "--ns", "ops", "--format", "json"])
        .env("AREEV_RUN_TOOL_CMD", "/bin/false")
        .env("AREEV_RUN_ALLOW_EXECUTOR", &addr)
        .env("AREEV_RUN_EXECUTOR_CACHE", cache.to_str().unwrap())
        .output()
        .expect("spawn areev");
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(out.status.success(), "{stdout}{}", String::from_utf8_lossy(&out.stderr));
    assert!(stdout.contains("\"runs_started\":1"), "{stdout}");
}

/// A pinned code plan whose blob loops far longer than a short override
/// should tolerate — the fixture `--executor-timeout` (#133) needs to prove
/// it actually shortens the ceiling a firing runs its Tier-C node under.
///
/// The blob busy-loops on shell builtins (`:`, `while`) rather than calling
/// an external command like `sleep`: `--allow-executor` runs the blob by
/// exec'ing its own file directly, and only the DIRECT child is guaranteed
/// killed on timeout — a `sleep`-shaped script can fork a grandchild that
/// outlives it (`areev-core::proc`'s module doc tracks process-group kill as
/// separate, not-yet-done work), which would make this test's timing bound
/// on a gap this crate does not close.
fn seed_sleepy_code_plan(db: &str) -> (String, String) {
    use areev_core::types::{Grain, Tool, ToolKind, Trigger, TriggerKind, Workflow};

    let mut m = areev_store::Areev::open(db).unwrap();
    let uri = m.put_blob(b"#!/bin/sh\nwhile :; do :; done\n").unwrap();
    let addr = uri.strip_prefix("cas://sha256:").unwrap().to_string();

    let def = Tool::new("slow_validate")
        .kind(ToolKind::Definition)
        .tool_description("a code-carrying tool that never finishes in time")
        .executor_uri(&uri)
        .namespace("ops");
    let dh = m.add(&def).unwrap();
    let wf = Workflow::new(vec!["slow_validate".into()])
        .bind("slow_validate", &dh.to_hex())
        .namespace("ops");
    let plan = m.add(&wf).unwrap();
    let t = Trigger::new(TriggerKind::Interval, &plan.to_hex()).interval_secs(60).namespace("ops");
    let trigger = m.add(&t).unwrap().to_hex();
    drop(m);
    (trigger[..12].to_string(), addr)
}

/// #133: the fixed 300s ceiling on a pinned code executor had no host
/// override on any surface, including `trigger run`. A 1s
/// `--executor-timeout` on a blob that sleeps for 30s must return well
/// inside this test's own bound, not the old default.
#[cfg(unix)]
#[test]
fn executor_timeout_flag_reaches_a_firings_pinned_code_node() {
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();
    let cache = dir.path().join("execache");
    let (_trigger, addr) = seed_sleepy_code_plan(db);

    let started = std::time::Instant::now();
    let (ok, out, err) = areev(&[
        "trigger", "run", "--db", db, "--ns", "ops",
        "--allow-executor", &addr, "--executor-cache", cache.to_str().unwrap(),
        "--executor-timeout", "1", "--format", "json",
    ]);
    assert!(
        started.elapsed() < std::time::Duration::from_secs(20),
        "the override must apply — the 300s default would still be sleeping"
    );
    assert!(ok, "a firing that starts still exits cleanly even if its node timed out: {out}{err}");
    assert!(out.contains("\"runs_started\":1"), "{out}");
}

#[test]
fn a_connector_named_as_a_grain_is_stored_and_refused_until_the_host_pins_it() {
    // The CLI half of #185: `--connector-tool` reaches the grain, and
    // `trigger run` refuses that code until this host pinned its address. The
    // evaluator's own resolve/pin/dispatch rules are pinned next door, in
    // areev-trigger's `grain_connector_tests`.
    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    // 1. the flag reaches the declaration. `trigger add` does not resolve the
    //    reference (that is TRG-E012 at firing time), so a literal is honest
    //    here — the same shape `--workflow` takes.
    let def_literal = "d".repeat(64);
    let (ok, out, err) = areev(&[
        "trigger", "add", "--db", db, "--ns", "ops", "--type", "polling",
        "--workflow", WF, "--observer", "mailbox", "--connector-tool", &def_literal,
        "--interval", "60", "--dedup-key", "/id",
        "--because", "the desk watches this mailbox, and its code is in the memory",
    ]);
    assert!(ok, "declaring a grain connector failed: {err}");
    let trigger = out.trim().rsplit(' ').next().unwrap().to_string();
    let (ok, shown, err) = areev(&["trigger", "show", &trigger, "--db", db, "--ns", "ops"]);
    assert!(ok, "{err}");
    assert!(shown.contains(&def_literal), "the declaration lost the reference: {shown}");

    // 2. and running it refuses by name. A pack is how a code-carrying
    //    Definition gets into a memory from the CLI, so this is also the
    //    deployment path end to end.
    let pack = dir.path().join("pack");
    std::fs::create_dir_all(pack.join("grains")).unwrap();
    std::fs::create_dir_all(pack.join("blobs")).unwrap();
    std::fs::write(pack.join("blobs/poll.sh"), b"#!/bin/sh\necho '{\"items\":[]}'\n").unwrap();
    std::fs::write(
        pack.join("grains/010-tool.json"),
        r#"{ "id": "poll", "type": "tool", "tool_name": "mailbox.poll",
             "kind": "definition", "tool_description": "read the mailbox",
             "created_at": 1788134400000, "executor_uri": "blob:poll" }"#,
    )
    .unwrap();
    std::fs::write(
        pack.join("grains/020-workflow.json"),
        r#"{ "id": "plan", "type": "workflow", "name": "triage", "nodes": ["triage"],
             "created_at": 1788134400000 }"#,
    )
    .unwrap();
    std::fs::write(
        pack.join("grains/030-trigger.json"),
        r#"{ "id": "watch", "type": "trigger", "kind": "polling", "workflow": "grain:plan",
             "connector": "mailbox", "connector_tool": "grain:poll", "interval_secs": 60,
             "dedup_key": ["/id"], "created_at": 1788134400000,
             "because": "the desk watches this mailbox" }"#,
    )
    .unwrap();
    std::fs::write(
        pack.join("pack.json"),
        r#"{ "pack": "mailbox", "version": "1.0.0", "namespace": "watch",
             "blobs": { "poll": "blobs/poll.sh" },
             "grains": [ "grains/010-tool.json", "grains/020-workflow.json",
                         "grains/030-trigger.json" ] }"#,
    )
    .unwrap();
    // Each case gets its own memory: a refused poll BACKS OFF (a stale pin is
    // treated like a connector failure), so a second attempt in the same
    // memory would be skipped as not-due rather than refused again.
    let install = |name: &str| -> (String, String) {
        let db = dir.path().join(name).to_str().unwrap().to_string();
        let (ok, installed, err) =
            areev(&["pack", "install", pack.to_str().unwrap(), "--db", &db, "--format", "json"]);
        assert!(ok, "pack install failed: {err}");
        let pin = serde_json::from_str::<serde_json::Value>(&installed).unwrap()
            ["allow_executor"][0]
            .as_str()
            .expect("a pinnable address")
            .to_string();
        (db, pin)
    };
    let (db, pin) = install("unpinned.db");
    let db = db.as_str();

    // A host that pinned NOTHING is refused for being that host, and told
    // which flag it is missing.
    let (_ok, out, err) =
        areev(&["trigger", "run", "--db", db, "--ns", "watch", "--format", "json"]);
    let both = format!("{out}{err}");
    assert!(both.contains("TRG-E012"), "expected a TRG-E012 refusal: {both}");
    assert!(both.contains("--allow-executor"), "the refusal must name the fix: {both}");

    // A host that pinned something ELSE is refused with the address to add,
    // so pinning it is a copy-paste rather than a hunt.
    let (wrong_db, _) = install("wrongpin.db");
    let (_ok, out, err) = areev(&[
        "trigger", "run", "--db", &wrong_db, "--ns", "watch", "--allow-executor",
        &"a".repeat(64), "--format", "json",
    ]);
    let both = format!("{out}{err}");
    assert!(both.contains("TRG-E012"), "{both}");
    assert!(both.contains(&pin), "the refusal must name the address to pin: {both}");

    // With the pin, the same poll runs the code the declaration names.
    //
    // Unix only, and not because of a test harness quirk: a pinned NATIVE blob
    // is a program, so it is platform-specific by construction — `docs/run.md`
    // says to pin per platform — and this pack carries a `#!/bin/sh` one.
    // Everything above is the part that must hold everywhere: the flag reaches
    // the declaration, and an unpinned or wrongly-pinned host refuses. What is
    // skipped here is only "and then it ran", which `grain_connector_tests`
    // pins the same way.
    #[cfg(unix)]
    {
        let (pinned_db, pin) = install("pinned.db");
        let (ok, out, err) = areev(&[
            "trigger", "run", "--db", &pinned_db, "--ns", "watch", "--allow-executor", &pin,
            "--format", "json",
        ]);
        assert!(ok, "a pinned connector must run: {err} {out}");
        assert!(!out.contains("TRG-E012"), "{out}");
        assert!(out.contains("\"claimed\": 1") || out.contains("\"claimed\":1"), "{out}");
    }
}

/// Editing a plan is a supersession that mints a NEW address, and a trigger
/// keeps starting the version it was declared against — correctly, because a
/// plan edit must not silently change what an unattended heartbeat runs. What
/// was missing is that nothing SAID so: a listing showed a trigger on a
/// three-edits-old plan exactly as it showed a current one, and the run
/// succeeded either way. Silence is the symptom of every trigger failure; it
/// must not also be the symptom of a correct default.
#[test]
fn an_edited_plan_is_reported_and_re_pointed_in_one_command() {
    use areev_core::types::{Grain, Tool, ToolKind, Trigger, TriggerKind, Workflow};

    let dir = TempDir::new().unwrap();
    let db = dir.path().join("t.db");
    let db = db.to_str().unwrap();

    let (v1, v2, trigger) = {
        let mut m = areev_store::Areev::open(db).unwrap();
        let def = Tool::new("greet").kind(ToolKind::Definition).namespace("ops");
        let dh = m.add(&def).unwrap();
        let wf = Workflow::new(vec!["greet".into()])
            .bind("greet", &dh.to_hex())
            .created_at(500)
            .namespace("ops");
        let v1 = m.add(&wf).unwrap().to_hex();
        let t = Trigger::new(TriggerKind::Interval, &v1).interval_secs(60).namespace("ops");
        let trigger = m.add(&t).unwrap().to_hex();

        // The edit.
        let mut next = Workflow::new(vec!["greet".into()])
            .bind("greet", &dh.to_hex())
            .created_at(600)
            .namespace("ops");
        let v1h = areev_core::error::Hash::from_hex(&v1).unwrap();
        let v2 = m.supersede(&v1h, &mut next).unwrap().to_hex();
        (v1, v2, trigger)
    };
    assert_ne!(v1, v2, "an edit mints a new address — that is the whole problem");

    // The listing names the drift AND the fix, in both renderings.
    let (ok, out, err) = areev(&["trigger", "list", "--db", db, "--ns", "ops"]);
    assert!(ok, "{err}");
    assert!(out.contains("plan superseded"), "the drift must be visible: {out}");
    assert!(out.contains(&v2[..12]), "…naming the plan to move to: {out}");
    assert!(err.contains("retarget"), "…and the command that moves it: {err}");

    let (ok, out, _) = areev(&["trigger", "list", "--db", db, "--ns", "ops", "--format", "json"]);
    assert!(ok);
    let rows: serde_json::Value = serde_json::from_str(&out).unwrap();
    let row = &rows["triggers"][0];
    assert_eq!(row["plan_superseded"], true, "{rows}");
    assert_eq!(row["plan_head"], serde_json::json!(v2), "{rows}");

    // Re-point with NO --workflow: the trigger follows its own plan's chain.
    // That is the ergonomic point — the operator's intent is almost always
    // "use the edited plan", and it should not require pasting a hash.
    let (ok, out, err) = areev(&[
        "trigger", "retarget", &trigger[..12], "--db", db, "--ns", "ops",
        "--because", "plan edited",
    ]);
    assert!(ok, "{err}");
    assert!(out.contains(&v2[..12]), "{out}");

    // Clean afterwards, and pointing at the edited plan.
    let (_, out, _) = areev(&["trigger", "list", "--db", db, "--ns", "ops", "--format", "json"]);
    let rows: serde_json::Value = serde_json::from_str(&out).unwrap();
    let row = &rows["triggers"][0];
    assert_eq!(row["workflow"], serde_json::json!(v2));
    assert!(row.get("plan_superseded").is_none(), "no longer stale: {rows}");

    // The evaluation state survives, because a supersession keeps the chain
    // ROOT that cursors and dedup fences are keyed on (#128). Without this a
    // re-pointed polling trigger would replay its whole history.
    let (_, out, _) = areev(&["trigger", "status", "--db", db, "--ns", "ops", "--format", "json"]);
    let st: serde_json::Value = serde_json::from_str(&out).unwrap();
    let row = &st["triggers"][0];
    assert_ne!(row["trigger"], serde_json::json!(trigger), "the trigger grain was superseded");
    assert_eq!(
        row["chain_root"],
        serde_json::json!(trigger),
        "…but state still keys on the original root: {st}"
    );

    // Re-pointing a trigger that is already current is refused, not a no-op
    // supersession: an audited write with nothing to change is noise in the
    // chain every later reader has to walk.
    let (ok, _, err) = areev(&[
        "trigger", "retarget", &row["trigger"].as_str().unwrap()[..12],
        "--db", db, "--ns", "ops", "--because", "again",
    ]);
    assert!(!ok, "a no-op retarget must be refused");
    assert!(err.contains("already points at the current plan"), "{err}");
}