marver 0.0.27

//! Turning a queued task into a live agent.
//!
//! ```text
//! provision worktrees  →  prepare the harness  →  create session  →  start agent
//! ```
//!
//! The order is not arbitrary: the session's working directory must exist before
//! tmux is asked to start there, and an agent's settings file appearing after it
//! starts would be read too late. The session name is recorded last, once there
//! is something to record.
//!
//! Launching is all-or-nothing. A half-launched task — worktrees but no session,
//! or a session marver has not recorded — is worse than one that plainly failed,
//! because nothing will ever clean it up.

use std::path::PathBuf;

use chrono::{DateTime, Utc};

use crate::agent;
use crate::brief::{self, Brief};
use crate::domain::Task;
use crate::harness::Harness;
use crate::hook;
use crate::scheduler::Launch;
use crate::store::Store;
use crate::tmux::{self, Tmux};
use crate::worktree::{Branches, Teardown, WorktreeManager};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("could not prepare worktrees: {0}")]
    Worktree(#[from] crate::worktree::Error),
    #[error("could not write hook settings: {0}")]
    Hooks(#[from] hook::Error),
    #[error("could not prepare the agent: {0}")]
    Harness(#[from] crate::harness::Error),
    #[error("tmux: {0}")]
    Tmux(#[from] tmux::Error),
    #[error(transparent)]
    Store(#[from] crate::store::Error),
    #[error("session {0} already exists")]
    SessionExists(String),
    #[error("session {0} reported no panes")]
    NoPane(String),
}

pub type Result<T> = std::result::Result<T, Error>;

/// What a successful launch produced.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Launched {
    pub session: String,
    /// The settings file written for this task, for harnesses that use one.
    pub settings: Option<PathBuf>,
    pub worktrees: Vec<PathBuf>,
}

pub struct Launcher {
    tmux: Tmux,
    worktrees: WorktreeManager,
    /// Path to the marver binary, baked into generated hook commands.
    hook_bin: PathBuf,
    /// Socket the daemon listens on.
    hook_socket: PathBuf,
    harness: Harness,
    /// Which marver the sessions it creates belong to; `None` until told.
    session_prefix: Option<String>,
    size: (u16, u16),
}

impl Launcher {
    pub fn new(
        tmux: Tmux,
        worktrees: WorktreeManager,
        hook_bin: impl Into<PathBuf>,
        hook_socket: impl Into<PathBuf>,
    ) -> Self {
        Self {
            tmux,
            worktrees,
            hook_bin: hook_bin.into(),
            hook_socket: hook_socket.into(),
            harness: Harness::claude(),
            session_prefix: None,
            size: tmux::DEFAULT_SIZE,
        }
    }

    /// Which marver the sessions this creates belong to, so two data
    /// directories do not both try to name their task 1 `marver-1`.
    pub fn session_prefix(mut self, prefix: impl Into<String>) -> Self {
        self.session_prefix = Some(prefix.into());
        self
    }

    /// Which agent to run. Defaults to Claude Code; tests use it to avoid
    /// starting a real one.
    pub fn harness(mut self, harness: Harness) -> Self {
        self.harness = harness;
        self
    }

    pub fn size(mut self, size: (u16, u16)) -> Self {
        self.size = size;
        self
    }

    /// Bring a task to life.
    pub fn launch(&self, store: &mut Store, task: &Task, now: DateTime<Utc>) -> Result<Launched> {
        let session = tmux::session_name(self.session_prefix.as_deref(), task.id);
        if self.tmux.has_session(&session) {
            return Err(Error::SessionExists(session));
        }

        let provisioned = self.worktrees.provision(store, task)?;
        let worktrees: Vec<PathBuf> = provisioned
            .iter()
            .filter_map(|link| link.worktree_path.clone())
            .collect();

        // From here on any failure has something to undo.
        match self.finish(store, task, &session, now) {
            Ok(settings) => Ok(Launched {
                session,
                settings,
                worktrees,
            }),
            Err(err) => {
                self.unwind(store, task, &session);
                Err(err)
            }
        }
    }

    /// Everything after provisioning, so the caller has one place to unwind
    /// from.
    fn finish(
        &self,
        store: &mut Store,
        task: &Task,
        session: &str,
        now: DateTime<Utc>,
    ) -> Result<Option<PathBuf>> {
        // Whatever this harness needs on disk and on its command line, decided
        // in one place — see `harness`.
        let start = self.harness.prepare(
            &task.workspace_dir,
            task.id,
            &self.hook_bin,
            &self.hook_socket,
            &task.prompt,
        )?;

        // A `task.md` beside the worktrees, saying what this directory is for.
        if let Ok(brief) = Brief::read(store, task.id)
            && let Err(err) = brief::write(&brief, &task.workspace_dir)
        {
            eprintln!("marverd: task {}: could not write task.md: {err}", task.id);
        }

        // The agent is the session's own process, not something typed at a
        // shell prompt inside it.
        self.tmux
            .new_session_running(session, &task.workspace_dir, self.size, &start.argv)?;

        // Still checked: a session with no pane means the agent never started,
        // and the caller needs to unwind rather than record a live session.
        if self.tmux.list_panes(session)?.is_empty() {
            return Err(Error::NoPane(session.to_string()));
        }

        store.set_session_name(task.id, session, now)?;
        Ok(start.settings)
    }

    /// Undo a partial launch. Best effort: it runs while already failing.
    fn unwind(&self, store: &Store, task: &Task, session: &str) {
        let _ = self.tmux.kill_session(session);
        // Discard: nothing was committed to the branch this is unwinding, and
        // a launch that failed must leave no trace to trip the next attempt.
        let _ = self.worktrees.teardown(store, task, Branches::Discard);
        let _ = store.clear_worktrees(task.id);
    }

    /// Stop a task's agent and remove what the launch created.
    pub fn shut_down(&self, store: &Store, task: &Task, branches: Branches) -> Result<Teardown> {
        // Nothing recorded means nothing was ever started under this task, so
        // there is no session to kill — only worktrees to take away.
        if let Some(session) = task.session_name.clone()
            && self.owns_session(task, &session)
        {
            // A session that goes between the check and the kill is the
            // outcome this wanted.
            if let Err(err) = self.tmux.kill_session(&session)
                && self.tmux.has_session(&session)
            {
                return Err(err.into());
            }
        }
        let teardown = self.worktrees.teardown(store, task, branches)?;
        if teardown.failed.is_empty() {
            store.clear_worktrees(task.id)?;
        }
        Ok(teardown)
    }

    /// Whether `session` is this task's, rather than a namesake.
    fn owns_session(&self, task: &Task, session: &str) -> bool {
        agent::owns_session(&self.tmux, task, session)
    }
}

#[cfg(test)]
pub(crate) mod testing {
    use super::*;
    use std::path::Path;

    /// A script standing in for the agent: it prints the argv it was given,
    /// one element per line, then stays alive as a real agent would.
    pub fn stub_agent(dir: &Path) -> Harness {
        let path = dir.join("stub-agent");
        std::fs::write(
            &path,
            "#!/bin/sh\nfor a in \"$@\"; do echo \"ARG[$a]\"; done\nexec sleep 300\n",
        )
        .expect("write stub agent");
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
            .expect("make stub executable");
        // Claude's reporting, so the settings file is still written and the
        // tests that read it keep testing what they were written to test.
        Harness {
            name: "stub".into(),
            program: path.to_string_lossy().into_owned(),
            args: Vec::new(),
            report: crate::harness::Report::ClaudeHooks,
        }
    }
}

impl Launch for Launcher {
    fn launch(&self, store: &mut Store, task: &Task) -> std::result::Result<(), String> {
        // The scheduler records the moment; the clock is only needed for the
        // session-name write, which happens within this call.
        Launcher::launch(self, store, task, Utc::now())
            .map(|_| ())
            .map_err(|err| err.to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::{Repo, TaskState};
    use crate::git::testing::init_repo;
    use crate::tmux::testing::TestServer;
    use std::path::Path;
    use tempfile::TempDir;

    fn at(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(secs, 0).expect("valid timestamp")
    }

    struct Fixture {
        _tmp: TempDir,
        server: TestServer,
        repos_dir: PathBuf,
        store: Store,
        launcher: Launcher,
    }

    impl Fixture {
        fn new() -> Self {
            let tmp = TempDir::new().unwrap();
            let server = TestServer::new();
            let repos_dir = tmp.path().join("repos");
            std::fs::create_dir_all(&repos_dir).unwrap();

            let launcher = Launcher::new(
                server.tmux.clone(),
                WorktreeManager::new(tmp.path().join("tasks")),
                PathBuf::from("/usr/local/bin/marver"),
                tmp.path().join("hooks.sock"),
            )
            .harness(testing::stub_agent(tmp.path()));

            Self {
                repos_dir,
                store: Store::open_in_memory().unwrap(),
                launcher,
                server,
                _tmp: tmp,
            }
        }

        fn repo(&self, name: &str) -> Repo {
            let path = self.repos_dir.join(name);
            init_repo(&path, "main");
            self.store.upsert_repo(&path, name, at(0)).unwrap()
        }

        fn task(&mut self, title: &str, prompt: &str, repos: &[Repo]) -> Task {
            let root = self.launcher.worktrees.workspace_root().to_path_buf();
            let ids: Vec<i64> = repos.iter().map(|r| r.id).collect();
            self.store
                .create_task(title, prompt, &root, &ids, at(0))
                .unwrap()
        }

        /// Poll a pane until it shows `needle`, so tests do not race the
        /// shell.
        fn wait_for_pane(&self, session: &str, needle: &str) -> String {
            let mut seen = String::new();
            for _ in 0..60 {
                if let Ok(panes) = self.server.tmux.list_panes(session)
                    && let Some(pane) = panes.first()
                    && let Ok(text) = self.server.tmux.capture_pane(pane)
                {
                    seen = text;
                    if seen.contains(needle) {
                        return seen;
                    }
                }
                std::thread::sleep(std::time::Duration::from_millis(50));
            }
            seen
        }
    }

    #[test]
    fn the_prompt_is_one_argument_however_it_is_written() {
        let tmp = TempDir::new().unwrap();
        let start = Harness::claude()
            .prepare(
                tmp.path(),
                1,
                Path::new("/bin/marver"),
                Path::new("/s.sock"),
                "fix the auth flow",
            )
            .unwrap();
        let argv: Vec<String> = start
            .argv
            .iter()
            .map(|a| a.to_string_lossy().into_owned())
            .collect();

        assert_eq!(argv[0], "claude");
        assert_eq!(argv[1], "--settings");
        assert_eq!(argv[3], "fix the auth flow");
    }

    #[test]
    fn a_prompt_with_spaces_and_quotes_survives_to_the_agent() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task(
            "Quoting",
            "fix the user's login flow",
            std::slice::from_ref(&repo),
        );

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();
        let pane = fx.wait_for_pane(&launched.session, "ARG[--settings]");

        assert!(
            pane.contains("ARG[fix the user's login flow]"),
            "the prompt must arrive as one argument, intact; pane was {pane:?}"
        );
    }

    #[test]
    fn a_prompt_cannot_escape_into_the_shell() {
        // Shell metacharacters, and — the case quoting could never cover — the
        // control bytes that a paste from an issue body or a terminal log
        // carries.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let canary = fx.repos_dir.join("PWNED");
        let prompt = format!(
            "fix\u{3}touch {} $(touch {}) `touch {}` '; touch {}; echo '",
            canary.display(),
            canary.display(),
            canary.display(),
            canary.display()
        );
        let task = fx.task("Nasty", &prompt, std::slice::from_ref(&repo));

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();
        fx.wait_for_pane(&launched.session, "ARG[--settings]");

        assert!(
            !canary.exists(),
            "the prompt must never reach a shell as executable input"
        );
    }

    #[test]
    fn the_agent_is_the_sessions_own_process() {
        // Not a command typed at a shell prompt inside it.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Direct", "do it", std::slice::from_ref(&repo));

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();
        fx.wait_for_pane(&launched.session, "ARG[do it]");

        let cmd = fx
            .server
            .tmux
            .run(&[
                "list-panes",
                "-t",
                &format!("={}", launched.session),
                "-F",
                "#{pane_current_command}",
            ])
            .unwrap();
        assert!(
            !cmd.trim().ends_with("sh")
                && !cmd.trim().ends_with("zsh")
                && !cmd.trim().contains("bash"),
            "the pane must not be running a shell, got {cmd:?}"
        );
    }

    #[test]
    fn launching_creates_worktree_session_and_settings() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix it", std::slice::from_ref(&repo));

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();

        assert_eq!(launched.session, "marver-1");
        assert_eq!(launched.worktrees.len(), 1);
        assert!(launched.worktrees[0].join("README.md").exists());
        assert!(
            launched.settings.as_ref().is_some_and(|p| p.exists()),
            "hook settings must exist"
        );
        assert!(fx.server.tmux.has_session(&launched.session));

        // The session sits in the workspace, so the agent sees every worktree.
        let cwd = fx.server.tmux.session_cwd(&launched.session).unwrap();
        assert_eq!(
            std::fs::canonicalize(cwd).unwrap(),
            std::fs::canonicalize(&task.workspace_dir).unwrap()
        );
    }

    #[test]
    fn the_session_name_is_recorded_so_the_agent_can_be_found_again() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix it", std::slice::from_ref(&repo));

        fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();

        let stored = fx.store.get_task(task.id).unwrap();
        assert_eq!(stored.session_name.as_deref(), Some("marver-1"));
        assert_eq!(stored.updated_at, at(5));
    }

    #[test]
    fn a_task_md_is_left_in_the_workspace_for_the_agent_to_read() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix the auth flow", std::slice::from_ref(&repo));

        fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();

        let written =
            std::fs::read_to_string(task.workspace_dir.join(crate::brief::FILE_NAME)).unwrap();
        assert!(written.contains("# Fix auth"), "{written}");
        assert!(written.contains("fix the auth flow"), "{written}");
        assert!(
            written.contains("- Branch: `marver/"),
            "written after provisioning, so it knows the branch: {written}"
        );
    }

    #[test]
    fn the_agent_command_is_actually_run() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix the thing", std::slice::from_ref(&repo));

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();
        let pane = fx.wait_for_pane(&launched.session, "ARG[--settings]");

        assert!(
            pane.contains("ARG[--settings]"),
            "the command must actually run; pane was {pane:?}"
        );
        assert!(
            pane.contains("ARG[fix the thing]"),
            "the prompt must reach the agent; pane was {pane:?}"
        );
    }

    #[test]
    fn hook_settings_carry_this_task_id() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix it", std::slice::from_ref(&repo));

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();
        let settings: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(launched.settings.unwrap()).unwrap())
                .unwrap();

        assert_eq!(
            settings["hooks"]["Stop"][0]["hooks"][0]["args"][2],
            task.id.to_string()
        );
    }

    #[test]
    fn a_multi_repo_task_gets_a_worktree_each() {
        let mut fx = Fixture::new();
        let api = fx.repo("api");
        let web = fx.repo("web");
        let task = fx.task("Cross cutting", "change both", &[api, web]);

        let launched = fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();

        assert_eq!(launched.worktrees.len(), 2);
        assert!(task.workspace_dir.join("api").exists());
        assert!(task.workspace_dir.join("web").exists());
    }

    #[test]
    fn a_task_with_no_repos_cannot_launch() {
        let mut fx = Fixture::new();
        let task = fx.task("Nothing", "nope", &[]);
        assert!(fx.launcher.launch(&mut fx.store, &task, at(5)).is_err());
        assert!(!fx.server.tmux.has_session("marver-1"));
    }

    #[test]
    fn a_failed_launch_leaves_nothing_running_or_on_disk() {
        let mut fx = Fixture::new();
        let good = fx.repo("api");
        // Registered but never initialised as a repo, so provisioning fails
        // partway.
        let broken_path = fx.repos_dir.join("broken");
        std::fs::create_dir_all(&broken_path).unwrap();
        let broken = fx.store.upsert_repo(&broken_path, "broken", at(0)).unwrap();
        let task = fx.task("Will fail", "x", &[good, broken]);

        assert!(fx.launcher.launch(&mut fx.store, &task, at(5)).is_err());

        assert!(!task.workspace_dir.exists(), "workspace must be gone");
        assert!(
            !fx.server.tmux.has_session("marver-1"),
            "no session may survive a failed launch"
        );
        assert_eq!(
            fx.store.get_task(task.id).unwrap().session_name,
            None,
            "nothing should claim a session that does not exist"
        );
    }

    #[test]
    fn two_data_directories_can_both_run_their_task_1() {
        // Both call it task 1 and both share a tmux server, so before the
        // prefix the second one could not start at all: the name was taken.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let mine = fx.task("First", "one", std::slice::from_ref(&repo));

        // A second marver, its own store and workspaces, the same tmux server.
        let other_tmp = TempDir::new().unwrap();
        let mut other_store = Store::open_in_memory().unwrap();
        // Its own repo: two marvers on the *same* repo also collide on the
        // branch name, which is a separate matter from the session name.
        let other_repo_path = fx.repos_dir.join("web");
        init_repo(&other_repo_path, "main");
        let other_repo = other_store
            .upsert_repo(&other_repo_path, "web", at(0))
            .unwrap();
        let theirs = other_store
            .create_task(
                "First",
                "one",
                &other_tmp.path().join("tasks"),
                &[other_repo.id],
                at(0),
            )
            .unwrap();
        assert_eq!(mine.id, theirs.id, "both are task 1");

        let other_launcher = Launcher::new(
            fx.server.tmux.clone(),
            WorktreeManager::new(other_tmp.path().join("tasks")),
            PathBuf::from("/usr/local/bin/marver"),
            other_tmp.path().join("hooks.sock"),
        )
        .harness(testing::stub_agent(other_tmp.path()))
        .session_prefix("bbbbbb");
        fx.launcher = fx.launcher.session_prefix("aaaaaa");

        let ours = fx.launcher.launch(&mut fx.store, &mine, at(5)).unwrap();
        let theirs = other_launcher
            .launch(&mut other_store, &theirs, at(5))
            .expect("a second marver's task 1 must not collide with the first");

        assert_ne!(ours.session, theirs.session);
        assert!(fx.server.tmux.has_session(&ours.session));
        assert!(fx.server.tmux.has_session(&theirs.session));
    }

    #[test]
    fn relaunching_a_live_task_is_refused() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix it", std::slice::from_ref(&repo));
        fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();

        let again = fx.launcher.launch(&mut fx.store, &task, at(6));
        assert!(
            matches!(again, Err(Error::SessionExists(_))),
            "a second launch must not adopt or clobber the live session"
        );
        assert!(
            task.workspace_dir.exists(),
            "the refusal must not destroy the running task's workspace"
        );
    }

    #[test]
    fn shutting_down_removes_the_session_and_the_worktrees() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Fix auth", "fix it", std::slice::from_ref(&repo));
        fx.launcher.launch(&mut fx.store, &task, at(5)).unwrap();
        let task = fx.store.get_task(task.id).unwrap();

        let teardown = fx
            .launcher
            .shut_down(&fx.store, &task, Branches::Keep)
            .unwrap();

        assert!(teardown.is_clean(), "{:?}", teardown.failed);
        assert!(!fx.server.tmux.has_session("marver-1"));
        assert!(!task.workspace_dir.exists());

        // The repo selection outlives the worktrees.
        let links = fx.store.list_task_repos(task.id).unwrap();
        assert_eq!(links.len(), 1);
        assert!(!links[0].is_provisioned());
    }

    #[test]
    fn shutting_down_spares_a_session_that_only_shares_the_name() {
        // Sessions are named from the task id alone, and one tmux server
        // serves the whole machine — so a second data directory holding its
        // own task 1 names the same session as this one.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Namesake", "x", std::slice::from_ref(&repo));

        // Someone else's session, sitting somewhere that is not this
        // workspace.
        let elsewhere = fx.repos_dir.clone();
        fx.server
            .tmux
            .new_session(&tmux::session_name(None, task.id), &elsewhere, (80, 24))
            .unwrap();

        let teardown = fx
            .launcher
            .shut_down(&fx.store, &task, Branches::Keep)
            .unwrap();

        assert!(teardown.is_clean(), "{:?}", teardown.failed);
        assert!(
            fx.server
                .tmux
                .has_session(&tmux::session_name(None, task.id)),
            "a session marver did not open must survive"
        );
    }

    #[test]
    fn shutting_down_a_task_that_never_launched_is_harmless() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.task("Never", "x", std::slice::from_ref(&repo));
        let teardown = fx
            .launcher
            .shut_down(&fx.store, &task, Branches::Keep)
            .unwrap();
        assert!(teardown.removed.is_empty());
    }

    #[test]
    fn the_scheduler_can_drive_the_launcher() {
        use crate::scheduler::Scheduler;

        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let a = fx.task("First", "one", std::slice::from_ref(&repo));
        let b = fx.task("Second", "two", std::slice::from_ref(&repo));

        // A cap of one: the second task must wait.
        let tick = Scheduler::new(1)
            .tick(&mut fx.store, &fx.launcher, at(5))
            .unwrap();

        assert_eq!(tick.started, [a.id]);
        assert_eq!(fx.store.get_task(a.id).unwrap().state, TaskState::Running);
        assert_eq!(fx.store.get_task(b.id).unwrap().state, TaskState::Queued);
        assert!(fx.server.tmux.has_session("marver-1"));
        assert!(!fx.server.tmux.has_session("marver-2"));
    }
}