sloop-daemon 0.5.0

#![allow(dead_code)]

use std::cell::RefCell;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde_json::Value;
use tempfile::TempDir;

pub struct FakeAgent {
    moves: Vec<FakeAgentMove>,
}

enum FakeAgentMove {
    BlockUntilReleased(String),
    Commit(String),
    Exit(i32),
    Note(String),
    Output(String),
    Sleep(Duration),
}

impl FakeAgent {
    pub fn new() -> Self {
        Self { moves: Vec::new() }
    }

    pub fn block_until_released(mut self, marker: &str) -> Self {
        self.moves
            .push(FakeAgentMove::BlockUntilReleased(marker.to_owned()));
        self
    }

    pub fn commit(mut self, message: &str) -> Self {
        self.moves.push(FakeAgentMove::Commit(message.to_owned()));
        self
    }

    pub fn exit(mut self, code: i32) -> Self {
        self.moves.push(FakeAgentMove::Exit(code));
        self
    }

    pub fn note(mut self, text: &str) -> Self {
        self.moves.push(FakeAgentMove::Note(text.to_owned()));
        self
    }

    pub fn output(mut self, text: &str) -> Self {
        self.moves.push(FakeAgentMove::Output(text.to_owned()));
        self
    }

    pub fn sleep(mut self, duration: Duration) -> Self {
        self.moves.push(FakeAgentMove::Sleep(duration));
        self
    }
}

pub struct World {
    root: TempDir,
    clock: TempDir,
    state: TempDir,
    runtime: TempDir,
    daemon_pids: RefCell<Vec<u32>>,
}

impl World {
    pub fn new() -> Self {
        let root = tempfile::tempdir().expect("create test directory");
        let status = Command::new("git")
            .args(["init", "--quiet"])
            .arg(root.path())
            .status()
            .expect("run git init");
        assert!(status.success(), "git init failed with {status}");
        let clock = tempfile::tempdir().expect("create test clock directory");
        let state = tempfile::tempdir().expect("create test state directory");
        let runtime = tempfile::tempdir().expect("create test runtime directory");
        let now_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system time is after the epoch")
            .as_millis() as i64;
        fs::write(clock.path().join("now_ms"), now_ms.to_string()).expect("initialize test clock");

        Self {
            root,
            clock,
            state,
            runtime,
            daemon_pids: RefCell::new(Vec::new()),
        }
    }

    pub fn configured() -> Self {
        let world = Self::new();
        let config_dir = world.root().join(".agents/sloop");
        fs::create_dir_all(&config_dir).expect("create Sloop config directory");
        fs::write(
            config_dir.join("config.yaml"),
            "version: 1\nscheduler:\n  max_parallel_tasks: 1\n",
        )
        .expect("write Sloop config");
        fs::create_dir(config_dir.join("projects")).expect("create project directory");
        fs::write(
            config_dir.join("projects/default.md"),
            "---\nid: default\ntitle: Default\n---\nTickets not assigned to another project.\n",
        )
        .expect("write default project");
        fs::create_dir(config_dir.join("tickets")).expect("create ticket directory");
        world
    }

    pub fn configure_fake_agent(&self, agent: FakeAgent) {
        self.configure_fake_agent_with_parallelism(agent, 1);
    }

    pub fn configure_fake_agent_with_parallelism(
        &self,
        agent: FakeAgent,
        max_parallel_tasks: usize,
    ) {
        self.configure_fake_agent_with_scheduler(agent, max_parallel_tasks, None, None);
    }

    pub fn configure_fake_agent_with_stall_report_after(
        &self,
        agent: FakeAgent,
        stall_report_after: &str,
    ) {
        self.configure_fake_agent_with_scheduler(agent, 1, Some(stall_report_after), None);
    }

    pub fn configure_fake_agent_with_stall_thresholds(
        &self,
        agent: FakeAgent,
        stall_report_after: &str,
        stall_after: &str,
    ) {
        self.configure_fake_agent_with_scheduler(
            agent,
            1,
            Some(stall_report_after),
            Some(stall_after),
        );
    }

