pull, rm, and rm workers

p/src/commands/pull.rs

@@ -1,6 +1,30 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
+use std::path::PathBuf;
+
+use crate::{config, db, sync};
 
 pub fn execute(job_id: &str, remote_path: &str, local_dest: Option<&str>) -> Result<()> {
-    let _ = (job_id, remote_path, local_dest);
-    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))?;
+
+    // Remote path is relative to the job's work directory.
+    let full_remote = format!("~/.p/workdirs/{}/{}", job.id, remote_path);
+
+    let local: PathBuf = match local_dest {
+        Some(d) => PathBuf::from(d),
+        None => std::env::current_dir().context("failed to get current directory")?,
+    };
+
+    eprintln!(
+        "Pulling '{}' from job {} on {} → {}",
+        remote_path,
+        job.short_id(),
+        job.worker,
+        local.display()
+    );
+
+    sync::pull_path(worker, &full_remote, &local)?;
+    Ok(())
 }

p/src/commands/rm.rs

@@ -1,6 +1,45 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
+
+use crate::{config, db, job::JobStatus, ssh};
 
 pub fn execute(job_id: &str, force: bool) -> Result<()> {
-    let _ = (job_id, force);
-    anyhow::bail!("not yet implemented")
+    let cfg = config::load()?;
+    let job = db::find(job_id)?.with_context(|| format!("job '{}' not found", job_id))?;
+    let sid = job.short_id().to_string();
+
+    if job.status == JobStatus::Running {
+        if !force {
+            anyhow::bail!(
+                "job {} is still running — use --force to remove it anyway\n\
+                 or 'p stop {}' to kill it first",
+                sid,
+                sid
+            );
+        }
+        // Kill the tmux session before wiping the files.
+        let session = format!("p-{}", sid);
+        let worker = cfg.resolve_worker(Some(&job.worker))?;
+        ssh::run_capture(
+            worker,
+            &format!("tmux kill-session -t '{}' 2>/dev/null || true", session),
+        )
+        .ok();
+    }
+
+    // Remove remote job dir and work dir (best-effort — warn if unreachable).
+    let remote_rm = format!("rm -rf ~/.p/jobs/{id} ~/.p/workdirs/{id}", id = job.id);
+    if let Some(worker) = cfg.get_worker(&job.worker) {
+        if let Err(e) = ssh::run_capture(worker, &remote_rm) {
+            eprintln!("warning: could not remove remote files: {}", e);
+            eprintln!(
+                "  you may need to manually delete ~/.p/jobs/{id} and ~/.p/workdirs/{id} on '{worker}'",
+                id = job.id,
+                worker = job.worker
+            );
+        }
+    }
+
+    db::delete(&job.id)?;
+    eprintln!("Job {} removed.", sid);
+    Ok(())
 }

p/src/commands/worker/rm.rs

@@ -1,6 +1,60 @@
 use anyhow::Result;
 
+use crate::{config, db, job::JobStatus};
+
 pub fn execute(name: &str) -> Result<()> {
-    let _ = name;
-    anyhow::bail!("not yet implemented")
+    let mut cfg = config::load()?;
+
+    if cfg.get_worker(name).is_none() {
+        let known: Vec<&str> = cfg.workers.iter().map(|w| w.name.as_str()).collect();
+        if known.is_empty() {
+            anyhow::bail!("no workers registered");
+        }
+        anyhow::bail!(
+            "unknown worker '{}'\n\nKnown workers: {}",
+            name,
+            known.join(", ")
+        );
+    }
+
+    // Refuse if there are running jobs on this worker.
+    let running: Vec<_> = db::list()?
+        .into_iter()
+        .filter(|j| j.worker == name && j.status == JobStatus::Running)
+        .collect();
+
+    if !running.is_empty() {
+        let ids: Vec<&str> = running.iter().map(|j| j.short_id()).collect();
+        anyhow::bail!(
+            "worker '{}' has {} running job(s): {}\n\
+             stop them first with 'p stop <id>' or wait for them to finish",
+            name,
+            running.len(),
+            ids.join(", ")
+        );
+    }
+
+    // Warn about leftover job records for this worker.
+    let leftover = db::list()?.into_iter().filter(|j| j.worker == name).count();
+    if leftover > 0 {
+        eprintln!(
+            "note: {} job record(s) from '{}' remain — use 'p rm <id>' to clean them up",
+            leftover, name
+        );
+    }
+
+    cfg.workers.retain(|w| w.name != name);
+
+    // If this was the default, promote the first remaining worker (if any).
+    if cfg.default_worker.as_deref() == Some(name) {
+        cfg.default_worker = cfg.workers.first().map(|w| w.name.clone());
+        match &cfg.default_worker {
+            Some(new) => eprintln!("Default worker updated to '{}'.", new),
+            None => eprintln!("No workers remaining."),
+        }
+    }
+
+    config::save(&cfg)?;
+    eprintln!("Worker '{}' removed.", name);
+    Ok(())
 }