attach, logs, and stop jobs

p/src/commands/attach.rs

@@ -1,6 +1,72 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
+use chrono::Utc;
+
+use crate::{
+    config, db,
+    job::{Job, JobStatus},
+    ssh,
+};
 
 pub fn execute(job_id: &str) -> Result<()> {
-    let _ = job_id;
-    anyhow::bail!("not yet implemented")
+    let cfg = config::load()?;
+    let job = db::find(job_id)?.with_context(|| format!("job '{}' not found", job_id))?;
+
+    if job.status != JobStatus::Running {
+        anyhow::bail!(
+            "job {} is not running (status: {})\n\
+             use 'p logs {}' to view its output",
+            job.short_id(),
+            job.status.as_str(),
+            job.short_id(),
+        );
+    }
+
+    let worker = cfg.resolve_worker(Some(&job.worker))?.clone();
+    let session = format!("p-{}", job.short_id());
+
+    let sid = job.short_id().to_string();
+    {
+        let sid = sid.clone();
+        ctrlc::set_handler(move || {
+            eprintln!(
+                "\nDetached. Use 'p attach {sid}' to re-attach or 'p logs {sid}' to view output."
+            );
+            std::process::exit(0);
+        })
+        .ok();
+    }
+
+    ssh::run_interactive(&worker, &format!("tmux attach -t '{}'", session))?;
+
+    // Same reconciliation logic as the run command.
+    let exit_code = ssh::read_job_exitcode(&worker, &job.id);
+    let now = Utc::now().timestamp();
+
+    if let Some(ec) = exit_code {
+        ssh::run_capture(
+            &worker,
+            &format!("tmux kill-session -t '{}' 2>/dev/null || true", session),
+        )
+        .ok();
+
+        db::save(&Job {
+            status: if ec == 0 {
+                JobStatus::Done
+            } else {
+                JobStatus::Failed
+            },
+            exit_code: Some(ec),
+            finished_at: Some(now),
+            ..job
+        })?;
+
+        eprintln!("Job {} finished with exit code {}.", sid, ec);
+    } else {
+        eprintln!(
+            "Detached from job {}. Use 'p attach {}' to re-attach.",
+            sid, sid
+        );
+    }
+
+    Ok(())
 }

p/src/commands/logs.rs

@@ -1,6 +1,41 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
+
+use crate::{config, db, job::JobStatus, ssh};
 
 pub fn execute(job_id: &str, follow: bool) -> Result<()> {
-    let _ = (job_id, follow);
-    anyhow::bail!("not yet implemented")
+    let cfg = config::load()?;
+    let job = db::find(job_id)?.with_context(|| format!("job '{}' not found", job_id))?;
+
+    let worker = cfg.resolve_worker(Some(&job.worker))?;
+    let log = format!("~/.p/jobs/{}/output.log", job.id);
+
+    if follow {
+        if job.status != JobStatus::Running {
+            // Job is done — just cat the log, no point following.
+            eprintln!(
+                "note: job {} is no longer running, showing full output",
+                job.short_id()
+            );
+            print_log(worker, &log)?;
+        } else {
+            // Stream live output. Ctrl+C kills ssh; the job keeps running.
+            ssh::run_output(worker, &format!("tail -n +1 -f '{}'", log))?;
+        }
+    } else {
+        print_log(worker, &log)?;
+    }
+
+    Ok(())
+}
+
+fn print_log(worker: &crate::config::WorkerConfig, log: &str) -> Result<()> {
+    let out = ssh::run_capture(
+        worker,
+        &format!(
+            "cat '{}' 2>/dev/null || echo '(no output captured yet)'",
+            log
+        ),
+    )?;
+    print!("{}", out);
+    Ok(())
 }

p/src/commands/stop.rs

@@ -1,6 +1,40 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
+use chrono::Utc;
+
+use crate::{
+    config, db,
+    job::{Job, JobStatus},
+    ssh,
+};
 
 pub fn execute(job_id: &str) -> Result<()> {
-    let _ = job_id;
-    anyhow::bail!("not yet implemented")
+    let cfg = config::load()?;
+    let job = db::find(job_id)?.with_context(|| format!("job '{}' not found", job_id))?;
+
+    if job.status != JobStatus::Running {
+        anyhow::bail!(
+            "job {} is not running (status: {})",
+            job.short_id(),
+            job.status.as_str()
+        );
+    }
+
+    let worker = cfg.resolve_worker(Some(&job.worker))?;
+    let session = format!("p-{}", job.short_id());
+    let sid = job.short_id().to_string();
+
+    // Kill the tmux session. This terminates the job process and run.sh.
+    ssh::run_capture(
+        worker,
+        &format!("tmux kill-session -t '{}' 2>/dev/null || true", session),
+    )?;
+
+    db::save(&Job {
+        status: JobStatus::Stopped,
+        finished_at: Some(Utc::now().timestamp()),
+        ..job
+    })?;
+
+    eprintln!("Job {} stopped.", sid);
+    Ok(())
 }

p/src/job.rs

@@ -19,6 +19,8 @@ pub enum JobStatus {
     Running,
     Done,
     Failed,
+    /// Explicitly killed via `p stop`.
+    Stopped,
     /// Status not yet reconciled with the worker.
     Unknown,
 }
@@ -29,6 +31,7 @@ impl JobStatus {
             Self::Running => "running",
             Self::Done => "done",
             Self::Failed => "failed",
+            Self::Stopped => "stopped",
             Self::Unknown => "unknown",
         }
     }
@@ -38,6 +41,7 @@ impl JobStatus {
             "running" => Self::Running,
             "done" => Self::Done,
             "failed" => Self::Failed,
+            "stopped" => Self::Stopped,
             _ => Self::Unknown,
         }
     }
@@ -51,12 +55,13 @@ impl Job {
         &self.id[..8.min(self.id.len())]
     }
 
-    /// Human-readable status: "running", "done [0]", "failed [1]", "unknown".
+    /// Human-readable status: "running", "done [0]", "failed [1]", "stopped", "unknown".
     pub fn status_display(&self) -> String {
         match self.status {
             JobStatus::Running => "running".to_string(),
             JobStatus::Done => format!("done [{}]", self.exit_code.unwrap_or(0)),
             JobStatus::Failed => format!("failed [{}]", self.exit_code.unwrap_or(-1)),
+            JobStatus::Stopped => "stopped".to_string(),
             JobStatus::Unknown => "unknown".to_string(),
         }
     }