fix(api): keep live runs tracked past stream ttl

gateway/platforms/api_server.py

@@ -61,7 +61,6 @@ from gateway.platforms.base import (
     validate_media_delivery_path,
 )
 from agent.redact import redact_sensitive_text
-from gateway.readiness import collect_runtime_readiness
 
 logger = logging.getLogger(__name__)
 
@@ -899,31 +898,10 @@ class APIServerAdapter(BasePlatformAdapter):
         # from a request flood (#7483).
         self._max_concurrent_runs: int = self._resolve_max_concurrent_runs()
         # Number of in-flight runs on the non-streaming chat/responses paths
-        # (the /v1/runs path tracks its own in-flight set via _run_streams).
+        # (the /v1/runs path tracks its own in-flight set via
+        # _active_run_tasks).
         self._inflight_agent_runs: int = 0
 
-    def _readiness_queue_depths(self) -> tuple[int, int, int]:
-        """Return queue counts only; readiness must never inspect payloads."""
-        process_depth = 0
-        delegation_depth = 0
-        try:
-            from tools.process_registry import process_registry
-
-            process_depth = process_registry.completion_queue.qsize()
-        except Exception:
-            pass
-        try:
-            from tools.async_delegation import list_async_delegations
-
-            delegation_depth = sum(
-                1
-                for record in list_async_delegations()
-                if record.get("status") in {"queued", "running"}
-            )
-        except Exception:
-            pass
-        return len(self._run_streams), process_depth, delegation_depth
-
     @staticmethod
     def _parse_cors_origins(value: Any) -> tuple[str, ...]:
         """Normalize configured CORS origins into a stable tuple."""
@@ -1420,17 +1398,8 @@ class APIServerAdapter(BasePlatformAdapter):
         # This endpoint is served BY the gateway process, so it is by definition
         # alive — gateway_running is True. Derive busy/drainable from the same
         # shared contract /api/status uses so the two surfaces never disagree.
