marver 0.0.19

//! Turning a queued task into a live agent.
//!
//! This is the glue the numbered components leave out: it composes the worktree
//! manager, tmux, and the hook receiver into the one operation the scheduler
//! actually needs.
//!
//! ```text
//! provision worktrees  →  write hook settings  →  create session  →  start agent
//! ```
//!
//! The order matters and is not arbitrary. Worktrees come first because the
//! session's working directory must exist before tmux is asked to start there.
//! Hook settings are written before the agent starts, since a settings file
//! appearing afterwards would be read too late. The session name is recorded
//! last, once there is genuinely something to record.
//!
//! Launching is all-or-nothing. A half-launched task — worktrees but no session,
//! or a session running an agent marver has not recorded — is worse than one
//! that plainly failed, because nothing will ever clean it up.

use std::ffi::OsString;
use std::path::{Path, PathBuf};

use chrono::{DateTime, Utc};

use crate::agent;
use crate::domain::Task;
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("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>;

/// How to start the agent inside a task's session.
#[derive(Debug, Clone)]
pub struct AgentCommand {
    pub program: String,
    /// Arguments placed before `--settings` and the prompt.
    pub args: Vec<String>,
}

impl AgentCommand {
    pub fn claude() -> Self {
        Self {
            program: "claude".to_string(),
            args: Vec::new(),
        }
    }

    /// The argument vector the agent is launched with.
    ///
    /// An argv, not a shell line. Quoting a prompt correctly is not enough when
    /// the result is *typed* into an interactive shell: control bytes in the
    /// text are consumed by the terminal's line discipline and by readline
    /// before the shell parses any quoting, so a prompt containing `\x03` or
    /// `\x15` — which survive a paste from an issue body or a terminal log —
    /// discards the half-typed line and leaves the remainder to run on its own.
    /// Handing tmux an argv means nothing interprets the prompt at all.
    pub fn argv(&self, settings: &Path, prompt: &str) -> Vec<OsString> {
        let mut argv: Vec<OsString> = vec![self.program.clone().into()];
        argv.extend(self.args.iter().map(OsString::from));
        argv.push("--settings".into());
        argv.push(settings.as_os_str().to_owned());
        argv.push(prompt.into());
        argv
    }
}

/// What a successful launch produced.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Launched {
    pub session: String,
    pub settings: 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,
    agent: AgentCommand,
    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(),
            agent: AgentCommand::claude(),
            size: tmux::DEFAULT_SIZE,
        }
    }

    /// Override the agent command. Tests use this to avoid starting a real agent.
    pub fn agent(mut self, agent: AgentCommand) -> Self {
        self.agent = agent;
        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(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<PathBuf> {
        let settings = hook::write_settings(
            &task.workspace_dir,
            task.id,
            &self.hook_bin,
            &self.hook_socket,
        )?;

        // 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,
            &self.agent.argv(&settings, &task.prompt),
        )?;

        // 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(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.
    ///
    /// The session is killed before the worktrees, since it is sitting in the
    /// directory being removed. A committed task can still have a live session
    /// — nothing kills one on review — so this order matters even long after
    /// the agent stopped working.
    pub fn shut_down(&self, store: &Store, task: &Task, branches: Branches) -> Result<Teardown> {
        let session = task
            .session_name
            .clone()
            .unwrap_or_else(|| tmux::session_name(task.id));
        if self.owns_session(task, &session) {
            // A session that goes between the check and the kill is the outcome
            // this wanted. Only one still standing afterwards is a failure.
            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.
    ///
    /// One implementation, in [`agent::owns_session`], because the question is
    /// the same whether marver is about to type into a session or kill it —
    /// and getting it wrong reaches into another marver's work. Not
    /// hypothetical: it happened the first time cleanup ran under test, against
    /// a live session belonging to the developer's own tasks.
    ///
    /// Declining costs a stale session; the other answer costs somebody's agent.
    fn owns_session(&self, task: &Task, session: &str) -> bool {
        agent::owns_session(&self.tmux, task, session)
    }
}

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

    /// 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.
    ///
    /// Printing each element separately is the point. A stub that echoed its
    /// arguments as one line could not tell "the prompt arrived as a single
    /// argument" from "the prompt was split into several" — which is exactly
    /// the difference an injection makes. Staying alive matters too: the agent
    /// is now the session's own process, so a stub that exits takes the session
    /// with it.
    pub fn stub_agent(dir: &Path) -> AgentCommand {
        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");
        AgentCommand {
            program: path.to_string_lossy().into_owned(),
            args: Vec::new(),
        }
    }
}

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 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"),
            )
            .agent(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 agent = AgentCommand::claude();
        let argv = agent.argv(Path::new("/s/settings.json"), "fix the auth flow");
        assert_eq!(
            argv,
            [
                "claude",
                "--settings",
                "/s/settings.json",
                "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. Typed at an interactive shell, \x03 discards the half-written
        // line and leaves the rest to run on its own; \x15 does the same via
        // readline. Nothing here is typed, so nothing interprets any of it.
        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. If a shell is what
        // the session runs, every byte of the prompt goes through the line
        // discipline first and the test above cannot hold.
        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.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 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();

        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 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. The first run of `marver cleanup`
        // under test killed a live session belonging to the developer's real
        // tasks this way. The working directory is what tells them apart.
        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(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(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"));
    }
}