[mcp] fix: release stop-all sessions concurrently

AGENTS.md

@@ -72,6 +72,7 @@ This file defines how coding agents should work in this repository.
 - Keep service daemon ownership safe: no stop, force-stop, or fallback path may signal a PID unless the auto-start ownership record verifies the running process.
 - Treat custom `job_id` values as reserved from the moment startup begins; duplicate starts must fail before another controller can begin keep-alive work.
 - Stop requests must not miss starting sessions; wait for startup to settle before returning `not found` or taking a stop-all snapshot.
+- Stop-all may release independent sessions concurrently, but must not duplicate release work for `stopping` sessions and must keep deterministic additive result fields.
 - Keep utilization backoff eco-safe: when telemetry is unavailable and `busy_threshold >= 0`, controllers should sleep instead of running keepalive compute. Only `busy_threshold=-1` is the explicit unconditional mode.
 - Avoid scattering platform-specific branching across unrelated modules; prefer one clear decision path then platform-specific controller classes.
 - Preserve simple controller flow: global controller orchestrates per-GPU controllers; single-GPU controllers handle device-level keep/release loops.

README.md

@@ -145,6 +145,9 @@ with GlobalGPUController(gpu_ids=[0, 1], vram_to_keep="750MB", interval=90, busy
   still starting is not reported as missing or skipped by stop-all.
 - Stop-all only covers sessions active or already starting when that request
   begins; later concurrent starts belong to a later stop request.
+- Stop-all releases independent sessions concurrently and reports outcomes in
+  deterministic snapshot order with the same `stopped`, `timed_out`, `failed`,
+  and `errors` fields.
 - Dashboard cards mirror that lifecycle state so a retained session shows
   `Releasing` or `Release failed` instead of being presented as a fully active
   keepalive.

docs/concepts/architecture.md

@@ -57,6 +57,8 @@ Elementwise keep-alive batches:
 - The keep-alive loop runs on daemon threads so the main process can exit fast.
 - `GlobalGPUController.release()` stops workers concurrently by way of threads, keeping
   shutdown time bounded even with many GPUs.
+- Service stop-all releases independent sessions concurrently after taking its
+  session snapshot, while aggregating results in deterministic snapshot order.
 - Service session state is intentionally conservative: `stop_keep` removes a
   session only after release succeeds. Timed-out sessions stay visible as
   `state="stopping"` until the background release finishes; failed releases stay

docs/guides/mcp.md

@@ -59,6 +59,8 @@ waits for startup to settle before deciding whether the job exists. Stop-all
 requests also wait for in-progress starts before taking their session snapshot.
 For stop-all, starts that begin after that request's initial snapshot are not
 stopped by that request.
+Stop-all releases the sessions in its snapshot concurrently and aggregates
+results in deterministic snapshot order using the same additive fields.
 
 ## REST quick examples
 

docs/plans/stop-all-release-concurrency.md

@@ -0,0 +1,48 @@
+# Stop-All Release Concurrency Plan
+
+## Background
+
+`KeepGPUServer.stop_keep(None)` now keeps lifecycle state truthful, waits for
+sessions that were already starting, and preserves deterministic stop-all
+snapshot semantics. After the snapshot, however, it releases sessions
+sequentially. Each `_release_with_timeout()` call can take up to 10 seconds, so
+multiple slow sessions can make `keep-gpu stop --all`, REST, JSON-RPC, or the
+dashboard time out before receiving the truthful stop result.
+
+Targeted `stop_keep(job_id)` is intentionally out of scope for this branch.
+
+## Goal
+
+Release independent sessions concurrently during stop-all while preserving the
+existing response contract, lifecycle state, late-release callbacks, and stable
+result ordering.
+
+## Design
+
+- Keep the current locked stop-all snapshot and `state="stopping"` marking.
+- Launch one stop worker per releasable session after the lock is released.
+- Each worker calls `_release_with_timeout()` with the same late-result callback
+  behavior used today.
+- Aggregate worker outcomes in the original stop-all snapshot order, not in
+  completion order, so response ordering remains deterministic.
+- Continue to report already-`stopping` sessions as `timed_out` without starting
+  another release.
+- Keep targeted stop behavior unchanged.
+
+## Todo
+
+- [x] Add a failing stop-all test showing release workers enter concurrently.
+- [x] Add a failing mixed-result test for concurrent stop-all aggregation:
+      success, timeout, and failure must all be reported in snapshot order.
+- [x] Add a stop-all late-callback test showing late success removes the session
+      and late failure keeps `state="stop_failed"`.
+- [x] Add a local-review regression for mixed full-snapshot ordering when a
+      newly timed-out session precedes an already-`stopping` session.
+- [x] Make existing stop-all timeout tests independent of release call order.
+- [x] Implement concurrent stop-all release workers without holding
+      `_sessions_lock` during controller release work.
+- [x] Update `AGENTS.md` and docs to state that stop-all releases independent
+      sessions concurrently while keeping additive result fields.
+- [x] Run targeted MCP tests, full tests, docs build, and pre-commit.
+- [ ] Open a GitHub PR, run local subagent review, resolve all comments, then
+      squash merge.

docs/reference/cli.md

@@ -71,6 +71,8 @@ Stop waits for in-progress starts to settle before returning `not found` or
 taking the stop-all snapshot, so starting sessions are not silently skipped.
 For `--all`, starts that begin after that command's initial snapshot are not
 stopped by that command.
+`--all` releases the sessions in its snapshot concurrently and prints results
+in deterministic snapshot order with the same additive response fields.
 
 ### `keep-gpu list-gpus`
 

src/keep_gpu/mcp/server.py

@@ -304,15 +304,19 @@ def stop_keep(
                 self._sessions_cond.wait()
             session_items = list(self._sessions.items())
             releasable_items = []
+            release_outcomes: List[Dict[str, Any]] = [
+                {} for _job_id, _session in session_items
+            ]
             result = self._empty_stop_result()
-            for job_id, session in session_items:
+            for index, (job_id, session) in enumerate(session_items):
                 if session.state == "stopping":
-                    result["timed_out"].append(job_id)
+                    release_outcomes[index] = {"state": "timed_out"}
                     continue
                 session.state = "stopping"
                 session.last_error = None
-                releasable_items.append((job_id, session))
-        for job_id, session in releasable_items:
+                releasable_items.append((index, job_id, session))
+
+        def _release_one(index: int, job_id: str, session: Session) -> None:
             try:
                 released = self._release_with_timeout(
                     session.controller,
@@ -323,18 +327,36 @@ def stop_keep(
             except Exception as exc:
                 error = str(exc)
                 self._mark_session(job_id, session, "stop_failed", error)
-                result["failed"].append(job_id)
-                result["errors"][job_id] = error
-                continue
+                release_outcomes[index] = {"state": "failed", "error": error}
+                return
             if released:
                 with self._sessions_lock:
                     if self._sessions.get(job_id) is session:
                         self._sessions.pop(job_id, None)
+                release_outcomes[index] = {"state": "stopped"}
+                return
+            self._mark_stop_timeout(job_id, session)
+            release_outcomes[index] = {"state": "timed_out"}
+
+        release_threads = []
+        for index, job_id, session in releasable_items:
+            thread = threading.Thread(
+                target=_release_one,
+                args=(index, job_id, session),
+            )
+            thread.start()
+            release_threads.append(thread)
+        for thread in release_threads:
+            thread.join()
+        for (job_id, _session), outcome in zip(session_items, release_outcomes):
+            state = outcome.get("state")
+            if state == "stopped":
                 result["stopped"].append(job_id)
-                continue
-            if not released:
-                self._mark_stop_timeout(job_id, session)
+            elif state == "timed_out":
                 result["timed_out"].append(job_id)
+            elif state == "failed":
+                result["failed"].append(job_id)
+                result["errors"][job_id] = outcome["error"]
         if result["stopped"] and not quiet:
             logger.info("Stopped sessions: %s", result["stopped"])
         if result["timed_out"] and not quiet:

tests/mcp/test_server.py

@@ -462,12 +462,10 @@ def test_stop_all_tracks_timeouts(monkeypatch):
     job_a = server.start_keep()["job_id"]
     job_b = server.start_keep()["job_id"]
 
-    outcomes = iter([True, False])
-    monkeypatch.setattr(
-        server,
-        "_release_with_timeout",
-        lambda controller, **_: next(outcomes),
-    )
+    def release_outcome(controller, **_):
+        return controller is not server._sessions[job_b].controller
+
+    monkeypatch.setattr(server, "_release_with_timeout", release_outcome)
 
     result = server.stop_keep()
     assert result["stopped"] == [job_a]
@@ -478,6 +476,115 @@ def test_stop_all_tracks_timeouts(monkeypatch):
     assert status_b["state"] == "stopping"
 
 
+def test_stop_all_orders_new_timeouts_before_later_already_stopping(monkeypatch):
+    server = make_server()
+    job_timeout = server.start_keep(gpu_ids=[0])["job_id"]
+    job_stopping = server.start_keep(gpu_ids=[1])["job_id"]
+
+    monkeypatch.setattr(server, "_release_with_timeout", lambda controller, **_: False)
+    targeted_result = server.stop_keep(job_stopping)
+    assert targeted_result["timed_out"] == [job_stopping]
+
+    stop_all_controllers = []
+
+    def timeout_release(controller, **_):
+        stop_all_controllers.append(controller)
+        return False
+
+    monkeypatch.setattr(server, "_release_with_timeout", timeout_release)
+
+    result = server.stop_keep()
+
+    assert result["timed_out"] == [job_timeout, job_stopping]
+    assert [controller.gpu_ids for controller in stop_all_controllers] == [[0]]
+
+
+def test_stop_all_release_workers_enter_concurrently(monkeypatch):
+    server = make_server()
+    job_ids = [
+        server.start_keep(gpu_ids=[0])["job_id"],
+        server.start_keep(gpu_ids=[1])["job_id"],
+        server.start_keep(gpu_ids=[2])["job_id"],
+    ]
+    entered_count = 0
+    entered_lock = threading.Lock()
+    all_entered = threading.Event()
+    release_gate = threading.Event()
+    stop_result = {}
+
+    def blocking_release(controller, **_):
+        nonlocal entered_count
+        with entered_lock:
+            entered_count += 1
+            if entered_count == len(job_ids):
+                all_entered.set()
+        release_gate.wait(timeout=1.0)
+        return True
+
+    monkeypatch.setattr(server, "_release_with_timeout", blocking_release)
+
+    stop_thread = threading.Thread(
+        target=lambda: stop_result.update(value=server.stop_keep())
+    )
+    stop_thread.start()
+
+    try:
+        assert all_entered.wait(timeout=2.0)
+    finally:
+        release_gate.set()
+        stop_thread.join(timeout=1.0)
+
+    assert stop_result["value"]["stopped"] == job_ids
+    assert all(server.status(job_id)["active"] is False for job_id in job_ids)
+
+
+def test_stop_all_concurrent_results_keep_snapshot_order(monkeypatch):
+    server = make_server()
+    job_success = server.start_keep(gpu_ids=[0])["job_id"]
+    job_timeout = server.start_keep(gpu_ids=[1])["job_id"]
+    job_failed = server.start_keep(gpu_ids=[2])["job_id"]
+    job_ids = [job_success, job_timeout, job_failed]
+    entered_count = 0
+    entered_lock = threading.Lock()
+    all_entered = threading.Event()
+    release_gate = threading.Event()
+    stop_result = {}
+
+    def release_outcome(controller, **_):
+        nonlocal entered_count
+        with entered_lock:
+            entered_count += 1
+            if entered_count == len(job_ids):
+                all_entered.set()
+        release_gate.wait(timeout=1.0)
+        if controller.gpu_ids == [1]:
+            return False
+        if controller.gpu_ids == [2]:
+            raise RuntimeError("release failed")
+        return True
+
+    monkeypatch.setattr(server, "_release_with_timeout", release_outcome)
+
+    stop_thread = threading.Thread(
+        target=lambda: stop_result.update(value=server.stop_keep())
+    )
+    stop_thread.start()
+
+    try:
+        assert all_entered.wait(timeout=2.0)
+    finally:
+        release_gate.set()
+        stop_thread.join(timeout=1.0)
+
+    assert stop_result["value"]["stopped"] == [job_success]
+    assert stop_result["value"]["timed_out"] == [job_timeout]
+    assert stop_result["value"]["failed"] == [job_failed]
+    assert stop_result["value"]["errors"] == {job_failed: "release failed"}
+    assert server.status(job_success)["active"] is False
+    assert server.status(job_timeout)["state"] == "stopping"
+    assert server.status(job_failed)["state"] == "stop_failed"
+
+
 def test_timed_out_stop_removes_session_after_background_release_succeeds(monkeypatch):
     release_gate = threading.Event()
 
@@ -537,6 +644,30 @@ def short_timeout(controller, **kwargs):
     assert status["last_error"] == "late release failed"
 
 
+def test_stop_all_late_callbacks_update_each_timed_out_session(monkeypatch):
+    server = make_server()
+    job_late_success = server.start_keep(gpu_ids=[0])["job_id"]
+    job_late_failure = server.start_keep(gpu_ids=[1])["job_id"]
+
+    def timeout_with_late_callback(controller, on_late_result, **_):
+        if controller.gpu_ids == [0]:
+            on_late_result(None)
+        else:
+            on_late_result(RuntimeError("late release failed"))
+        return False
+
+    monkeypatch.setattr(server, "_release_with_timeout", timeout_with_late_callback)
+
+    result = server.stop_keep()
+
+    assert result["timed_out"] == [job_late_success, job_late_failure]
+    assert server.status(job_late_success)["active"] is False
+    status = server.status(job_late_failure)
+    assert status["active"] is True
+    assert status["state"] == "stop_failed"
+    assert status["last_error"] == "late release failed"
+
+
 def test_timed_out_stop_preserves_failure_from_timeout_race(monkeypatch):
     server = make_server()
     job_id = server.start_keep()["job_id"]