fix(jobs): job model ownership survives pause/resume; delete-while-paused reconciles

The exclusive snapshot is now per-job and captured once, the first time the
job acquires the slot — a pause/resume cycle no longer re-baselines it, so a
model the job loaded before pausing is still recognized as job-introduced and
cleaned up on a later interrupt or delete. Deleting a paused or interrupted
job routes through the worker for the same snapshot reconcile instead of being
erased without cleanup. Jobs that end completed/failed commit their model
side-effects (snapshot discarded). Reconcile is keyed by job id and is a no-op
for jobs that never held the slot.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Z9WMqprhqjWEDuv1gnGaH

Author: pwilkin

docs/dev/job-system.md

@@ -77,16 +77,23 @@ A single worker runs one job at a time.
 - **pause** — stop *after the current step*; the job goes `paused` at the next
   step boundary and (if exclusive) releases the slot. Pausing a still-`queued`
   job takes effect immediately: it is removed from the queue and persisted as
-  `paused` before the call returns.
+  `paused` before the call returns. Pause does **not** commit the job's model
+  side-effects: models the job loaded stay resident for the resume, and the
+  job keeps owning them (see below).
 - **interrupt** — cancel the *current step now* (kills an in-flight `load`, and
   aborts an in-flight `chat` at the HTTP layer rather than waiting for the
   backend to reply); the step returns to `pending`, the job goes `interrupted`.
   Interrupting a still-`queued` job likewise dequeues and persists it as
   `interrupted` immediately. An interrupted exclusive job unloads only the
   model(s) it introduced before releasing the slot: the reconcile snapshots the
-  router's loaded models (with their pin state) when the exclusive session
-  begins, unloads `(current set − snapshot)` — including models the job pinned
-  itself — and restores the recorded pin state of surviving pre-job models.
+  router's loaded models (with their pin state) **once, the first time the job
+  acquires the exclusive slot**, and that snapshot is the job's for its whole
+  lifetime — a pause/resume cycle does not re-baseline it, so a model the job
+  loaded before pausing is still recognized as job-introduced after resume.
+  Reconcile unloads `(current set − snapshot)` — including models the job
+  pinned itself — and restores the recorded pin state of surviving pre-job
+  models. A job that ends `completed` or `failed` commits its model
+  side-effects instead (its snapshot is discarded, resident models stay).
   **Guarantee scope:** a model that was resident before the job is preserved
   only if it is *still resident* at reconcile time. If a job `load` evicted it
   (loaded-model cap) or replaced it with different options, the reconcile does
@@ -97,10 +104,12 @@ A single worker runs one job at a time.
 - **delete** — removes the job. Deleting an active job persists a deletion
   tombstone *before* the call returns, then interrupts it and defers the actual
   removal until the worker has finished cleanup (reconcile unload), so a deleted
-  exclusive job never leaks a resident model. The tombstone makes the deletion
-  durable: a crash between the acknowledgement and the final removal does not
-  resurrect the job on restart, and a tombstoned job is already invisible to
-  `GET`/list.
+  exclusive job never leaks a resident model. Deleting a `paused` or
+  `interrupted` job takes the same tombstone-then-worker-cleanup path, so a
+  model introduced before a pause is still unloaded. The tombstone makes the
+  deletion durable: a crash between the acknowledgement and the final removal
+  does not resurrect the job on restart, and a tombstoned job is already
+  invisible to `GET`/list.
 - **query** — the full job record (status, per-step state, context).
 
 Chat interrupts are genuine aborts: the chat op passes its cancel flag through

src/cpp/include/lemon/jobs/job_ops.h

