From 47cd5f00abfbc3308d1f7fbf06f592fb51104775 Mon Sep 17 00:00:00 2001 From: ananthanandanan Date: Wed, 5 Aug 2026 20:54:11 +0530 Subject: [PATCH] feat: read session history over HTTP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /run lets a client continue a thread, but nothing let it read one back — sessions JSONL was only reachable from the CLI, so a browser client that reloaded lost the conversation unless it mirrored every event into a store of its own. Adds two endpoints over the store the harness already owns: - GET /sessions — every session's metadata, the same shape `reigner session list --json` emits. - GET /sessions/{id}/events — one session's transcript as the same event envelopes /run streams, so a replay is byte-identical to what the client would have seen live. ?limit=N returns the tail via a bounded deque; total still counts the whole transcript. Also fixes a live 500: store.exists() validates the id before touching disk, so a session_id with path separators raised InvalidSessionId out of POST /run. Unknown and malformed ids now both answer 404 through one shared guard, which runs before the read — load_events is a generator, so its validation would otherwise fire inside the loop and read as a torn line. An unreadable stored transcript answers 422 rather than silently returning a short one. `serve` lists all four routes on startup and warns on stderr when bound to anything other than loopback: these endpoints serve every transcript on disk, so an accidental exposure costs more than it used to. Closes #130 --- docs/guide/usage.md | 30 ++++++- reigner/cli/serve.py | 32 ++++++- reigner/server/fastapi_app.py | 150 +++++++++++++++++++++++++++---- tests/cli/test_serve.py | 75 ++++++++++++++++ tests/server/test_fastapi_app.py | 142 +++++++++++++++++++++++++++++ 5 files changed, 410 insertions(+), 19 deletions(-) create mode 100644 tests/cli/test_serve.py diff --git a/docs/guide/usage.md b/docs/guide/usage.md index d3c9e74..37703f6 100644 --- a/docs/guide/usage.md +++ b/docs/guide/usage.md @@ -80,6 +80,7 @@ output in Section 3 is marked representative. | Inspect artifacts/index | ✅ | `reigner inspect {artifacts,index}` | [Section 3.4](#34-inspect-the-project--reigner-inspect) | | Sessions — list/show/tree/fork/replay | ✅ | `reigner session …` | [Section 3.5](#35-sessions-list--show--tree--fork--replay--reigner-session) | | Serve — HTTP / SSE | ✅ | `reigner serve --http` | [Section 3.6](#36-serve-the-agent--reigner-serve) | +| Serve — read session history | ✅ | `GET /sessions` · `GET /sessions/{id}/events` | [Section 3.6](#36-serve-the-agent--reigner-serve) | | Serve — MCP export | ⏳ | `reigner serve --mcp` | [Section 3.6](#36-serve-the-agent--reigner-serve) | | Plugins — metrics, PII redact | ✅ | `plugins:` in `reigner.yaml` | [Section 3.7](#37-plugins) | | Skills (on-demand modules) | ✅ | `role.skills:` in `reigner.yaml` | [Section 3.9](#39-skills--on-demand-instruction-modules) | @@ -1028,19 +1029,44 @@ Expose the configured agent over HTTP with SSE streaming (needs ```console $ reigner serve --http # representative output -· listening on http://127.0.0.1:8000 (POST /run · GET /health) +· listening on http://127.0.0.1:8000 + (POST /run · GET /sessions · GET /sessions/{id}/events · GET /health) ``` -Two endpoints: +Four endpoints: - `GET /health` → `{"status": "ok", "name": ..., "model": ...}` — liveness + identity probe. - `POST /run` → an SSE stream of the same typed events as `chat --json`. Body: `{"query": "...", "session_id": "optional", "profile": "full"}`. +- `GET /sessions` → `{"sessions": [...]}`, one entry per session in the store + with the same fields `reigner session list --json` emits (`session_id`, + `parent_id`, `title`, `created`, `last_updated`, `event_count`, + `schema_version`). +- `GET /sessions/{id}/events` → `{"session_id", "total", "truncated", "events"}` + — a session's stored transcript, in write order, as the same event envelopes + `/run` streams. Add `?limit=N` for just the last N events; `total` still + counts the whole transcript so you can tell what you skipped. Flags: `--host` (defaults to loopback; set `0.0.0.0` to expose), `--port` (default `8000`), `-c` for a non-default config. +Reading history back is what lets a browser client survive a reload: `POST /run` +returns a `session_id` on every frame, and `GET /sessions/{id}/events` replays +that thread later. Note that `EventSource` can't drive `/run` — it only issues +GETs — so browser clients call `fetch()` and parse the stream themselves. + +Unknown and malformed session ids both answer `404`; a stored transcript that +won't parse answers `422` rather than silently returning a short one. + +!!! warning "No auth, CORS, or rate limiting" + + The server ships none of it, deliberately — put a gateway in front before + exposing it. The session endpoints raise the stakes: an exposed server + serves every transcript on disk, not just the ability to ask a question. + `reigner serve` prints a reminder to stderr whenever you bind to anything + other than loopback. + ⏳ **MCP export is not implemented yet** — `--mcp` exits cleanly rather than pretending to work: diff --git a/reigner/cli/serve.py b/reigner/cli/serve.py index a58f67e..fdc7689 100644 --- a/reigner/cli/serve.py +++ b/reigner/cli/serve.py @@ -10,6 +10,7 @@ from __future__ import annotations +import ipaddress from pathlib import Path import typer @@ -100,5 +101,34 @@ def _run_http(cfg: ReignerConfig, harness: Harness, *, host: str, port: int) -> app = create_app(harness, name=cfg.name, model=model) typer.echo(f"· reigner http server — {cfg.name} ({model})") - typer.echo(f"· listening on http://{host}:{port} (POST /run · GET /health)") + if not _is_loopback(host): + typer.echo( + f"! bound to {host} with no auth, CORS, or rate limiting\n" + f" — put a gateway in front before exposing this.", + err=True, + ) + typer.echo(f"· listening on http://{host}:{port}") + typer.echo(f" ({' · '.join(_ROUTES)})") uvicorn.run(app, host=host, port=port) + + +_ROUTES = ( + "POST /run", + "GET /sessions", + "GET /sessions/{id}/events", + "GET /health", +) + + +def _is_loopback(host: str) -> bool: + """Whether ``host`` keeps the server on this machine. + + Anything else reaches the network, and the session endpoints serve every + stored transcript — so that bind earns a warning. Unparseable hosts (a + name, not an address) are treated as exposed: warning on a loopback alias + is cheap, staying quiet on a public bind is not. + """ + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return host == "localhost" diff --git a/reigner/server/fastapi_app.py b/reigner/server/fastapi_app.py index 834a2db..ed2afe8 100644 --- a/reigner/server/fastapi_app.py +++ b/reigner/server/fastapi_app.py @@ -1,23 +1,36 @@ """Optional FastAPI HTTP server — deploy a Reigner agent as a service. -Two endpoints over one shared, immutable :class:`Harness`: +Four endpoints over one shared, immutable :class:`Harness`: - ``POST /run`` — stream the agent's event protocol as Server-Sent Events. An optional ``session_id`` resumes a durable session; absent, a new one is minted. The new/resumed id rides every frame (it's on the event envelope), so a client reads it off the first frame to continue later. +- ``GET /sessions`` — list every session in the store with its metadata. +- ``GET /sessions/{id}/events`` — replay one session's stored transcript, so a + client that reloads can restore the thread instead of mirroring every event + into a store of its own. - ``GET /health`` — liveness + identity probe for load balancers and operators. The server adds no new event types and owns no output path of its own: every frame is ``to_json(event)``, the exact bytes the CLI's ``--json`` -mode emits, just wrapped in SSE framing. Build with :func:`create_app`; the -``serve`` CLI command injects a live harness and the display strings for -``/health``. +mode emits, just wrapped in SSE framing. The replay endpoint returns those same +envelopes as JSON. Build with :func:`create_app`; the ``serve`` CLI command +injects a live harness and the display strings for ``/health``. + +**No auth, CORS, or rate limiting ships here** — that's deliberate scope, not an +oversight. Put a gateway in front before exposing this. Note that the read +endpoints raise the stakes of an accidental exposure: they serve every stored +transcript, not just the ability to ask a question. Errors: -- Empty ``query`` / bad ``profile`` enum → 422 from Pydantic, before any stream. -- Unknown ``session_id`` → 404, before any stream. +- Empty ``query`` / bad ``profile`` enum / ``limit`` below 1 → 422 from + Pydantic, before any work. +- Unknown or malformed ``session_id`` → 404. Both collapse to 404 on purpose: + an id that can't name a file isn't on disk either, and a uniform answer tells + a prober nothing about the store's layout. +- An unreadable stored transcript (torn line, foreign JSONL) → 422. - A failure *after* the stream opens (headers already sent, so no HTTP status is available) → a terminal ``error`` frame, then the stream closes. The loop already yields :class:`ErrorEvent` for adapter faults; this wraps anything @@ -26,16 +39,19 @@ from __future__ import annotations +import json +from collections import deque from collections.abc import AsyncIterator +from dataclasses import asdict from typing import Any -from fastapi import FastAPI, HTTPException, Request +from fastapi import FastAPI, HTTPException, Query, Request from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field from reigner.harness.agent import Harness, Session from reigner.harness.events import ErrorEvent, Event, to_json -from reigner.sessions.store import SessionNotFound +from reigner.sessions.store import InvalidSessionId, SessionStore from reigner.types import Profile @@ -54,6 +70,86 @@ class RunRequest(BaseModel): state: dict[str, Any] = Field(default_factory=dict) +class SessionsResponse(BaseModel): + """Body of ``GET /sessions`` — every session's stored metadata. + + ``sessions`` holds :class:`~reigner.sessions.store.SessionMeta` as plain + dicts rather than a mirrored Pydantic model, so the payload can't drift from + the dataclass. It's the same shape ``reigner session list --json`` emits. + """ + + sessions: list[dict[str, Any]] + + +class EventsResponse(BaseModel): + """Body of ``GET /sessions/{id}/events`` — one session's transcript. + + ``events`` are the stored envelopes, in write order. They stay plain dicts + for the same reason ``/run`` frames are raw ``to_json`` output: re-modelling + them here would introduce a second schema that could silently disagree with + the event protocol. + + ``total`` counts the whole transcript, not the returned window, so a client + that passed ``limit`` can tell how much it didn't ask for; ``truncated`` + says whether it's looking at a suffix. + """ + + session_id: str + total: int + truncated: bool + events: list[dict[str, Any]] + + +def _session_or_404(store: SessionStore, session_id: str) -> None: + """Assert a session is readable, or raise 404. + + Unknown and structurally invalid ids both answer 404 — see the module + docstring for why they're deliberately indistinguishable. + + Call this *before* reading. :meth:`SessionStore.load_events` is a generator, + so its id validation only fires on the first ``next()`` — inside the read + loop, where an :class:`InvalidSessionId` would be mistaken for a torn line + and surface as a 422. + """ + try: + exists = store.exists(session_id) + except InvalidSessionId: + exists = False + if not exists: + raise HTTPException(status_code=404, detail=f"session {session_id!r} not found") + + +def _read_events( + store: SessionStore, session_id: str, limit: int | None +) -> tuple[list[Event], int]: + """Return ``(window, total)`` — the last ``limit`` events, in write order. + + A ``deque`` bounded by ``limit`` keeps the tail while the file streams past, + so ``total`` stays honest on a long session without holding all of it in + memory. ``limit=None`` returns everything. + + An unreadable row is a 422 rather than a skip: a transcript someone restores + a conversation from is complete, or it's declared broken — never quietly + short. The position reported counts events, not physical lines, since + ``load_events`` drops blank ones. + """ + window: deque[Event] | list[Event] = deque(maxlen=limit) if limit else [] + total = 0 + try: + for event in store.load_events(session_id): + window.append(event) + total += 1 + except (ValueError, TypeError) as exc: + # UnknownEventType, SchemaVersionMismatch and JSONDecodeError are all + # ValueError; a row that parses but doesn't fit its event class raises + # TypeError out of ``cls(**raw)``. + raise HTTPException( + status_code=422, + detail=f"session {session_id!r} unreadable at event {total + 1}: {exc}", + ) from exc + return list(window), total + + def _resolve_session(harness: Harness, req: RunRequest) -> Session: """New session, or resume an existing one — raising 404 if it's unknown. @@ -63,18 +159,18 @@ def _resolve_session(harness: Harness, req: RunRequest) -> Session: """ if req.session_id is None: return harness.session(state=req.state, profile=req.profile) - try: - return Session.load(req.session_id, harness=harness) - except SessionNotFound as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc + _session_or_404(harness.store, req.session_id) + return Session.load(req.session_id, harness=harness) def _frame(event: Event) -> str: """Render one event as an SSE message: a named ``event:`` line + ``data:``. - The type is also inside ``data`` (it's an event field), so non-browser - clients can ignore the ``event:`` line; browser ``EventSource`` clients can - ``addEventListener(, ...)``. + The type is also inside ``data`` (it's an event field), so a client can + dispatch on either. Browsers can't use ``EventSource`` here — it only issues + GETs and ``/run`` is a POST — so they call ``fetch()`` and parse the stream + themselves; the ``event:`` line is there for the SSE tooling that does read + it, and is safe to ignore. """ return f"event: {event.type}\ndata: {to_json(event)}\n\n" @@ -126,7 +222,29 @@ async def run(req: RunRequest, request: Request) -> StreamingResponse: headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) + @app.get("/sessions") + async def sessions() -> SessionsResponse: + return SessionsResponse(sessions=[asdict(m) for m in harness.store.list()]) + + @app.get("/sessions/{session_id}/events") + async def session_events( + session_id: str, + limit: int | None = Query( + None, + ge=1, + description="Return only the last N events. Omit for the whole transcript.", + ), + ) -> EventsResponse: + _session_or_404(harness.store, session_id) + window, total = _read_events(harness.store, session_id, limit) + return EventsResponse( + session_id=session_id, + total=total, + truncated=len(window) < total, + events=[json.loads(to_json(e)) for e in window], + ) + return app -__all__ = ["RunRequest", "create_app"] +__all__ = ["EventsResponse", "RunRequest", "SessionsResponse", "create_app"] diff --git a/tests/cli/test_serve.py b/tests/cli/test_serve.py new file mode 100644 index 0000000..d256999 --- /dev/null +++ b/tests/cli/test_serve.py @@ -0,0 +1,75 @@ +"""Startup-surface tests for `reigner serve --http`. + +``uvicorn.run`` is monkeypatched away — these assert on what the operator sees +printed before the server blocks, not on a live socket. +""" + +from __future__ import annotations + +import pytest + +from reigner.cli import serve as serve_mod +from reigner.config import ModelConfig, ReignerConfig, SettingsConfig +from reigner.harness.adapters.base import ModelAction +from reigner.harness.agent import Harness +from tests.harness.test_loop import FakeAdapter, _final + + +@pytest.fixture +def run_http(monkeypatch: pytest.MonkeyPatch): + """Call ``_run_http`` with uvicorn stubbed; return the host/port it got.""" + import uvicorn + + called: dict[str, object] = {} + monkeypatch.setattr(uvicorn, "run", lambda app, **kw: called.update(kw)) + + def _call(*, host: str = "127.0.0.1", port: int = 8000) -> dict[str, object]: + cfg = ReignerConfig(name="test_agent", model=ModelConfig(provider="openai", name="gpt-5.5")) + actions: list[ModelAction | Exception] = [_final("hi")] + harness = Harness( + adapter=FakeAdapter(actions=actions), role="TEST", settings=SettingsConfig() + ) + serve_mod._run_http(cfg, harness, host=host, port=port) + return called + + return _call + + +def test_startup_lists_every_route(run_http, capsys: pytest.CaptureFixture[str]) -> None: + run_http() + out = capsys.readouterr().out + assert "listening on http://127.0.0.1:8000" in out + for route in ("POST /run", "GET /sessions", "GET /sessions/{id}/events", "GET /health"): + assert route in out + + +def test_loopback_bind_prints_no_warning(run_http, capsys: pytest.CaptureFixture[str]) -> None: + run_http() + assert capsys.readouterr().err == "" + + +@pytest.mark.parametrize("host", ["0.0.0.0", "192.168.1.20", "::"]) +def test_exposed_bind_warns_on_stderr( + run_http, capsys: pytest.CaptureFixture[str], host: str +) -> None: + run_http(host=host) + err = capsys.readouterr().err + assert "no auth, CORS, or rate limiting" in err + assert "gateway" in err + + +@pytest.mark.parametrize("host", ["127.0.0.1", "localhost", "::1"]) +def test_loopback_spellings_are_all_quiet( + run_http, capsys: pytest.CaptureFixture[str], host: str +) -> None: + run_http(host=host) + assert capsys.readouterr().err == "" + + +def test_hostname_binds_are_treated_as_exposed( + run_http, capsys: pytest.CaptureFixture[str] +) -> None: + # A name we can't resolve to an address gets the warning — a spurious + # caution costs nothing, silence on a public bind costs a lot. + run_http(host="agent.internal") + assert "no auth" in capsys.readouterr().err diff --git a/tests/server/test_fastapi_app.py b/tests/server/test_fastapi_app.py index 6eabab5..29b7945 100644 --- a/tests/server/test_fastapi_app.py +++ b/tests/server/test_fastapi_app.py @@ -151,3 +151,145 @@ def test_mid_stream_failure_yields_terminal_error_frame(make_app: AppBuilder) -> assert name == "error" assert data["recoverable"] is False assert "boom" in data["error"] + + +def test_malformed_session_id_is_404_not_500(make_app: AppBuilder) -> None: + # store.exists() validates the id before checking disk, so an id with path + # separators used to escape as InvalidSessionId -> 500. + client, _ = make_app() + resp = client.post("/run", json={"query": "hi", "session_id": "../etc/passwd"}) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# /sessions +# --------------------------------------------------------------------------- + + +def _seed(client: TestClient, query: str = "q1") -> str: + """Run once through the API and return the session id it minted.""" + resp = client.post("/run", json={"query": query}) + session_id = _frames(resp.text)[0][1]["session_id"] + assert isinstance(session_id, str) + return session_id + + +def test_sessions_lists_stored_metadata(make_app: AppBuilder) -> None: + client, _ = make_app([_final("a"), _final("b")]) + first = _seed(client, "q1") + second = _seed(client, "q2") + + resp = client.get("/sessions") + assert resp.status_code == 200 + listed = resp.json()["sessions"] + assert {s["session_id"] for s in listed} == {first, second} + # The payload is SessionMeta verbatim — same shape as `session list --json`. + assert set(listed[0]) == { + "session_id", + "parent_id", + "title", + "created", + "last_updated", + "event_count", + "schema_version", + } + assert listed[0]["event_count"] > 0 + + +def test_sessions_is_empty_before_any_run(make_app: AppBuilder) -> None: + client, _ = make_app() + assert client.get("/sessions").json() == {"sessions": []} + + +# --------------------------------------------------------------------------- +# /sessions/{id}/events +# --------------------------------------------------------------------------- + + +def test_events_replays_the_stored_transcript(make_app: AppBuilder) -> None: + client, _ = make_app([_final("the answer is 42")]) + session_id = _seed(client, "what is the answer?") + + resp = client.get(f"/sessions/{session_id}/events") + assert resp.status_code == 200 + body = resp.json() + + assert body["session_id"] == session_id + assert body["truncated"] is False + assert body["total"] == len(body["events"]) + # Same envelopes /run frames carry, so they round-trip through from_json. + for event in body["events"]: + assert from_json(json.dumps(event)) is not None + assert event["session_id"] == session_id + assert body["events"][0]["type"] == "user_query" + assert body["events"][-1]["type"] == "final_answer" + assert body["events"][-1]["text"] == "the answer is 42" + + +def test_events_matches_the_frames_run_streamed(make_app: AppBuilder) -> None: + # The point of the endpoint: a reloading client reconstructs exactly what it + # would have seen live. + client, _ = make_app([_final("done")]) + streamed = client.post("/run", json={"query": "ping"}) + frames = [data for _, data in _frames(streamed.text)] + session_id = frames[0]["session_id"] + + replayed = client.get(f"/sessions/{session_id}/events").json()["events"] + assert replayed == frames + + +def test_events_limit_returns_the_tail_in_write_order(make_app: AppBuilder) -> None: + client, _ = make_app([_final("done")]) + session_id = _seed(client) + full = client.get(f"/sessions/{session_id}/events").json() + + resp = client.get(f"/sessions/{session_id}/events", params={"limit": 1}) + body = resp.json() + assert body["events"] == full["events"][-1:] + assert body["total"] == full["total"] # total counts the transcript, not the window + assert body["truncated"] is True + + +def test_events_limit_wider_than_the_session_is_not_truncated(make_app: AppBuilder) -> None: + client, _ = make_app([_final("done")]) + session_id = _seed(client) + + body = client.get(f"/sessions/{session_id}/events", params={"limit": 999}).json() + assert body["truncated"] is False + assert len(body["events"]) == body["total"] + + +def test_events_limit_below_one_is_422(make_app: AppBuilder) -> None: + client, _ = make_app() + session_id = _seed(client) + assert client.get(f"/sessions/{session_id}/events", params={"limit": 0}).status_code == 422 + + +def test_events_unknown_session_is_404(make_app: AppBuilder) -> None: + client, _ = make_app() + resp = client.get("/sessions/does-not-exist/events") + assert resp.status_code == 404 + assert "does-not-exist" in resp.json()["detail"] + + +def test_events_malformed_session_id_is_404(make_app: AppBuilder) -> None: + # Must be caught before the read: load_events is a generator, so its id + # validation would otherwise fire inside the loop and read as a torn line. + client, _ = make_app() + assert client.get("/sessions/not.a.valid.id/events").status_code == 404 + + +def test_events_torn_transcript_is_422(make_app: AppBuilder) -> None: + client, harness = make_app([_final("done")]) + session_id = _seed(client) + path = harness.store.root / f"{session_id}.jsonl" + path.write_text( + path.read_text(encoding="utf-8") + '{"type": "not_an_event"}\n', + encoding="utf-8", + ) + + resp = client.get(f"/sessions/{session_id}/events") + assert resp.status_code == 422 + detail = resp.json()["detail"] + assert session_id in detail + assert "not_an_event" in detail