    fn configure_fake_agent_with_scheduler(
        &self,
        agent: FakeAgent,
        max_parallel_tasks: usize,
        stall_report_after: Option<&str>,
        stall_after: Option<&str>,
    ) {
        let flow_directory = self.root().join(".agents/sloop/flows");
        fs::create_dir_all(&flow_directory).expect("create flow directory");
        fs::write(
            flow_directory.join("default.yaml"),
            "stages:\n  - { name: build, action: agent, result_check: { exec: ['true'] } }\n  - { name: merge, action: { builtin: merge } }\n",
        )
        .expect("write default test flow");
        let script = self.root().join("fake-agent.sh");
        let mut body = String::from("#!/bin/sh\nset -eu\n");

        for agent_move in agent.moves {
            match agent_move {
                FakeAgentMove::BlockUntilReleased(marker) => {
                    let reached = self.fake_agent_marker_path(&marker, "reached");
                    let release = self.fake_agent_marker_path(&marker, "release");
                    for path in [&reached, &release] {
                        fs::create_dir_all(path.parent().expect("fake-agent marker parent"))
                            .expect("create fake-agent marker directory");
                        let _ = fs::remove_file(path);
                    }
                    body.push_str(&format!(
                        ": > {reached}\nwhile [ ! -e {release} ]; do sleep 0.01; done\n",
                        reached = shell_quote(&reached.to_string_lossy()),
                        release = shell_quote(&release.to_string_lossy()),
                    ));
                }
                FakeAgentMove::Commit(message) => body.push_str(&format!(
                    "git -c user.name=sloop-test-agent -c user.email=sloop-test-agent@example.invalid commit --quiet --allow-empty -m {}\n",
                    shell_quote(&message),
                )),
                FakeAgentMove::Exit(code) => body.push_str(&format!("exit {code}\n")),
                FakeAgentMove::Note(text) => body.push_str(&format!(
                    "{} --json note {} >/dev/null\n",
                    shell_quote(env!("CARGO_BIN_EXE_sloop")),
                    shell_quote(&text),
                )),
                FakeAgentMove::Output(text) => body.push_str(&format!(
                    "printf %s {}\n",
                    shell_quote(&text),
                )),
                FakeAgentMove::Sleep(duration) => {
                    body.push_str(&format!("sleep {}\n", duration.as_secs_f64()));
                }
            }
        }
        fs::write(&script, body).expect("write fake-agent script");

        fs::write(
            self.root().join(".agents/sloop/config.yaml"),
            format!(
                "version: 1\nscheduler:\n  max_parallel_tasks: {max_parallel_tasks}\n{}{}agent:\n  default_target: fake\n  targets:\n    fake:\n      cmd: [\"sh\", {}, \"{{prompt}}\"]\n",
                stall_report_after.map_or_else(String::new, |duration| format!("  stall_report_after: {duration}\n")),
                stall_after.map_or_else(String::new, |duration| format!("  stall_after: {duration}\n")),
                serde_json::to_string(&script.to_string_lossy()).expect("serialize fake-agent path"),
            ),
        )
        .expect("write fake-agent config");
    }

    pub fn fake_agent_reached(&self, marker: &str) -> bool {
        self.fake_agent_marker_path(marker, "reached").is_file()
    }

    pub fn release(&self, marker: &str) {
        let path = self.fake_agent_marker_path(marker, "release");
        fs::create_dir_all(path.parent().expect("fake-agent release parent"))
            .expect("create fake-agent release directory");
        fs::write(path, b"").expect("release fake agent");
    }

    fn fake_agent_marker_path(&self, marker: &str, state: &str) -> PathBuf {
        self.state_dir()
            .join("fake-agent")
            .join(format!("{marker}.{state}"))
    }

    pub fn root(&self) -> &Path {
        self.root.path()
    }

    pub fn state_dir(&self) -> PathBuf {
        self.state
            .path()
            .join("sloop/repositories")
            .join(self.repository_key())
    }

