From 27870e7953356129332ebd1e881ff799573acccb Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 19:46:13 +0200 Subject: [PATCH] [F087,F088] enforce panel token on live-chat bridges (Phase 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a CEO-bound, header-token-only gate (require_panel_token) at the route level of the prompter_live + secretary_live bridges, which were the only panel-facing API surface that ran unauthenticated. It mirrors the WS _require_panel_token and _check_agent_auth_token contracts: in dev (ROBOCO_AGENT_AUTH_REQUIRED unset) a missing token is allowed; a presented-but-forged token is rejected even in dev; in prod nginx already injects the CEO-signed X-Agent-Token on /api/ for GET + POST, so the SSE stream (EventSource can't set headers) and the POSTs are now checked instead of anonymous. Applied to start/stream/status/messages/stop on both routers; preview_live_batch switched from CurrentAgentContext+noqa to the route-level gate (genuinely auth-only). confirm/confirm-batch/re-interview keep CurrentAgentContext (they use agent.identity). The container->relay /events callback is intentionally left ungated (internal Docker network, opaque session id) — gated by a test sentinel so Option B (spawn+SDK token wiring) is a deliberate future decision. No panel/nginx/spawn/SDK changes; master merge invariant untouched. 22 new TDD auth tests, 492 api tests green. --- roboco/api/deps.py | 28 ++- roboco/api/routes/prompter_live.py | 17 +- roboco/api/routes/secretary_live.py | 13 +- tests/unit/api/test_prompter_live_auth.py | 276 +++++++++++++++++++++ tests/unit/api/test_secretary_live_auth.py | 190 ++++++++++++++ 5 files changed, 510 insertions(+), 14 deletions(-) create mode 100644 tests/unit/api/test_prompter_live_auth.py create mode 100644 tests/unit/api/test_secretary_live_auth.py diff --git a/roboco/api/deps.py b/roboco/api/deps.py index 65e382aa..020a59b6 100644 --- a/roboco/api/deps.py +++ b/roboco/api/deps.py @@ -15,7 +15,7 @@ from fastapi import Depends, Header, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from roboco.agents_config import verify_agent_token +from roboco.agents_config import CEO_AGENT_ID, verify_agent_token from roboco.api.schemas.optimal import PaginationParams from roboco.db.base import get_db from roboco.db.tables import AgentTable @@ -232,6 +232,32 @@ def _check_agent_auth_token( ) +def require_panel_token( + x_agent_token: Annotated[str | None, Header(alias="X-Agent-Token")] = None, +) -> None: + """Panel (CEO) HMAC gate for the live-chat bridges. + + The HTTP analog of the WS ``_require_panel_token``: the panel is the only + caller of the live intake/secretary chat, nginx injects the CEO-signed + ``X-Agent-Token`` on ``/api/`` in prod, and browser ``EventSource`` cannot + set headers — so the gate is token-only (no ``X-Agent-ID``; the stream is + session-keyed and the panel is the sole client). In dev + (``ROBOCO_AGENT_AUTH_REQUIRED`` unset) a missing token is allowed; a + presented-but-forged token is still rejected, matching + ``_check_agent_auth_token`` and the WS gate. + """ + if _auth_required() and not x_agent_token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing X-Agent-Token header (auth required)", + ) + if x_agent_token and not verify_agent_token(x_agent_token, CEO_AGENT_ID, "ceo", ""): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid X-Agent-Token — signature mismatch.", + ) + + async def _resolve_agent_identity( db: DbSession, x_agent_id: str, x_agent_role: str ) -> tuple[UUID, str]: diff --git a/roboco/api/routes/prompter_live.py b/roboco/api/routes/prompter_live.py index 87d51dd3..122c049d 100644 --- a/roboco/api/routes/prompter_live.py +++ b/roboco/api/routes/prompter_live.py @@ -20,13 +20,14 @@ import json from typing import TYPE_CHECKING, Any from uuid import UUID, uuid4 -from fastapi import APIRouter, HTTPException, Request, status +from fastapi import APIRouter, Depends, HTTPException, Request, status from sse_starlette import EventSourceResponse from roboco.api.deps import ( CurrentAgentContext, DbSession, get_orchestrator, + require_panel_token, require_pm_or_above, ) from roboco.api.schemas.prompter_live import ( @@ -74,6 +75,7 @@ def _translate_service_error(e: ServiceError) -> HTTPException: "/live/start", response_model=StartLiveResponse, status_code=status.HTTP_201_CREATED, + dependencies=[Depends(require_panel_token)], ) async def start_live(body: StartLiveRequest, db: DbSession) -> StartLiveResponse: """Spawn the intake agent for a new chat and return its session id. @@ -117,7 +119,7 @@ async def start_live(body: StartLiveRequest, db: DbSession) -> StartLiveResponse return StartLiveResponse(session_id=session_id) -@router.get("/live/{session_id}/stream") +@router.get("/live/{session_id}/stream", dependencies=[Depends(require_panel_token)]) async def stream(session_id: str, request: Request) -> EventSourceResponse: """Stream the agent's live events (token deltas, tool calls) to the panel.""" registry = get_live_registry() @@ -131,7 +133,7 @@ async def stream(session_id: str, request: Request) -> EventSourceResponse: return EventSourceResponse(events(), ping=15) -@router.get("/live/{session_id}/status") +@router.get("/live/{session_id}/status", dependencies=[Depends(require_panel_token)]) async def session_status(session_id: str) -> dict[str, bool]: """Report whether a live intake session is still running. @@ -142,7 +144,7 @@ async def session_status(session_id: str) -> dict[str, bool]: return {"alive": get_live_registry().is_alive(session_id)} -@router.post("/live/{session_id}/messages") +@router.post("/live/{session_id}/messages", dependencies=[Depends(require_panel_token)]) async def send_message(session_id: str, body: LiveMessageRequest) -> dict[str, bool]: """Deliver the human's message to the running intake agent.""" delivered = await get_live_registry().deliver(session_id, body.text) @@ -157,7 +159,7 @@ async def send_message(session_id: str, body: LiveMessageRequest) -> dict[str, b return {"delivered": True} -@router.post("/live/{session_id}/stop") +@router.post("/live/{session_id}/stop", dependencies=[Depends(require_panel_token)]) async def stop_live(session_id: str) -> dict[str, bool]: """Reap the live intake session (panel close, or draft confirmed).""" await get_orchestrator().reap_intake_session(session_id) @@ -215,12 +217,13 @@ async def confirm_live( return {"task_id": str(task_id)} -@router.post("/live/{session_id}/preview-batch") +@router.post( + "/live/{session_id}/preview-batch", dependencies=[Depends(require_panel_token)] +) async def preview_live_batch( session_id: str, # noqa: ARG001 — kept for route symmetry; preview is pure body: BatchPreviewRequest, db: DbSession, - agent: CurrentAgentContext, # noqa: ARG001 — auth context only ) -> dict[str, Any]: """Compute a MegaTask's waves from the proposed drafts WITHOUT creating it. diff --git a/roboco/api/routes/secretary_live.py b/roboco/api/routes/secretary_live.py index 0d60f0dc..ea43842b 100644 --- a/roboco/api/routes/secretary_live.py +++ b/roboco/api/routes/secretary_live.py @@ -19,10 +19,10 @@ import json from typing import TYPE_CHECKING, Any from uuid import uuid4 -from fastapi import APIRouter, HTTPException, Request, status +from fastapi import APIRouter, Depends, HTTPException, Request, status from sse_starlette import EventSourceResponse -from roboco.api.deps import get_orchestrator +from roboco.api.deps import get_orchestrator, require_panel_token from roboco.api.schemas.secretary_live import ( AgentEvent, LiveMessageRequest, @@ -41,6 +41,7 @@ router = APIRouter() "/live/start", response_model=StartSecretaryResponse, status_code=status.HTTP_201_CREATED, + dependencies=[Depends(require_panel_token)], ) async def start_live(body: StartSecretaryRequest) -> StartSecretaryResponse: """Spawn the Secretary agent for a new chat and return its session id.""" @@ -57,7 +58,7 @@ async def start_live(body: StartSecretaryRequest) -> StartSecretaryResponse: return StartSecretaryResponse(session_id=session_id) -@router.get("/live/{session_id}/stream") +@router.get("/live/{session_id}/stream", dependencies=[Depends(require_panel_token)]) async def stream(session_id: str, request: Request) -> EventSourceResponse: """Stream the Secretary's live events (token deltas, tool calls) to the panel.""" registry = get_live_registry() @@ -71,13 +72,13 @@ async def stream(session_id: str, request: Request) -> EventSourceResponse: return EventSourceResponse(events(), ping=15) -@router.get("/live/{session_id}/status") +@router.get("/live/{session_id}/status", dependencies=[Depends(require_panel_token)]) async def session_status(session_id: str) -> dict[str, bool]: """Report whether a live Secretary session is still running.""" return {"alive": get_live_registry().is_alive(session_id)} -@router.post("/live/{session_id}/messages") +@router.post("/live/{session_id}/messages", dependencies=[Depends(require_panel_token)]) async def send_message(session_id: str, body: LiveMessageRequest) -> dict[str, bool]: """Deliver the CEO's message to the running Secretary agent.""" delivered = await get_live_registry().deliver(session_id, body.text) @@ -92,7 +93,7 @@ async def send_message(session_id: str, body: LiveMessageRequest) -> dict[str, b return {"delivered": True} -@router.post("/live/{session_id}/stop") +@router.post("/live/{session_id}/stop", dependencies=[Depends(require_panel_token)]) async def stop_live(session_id: str) -> dict[str, bool]: """Reap the live Secretary session.""" await get_orchestrator().reap_secretary_session(session_id) diff --git a/tests/unit/api/test_prompter_live_auth.py b/tests/unit/api/test_prompter_live_auth.py new file mode 100644 index 00000000..f739d0c1 --- /dev/null +++ b/tests/unit/api/test_prompter_live_auth.py @@ -0,0 +1,276 @@ +"""Token enforcement on the live intake chat (Phase 5). + +The panel-facing ``prompter_live`` routes used to take no auth dependency — the +SSE stream carried no identity (browser ``EventSource`` can't set headers) and +``start``/``status``/``messages``/``stop`` accepted anonymous calls. The fix +adds a CEO-bound, header-token-only gate (``require_panel_token``) at the route +level: in prod nginx injects the CEO-signed ``X-Agent-Token`` on ``/api/``, and +in dev a missing token is allowed while a presented-but-forged one is still +rejected (matching ``_check_agent_auth_token`` and the WS gate). + +These tests mount the router on a bare ``FastAPI()`` (no ``setup_middleware``), +so the gate must raise ``HTTPException(401)`` — the same shape as the a2a auth +tests. +""" + +from __future__ import annotations + +from http import HTTPStatus +from typing import TYPE_CHECKING, Any +from uuid import uuid4 + +import httpx +import pytest +import pytest_asyncio +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.agents_config import issue_panel_token +from roboco.api import deps +from roboco.api.routes.prompter_live import router as prompter_live_router +from roboco.db.base import get_db +from roboco.services import prompter_live + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + +_SECRET = "test-secret-for-prompter-live-auth" +_HTTP_401 = HTTPStatus.UNAUTHORIZED + + +class _FakeOrchestrator: + """Records spawn/reap calls; stands in for the real orchestrator singleton.""" + + def __init__(self) -> None: + self.spawned: list[dict[str, Any]] = [] + self.reaped: list[str] = [] + + async def start_intake_session( + self, + session_id: str, + *, + project_slug: str | None = None, + product_id: str | None = None, + project_ids: list[str] | None = None, + initial_message: str | None = None, + ) -> None: + self.spawned.append( + { + "session_id": session_id, + "project_slug": project_slug, + "product_id": product_id, + "project_ids": project_ids, + "initial_message": initial_message, + } + ) + + async def reap_intake_session(self, session_id: str) -> None: + self.reaped.append(session_id) + + +@pytest_asyncio.fixture +async def auth_client( + monkeypatch: pytest.MonkeyPatch, +) -> AsyncIterator[AsyncClient]: + """Mounted router + fake orchestrator + empty registry; no auth env set. + + Each test monkeypatches ``ROBOCO_AGENT_AUTH_SECRET`` and + ``ROBOCO_AGENT_AUTH_REQUIRED`` to pick dev vs strict mode. The registry is + empty so the SSE stream over an unknown session yields nothing (200), + ``status`` reports dead, ``messages`` 404s, and ``/events`` reports + ``pushed: false`` — all non-401, which is what the "gate passed" assertions + need. + """ + orch = _FakeOrchestrator() + monkeypatch.setattr(deps._ServiceHolder, "orchestrator", orch) + + def container_handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True}) + + mock_client = httpx.AsyncClient(transport=httpx.MockTransport(container_handler)) + registry = prompter_live.PrompterLiveRegistry(http_client=mock_client) + prompter_live._RegistryHolder.instance = registry + + async def _fake_db() -> AsyncIterator[object]: + yield object() + + app = FastAPI() + app.include_router(prompter_live_router, prefix="/api/prompter") + app.dependency_overrides[get_db] = _fake_db + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + yield client + + prompter_live._RegistryHolder.instance = None + await mock_client.aclose() + app.dependency_overrides.clear() + + +def _strict(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + + +def _dev(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + + +def _start_body() -> dict[str, Any]: + return {"product_id": str(uuid4()), "initial_message": "build X"} + + +# --------------------------------------------------------------------------- +# /live/start +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_start_rejects_missing_token_when_required( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post("/api/prompter/live/start", json=_start_body()) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_start_rejects_forged_token_when_required( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post( + "/api/prompter/live/start", + json=_start_body(), + headers={"X-Agent-Token": "forged-not-a-real-hmac"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_start_accepts_valid_panel_token( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post( + "/api/prompter/live/start", + json=_start_body(), + headers={"X-Agent-Token": issue_panel_token()}, + ) + assert r.status_code == HTTPStatus.CREATED # gate passed -> 201 + + +@pytest.mark.asyncio +async def test_start_rejects_forged_token_even_in_dev( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A presented-but-forged token is rejected even in header-trust mode.""" + _dev(monkeypatch) + r = await auth_client.post( + "/api/prompter/live/start", + json=_start_body(), + headers={"X-Agent-Token": "forged-not-a-real-hmac"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_start_dev_mode_missing_token_succeeds( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _dev(monkeypatch) + r = await auth_client.post("/api/prompter/live/start", json=_start_body()) + assert r.status_code == HTTPStatus.CREATED # dev flow preserved + + +# --------------------------------------------------------------------------- +# /live/{id}/stream (SSE) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_stream_rejects_missing_token_when_required( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.get("/api/prompter/live/unknown/stream") + assert r.status_code == _HTTP_401 # 401 before the EventSourceResponse starts + + +@pytest.mark.asyncio +async def test_stream_accepts_valid_panel_token( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.get( + "/api/prompter/live/unknown/stream", + headers={"X-Agent-Token": issue_panel_token()}, + ) + assert r.status_code == HTTPStatus.OK # unknown session -> empty stream -> 200 + + +# --------------------------------------------------------------------------- +# /live/{id}/status, /messages, /stop +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("method", "path", "json"), + [ + ("GET", "/api/prompter/live/unknown/status", None), + ("POST", "/api/prompter/live/unknown/messages", {"text": "hi"}), + ("POST", "/api/prompter/live/sess/stop", None), + ], +) +@pytest.mark.asyncio +async def test_status_send_stop_reject_missing_token_when_required( + auth_client: AsyncClient, + monkeypatch: pytest.MonkeyPatch, + method: str, + path: str, + json: dict[str, Any] | None, +) -> None: + _strict(monkeypatch) + if method == "GET": + r = await auth_client.get(path) + else: + r = await auth_client.post(path, json=json) + assert r.status_code == _HTTP_401 + + +# --------------------------------------------------------------------------- +# /live/{id}/preview-batch — switched from CurrentAgentContext to the panel gate +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_preview_batch_passes_with_valid_token_in_strict_mode( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post( + "/api/prompter/live/s1/preview-batch", + json={"drafts": [{"title": "A"}, {"title": "B"}]}, + headers={"X-Agent-Token": issue_panel_token()}, + ) + # Gate passed -> 200 with waves (preview is pure compute; no session needed). + assert r.status_code == HTTPStatus.OK + + +# --------------------------------------------------------------------------- +# /live/{id}/events — container -> relay, intentionally UNGATED (scope sentinel) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_events_ungated_in_strict_mode( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """``/events`` is the container->relay callback on the internal Docker + network (opaque session id). Option A leaves it ungated; this test pins + that decision so a future gating change can't land silently.""" + _strict(monkeypatch) + r = await auth_client.post( + "/api/prompter/live/unknown/events", json={"kind": "text"} + ) + assert r.status_code == HTTPStatus.OK # ungated -> 200 (pushed: false) + assert r.json() == {"pushed": False} diff --git a/tests/unit/api/test_secretary_live_auth.py b/tests/unit/api/test_secretary_live_auth.py new file mode 100644 index 00000000..36204605 --- /dev/null +++ b/tests/unit/api/test_secretary_live_auth.py @@ -0,0 +1,190 @@ +"""Token enforcement on the live Secretary chat (Phase 5). + +Mirror of ``test_prompter_live_auth`` for the ``secretary_live`` router, which +previously had zero auth on any endpoint. The same ``require_panel_token`` gate +applies at the route level. The Secretary's *authority* (directive execution) +is gated separately at ``/api/secretary/directives``; this only closes the +live-chat transport. +""" + +from __future__ import annotations + +from http import HTTPStatus +from typing import TYPE_CHECKING, Any + +import httpx +import pytest +import pytest_asyncio +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.agents_config import issue_panel_token +from roboco.api import deps +from roboco.api.routes.secretary_live import router as secretary_live_router +from roboco.services import prompter_live + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + +_SECRET = "test-secret-for-secretary-live-auth" +_HTTP_401 = HTTPStatus.UNAUTHORIZED + + +class _FakeOrchestrator: + def __init__(self) -> None: + self.spawned: list[dict[str, Any]] = [] + self.reaped: list[str] = [] + + async def start_secretary_session( + self, session_id: str, *, initial_message: str | None = None + ) -> None: + self.spawned.append( + {"session_id": session_id, "initial_message": initial_message} + ) + + async def reap_secretary_session(self, session_id: str) -> None: + self.reaped.append(session_id) + + +@pytest_asyncio.fixture +async def auth_client( + monkeypatch: pytest.MonkeyPatch, +) -> AsyncIterator[AsyncClient]: + orch = _FakeOrchestrator() + monkeypatch.setattr(deps._ServiceHolder, "orchestrator", orch) + + def container_handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True}) + + mock_client = httpx.AsyncClient(transport=httpx.MockTransport(container_handler)) + registry = prompter_live.PrompterLiveRegistry(http_client=mock_client) + prompter_live._RegistryHolder.instance = registry + + app = FastAPI() + app.include_router(secretary_live_router, prefix="/api/secretary") + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + yield client + + prompter_live._RegistryHolder.instance = None + await mock_client.aclose() + + +def _strict(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + + +def _dev(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + + +def _start_body() -> dict[str, Any]: + return {"initial_message": "hi"} + + +@pytest.mark.asyncio +async def test_start_rejects_missing_token_when_required( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post("/api/secretary/live/start", json=_start_body()) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_start_rejects_forged_token_when_required( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post( + "/api/secretary/live/start", + json=_start_body(), + headers={"X-Agent-Token": "forged-not-a-real-hmac"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_start_accepts_valid_panel_token( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.post( + "/api/secretary/live/start", + json=_start_body(), + headers={"X-Agent-Token": issue_panel_token()}, + ) + assert r.status_code == HTTPStatus.CREATED + + +@pytest.mark.asyncio +async def test_start_rejects_forged_token_even_in_dev( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _dev(monkeypatch) + r = await auth_client.post( + "/api/secretary/live/start", + json=_start_body(), + headers={"X-Agent-Token": "forged-not-a-real-hmac"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_stream_rejects_missing_token_when_required( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.get("/api/secretary/live/unknown/stream") + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_stream_accepts_valid_panel_token( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + _strict(monkeypatch) + r = await auth_client.get( + "/api/secretary/live/unknown/stream", + headers={"X-Agent-Token": issue_panel_token()}, + ) + assert r.status_code == HTTPStatus.OK + + +@pytest.mark.parametrize( + ("method", "path", "json"), + [ + ("GET", "/api/secretary/live/unknown/status", None), + ("POST", "/api/secretary/live/unknown/messages", {"text": "hi"}), + ("POST", "/api/secretary/live/sess/stop", None), + ], +) +@pytest.mark.asyncio +async def test_status_send_stop_reject_missing_token_when_required( + auth_client: AsyncClient, + monkeypatch: pytest.MonkeyPatch, + method: str, + path: str, + json: dict[str, Any] | None, +) -> None: + _strict(monkeypatch) + if method == "GET": + r = await auth_client.get(path) + else: + r = await auth_client.post(path, json=json) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_events_ungated_in_strict_mode( + auth_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """``/events`` is the container->relay callback; intentionally ungated + (scope sentinel mirroring the prompter test).""" + _strict(monkeypatch) + r = await auth_client.post( + "/api/secretary/live/unknown/events", json={"kind": "text"} + ) + assert r.status_code == HTTPStatus.OK + assert r.json() == {"pushed": False}