-        api_depth, process_depth, delegation_depth = self._readiness_queue_depths()
-        readiness = collect_runtime_readiness(
-            model_name=self._model_name,
-            runtime_status=runtime,
-            api_run_queue_depth=api_depth,
-            process_completion_queue_depth=process_depth,
-            delegation_completion_queue_depth=delegation_depth,
-        )
         return web.json_response({
-            "status": readiness["status"],
-            "readiness": readiness,
+            "status": "ok",
             "platform": "hermes-agent",
             "version": _hermes_version(),
             "gateway_state": gw_state,
@@ -3998,14 +3967,18 @@ class APIServerAdapter(BasePlatformAdapter):
 
         The cap bounds total in-flight agent activity across every
         agent-serving endpoint: the non-streaming chat/responses paths
-        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` streaming
-        path (tracked by ``_run_streams``). A configured value of 0 disables
-        the cap entirely.
+        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` path
+        (tracked by live entries in ``_active_run_tasks``).  Stream queues are
+        transport state and may disappear while their underlying run remains
+        active, so they must not define run concurrency. A configured value of
+        0 disables the cap entirely.
         """
         limit = self._max_concurrent_runs
         if limit <= 0:
             return None
-        inflight = self._inflight_agent_runs + len(self._run_streams)
+        inflight = self._inflight_agent_runs + sum(
+            not task.done() for task in self._active_run_tasks.values()
+        )
         if inflight >= limit:
             return web.json_response(
                 _openai_error(
@@ -4702,7 +4675,7 @@ class APIServerAdapter(BasePlatformAdapter):
         return web.json_response({"run_id": run_id, "status": "stopping"})
 
     async def _sweep_orphaned_runs(self) -> None:
-        """Periodically clean up run streams that were never consumed."""
+        """Periodically clean up expired transport state for inactive runs."""
         while True:
             await asyncio.sleep(60)
             now = time.time()
@@ -4710,6 +4683,10 @@ class APIServerAdapter(BasePlatformAdapter):
                 run_id
                 for run_id, created_at in list(self._run_streams_created.items())
                 if now - created_at > self._RUN_STREAM_TTL
+                and (
+                    (task := self._active_run_tasks.get(run_id)) is None
+                    or task.done()
+                )
             ]
             for run_id in stale:
                 logger.debug("[api_server] sweeping orphaned run %s", run_id)

gateway/readiness.py

@@ -1,130 +0,0 @@
-"""Bounded, non-destructive readiness probes for authenticated health surfaces."""
-
-from __future__ import annotations
-
-import shutil
-import sqlite3
-from pathlib import Path
-from typing import Any
-
-import yaml
-
-from hermes_constants import get_hermes_home
-
-
-_DISK_DEGRADED_PERCENT = 90.0
-
-
-def _check(status: str, detail: str | None = None, **extra: Any) -> dict[str, Any]:
-    result: dict[str, Any] = {"status": status}
-    if detail:
-        result["detail"] = detail
-    result.update(extra)
-    return result
-
-
-def _probe_state_db(home: Path) -> dict[str, Any]:
-    path = home / "state.db"
-    if not path.exists():
-        return _check("ok", "not initialized")
-    try:
-        # mode=rw prevents a health probe from creating or repairing state.
-        uri = f"file:{path.as_posix()}?mode=rw"
-        with sqlite3.connect(uri, uri=True, timeout=1.0) as conn:
-            row = conn.execute("PRAGMA quick_check(1)").fetchone()
-            conn.execute("BEGIN IMMEDIATE")
-            conn.execute("ROLLBACK")
-        if not row or str(row[0]).lower() != "ok":
-            return _check("degraded", "integrity check failed")
-        return _check("ok")
-    except Exception as exc:
-        return _check("degraded", type(exc).__name__)
-
-
-def _probe_config(home: Path) -> dict[str, Any]:
-    path = home / "config.yaml"
-    if not path.exists():
-        return _check("ok", "using defaults")
-    try:
-        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
-        if raw is not None and not isinstance(raw, dict):
-            return _check("degraded", "top level is not a mapping")
-        return _check("ok")
-    except Exception as exc:
-        # Recommendation 15 may provide a durable snapshot on a sibling branch.
-        # Recognize it without depending on its implementation or exposing paths.
-        lkg_candidates = (
-            home / "config.last-known-good.yaml",
-            home / ".config.last-known-good.yaml",
-        )
-        if any(candidate.is_file() for candidate in lkg_candidates):
-            return _check("degraded", "invalid config; last-known-good available")
-        return _check("degraded", f"invalid config ({type(exc).__name__})")
-
-
-def _probe_disk(home: Path) -> dict[str, Any]:
-    try:
-        usage = shutil.disk_usage(home)
-        used_pct = round((usage.used / usage.total) * 100, 1) if usage.total else 0.0
-        status = "degraded" if used_pct >= _DISK_DEGRADED_PERCENT else "ok"
-        return _check(status, used_percent=used_pct, free_bytes=usage.free)
-    except Exception as exc:
-        return _check("degraded", type(exc).__name__)
-
-
-def _probe_gateway(runtime_status: dict[str, Any]) -> dict[str, Any]:
-    state = str(runtime_status.get("gateway_state") or "unknown")
-    platforms = runtime_status.get("platforms")
-    connected = 0
-    configured = 0
-    if isinstance(platforms, dict):
-        configured = len(platforms)
-        connected = sum(
-            1
-            for value in platforms.values()
-            if isinstance(value, dict)
-            and str(value.get("state") or value.get("status") or "").lower()
-            in {"connected", "running", "ok"}
-        )
-    status = "ok" if state in {"running", "draining"} else "degraded"
-    return _check(status, state=state, connected_platforms=connected, platforms=configured)
-
-
-def collect_runtime_readiness(
-    *,
-    model_name: str,
-    runtime_status: dict[str, Any] | None,
-    api_run_queue_depth: int = 0,
-    process_completion_queue_depth: int = 0,
-    delegation_completion_queue_depth: int = 0,
-) -> dict[str, Any]:
-    """Return bounded readiness diagnostics without mutating runtime state.
-
-    The detailed health endpoint is authenticated. Even there, probes expose
-    status and counts only: never config values, credentials, paths, commands,
-    queue payloads, or exception messages.
-    """
-    home = get_hermes_home()
-    runtime = runtime_status if isinstance(runtime_status, dict) else {}
-    checks = {
-        "state_db": _probe_state_db(home),
-        "config": _probe_config(home),
-        "provider": _check("ok" if str(model_name or "").strip() else "degraded"),
-        "disk": _probe_disk(home),
-        "gateway": _probe_gateway(runtime),
-        "scheduler": _check(
-            "ok" if runtime.get("updated_at") and runtime.get("gateway_state") == "running" else "degraded",
-            "gateway ticker status unavailable" if not runtime.get("updated_at") else None,
-        ),
-        "background_queues": _check(
-            "ok",
-            api_runs=max(0, int(api_run_queue_depth)),
-            process_completions=max(0, int(process_completion_queue_depth)),
-            delegation_completions=max(0, int(delegation_completion_queue_depth)),
-        ),
-    }
-    overall = "ok" if all(item.get("status") == "ok" for item in checks.values()) else "degraded"
-    return {"status": overall, "checks": checks}
-
-
-__all__ = ["collect_runtime_readiness"]

tests/gateway/test_api_server.py

@@ -568,14 +568,29 @@ class TestConcurrencyCap:
         assert resp.headers.get("Retry-After")
 
     def test_cap_counts_both_buckets(self):
-        # /v1/runs (tracked by _run_streams) + chat/responses (inflight)
+        # /v1/runs (tracked by live tasks) + chat/responses (inflight)
         adapter = _make_adapter()
         adapter._max_concurrent_runs = 4
         adapter._inflight_agent_runs = 2
-        adapter._run_streams = {"r1": object(), "r2": object()}
-        resp = adapter._concurrency_limited_response()
-        assert resp is not None
-        assert resp.status == 429
+
+        async def _assert_live_tasks_are_counted_without_streams():
+            blocker = asyncio.Event()
+
+            async def _live_run():
+                await blocker.wait()
+
+            tasks = [asyncio.create_task(_live_run()) for _ in range(2)]
+            adapter._active_run_tasks = {f"r{i}": task for i, task in enumerate(tasks)}
+            adapter._run_streams = {}
+            try:
+                resp = adapter._concurrency_limited_response()
+                assert resp is not None
+                assert resp.status == 429
+            finally:
+                blocker.set()
+                await asyncio.gather(*tasks)
+
+        asyncio.run(_assert_live_tasks_are_counted_without_streams())
 
     def test_zero_disables_cap(self):
         adapter = _make_adapter()
@@ -764,8 +779,7 @@ class TestHealthDetailedEndpoint:
                 resp = await cli.get("/health/detailed")
                 assert resp.status == 200
                 data = await resp.json()
-                assert data["status"] == "degraded"
-                assert data["readiness"]["checks"]["gateway"]["status"] == "degraded"
+                assert data["status"] == "ok"
                 assert data["gateway_state"] is None
                 assert data["platforms"] == {}
                 # No runtime file ⇒ state None ⇒ not busy, not drainable.
@@ -790,36 +804,6 @@ class TestHealthDetailedEndpoint:
                 resp = await cli.get("/health/detailed", headers=headers)
                 assert resp.status == 200
 
-    @pytest.mark.asyncio
-    async def test_health_detailed_reports_runtime_readiness(self, adapter):
-        """Detailed health exposes bounded readiness probes without changing /health."""
-        app = _create_app(adapter)
-        expected = {
-            "status": "degraded",
-            "checks": {
-                "state_db": {"status": "ok"},
-                "config": {"status": "degraded", "detail": "using last-known-good"},
-            },
-        }
-        with patch("gateway.status.read_runtime_status", return_value={"gateway_state": "running"}), \
-             patch("gateway.platforms.api_server.collect_runtime_readiness", return_value=expected):
-            async with TestClient(TestServer(app)) as cli:
-                resp = await cli.get("/health/detailed")
-                assert resp.status == 200
-                data = await resp.json()
-                assert data["status"] == "degraded"
-                assert data["readiness"] == expected
-
-    @pytest.mark.asyncio
-    async def test_public_health_does_not_run_readiness_probes(self, adapter):
-        app = _create_app(adapter)
-        with patch("gateway.platforms.api_server.collect_runtime_readiness") as probe:
-            async with TestClient(TestServer(app)) as cli:
-                resp = await cli.get("/health")
-                assert resp.status == 200
-                assert (await resp.json())["status"] == "ok"
-        probe.assert_not_called()
-
 
 # ---------------------------------------------------------------------------
 # /v1/models endpoint

tests/gateway/test_api_server_runs.py

@@ -442,6 +442,100 @@ class TestRunEvents:
         assert resp.status == 401
 
 
+# ---------------------------------------------------------------------------
+# Run lifecycle TTL sweeping
+# ---------------------------------------------------------------------------
+
+
+class TestRunLifecycleSweep:
+    @pytest.mark.asyncio
+    async def test_expired_live_run_remains_counted_approvable_and_stoppable(self, adapter):
+        """Stream TTL must not detach control state from a still-running task."""
+        app = _create_runs_app(adapter)
+        adapter._max_concurrent_runs = 1
+
+        async with TestClient(TestServer(app)) as cli:
+            with patch.object(adapter, "_create_agent") as mock_create:
+                mock_agent, agent_ready, _ = _make_slow_agent()
+                mock_create.return_value = mock_agent
+
+                start_resp = await cli.post("/v1/runs", json={"input": "hello"})
+                assert start_resp.status == 202
+                run_id = (await start_resp.json())["run_id"]
+                assert agent_ready.wait(timeout=3.0)
+
+                task = adapter._active_run_tasks[run_id]
+                assert isinstance(task, asyncio.Task)
+                assert not task.done()
+
+                pending = approval_mod._ApprovalEntry({
+                    "command": "bash -c long-running",
+                    "description": "approval after stream TTL",
+                    "pattern_keys": ["shell-c"],
+                })
+                with approval_mod._lock:
+                    approval_mod._gateway_queues[run_id] = [pending]
+
+                adapter._run_streams_created[run_id] -= adapter._RUN_STREAM_TTL + 1
+                # Exercise one real sweeper iteration without waiting 60 seconds.
+                with patch(
+                    "gateway.platforms.api_server.asyncio.sleep",
+                    side_effect=[None, asyncio.CancelledError()],
+                ):
+                    with pytest.raises(asyncio.CancelledError):
+                        await adapter._sweep_orphaned_runs()
+
+                assert adapter._active_run_tasks[run_id] is task
+                assert adapter._active_run_agents[run_id] is mock_agent
+                assert run_id in adapter._run_streams
+                assert adapter._run_approval_sessions[run_id] == run_id
+
+                limited = adapter._concurrency_limited_response()
+                assert limited is not None
+                assert limited.status == 429
+
+                approval_resp = await cli.post(
+                    f"/v1/runs/{run_id}/approval",
+                    json={"choice": "once"},
+                )
+                assert approval_resp.status == 200
+                assert pending.event.is_set()
+                assert pending.result == "once"
+
+                stop_resp = await cli.post(f"/v1/runs/{run_id}/stop")
+                assert stop_resp.status == 200
+                mock_agent.interrupt.assert_called_once_with("Stop requested via API")
+
+    @pytest.mark.asyncio
+    async def test_expired_orphan_run_state_is_reaped(self, adapter):
+        run_id = "run_expired_orphan"
+        adapter._run_streams[run_id] = asyncio.Queue()
+        adapter._run_streams_created[run_id] = 0
+        adapter._run_approval_sessions[run_id] = run_id
+
+        pending = approval_mod._ApprovalEntry({
+            "command": "bash -c orphaned",
+            "description": "orphaned approval",
+            "pattern_keys": ["shell-c"],
+        })
+        with approval_mod._lock:
+            approval_mod._gateway_queues[run_id] = [pending]
+
+        with patch(
+            "gateway.platforms.api_server.asyncio.sleep",
+            side_effect=[None, asyncio.CancelledError()],
+        ):
+            with pytest.raises(asyncio.CancelledError):
+                await adapter._sweep_orphaned_runs()
+
+        assert run_id not in adapter._run_streams
+        assert run_id not in adapter._run_streams_created
+        assert run_id not in adapter._run_approval_sessions
+        assert pending.event.is_set()
+        with approval_mod._lock:
+            assert run_id not in approval_mod._gateway_queues
+
+
 # ---------------------------------------------------------------------------
 # POST /v1/runs/{run_id}/stop — interrupt a running agent
 # ---------------------------------------------------------------------------

tests/gateway/test_readiness.py

@@ -1,103 +0,0 @@
-from __future__ import annotations
-
-import json
-import os
-import sqlite3
-from pathlib import Path
-
-from gateway.readiness import collect_runtime_readiness
-
-
-def test_collect_runtime_readiness_reports_healthy_local_runtime(tmp_path, monkeypatch):
-    home = tmp_path / ".hermes"
-    home.mkdir()
-    (home / "config.yaml").write_text(
-        "model:\n  provider: openrouter\n  model: test/model\n",
-        encoding="utf-8",
-    )
-    with sqlite3.connect(home / "state.db") as conn:
-        conn.execute("CREATE TABLE probe (id INTEGER PRIMARY KEY)")
-    monkeypatch.setenv("HERMES_HOME", str(home))
-
-    result = collect_runtime_readiness(
-        model_name="test/model",
-        runtime_status={
-            "gateway_state": "running",
-            "platforms": {"telegram": {"state": "connected"}},
-            "updated_at": "2026-07-09T00:00:00Z",
-        },
-        api_run_queue_depth=2,
-    )
-
-    assert result["status"] == "ok"
-    assert result["checks"]["state_db"]["status"] == "ok"
-    assert result["checks"]["config"]["status"] == "ok"
-    assert result["checks"]["provider"]["status"] == "ok"
-    assert result["checks"]["gateway"]["status"] == "ok"
-    assert result["checks"]["background_queues"]["api_runs"] == 2
-    assert result["checks"]["disk"]["status"] in {"ok", "degraded"}
-
-
-def test_collect_runtime_readiness_degrades_on_invalid_config_and_stopped_gateway(
-    tmp_path, monkeypatch
-):
-    home = tmp_path / ".hermes"
-    home.mkdir()
-    (home / "config.yaml").write_text("model: [unterminated", encoding="utf-8")
-    monkeypatch.setenv("HERMES_HOME", str(home))
-
-    result = collect_runtime_readiness(
-        model_name="",
-        runtime_status={"gateway_state": "stopped", "platforms": {}},
-    )
-
-    assert result["status"] == "degraded"
-    assert result["checks"]["config"]["status"] == "degraded"
-    assert result["checks"]["provider"]["status"] == "degraded"
-    assert result["checks"]["gateway"]["status"] == "degraded"
-    # Readiness is diagnostic data, not an exception or a destructive repair.
-    assert (home / "config.yaml").read_text(encoding="utf-8") == "model: [unterminated"
-
-
-def test_collect_runtime_readiness_marks_corrupt_state_db_degraded(tmp_path, monkeypatch):
-    home = tmp_path / ".hermes"
-    home.mkdir()
-    (home / "config.yaml").write_text("{}\n", encoding="utf-8")
-    (home / "state.db").write_bytes(b"not sqlite")
-    monkeypatch.setenv("HERMES_HOME", str(home))
-
-    result = collect_runtime_readiness(model_name="configured-model", runtime_status={})
-
-    assert result["status"] == "degraded"
-    assert result["checks"]["state_db"]["status"] == "degraded"
-    assert "detail" in result["checks"]["state_db"]
-
-
-def test_collect_runtime_readiness_never_exposes_config_values(tmp_path, monkeypatch):
-    home = tmp_path / ".hermes"
-    home.mkdir()
-    secret = "do-not-return-this-value"
-    (home / "config.yaml").write_text(
-        f"model:\n  provider: openrouter\nprivate_value: {secret}\n",
-        encoding="utf-8",
-    )
-    monkeypatch.setenv("HERMES_HOME", str(home))
-
-    result = collect_runtime_readiness(model_name="model", runtime_status={})
-
-    assert secret not in json.dumps(result)
-    assert str(home) not in json.dumps(result)
-    assert result["checks"]["config"]["status"] == "ok"
-
-
-def test_collect_runtime_readiness_uses_active_profile_home(tmp_path, monkeypatch):
-    profile_home = tmp_path / "profiles" / "coder"
-    profile_home.mkdir(parents=True)
-    (profile_home / "config.yaml").write_text("{}\n", encoding="utf-8")
-    monkeypatch.setenv("HERMES_HOME", str(profile_home))
-
-    result = collect_runtime_readiness(model_name="model", runtime_status={})
-
-    assert result["checks"]["config"]["status"] == "ok"
-    assert not (tmp_path / ".hermes" / "state.db").exists()
-    assert os.environ["HERMES_HOME"] == str(profile_home)