    pub fn runtime_dir(&self) -> PathBuf {
        let key = self.repository_key();
        let hash = key.rsplit_once('-').expect("repository key has hash").1;
        self.runtime.path().join("sloop").join(hash)
    }

    fn repository_key(&self) -> String {
        let root = self
            .root()
            .canonicalize()
            .expect("canonicalize test repository");
        sloop::paths::repository_key(&root)
    }

    pub fn operator_socket(&self) -> PathBuf {
        self.runtime_dir().join("operator.sock")
    }

    pub fn lock_path(&self) -> PathBuf {
        self.state_dir().join("daemon.lock")
    }

    pub fn daemon_log(&self) -> PathBuf {
        self.state_dir().join("logs/daemon.ndjson")
    }

    pub fn db_path(&self) -> PathBuf {
        self.state_dir().join("sloop.db")
    }

    pub fn worker_socket(&self, run: &str) -> PathBuf {
        // Mirrors `worker_socket_path`: short id, directly in the runtime
        // directory, so the path fits the 104-byte macOS socket cap.
        let short = run.get(..8).unwrap_or(run);
        self.runtime_dir().join(format!("w{short}.sock"))
    }

    pub fn now_ms(&self) -> i64 {
        fs::read_to_string(self.clock.path().join("now_ms"))
            .expect("read test clock")
            .trim()
            .parse()
            .expect("test clock is an integer")
    }

    pub fn tick(&self, duration: Duration) {
        let now_ms = self.now_ms();
        fs::write(
            self.clock.path().join("now_ms"),
            (now_ms + duration.as_millis() as i64).to_string(),
        )
        .expect("advance test clock");
    }

    /// Runs sloop with `--json`, as agents and scripts should: envelopes on
    /// stdout/stderr. Tests parse them with `json_stdout`.
    pub fn sloop(&self, args: &[&str]) -> Output {
        self.sloop_command(&Self::with_json(args))
            .output()
            .expect("run sloop")
    }

    pub fn spawn_sloop(&self, args: &[&str]) -> std::process::Child {
        self.sloop_command(&Self::with_json(args))
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .expect("spawn sloop")
    }

    pub fn sloop_in(&self, directory: &Path, args: &[&str]) -> Output {
        self.sloop_command_in(directory, &Self::with_json(args))
            .output()
            .expect("run sloop")
    }

    pub fn sloop_with_runtime(&self, args: &[&str], runtime: &Path) -> Output {
        self.sloop_command(&Self::with_json(args))
            .env("XDG_RUNTIME_DIR", runtime)
            .output()
            .expect("run sloop with alternate runtime")
    }

    pub fn sloop_with_binary(&self, binary: &Path, args: &[&str]) -> Output {
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        loop {
            match self
                .sloop_command_with_binary(binary, self.root(), &Self::with_json(args))
                .output()
            {
                Ok(output) => return output,
                Err(error)
                    if error.raw_os_error() == Some(libc::ETXTBSY)
                        && std::time::Instant::now() < deadline =>
                {
                    thread::sleep(Duration::from_millis(10));
                }
                Err(error) => panic!("run alternate sloop binary: {error}"),
            }
        }
    }

    /// Runs sloop without `--json`: the human-readable default output.
    pub fn sloop_plain(&self, args: &[&str]) -> Output {
        self.sloop_command(args).output().expect("run sloop")
    }