@@ -24,10 +24,11 @@ class OpRegistry {
     const OpHandler* find(const std::string& name) const;
     std::set<std::string> names() const;
 
-    std::function<bool(CancelFlag*)> begin_exclusive;
+    std::function<bool(const std::string& job_id, CancelFlag*)> begin_exclusive;
     std::function<void()> end_exclusive;
 
-    std::function<void()> reconcile_unload;
+    std::function<void(const std::string& job_id)> reconcile_unload;
+    std::function<void(const std::string& job_id)> discard_exclusive;
 
 private:
     std::map<std::string, OpHandler> handlers_;
@@ -43,9 +44,10 @@ struct OpProviders {
     std::function<json(const json& params, CancelFlag& cancel)> unload_op;
     std::function<json(const json& params, CancelFlag& cancel)> chat_op;
 
-    std::function<bool(CancelFlag*)> begin_exclusive;
+    std::function<bool(const std::string& job_id, CancelFlag*)> begin_exclusive;
     std::function<void()> end_exclusive;
-    std::function<void()> reconcile_unload;
+    std::function<void(const std::string& job_id)> reconcile_unload;
+    std::function<void(const std::string& job_id)> discard_exclusive;
 };
 
 OpRegistry build_op_registry(OpProviders providers);

src/cpp/server/jobs/job_manager.cpp

@@ -359,6 +359,17 @@ bool JobManager::remove(const std::string& id, bool& active_out) {
                           << " (tombstoned; erased after cleanup)" << std::endl;
         return true;
     }
+    if (it->second.status == JobStatus::Paused || it->second.status == JobStatus::Interrupted) {
+        it->second.deleted = true;
+        control_for_locked(id)->delete_requested.store(true);
+        persist_locked();
+        enqueue_locked(id);
+        cv_.notify_all();
+        LOG(INFO, "Jobs") << "delete requested for " << to_string(it->second.status)
+                          << " job " << id << " (tombstoned; erased after reconcile)"
+                          << std::endl;
+        return true;
+    }
     jobs_.erase(id);
     controls_.erase(id);
     order_.erase(std::remove(order_.begin(), order_.end(), id), order_.end());
@@ -373,6 +384,7 @@ void JobManager::worker_main() {
         std::string id;
         std::shared_ptr<Control> ctrl;
         bool exclusive = false;
+        bool cleanup_only = false;
         {
             std::unique_lock<std::mutex> lock(mutex_);
             cv_.wait(lock, [&] { return stop_ || !queue_.empty(); });
@@ -381,23 +393,41 @@ void JobManager::worker_main() {
             id = queue_.front();
             queue_.pop_front();
             auto it = jobs_.find(id);
-            if (it == jobs_.end() || it->second.status != JobStatus::Queued) continue;
+            if (it == jobs_.end()) continue;
             ctrl = control_for_locked(id);
-            it->second.status = JobStatus::Running;
-            if (it->second.started_at.empty()) it->second.started_at = iso_now();
-            active_id_ = id;
-            exclusive = job_needs_exclusive(it->second, registry_);
+            if (it->second.deleted) {
+                cleanup_only = true;
+            } else if (it->second.status != JobStatus::Queued) {
+                continue;
+            } else {
+                it->second.status = JobStatus::Running;
+                if (it->second.started_at.empty()) it->second.started_at = iso_now();
+                active_id_ = id;
+                exclusive = job_needs_exclusive(it->second, registry_);
+                persist_locked();
+                LOG(INFO, "Jobs") << "running job " << id << " from step '" << it->second.cursor
+                                  << "'" << std::endl;
+            }
+        }
+
+        if (cleanup_only) {
+            if (registry_.reconcile_unload) registry_.reconcile_unload(id);
+            std::lock_guard<std::mutex> lock(mutex_);
+            jobs_.erase(id);
+            controls_.erase(id);
+            order_.erase(std::remove(order_.begin(), order_.end(), id), order_.end());
+            queue_.erase(std::remove(queue_.begin(), queue_.end(), id), queue_.end());
             persist_locked();
-            LOG(INFO, "Jobs") << "running job " << id << " from step '" << it->second.cursor << "'"
-                              << std::endl;
+            LOG(INFO, "Jobs") << "erased deleted job " << id << " after reconcile" << std::endl;
+            continue;
         }
 
         struct ExclusiveGuard {
             const OpRegistry& reg;
             const std::string& id;
             bool held = false;
             bool begin(CancelFlag* cancel) {
-                if (reg.begin_exclusive && !reg.begin_exclusive(cancel)) return false;
+                if (reg.begin_exclusive && !reg.begin_exclusive(id, cancel)) return false;
                 held = true;
                 LOG(INFO, "Jobs") << "job " << id << " acquired exclusive slot" << std::endl;
                 return true;
@@ -421,18 +451,23 @@ void JobManager::worker_main() {
                               << " interrupted while waiting for the exclusive slot" << std::endl;
         }
         bool interrupted = false;
+        bool terminal = false;
         {
             std::lock_guard<std::mutex> lock(mutex_);
             auto it = jobs_.find(id);
-            interrupted = it != jobs_.end() && it->second.status == JobStatus::Interrupted;
+            if (it != jobs_.end()) {
+                interrupted = it->second.status == JobStatus::Interrupted;
+                terminal = is_terminal(it->second.status);
+            }
             active_id_.clear();
         }
 
-        if (guard.held && interrupted && registry_.reconcile_unload) {
-            registry_.reconcile_unload();
-            LOG(INFO, "Jobs") << "job " << id << " interrupted — unloaded resident model(s)"
+        if (interrupted && registry_.reconcile_unload) {
+            registry_.reconcile_unload(id);
+            LOG(INFO, "Jobs") << "job " << id << " interrupted — reconciled job-loaded model(s)"
                               << std::endl;
         }
+        if (terminal && registry_.discard_exclusive) registry_.discard_exclusive(id);
 
         if (ctrl->delete_requested.load()) {
             std::lock_guard<std::mutex> lock(mutex_);

src/cpp/server/jobs/job_ops.cpp

@@ -73,6 +73,7 @@ OpRegistry build_op_registry(OpProviders providers) {
     reg.begin_exclusive = providers.begin_exclusive;
     reg.end_exclusive = providers.end_exclusive;
     reg.reconcile_unload = providers.reconcile_unload;
+    reg.discard_exclusive = providers.discard_exclusive;
 
     return reg;
 }

src/cpp/server/server.cpp

@@ -500,24 +500,38 @@ Server::Server(std::shared_ptr<RuntimeConfig> config, const std::string& cache_d
             }
             return lemon::jobs::json::parse(response.dump());
         };
-        auto exclusive_snapshot = std::make_shared<std::map<std::string, bool>>();
+        auto job_snapshots =
+            std::make_shared<std::map<std::string, std::map<std::string, bool>>>();
         auto snapshot_mutex = std::make_shared<std::mutex>();
-        providers.begin_exclusive = [this, exclusive_snapshot, snapshot_mutex](
+        providers.begin_exclusive = [this, job_snapshots, snapshot_mutex](
+                                        const std::string& job_id,
                                         lemon::jobs::CancelFlag* cancel) -> bool {
             if (!router_->begin_exclusive(cancel)) return false;
-            auto snap = router_->snapshot_loaded_models();
             std::lock_guard<std::mutex> lk(*snapshot_mutex);
-            *exclusive_snapshot = std::move(snap);
+            if (!job_snapshots->count(job_id))
+                (*job_snapshots)[job_id] = router_->snapshot_loaded_models();
             return true;
         };
         providers.end_exclusive = [this] { router_->end_exclusive(); };
-        providers.reconcile_unload = [this, exclusive_snapshot, snapshot_mutex] {
+        providers.reconcile_unload = [this, job_snapshots, snapshot_mutex](
+                                         const std::string& job_id) {
             std::map<std::string, bool> keep;
+            bool have = false;
             {
                 std::lock_guard<std::mutex> lk(*snapshot_mutex);
-                keep = *exclusive_snapshot;
+                auto it = job_snapshots->find(job_id);
+                if (it != job_snapshots->end()) {
+                    keep = std::move(it->second);
+                    job_snapshots->erase(it);
+                    have = true;
+                }
             }
-            router_->unload_models_not_in(keep);
+            if (have) router_->unload_models_not_in(keep);
+        };
+        providers.discard_exclusive = [job_snapshots, snapshot_mutex](
+                                          const std::string& job_id) {
+            std::lock_guard<std::mutex> lk(*snapshot_mutex);
+            job_snapshots->erase(job_id);
         };
         job_manager_ = std::make_unique<lemon::jobs::JobManager>(
             lemon::utils::get_cache_dir(), lemon::jobs::build_op_registry(std::move(providers)));

test/server_jobs.py

@@ -1065,6 +1065,99 @@ def test_pin_state_of_preexisting_model_restored_after_job(self):
             "the pre-job model did not actually survive the job (it was reloaded)",
         )
 
+    def test_job_ownership_survives_pause_resume_interrupt(self):
+        backend = self.require_real_backend()
+        steps = [
+            {
+                "id": "ld",
+                "op": "load",
+                "params": {
+                    "model": TEST_MODEL,
+                    "llamacpp_backend": backend,
+                    "ctx_size": 2048,
+                },
+            },
+            {"id": "hold", "op": "sleep", "params": {"ms": 4000}},
+            {"id": "hold2", "op": "sleep", "params": {"ms": 20000}},
+        ]
+        job = self.create_job("pause-resume-interrupt", steps)
+        jid = job["id"]
+        self.poll_cursor(jid, "hold", timeout=60)
+        self.assertEqual(
+            requests.get(f"{BASE}/health", timeout=5).json().get("model_loaded"),
+            TEST_MODEL,
+        )
+
+        r = requests.post(f"{BASE}/jobs/{jid}/pause", timeout=10)
+        self.assertEqual(r.status_code, 200, r.text)
+        self.poll_status(jid, "paused", timeout=15)
+        self.assertEqual(
+            requests.get(f"{BASE}/health", timeout=5).json().get("model_loaded"),
+            TEST_MODEL,
+            "pause must keep the job's models resident for the resume",
+        )
+
+        r = requests.post(f"{BASE}/jobs/{jid}/resume", timeout=10)
+        self.assertEqual(r.status_code, 200, r.text)
+        self.poll_cursor(jid, "hold2", timeout=30)
+
+        r = requests.post(f"{BASE}/jobs/{jid}/interrupt", timeout=10)
+        self.assertEqual(r.status_code, 200, r.text)
+        self.poll_status(jid, "interrupted", timeout=20)
+        self.assertNotEqual(
+            requests.get(f"{BASE}/health", timeout=5).json().get("model_loaded"),
+            TEST_MODEL,
+            "a pause/resume cycle re-baselined the snapshot: the interrupt "
+            "preserved a model the job itself introduced",
+        )
+
+    def test_delete_while_paused_cleans_up(self):
+        backend = self.require_real_backend()
+        steps = [
+            {
+                "id": "ld",
+                "op": "load",
+                "params": {
+                    "model": TEST_MODEL,
+                    "llamacpp_backend": backend,
+                    "ctx_size": 2048,
+                },
+            },
+            {"id": "hold", "op": "sleep", "params": {"ms": 4000}},
+            {"id": "hold2", "op": "sleep", "params": {"ms": 20000}},
+        ]
+        job = self.create_job("delete-while-paused", steps)
+        jid = job["id"]
+        self.poll_cursor(jid, "hold", timeout=60)
+
+        r = requests.post(f"{BASE}/jobs/{jid}/pause", timeout=10)
+        self.assertEqual(r.status_code, 200, r.text)
+        self.poll_status(jid, "paused", timeout=15)
+        self.assertEqual(
+            requests.get(f"{BASE}/health", timeout=5).json().get("model_loaded"),
+            TEST_MODEL,
+        )
+
+        r = requests.delete(f"{BASE}/jobs/{jid}", timeout=10)
+        self.assertEqual(r.status_code, 200, r.text)
+        self.assertEqual(self.get_job(jid).status_code, 404)
+        self.poll_gone(jid)
+
+        deadline = time.time() + 20
+        while time.time() < deadline:
+            loaded = (
+                requests.get(f"{BASE}/health", timeout=5).json().get("model_loaded")
+            )
+            if loaded != TEST_MODEL:
+                break
+            time.sleep(0.2)
+        self.assertNotEqual(
+            requests.get(f"{BASE}/health", timeout=5).json().get("model_loaded"),
+            TEST_MODEL,
+            "deleting a paused job skipped reconciliation and leaked the "
+            "model the job introduced",
+        )
+
     def test_bench_shaped_sweep(self):
         backend = self.require_real_backend()