[harness] feat: 新增 GET /agent/runtime 管理员运行状态端点

app/main.py

@@ -452,6 +452,11 @@ async def request_context_middleware(request: Request, call_next):
 app.include_router(notes_router)
 app.include_router(tasks_ws_router)
 
+# Agent runtime status - admin-only observability endpoint
+from app.routers.agent_runtime import router as agent_runtime_router  # noqa: E402
+
+app.include_router(agent_runtime_router)
+
 # Plan 0021: Cloud drive router (with graceful degradation)
 try:
     from app.routers.cloud import router as cloud_router

app/routers/agent_runtime.py

@@ -0,0 +1,41 @@
+"""Admin-only endpoint exposing AgentHarness runtime status.
+
+``GET /agent/runtime`` returns a snapshot of the harness's in-memory state
+(``harness.health()``): circuit breaker, per-tool metrics, scheduler stats,
+registered agents, and tool discovery. Intended for the ops/observability
+panel - it is a transient snapshot, not a persisted history.
+
+Returns 503 with the root cause when the harness failed to start or was
+skipped (e.g. LLM not configured).
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+
+from app.routers.auth import require_admin
+
+router = APIRouter(prefix="/agent", tags=["agent-runtime"])
+
+
+@router.get("/runtime")
+async def get_agent_runtime(
+    request: Request,
+    _uid: int = Depends(require_admin),
+) -> dict:
+    """Return a snapshot of the AgentHarness runtime state (admin-only)."""
+    status = getattr(request.app.state, "agent_harness_status", "unknown")
+    error = getattr(request.app.state, "agent_harness_error", None)
+    harness = getattr(request.app.state, "agent_harness", None)
+
+    if status != "started" or harness is None:
+        if status == "failed" and error:
+            raise HTTPException(status_code=503, detail=f"Agent 服务启动失败: {error}")
+        if status == "skipped":
+            raise HTTPException(
+                status_code=503,
+                detail=f"Agent 服务未启动: {error or 'LLM 未配置'}",
+            )
+        raise HTTPException(status_code=503, detail="Agent 服务暂不可用")
+
+    return await harness.health()

app/test/test_agent_runtime_api.py

@@ -0,0 +1,96 @@
+"""Tests for GET /agent/runtime - admin-only agent harness status endpoint."""
+
+from __future__ import annotations
+
+import pytest
+import pytest_asyncio
+from fastapi import HTTPException
+from httpx import ASGITransport, AsyncClient
+from unittest.mock import AsyncMock, MagicMock
+
+from app.main import app
+from app.routers.auth import require_admin
+
+
+@pytest_asyncio.fixture
+async def client():
+    """ASGI test client - no lifespan, so app.state is set manually per test."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        yield ac
+    app.dependency_overrides.clear()
+
+
+def _set_harness(status: str, health: dict | None = None, error: str | None = None) -> None:
+    """Inject a mock harness + status onto app.state."""
+    mock_harness = MagicMock()
+    mock_harness.started = status == "started"
+    mock_harness.health = AsyncMock(return_value=health or {})
+    app.state.agent_harness = mock_harness
+    app.state.agent_harness_status = status
+    app.state.agent_harness_error = error
+
+
+def _as_admin(uid: int = 1) -> None:
+    """Override require_admin so the caller is treated as an admin."""
+
+    async def _override() -> int:
+        return uid
+
+    app.dependency_overrides[require_admin] = _override
+
+
+def _as_non_admin() -> None:
+    """Override require_admin to reject (simulates a non-admin caller)."""
+
+    async def _override() -> int:
+        raise HTTPException(status_code=403, detail="需要管理员权限")
+
+    app.dependency_overrides[require_admin] = _override
+
+
+class TestGetAgentRuntime:
+    @pytest.mark.asyncio
+    async def test_started_returns_health_snapshot(self, client):
+        _as_admin()
+        _set_harness(
+            "started",
+            health={
+                "status": "running",
+                "registered_agents": ["chat", "memory", "quiz"],
+                "sessions_active": 2,
+                "circuit_breaker": {"state": "closed", "failures": 0},
+            },
+        )
+
+        resp = await client.get("/agent/runtime")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["status"] == "running"
+        assert "chat" in data["registered_agents"]
+        assert data["sessions_active"] == 2
+
+    @pytest.mark.asyncio
+    async def test_skipped_returns_503(self, client):
+        _as_admin()
+        _set_harness("skipped", error="LLM 未配置")
+
+        resp = await client.get("/agent/runtime")
+        assert resp.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_failed_returns_503_with_cause(self, client):
+        _as_admin()
+        _set_harness("failed", error="ModuleNotFoundError: No module named 'langgraph'")
+
+        resp = await client.get("/agent/runtime")
+        assert resp.status_code == 503
+        assert "langgraph" in resp.json()["detail"]
+
+    @pytest.mark.asyncio
+    async def test_non_admin_returns_403(self, client):
+        _as_non_admin()
+        _set_harness("started", health={"status": "running"})
+
+        resp = await client.get("/agent/runtime")
+        assert resp.status_code == 403