    fn with_json<'a>(args: &[&'a str]) -> Vec<&'a str> {
        // Prepended so verbs with trailing arguments (`note`) cannot
        // swallow the flag.
        let mut with_flag = Vec::with_capacity(args.len() + 1);
        with_flag.push("--json");
        with_flag.extend_from_slice(args);
        with_flag
    }

    pub fn start_daemon(&self) -> Value {
        let output = self.sloop(&["daemon"]);
        assert!(
            output.status.success(),
            "daemon failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        let response = Self::json_stdout(&output);
        let pid = response["data"]["pid"].as_u64().expect("daemon pid") as u32;
        let mut daemon_pids = self.daemon_pids.borrow_mut();
        if !daemon_pids.contains(&pid) {
            daemon_pids.push(pid);
        }
        drop(daemon_pids);
        response
    }

    pub fn start_daemon_with_binary(&self, binary: &Path) -> Value {
        let output = self.sloop_with_binary(binary, &["daemon"]);
        assert!(
            output.status.success(),
            "daemon failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        let response = Self::json_stdout(&output);
        let pid = response["data"]["pid"].as_u64().expect("daemon pid") as u32;
        let mut daemon_pids = self.daemon_pids.borrow_mut();
        if !daemon_pids.contains(&pid) {
            daemon_pids.push(pid);
        }
        drop(daemon_pids);
        response
    }

    pub fn kill_daemon(&self, pid: u32) {
        let status = Command::new("kill")
            .args(["-9", &pid.to_string()])
            .status()
            .expect("kill daemon");
        assert!(status.success(), "kill failed with {status}");
        wait_until("the crashed daemon exits", || !process_alive(pid));
    }

    pub fn kill_process_group(&self, leader: u32) {
        let status = Command::new("kill")
            .args(["-9", "--", &format!("-{leader}")])
            .status()
            .expect("kill process group");
        assert!(status.success(), "kill process group failed with {status}");
        wait_until("the process group leader exits", || !process_alive(leader));
    }

    /// The internal id of the `position`-th run the daemon claimed, counting
    /// from 1. Ids are minted randomly, so tests name runs by the order they
    /// were created rather than by a predictable literal.
    ///
    /// Returns a placeholder when that run does not exist yet. Polling helpers
    /// call this before a claim lands, and a path built from the placeholder
    /// simply does not exist — which is the answer such a poll wants.
    pub fn run_id(&self, position: usize) -> String {
        self.run_ids()
            .into_iter()
            .nth(position - 1)
            .unwrap_or_else(|| "pending".into())
    }

    /// The `<ticket>-r<attempt>` alias of the `position`-th run.
    pub fn run_alias(&self, position: usize) -> String {
        let connection = rusqlite::Connection::open(self.db_path()).expect("open state database");
        connection
            .query_row(
                "SELECT ticket_id || '-r' || attempt FROM runs WHERE id = ?1",
                [self.run_id(position)],
                |row| row.get(0),
            )
            .unwrap_or_else(|_| "pending".into())
    }

    /// Every run id in claim order.
    pub fn run_ids(&self) -> Vec<String> {
        let Ok(connection) = rusqlite::Connection::open(self.db_path()) else {
            return Vec::new();
        };
        let Ok(mut statement) =
            connection.prepare("SELECT id FROM runs ORDER BY created_at_ms, rowid")
        else {
            return Vec::new();
        };
        let Ok(rows) = statement.query_map([], |row| row.get::<_, String>(0)) else {
            return Vec::new();
        };
        rows.filter_map(Result::ok).collect()
    }

    /// The worktree directory the `position`-th run was given. Worktrees are
    /// named by the short form of the internal id.
    pub fn run_worktree(&self, position: usize) -> PathBuf {
        let id = self.run_id(position);
        self.root()
            .join(".worktrees")
            .join(id.get(..8).unwrap_or(&id))
    }

    pub fn run_process_id(&self, run_id: &str) -> u32 {
        let connection = rusqlite::Connection::open(self.db_path()).expect("open state database");
        let pid: i64 = connection
            .query_row("SELECT pid FROM runs WHERE id = ?1", [run_id], |row| {
                row.get(0)
            })
            .expect("read run pid");
        u32::try_from(pid).expect("run pid fits u32")
    }

    /// The run's lease expiry, or `None` once the lease row is gone —
    /// settlement deletes it.
    pub fn lease_expires_at_ms(&self, run_id: &str) -> Option<i64> {
        let connection = rusqlite::Connection::open(self.db_path()).expect("open state database");
        connection
            .query_row(
                "SELECT expires_at_ms FROM leases WHERE run_id = ?1",
                [run_id],
                |row| row.get(0),
            )
            .ok()
    }

    pub fn run_state(&self, run_id: &str) -> String {
        let connection = rusqlite::Connection::open(self.db_path()).expect("open state database");
        connection
            .query_row("SELECT state FROM runs WHERE id = ?1", [run_id], |row| {
                row.get(0)
            })
            .expect("read run state")
    }

    pub fn run_note_count(&self, run_id: &str) -> i64 {
        let connection = rusqlite::Connection::open(self.db_path()).expect("open state database");
        connection
            .query_row(
                "SELECT COUNT(*) FROM notes WHERE run_id = ?1",
                [run_id],
                |row| row.get(0),
            )
            .expect("count run notes")
    }

    pub fn run_evidence(&self, run_id: &str, kind: &str) -> Option<Value> {
        let connection = rusqlite::Connection::open(self.db_path()).expect("open state database");
        let mut statement = connection
            .prepare("SELECT data_json FROM run_evidence WHERE run_id = ?1 AND kind = ?2")
            .expect("prepare evidence query");
        let mut rows = statement.query([run_id, kind]).expect("query run evidence");
        rows.next().expect("read run evidence").map(|row| {
            let data: String = row.get(0).expect("read evidence JSON");
            serde_json::from_str(&data).expect("evidence is JSON")
        })
    }

    pub fn arm_test_hook(&self, name: &str) {
        let directory = self.state.path().join("test-hooks");
        fs::create_dir_all(&directory).expect("create test hook directory");
        for state in ["reached", "release"] {
            let _ = fs::remove_file(directory.join(format!("{name}.{state}")));
        }
        fs::write(directory.join(format!("{name}.armed")), b"").expect("arm test hook");
    }

    pub fn test_hook_reached(&self, name: &str) -> bool {
        self.state
            .path()
            .join("test-hooks")
            .join(format!("{name}.reached"))
            .is_file()
    }

    pub fn release_test_hook(&self, name: &str) {
        fs::write(
            self.state
                .path()
                .join("test-hooks")
                .join(format!("{name}.release")),
            b"",
        )
        .expect("release test hook");
    }

    pub fn operator_exchange(&self, request: &str) -> Value {
        Self::socket_exchange(&self.operator_socket(), request)
    }

    pub fn wait_snapshot(&self, run: &str) -> Value {
        self.operator_exchange(
            &serde_json::json!({
                "v": 1,
                "id": "req-wait-snapshot",
                "verb": "wait",
                "args": {"run": run},
                "token": null,
            })
            .to_string(),
        )
    }

    pub fn show_snapshot(&self, reference: &str) -> Value {
        World::json_stdout(&self.sloop(&["show", reference]))["data"]["value"].clone()
    }

    /// Sends one raw envelope line over a Unix socket and returns the reply.
    /// Lets tests speak to per-run worker sockets directly.
    pub fn socket_exchange(socket: &Path, request: &str) -> Value {
        let mut stream = UnixStream::connect(socket).expect("connect to socket");
        stream.write_all(request.as_bytes()).expect("write request");
        stream.write_all(b"\n").expect("finish request");

        let mut response = String::new();
        BufReader::new(stream)
            .read_line(&mut response)
            .expect("read response");
        serde_json::from_str(response.trim_end()).expect("response is JSON")
    }

    pub fn write_ticket(&self, name: &str, body: &str) -> PathBuf {
        let relative = PathBuf::from(".agents/sloop/tickets").join(name);
        let title = name.strip_suffix(".md").unwrap_or(name).replace('-', " ");
        let content = if body.starts_with("---\n") {
            body.replacen("---\n", &format!("---\nname: {title}\nblocked_by: []\n"), 1)
        } else {
            format!("---\nname: {title}\nblocked_by: []\n---\n{body}")
        };
        fs::write(self.root().join(&relative), content).expect("write ticket");
        relative
    }

    /// Commits everything in the world's repository so worktrees have a HEAD
    /// to branch from.
    pub fn commit_all(&self, message: &str) {
        let status = Command::new("git")
            .args([
                "-c",
                "user.name=sloop-test",
                "-c",
                "user.email=sloop-test@example.invalid",
                "add",
                "-A",
            ])
            .current_dir(self.root())
            .status()
            .expect("run git add");
        assert!(status.success(), "git add failed with {status}");
        let status = Command::new("git")
            .args([
                "-c",
                "user.name=sloop-test",
                "-c",
                "user.email=sloop-test@example.invalid",
                "commit",
                "--quiet",
                "-m",
                message,
            ])
            .current_dir(self.root())
            .status()
            .expect("run git commit");
        assert!(status.success(), "git commit failed with {status}");
    }

    pub fn json_stdout(output: &Output) -> Value {
        serde_json::from_slice(&output.stdout).unwrap_or_else(|error| {
            panic!(
                "stdout is JSON: {error}; status={}; stdout={}; stderr={}",
                output.status,
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr),
            )
        })
    }

    /// Parses whichever stream carried the envelope; errors land on stderr.
    pub fn json_stdout_or_stderr(output: &Output) -> Value {
        if output.stdout.is_empty() {
            serde_json::from_slice(&output.stderr).expect("stderr is JSON")
        } else {
            serde_json::from_slice(&output.stdout).expect("stdout is JSON")
        }
    }

    pub fn worker_exchange(&self, args: &[&str], response: Value) -> (Output, Value) {
        let socket = self.root().join("worker.sock");
        let listener = UnixListener::bind(&socket).expect("bind worker socket");
        let server = thread::spawn(move || {
            let (stream, _) = listener.accept().expect("accept worker request");
            let mut request = String::new();
            BufReader::new(&stream)
                .read_line(&mut request)
                .expect("read worker request");

            let mut stream = stream;
            if serde_json::to_writer(&mut stream, &response).is_ok() {
                let _ = stream.write_all(b"\n");
            }

            serde_json::from_str(request.trim()).expect("worker request is JSON")
        });

        let output = self
            .sloop_command(args)
            .env("SLOOP_SOCKET", &socket)
            .env("SLOOP_TOKEN", "test-worker-token")
            .output()
            .expect("run worker command");

        // Unblock the fixture when an unimplemented client never opened the socket.
        if let Ok(mut stream) = UnixStream::connect(&socket) {
            let _ = stream.write_all(b"{}\n");
        }

        let request = server.join().expect("worker server thread");
        (output, request)
    }

    fn sloop_command(&self, args: &[&str]) -> Command {
        self.sloop_command_in(self.root(), args)
    }

    fn sloop_command_in(&self, directory: &Path, args: &[&str]) -> Command {
        self.sloop_command_with_binary(Path::new(env!("CARGO_BIN_EXE_sloop")), directory, args)
    }

    fn sloop_command_with_binary(&self, binary: &Path, directory: &Path, args: &[&str]) -> Command {
        let mut command = Command::new(binary);
        command
            .args(args)
            .current_dir(directory)
            .env("HOME", self.root().join("home"))
            .env("XDG_STATE_HOME", self.state.path())
            .env("XDG_RUNTIME_DIR", self.runtime.path())
            .env_remove("SLOOP_SOCKET")
            .env_remove("SLOOP_TOKEN")
            .env("SLOOP_TEST_CLOCK_PATH", self.clock.path().join("now_ms"))
            .env("SLOOP_TEST_HOOK_DIR", self.state.path().join("test-hooks"));
        command
    }
}

