fix(compression): reject stale runtime before async job

api/routes.py

@@ -23813,6 +23813,10 @@ def _manual_compression_status_payload(job):
         payload["ok"] = False
         payload["error"] = job.get("error") or "Compression failed"
         payload["error_status"] = int(job.get("error_status") or 400)
+        if job.get("error_type"):
+            payload["type"] = job["error_type"]
+        if job.get("retryable") is not None:
+            payload["retryable"] = bool(job["retryable"])
     elif status == "cancelled":
         payload["ok"] = False
         payload["error"] = job.get("error") or "Compression cancelled"
@@ -23847,6 +23851,8 @@ def _run_manual_compression_job(sid, body):
                         "status": "error",
                         "error": str((payload or {}).get("error") or "Compression failed"),
                         "error_status": status,
+                        "error_type": (payload or {}).get("type"),
+                        "retryable": (payload or {}).get("retryable"),
                         "updated_at": now,
                     }
                 )
@@ -23858,6 +23864,21 @@ def _run_manual_compression_job(sid, body):
                         "updated_at": now,
                     }
                 )
+    except AgentRuntimeChangedError as exc:
+        logger.warning("Manual compression worker found stale Agent runtime for session %s", sid)
+        with _MANUAL_COMPRESSION_JOBS_LOCK:
+            job = _MANUAL_COMPRESSION_JOBS.get(sid)
+            if job:
+                job.update(
+                    {
+                        "status": "error",
+                        "error": str(exc),
+                        "error_status": 409,
+                        "error_type": "agent_runtime_stale",
+                        "retryable": True,
+                        "updated_at": time.time(),
+                    }
+                )
     except Exception as exc:
         logger.warning("Manual compression worker failed for session %s: %s", sid, exc)
         with _MANUAL_COMPRESSION_JOBS_LOCK:
@@ -23889,6 +23910,23 @@ def _handle_session_compress_start(handler, body):
     if getattr(s, "active_stream_id", None):
         return bad(handler, "Session is still streaming; wait for the current turn to finish.", 409)
 
+    # Reject a stale local Agent runtime before creating the asynchronous job.
+    # The worker has its own guarded synchronous path, but waiting for that
+    # worker would already mutate the job state and would collapse the typed
+    # retryable stale-runtime response into a generic worker error.
+    try:
+        ensure_agent_runtime_current()
+    except AgentRuntimeChangedError as exc:
+        return j(
+            handler,
+            {
+                "error": str(exc),
+                "type": "agent_runtime_stale",
+                "retryable": True,
+            },
+            status=409,
+        )
+
     focus_topic = str(body.get("focus_topic") or body.get("topic") or "").strip()[:500] or None
     job_body = {"session_id": sid}
     if focus_topic:

tests/test_sprint46.py

@@ -161,6 +161,75 @@ def test_session_compress_stale_runtime_returns_typed_409_before_mutation(
     assert loaded_after.compact() == before
 
 
+def test_session_compress_start_stale_runtime_returns_409_before_job_creation(
+    monkeypatch, cleanup_test_sessions
+):
+    import api.routes as routes
+
+    sid = _make_session()
+    cleanup_test_sessions.append(sid)
+    with routes._MANUAL_COMPRESSION_JOBS_LOCK:
+        routes._MANUAL_COMPRESSION_JOBS.pop(sid, None)
+
+    monkeypatch.setattr(
+        routes,
+        "ensure_agent_runtime_current",
+        lambda: (_ for _ in ()).throw(
+            routes.AgentRuntimeChangedError("restart required")
+        ),
+    )
+
+    handler = _FakeHandler()
+    routes._handle_session_compress_start(handler, {"session_id": sid})
+
+    assert handler.status == 409
+    assert json.loads(handler.wfile.getvalue().decode("utf-8")) == {
+        "error": "restart required",
+        "type": "agent_runtime_stale",
+        "retryable": True,
+    }
+    with routes._MANUAL_COMPRESSION_JOBS_LOCK:
+        assert sid not in routes._MANUAL_COMPRESSION_JOBS
+
+
+def test_session_compress_worker_preserves_stale_runtime_taxonomy(monkeypatch):
+    import api.routes as routes
+
+    sid = "stale-worker-session"
+    started_at = time.time()
+    with routes._MANUAL_COMPRESSION_JOBS_LOCK:
+        routes._MANUAL_COMPRESSION_JOBS[sid] = {
+            "session_id": sid,
+            "status": "running",
+            "started_at": started_at,
+            "updated_at": started_at,
+        }
+
+    monkeypatch.setattr(
+        routes,
+        "_handle_session_compress",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            routes.AgentRuntimeChangedError("restart required")
+        ),
+    )
+
+    routes._run_manual_compression_job(sid, {"session_id": sid})
+
+    handler = _FakeHandler()
+    routes._handle_session_compress_status(handler, sid)
+    payload = handler.payload()
+    assert handler.status == 200
+    assert payload["ok"] is False
+    assert payload["status"] == "error"
+    assert payload["session_id"] == sid
+    assert payload["error"] == "restart required"
+    assert payload["error_status"] == 409
+    assert payload["type"] == "agent_runtime_stale"
+    assert payload["retryable"] is True
+    with routes._MANUAL_COMPRESSION_JOBS_LOCK:
+        routes._MANUAL_COMPRESSION_JOBS.pop(sid, None)
+
+
 def test_session_compress_roundtrip(monkeypatch, cleanup_test_sessions):
     created = cleanup_test_sessions
     original_messages = [