fn shell_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', "'\"'\"'"))
}

/// Whether a PID currently exists (signal 0 probe).
pub fn process_alive(pid: u32) -> bool {
    Command::new("kill")
        .args(["-0", &pid.to_string()])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}

/// Polls an observable condition until it holds or a deadline passes. Tests
/// must wait on state, never sleep and hope.
pub fn wait_until(what: &str, mut condition: impl FnMut() -> bool) {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    while std::time::Instant::now() < deadline {
        if condition() {
            return;
        }
        thread::sleep(std::time::Duration::from_millis(25));
    }
    panic!("timed out waiting until {what}");
}

/// Like `wait_until`, with a 20-second deadline for probes on multi-second
/// timers such as the daemon liveness tick.
pub fn wait_until_slow(what: &str, mut condition: impl FnMut() -> bool) {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
    while std::time::Instant::now() < deadline {
        if condition() {
            return;
        }
        thread::sleep(std::time::Duration::from_millis(100));
    }
    panic!("timed out waiting until {what}");
}

/// Puts a database back into the shape it had before triggers were called
/// triggers: `activations`/`activation_filters`, `activation_id` on everything
/// that points at them, `A<ordinal>` ids, and the lease ownership token that
/// embeds one. Mirrors `REVERT_TRIGGER_RENAME` in `db::migrations`, which does
/// the same for the unit fixtures — keep the two in step.
///
/// Callers still have to set `PRAGMA user_version` themselves: which older
/// version they are impersonating is theirs to decide, and this only undoes the
/// one step.
///
/// The transaction is load-bearing. `defer_foreign_keys` lasts only to the end
/// of the enclosing transaction, and the rewrite points `runs.activation_id` at
/// ids that do not exist yet partway through.
pub fn revert_trigger_rename(world: &World) {
    let connection = rusqlite::Connection::open(world.db_path()).expect("open state database");
    connection
        .execute_batch(
            "BEGIN IMMEDIATE;
             PRAGMA defer_foreign_keys = ON;

             UPDATE leases
                SET owner_id = json_object(
                        'activation', 'A' || SUBSTR(json_extract(owner_id, '$.trigger'), 3),
                        'owner', json_extract(owner_id, '$.owner'))
              WHERE json_valid(owner_id)
                AND json_extract(owner_id, '$.trigger') GLOB 'TR[0-9]*';

             UPDATE trigger_filters SET trigger_id = 'A' || SUBSTR(trigger_id, 3)
              WHERE trigger_id GLOB 'TR[0-9]*';
             UPDATE runs SET trigger_id = 'A' || SUBSTR(trigger_id, 3)
              WHERE trigger_id GLOB 'TR[0-9]*';
             UPDATE triggers SET id = 'A' || SUBSTR(id, 3) WHERE id GLOB 'TR[0-9]*';

             DROP INDEX runs_by_trigger;

             ALTER TABLE runs RENAME COLUMN trigger_id TO activation_id;
             ALTER TABLE trigger_filters RENAME COLUMN trigger_id TO activation_id;
             ALTER TABLE trigger_filters RENAME TO activation_filters;
             ALTER TABLE triggers RENAME TO activations;

             CREATE INDEX runs_by_activation ON runs(activation_id, created_at_ms);

             UPDATE id_counters SET kind = 'activation' WHERE kind = 'trigger';
             COMMIT;",
        )
        .expect("plant the pre-rename trigger shape");
}

impl Drop for World {
    fn drop(&mut self) {
        // Layer 1: a clean stop through the public verb; never autostarts.
        let _ = self.sloop_command(&["--json", "stop", "--force"]).output();

        // Layer 2: identity-checked kill of whatever owns the lockfile,
        // catching daemons autostarted by arbitrary verbs during the test.
        let lock = self.lock_path();
        if let Some(identity) = sloop::daemon::read_lock_identity(&lock) {
            let cmdline = fs::read(format!("/proc/{}/cmdline", identity.pid)).unwrap_or_default();
            if String::from_utf8_lossy(&cmdline).contains("sloop") {
                let _ = Command::new("kill")
                    .args(["-9", &identity.pid.to_string()])
                    .status();
            }
        }

        // Layer 3: pids tests registered explicitly (kept as a backstop for
        // daemons whose lockfile was already deleted).
        for pid in self.daemon_pids.get_mut().drain(..) {
            let _ = Command::new("kill")
                .arg(pid.to_string())
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null())
                .status();
        }
    }
}