v0.19.0 backend: /api/docs/sessions/* endpoints + TTL cache

Adds the four read endpoints the v0.19.0 /docs/sessions/* surface
mounts on top of:

- GET /api/docs/sessions/manifest      → sessions.json (title map)
- GET /api/docs/sessions/about         → README.md
- GET /api/docs/sessions/<NNNN>/index  → per-session transcript list
- GET /api/docs/sessions/<NNNN>/<file> → transcript body

The framework mediates the gitea fetch so the rendered surface
inherits the same chrome as the v0.14.0 /docs route and the browser
makes no cross-origin call. Reads are aggressively cached in-process
(60s for the manifest, 5min for content) so the framework doesn't
hammer git.wiggleverse.org under normal traffic. Negative results
(gitea 404) are also cached at the content TTL to absorb the
expected empty-state at deploy-time (the parallel
ohm-session-history repo restructure ships in driver subsession
0017.2). All four endpoints are anonymous-reachable, sibling to
/api/philosophy and /api/docs.

Path validation gates the network: only /^\d{4}$/ session dirs and
the full SESSION-NNNN.M-TRANSCRIPT-...md filename shape pass to the
upstream. Legacy flat-root names (e.g. SESSION-A-TRANSCRIPT.md) and
path-traversal attempts are rejected 400 before any fetch.

18 new tests cover the happy paths, 404 empty-states, 502 upstream
errors, path-validation rejections, and the cache-hit-within-TTL
contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-05-28 09:07:01 -07:00
parent ac3513a686
commit 39e57706d9
3 changed files with 855 additions and 0 deletions
+118
View File
@@ -15,6 +15,7 @@ import json
from typing import Any from typing import Any
from fastapi import APIRouter, HTTPException, Request from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import PlainTextResponse, Response
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from . import ( from . import (
@@ -29,6 +30,7 @@ from . import (
db, db,
device_trust as device_trust_mod, device_trust as device_trust_mod,
docs as docs_mod, docs as docs_mod,
docs_sessions,
entry as entry_mod, entry as entry_mod,
cache, cache,
funder, funder,
@@ -143,6 +145,122 @@ def make_router(
payload = docs_mod.load() payload = docs_mod.load()
return {"body": payload["body"]} return {"body": payload["body"]}
# ---------------------------------------------------------------
# v0.19.0 / roadmap item #30 — /api/docs/sessions/*
#
# The framework mediates reads against the public
# `wiggleverse/ohm-session-history` gitea repo so the rendered
# `/docs/sessions/*` surface inherits the same chrome as the
# /docs/user-guide route and doesn't require a cross-origin
# gesture from the frontend. See backend/app/docs_sessions.py
# for the cache shape and env knobs.
#
# The route mapping for the three `status` values returned by
# the fetchers:
#
# "ok" → HTTP 200, payload as documented per endpoint
# "404" → HTTP 200/404 depending on the endpoint (the
# manifest's empty state is 200 + {} so the
# frontend can short-circuit without an error
# banner; transcripts/about return 404 so the
# frontend can render its own empty-state)
# "error" → HTTP 502, {"error": ..., "detail": ...} so the
# frontend retry surface reads as "couldn't reach
# the session-history repo" rather than as a
# generic 5xx.
# ---------------------------------------------------------------
@router.get("/api/docs/sessions/manifest")
async def get_sessions_manifest() -> dict[str, Any]:
result = await docs_sessions.fetch_manifest()
if result["status"] == "ok":
return result["manifest"]
if result["status"] == "404":
# Empty-state contract: render no session rows in the
# flyout but don't show an error banner. The frontend
# treats `{}` as "no sessions published yet".
return {}
raise HTTPException(
status_code=502,
detail={
"error": "session-history fetch failed",
"detail": result.get("detail", "unknown"),
},
)
@router.get("/api/docs/sessions/about")
async def get_sessions_about() -> Response:
result = await docs_sessions.fetch_about()
if result["status"] == "ok":
return PlainTextResponse(
content=result["body"],
media_type="text/markdown; charset=utf-8",
)
if result["status"] == "404":
raise HTTPException(
status_code=404,
detail="session-history README not yet published",
)
raise HTTPException(
status_code=502,
detail={
"error": "session-history fetch failed",
"detail": result.get("detail", "unknown"),
},
)
@router.get("/api/docs/sessions/{nnnn}/index")
async def get_sessions_index(nnnn: str) -> dict[str, Any]:
if not docs_sessions._is_valid_session_dir(nnnn):
# 400 over 404: the request itself is malformed (the
# session directory name doesn't match `^\d{4}$`),
# distinct from "no such session published yet".
raise HTTPException(status_code=400, detail="invalid session directory")
result = await docs_sessions.fetch_session_index(nnnn)
if result["status"] == "ok":
return {"files": result["files"]}
if result["status"] == "404":
raise HTTPException(
status_code=404,
detail="no transcripts published for this session",
)
raise HTTPException(
status_code=502,
detail={
"error": "session-history fetch failed",
"detail": result.get("detail", "unknown"),
},
)
@router.get("/api/docs/sessions/{nnnn}/{filename}")
async def get_sessions_transcript(nnnn: str, filename: str) -> Response:
# Path-shape validation before any network — refuses anything
# that would resolve outside the `NNNN/SESSION-...md` layout
# (e.g. legacy `SESSION-A-TRANSCRIPT.md` at the repo root,
# `../etc/passwd`, or any non-numeric session dir).
if not docs_sessions._is_valid_session_dir(nnnn):
raise HTTPException(status_code=400, detail="invalid session directory")
if not docs_sessions._is_valid_transcript_filename(filename):
raise HTTPException(status_code=400, detail="invalid transcript filename")
result = await docs_sessions.fetch_transcript(nnnn, filename)
if result["status"] == "ok":
return PlainTextResponse(
content=result["body"],
media_type="text/markdown; charset=utf-8",
)
if result["status"] == "404":
raise HTTPException(
status_code=404,
detail="transcript not found",
)
raise HTTPException(
status_code=502,
detail={
"error": "session-history fetch failed",
"detail": result.get("detail", "unknown"),
},
)
# --------------------------------------------------------------- # ---------------------------------------------------------------
# Auth surface — reads role from our users table per §6. # Auth surface — reads role from our users table per §6.
# --------------------------------------------------------------- # ---------------------------------------------------------------
+358
View File
@@ -0,0 +1,358 @@
"""§14 + roadmap item #30 — on-site sessions-history browser source.
Sibling of `docs.py` / `philosophy.py` but with a different read shape:
the bodies here live in the **public** `wiggleverse/ohm-session-history`
gitea repo (transcripts of every OHM build session, published per the
ohm-infra SESSION-PROTOCOL.md), not on disk. The framework mediates
the gitea fetch on behalf of the browser so the rendered `/docs/sessions/*`
surface inherits the same chrome as `/philosophy` and `/docs/user-guide`
and stays free of any cross-origin gestures from the frontend.
Three read endpoints, all anonymous-reachable:
GET /api/docs/sessions/manifest sessions.json (title manifest)
GET /api/docs/sessions/about README.md (the about page)
GET /api/docs/sessions/<NNNN>/<file> a transcript body
GET /api/docs/sessions/<NNNN>/index per-session file listing
All three sit behind a small in-process TTL cache (manifest TTL default
60 s, content TTL default 300 s). Negative results (404 from gitea) are
also cached at the content TTL to avoid hammering gitea when a
deployment hasn't yet been populated with transcripts. The cache key
is the URL path on the gitea raw base (or the contents API for the
per-session listing); the cache lives in-process, plain dict +
`time.monotonic()` check, no external dep.
Env knobs:
OHM_SESSION_HISTORY_RAW_BASE
Override the gitea raw base URL. Default points at OHM's canonical
transcript repo:
https://git.wiggleverse.org/wiggleverse/ohm-session-history/raw/branch/main
The framework-default value is OHM-flavored because OHM is the
only deployment to date a deployment running its own
transcript repo overrides this via flotilla's overlay.
OHM_SESSION_HISTORY_CONTENTS_BASE
Override the gitea contents-API base URL (for the per-session
listing endpoint, which enumerates files inside a `NNNN/` folder).
Default:
https://git.wiggleverse.org/api/v1/repos/wiggleverse/ohm-session-history/contents
OHM_DOCS_SESSIONS_MANIFEST_TTL_SEC
Cache TTL for the manifest (default 60 s). The manifest is small
and changes when a new session is added; 60 s strikes a balance
between freshness and gitea load.
OHM_DOCS_SESSIONS_CONTENT_TTL_SEC
Cache TTL for transcript bodies + README + per-session listings
(default 300 s = 5 minutes). Transcripts are append-only once
published, so 5 minutes of staleness is harmless.
§3 invariant 1 is preserved: the framework holds no secret bytes; the
gitea repo is public, the fetch carries no auth header.
"""
from __future__ import annotations
import logging
import os
import re
import threading
import time
from typing import Any
import httpx
log = logging.getLogger(__name__)
_DEFAULT_RAW_BASE = (
"https://git.wiggleverse.org/wiggleverse/ohm-session-history/raw/branch/main"
)
_DEFAULT_CONTENTS_BASE = (
"https://git.wiggleverse.org/api/v1/repos/wiggleverse/ohm-session-history/contents"
)
_DEFAULT_MANIFEST_TTL_SEC = 60.0
_DEFAULT_CONTENT_TTL_SEC = 300.0
# The transcript filename shape per SESSION-PROTOCOL.md §1. The
# `<start>--<end>` suffix is optional so legacy renamed-letter
# transcripts (e.g. `SESSION-0009.0-TRANSCRIPT.md` without timestamps)
# remain reachable. The `\.\d+(\.\d+)*` after the session number
# accommodates `0017.0`, `0017.1`, `0017.1.1`, etc.
_TRANSCRIPT_FILENAME_RE = re.compile(
r"^SESSION-\d{4}\.\d+(\.\d+)*-TRANSCRIPT"
r"(-\d{4}-\d{2}-\d{2}T\d{2}-\d{2}--\d{4}-\d{2}-\d{2}T\d{2}-\d{2})?"
r"\.md$"
)
_SESSION_DIR_RE = re.compile(r"^\d{4}$")
_HTTP_TIMEOUT_SEC = 5.0
def _env_float(name: str, default: float) -> float:
raw = os.environ.get(name, "").strip()
if not raw:
return default
try:
return float(raw)
except ValueError:
log.warning("invalid %s=%r — falling back to %s", name, raw, default)
return default
def _raw_base() -> str:
return os.environ.get("OHM_SESSION_HISTORY_RAW_BASE", "").strip() or _DEFAULT_RAW_BASE
def _contents_base() -> str:
return (
os.environ.get("OHM_SESSION_HISTORY_CONTENTS_BASE", "").strip()
or _DEFAULT_CONTENTS_BASE
)
def _manifest_ttl() -> float:
return _env_float("OHM_DOCS_SESSIONS_MANIFEST_TTL_SEC", _DEFAULT_MANIFEST_TTL_SEC)
def _content_ttl() -> float:
return _env_float("OHM_DOCS_SESSIONS_CONTENT_TTL_SEC", _DEFAULT_CONTENT_TTL_SEC)
# ---------------------------------------------------------------------------
# In-process TTL cache
# ---------------------------------------------------------------------------
#
# Plain dict + `time.monotonic()` check, no external dep. The cache
# value is a `(stored_at, payload)` tuple; `payload` may carry an
# error-shape sentinel for negative caching (404s). Lock guards
# read-modify-write across worker tasks; entries are immutable once
# stored so reads under the lock are fast.
_lock = threading.Lock()
_cache: dict[str, tuple[float, dict[str, Any]]] = {}
def _cache_get(key: str, ttl_sec: float) -> dict[str, Any] | None:
with _lock:
entry = _cache.get(key)
if entry is None:
return None
stored_at, payload = entry
if time.monotonic() - stored_at > ttl_sec:
# Don't evict here; let _cache_put overwrite on next fetch.
# The stale entry is gated by the TTL check, so it stays
# invisible to readers regardless.
return None
return payload
def _cache_put(key: str, payload: dict[str, Any]) -> None:
with _lock:
_cache[key] = (time.monotonic(), payload)
def reset_cache() -> None:
"""Drop every cached entry. Test seam — not called in production."""
with _lock:
_cache.clear()
# ---------------------------------------------------------------------------
# Public fetch surface
# ---------------------------------------------------------------------------
#
# Each fetcher returns a `{status, ...}` dict. `status` is one of:
# "ok" — payload field carries the body
# "404" — gitea returned 404 (or content was missing)
# "error" — gitea returned 5xx, timed out, or returned malformed data
#
# The route layer maps these onto HTTP responses; keeping the mapping
# out of this module makes the cache transparent to the test harness.
async def _http_get(url: str) -> tuple[int, str]:
"""Perform a single GET against `url`; return (status_code, body).
On timeout or network error, returns (599, error_message). The 599
pseudo-status maps to a 502 at the route layer the same way an
upstream 5xx does the caller doesn't care which leg of the
network broke.
"""
try:
async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT_SEC) as client:
r = await client.get(url)
return r.status_code, r.text
except httpx.HTTPError as e:
log.warning("gitea fetch failed for %s: %s", url, e)
return 599, f"fetch error: {e}"
def _is_valid_session_dir(nnnn: str) -> bool:
return bool(_SESSION_DIR_RE.match(nnnn))
def _is_valid_transcript_filename(filename: str) -> bool:
return bool(_TRANSCRIPT_FILENAME_RE.match(filename))
async def fetch_manifest() -> dict[str, Any]:
"""Fetch and parse `sessions.json` from the public repo.
Returns one of:
{"status": "ok", "manifest": {...}} successful parse
{"status": "404"} gitea 404 (empty state)
{"status": "error", "detail": "..."} 5xx / timeout / bad JSON
"""
cache_key = "manifest"
cached = _cache_get(cache_key, _manifest_ttl())
if cached is not None:
return cached
url = f"{_raw_base()}/sessions.json"
status, body = await _http_get(url)
if status == 200:
try:
import json
data = json.loads(body)
except (json.JSONDecodeError, ValueError) as e:
payload: dict[str, Any] = {
"status": "error",
"detail": f"sessions.json malformed: {e}",
}
# Don't cache parse errors — give the upstream a chance to
# fix the file without waiting for TTL expiry.
return payload
if not isinstance(data, dict):
return {
"status": "error",
"detail": "sessions.json is not a JSON object",
}
payload = {"status": "ok", "manifest": data}
_cache_put(cache_key, payload)
return payload
if status == 404:
payload = {"status": "404"}
_cache_put(cache_key, payload)
return payload
return {"status": "error", "detail": f"upstream returned {status}"}
async def fetch_about() -> dict[str, Any]:
"""Fetch the repo's README.md (rendered as the /docs/sessions/about page).
Returns one of:
{"status": "ok", "body": "..."}
{"status": "404"}
{"status": "error", "detail": "..."}
"""
cache_key = "about:README.md"
cached = _cache_get(cache_key, _content_ttl())
if cached is not None:
return cached
url = f"{_raw_base()}/README.md"
status, body = await _http_get(url)
if status == 200:
payload: dict[str, Any] = {"status": "ok", "body": body}
_cache_put(cache_key, payload)
return payload
if status == 404:
payload = {"status": "404"}
_cache_put(cache_key, payload)
return payload
return {"status": "error", "detail": f"upstream returned {status}"}
async def fetch_transcript(nnnn: str, filename: str) -> dict[str, Any]:
"""Fetch a single transcript body from `{nnnn}/{filename}` in the repo.
The caller is expected to have validated `nnnn` and `filename`
against `_is_valid_session_dir` / `_is_valid_transcript_filename`
before calling this invalid paths shouldn't reach the network.
"""
cache_key = f"transcript:{nnnn}/{filename}"
cached = _cache_get(cache_key, _content_ttl())
if cached is not None:
return cached
url = f"{_raw_base()}/{nnnn}/{filename}"
status, body = await _http_get(url)
if status == 200:
payload: dict[str, Any] = {"status": "ok", "body": body}
_cache_put(cache_key, payload)
return payload
if status == 404:
payload = {"status": "404"}
_cache_put(cache_key, payload)
return payload
return {"status": "error", "detail": f"upstream returned {status}"}
async def fetch_session_index(nnnn: str) -> dict[str, Any]:
"""List the transcript filenames inside the `{nnnn}/` folder.
Uses gitea's contents API (one HTTP per session-index page-view per
cache-TTL) rather than the raw URL there's no flat way to list a
folder via the raw mount.
Returns one of:
{"status": "ok", "files": ["SESSION-...md", ...]}
{"status": "404"}
{"status": "error", "detail": "..."}
Only filenames that match `_is_valid_transcript_filename` are
surfaced sibling files (e.g. an attached `notes.md`) are ignored
so the /docs/sessions/<NNNN> page never lists a non-transcript
masquerading as one.
"""
cache_key = f"index:{nnnn}"
cached = _cache_get(cache_key, _content_ttl())
if cached is not None:
return cached
url = f"{_contents_base()}/{nnnn}"
status, body = await _http_get(url)
if status == 200:
try:
import json
data = json.loads(body)
except (json.JSONDecodeError, ValueError) as e:
return {
"status": "error",
"detail": f"contents API response malformed: {e}",
}
if not isinstance(data, list):
return {
"status": "error",
"detail": "contents API returned non-list",
}
files: list[str] = []
for entry in data:
if not isinstance(entry, dict):
continue
if entry.get("type") != "file":
continue
name = entry.get("name")
if not isinstance(name, str):
continue
if _is_valid_transcript_filename(name):
files.append(name)
files.sort()
payload: dict[str, Any] = {"status": "ok", "files": files}
_cache_put(cache_key, payload)
return payload
if status == 404:
payload = {"status": "404"}
_cache_put(cache_key, payload)
return payload
return {"status": "error", "detail": f"upstream returned {status}"}
@@ -0,0 +1,379 @@
"""v0.19.0 / roadmap item #30 — `/api/docs/sessions/*` endpoints.
The framework mediates reads against the public
`wiggleverse/ohm-session-history` gitea repo so the rendered
`/docs/sessions/*` surface inherits the same chrome as `/docs/user-guide`.
This test suite covers the four endpoints + the in-process TTL cache,
mocking the upstream HTTP via `httpx.MockTransport` (the same shape the
rest of the test suite uses for Gitea).
The tests do NOT spin up the full FakeGitea they only need to mock
the gitea raw URL surface (and the contents API for the session-index
endpoint). Each test owns its mock transport so we can dial in 200 /
404 / 5xx / timeout responses per case.
Path-validation tests intentionally bypass the network a malformed
`nnnn` or `filename` MUST be rejected at the route layer before any
upstream call is made.
"""
from __future__ import annotations
import json
import httpx
import pytest
from fastapi.testclient import TestClient
from app import docs_sessions
# Reuse the proven app-construction fixtures from the proposal vertical
# (same shape every test file in this repo uses).
from test_propose_vertical import ( # noqa: F401
app_with_fake_gitea,
tmp_env,
)
# ---------------------------------------------------------------------------
# Test scaffolding
# ---------------------------------------------------------------------------
class _UpstreamHandler:
"""Records every URL the docs_sessions module fetched and returns
canned responses keyed by URL substring. Allows the test to assert
on call count (for cache verification) without needing a full Gitea
simulator.
`calls` tracks only URLs that hit the session-history host (the
`OHM_SESSION_HISTORY_*` bases) so reconciler/Gitea-side calls which
also pass through this handler because `httpx.AsyncClient` is a
shared attribute the gitea-side fixture also monkeypatches don't
inflate the count we use for cache-hit assertions.
"""
_SESSION_HOST_MARKERS = ("ohm-session-history", "wiggleverse/ohm-session-history")
def __init__(self, responses: dict[str, tuple[int, str]]):
self.responses = responses
self.calls: list[str] = []
def __call__(self, request: httpx.Request) -> httpx.Response:
url = str(request.url)
if any(m in url for m in self._SESSION_HOST_MARKERS):
self.calls.append(url)
for key, (status, body) in self.responses.items():
if key in url:
return httpx.Response(status, text=body)
# Default: 404. Lets tests skip declaring "the rest is 404".
return httpx.Response(404, text="not found")
@pytest.fixture
def patched_httpx(monkeypatch):
"""Provide a hook the test can call to install a MockTransport.
Returns a closure: `install(handler)` patches
`app.docs_sessions.httpx.AsyncClient` so every constructed client
uses the handler's transport.
NB: the upstream `app_with_fake_gitea` fixture also patches
`httpx.AsyncClient` (to route gitea calls to a FakeGitea handler),
and because `httpx` is a single shared module, that patch mutates
the *same* `AsyncClient` attribute we're about to overwrite. We
therefore import the unpatched class directly from the
`httpx._client` module so our install path can construct a fresh
real client around our MockTransport without going through the
FakeGitea wrapper.
"""
from httpx._client import AsyncClient as RealAsyncClient
def install(handler):
def patched(*args, **kwargs):
kwargs["transport"] = httpx.MockTransport(handler)
return RealAsyncClient(*args, **kwargs)
monkeypatch.setattr("app.docs_sessions.httpx.AsyncClient", patched)
return handler
yield install
@pytest.fixture
def app(app_with_fake_gitea):
"""Wrap the shared app fixture, resetting the docs-sessions cache so
cross-test state can't leak. Returns just the FastAPI app — the
fake-Gitea handle is irrelevant for the docs-sessions surface.
"""
docs_sessions.reset_cache()
fastapi_app, _fake = app_with_fake_gitea
return fastapi_app
# ---------------------------------------------------------------------------
# Manifest endpoint
# ---------------------------------------------------------------------------
def test_manifest_happy_path(app, patched_httpx):
manifest_body = json.dumps(
{
"0001": {"title": "Bootstrap"},
"0014": {"title": "Wave 7 driver"},
}
)
patched_httpx(
_UpstreamHandler({"sessions.json": (200, manifest_body)})
)
with TestClient(app) as client:
r = client.get("/api/docs/sessions/manifest")
assert r.status_code == 200, r.text
payload = r.json()
assert payload == {
"0001": {"title": "Bootstrap"},
"0014": {"title": "Wave 7 driver"},
}
def test_manifest_empty_state(app, patched_httpx):
"""A 404 from gitea means the manifest hasn't been published yet.
The endpoint returns HTTP 200 + `{}` so the frontend can render the
no-sessions-yet state without an error banner.
"""
patched_httpx(_UpstreamHandler({"sessions.json": (404, "not found")}))
with TestClient(app) as client:
r = client.get("/api/docs/sessions/manifest")
assert r.status_code == 200, r.text
assert r.json() == {}
def test_manifest_upstream_5xx_returns_502(app, patched_httpx):
patched_httpx(_UpstreamHandler({"sessions.json": (500, "internal")}))
with TestClient(app) as client:
r = client.get("/api/docs/sessions/manifest")
assert r.status_code == 502, r.text
body = r.json()
assert body["detail"]["error"] == "session-history fetch failed"
# ---------------------------------------------------------------------------
# About endpoint
# ---------------------------------------------------------------------------
def test_about_happy_path(app, patched_httpx):
readme = "# OHM session history\n\nWelcome.\n"
patched_httpx(_UpstreamHandler({"README.md": (200, readme)}))
with TestClient(app) as client:
r = client.get("/api/docs/sessions/about")
assert r.status_code == 200, r.text
assert "text/markdown" in r.headers["content-type"]
assert r.text == readme
def test_about_404(app, patched_httpx):
patched_httpx(_UpstreamHandler({"README.md": (404, "")}))
with TestClient(app) as client:
r = client.get("/api/docs/sessions/about")
assert r.status_code == 404, r.text
def test_about_upstream_5xx_returns_502(app, patched_httpx):
patched_httpx(_UpstreamHandler({"README.md": (503, "down")}))
with TestClient(app) as client:
r = client.get("/api/docs/sessions/about")
assert r.status_code == 502, r.text
# ---------------------------------------------------------------------------
# Transcript endpoint
# ---------------------------------------------------------------------------
def test_transcript_happy_path(app, patched_httpx):
body = "# Session 0017.1 — Transcript\n\nbody.\n"
fname = "SESSION-0017.1-TRANSCRIPT-2026-05-28T08-50--2026-05-28T11-20.md"
patched_httpx(_UpstreamHandler({fname: (200, body)}))
with TestClient(app) as client:
r = client.get(f"/api/docs/sessions/0017/{fname}")
assert r.status_code == 200, r.text
assert "text/markdown" in r.headers["content-type"]
assert r.text == body
def test_transcript_404(app, patched_httpx):
fname = "SESSION-9999.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md"
patched_httpx(_UpstreamHandler({})) # everything 404s
with TestClient(app) as client:
r = client.get(f"/api/docs/sessions/9999/{fname}")
assert r.status_code == 404, r.text
def test_transcript_rejects_invalid_session_dir(app, patched_httpx):
"""`nnnn` must be exactly 4 digits. `abcd` fails before any
network call.
"""
handler = _UpstreamHandler({})
patched_httpx(handler)
with TestClient(app) as client:
r = client.get(
"/api/docs/sessions/abcd/"
"SESSION-0001.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md"
)
assert r.status_code == 400, r.text
assert handler.calls == [], "rejected path must not hit the network"
def test_transcript_rejects_path_traversal(app, patched_httpx):
"""A filename that doesn't match the SESSION-NNNN.M-TRANSCRIPT regex
is rejected. `../etc/passwd` doesn't match; neither does the legacy
flat-root `SESSION-A-TRANSCRIPT.md`.
"""
handler = _UpstreamHandler({})
patched_httpx(handler)
with TestClient(app) as client:
# Path traversal — but FastAPI normalizes `..` in the path before
# routing, so this resolves to /api/docs/sessions/0001/etc/passwd
# which routes to the same handler with filename=etc/passwd, and
# gets rejected as an invalid transcript filename. Even if the
# normalization didn't apply (some intermediary), the regex
# check rejects anything not matching the SESSION- prefix.
r = client.get(
"/api/docs/sessions/0001/etc%2Fpasswd"
)
# 400 (filename validation) or 404 (path didn't match the
# route); both reject before any network call. Either is
# acceptable — what matters is that we never fetched it.
assert r.status_code in (400, 404), r.text
assert handler.calls == [], "rejected path must not hit the network"
def test_transcript_rejects_legacy_flat_filename(app, patched_httpx):
"""Legacy `SESSION-A-TRANSCRIPT.md` (letter form) doesn't match the
numeric regex by design, since post-#23 transcripts live in
`NNNN/` folders with numeric names. Reject 400.
"""
handler = _UpstreamHandler({})
patched_httpx(handler)
with TestClient(app) as client:
r = client.get("/api/docs/sessions/0001/SESSION-A-TRANSCRIPT.md")
assert r.status_code == 400, r.text
assert handler.calls == [], "rejected path must not hit the network"
def test_transcript_upstream_5xx_returns_502(app, patched_httpx):
fname = "SESSION-0001.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md"
patched_httpx(_UpstreamHandler({fname: (502, "bad gateway")}))
with TestClient(app) as client:
r = client.get(f"/api/docs/sessions/0001/{fname}")
assert r.status_code == 502, r.text
# ---------------------------------------------------------------------------
# Session-index endpoint
# ---------------------------------------------------------------------------
def test_session_index_happy_path(app, patched_httpx):
"""The contents API returns a JSON list of file entries. The
endpoint filters to entries that match the transcript regex and
sorts them.
"""
# Two transcripts (driver + subagent) + a non-transcript sibling
# that must be filtered out.
listing = json.dumps(
[
{
"name": "SESSION-0017.0-TRANSCRIPT-"
"2026-05-28T08-30--2026-05-28T12-00.md",
"type": "file",
},
{
"name": "SESSION-0017.1-TRANSCRIPT-"
"2026-05-28T08-50--2026-05-28T11-20.md",
"type": "file",
},
{"name": "notes.md", "type": "file"}, # not a transcript
{"name": "attached-dir", "type": "dir"}, # not a file
]
)
patched_httpx(_UpstreamHandler({"/contents/0017": (200, listing)}))
with TestClient(app) as client:
r = client.get("/api/docs/sessions/0017/index")
assert r.status_code == 200, r.text
files = r.json()["files"]
assert files == [
"SESSION-0017.0-TRANSCRIPT-2026-05-28T08-30--2026-05-28T12-00.md",
"SESSION-0017.1-TRANSCRIPT-2026-05-28T08-50--2026-05-28T11-20.md",
]
def test_session_index_404(app, patched_httpx):
patched_httpx(_UpstreamHandler({})) # everything 404s
with TestClient(app) as client:
r = client.get("/api/docs/sessions/9999/index")
assert r.status_code == 404, r.text
def test_session_index_rejects_invalid_session_dir(app, patched_httpx):
handler = _UpstreamHandler({})
patched_httpx(handler)
with TestClient(app) as client:
r = client.get("/api/docs/sessions/abc/index")
assert r.status_code == 400, r.text
assert handler.calls == [], "rejected path must not hit the network"
# ---------------------------------------------------------------------------
# Cache behavior
# ---------------------------------------------------------------------------
def test_manifest_cache_hits_within_ttl(app, patched_httpx, monkeypatch):
"""Two consecutive manifest calls within the TTL window should
issue exactly one HTTP request to gitea.
"""
# Generous TTL so the test never races.
monkeypatch.setenv("OHM_DOCS_SESSIONS_MANIFEST_TTL_SEC", "60")
handler = _UpstreamHandler(
{"sessions.json": (200, json.dumps({"0001": {"title": "x"}}))}
)
patched_httpx(handler)
with TestClient(app) as client:
r1 = client.get("/api/docs/sessions/manifest")
r2 = client.get("/api/docs/sessions/manifest")
assert r1.status_code == 200
assert r2.status_code == 200
assert len(handler.calls) == 1, (
f"expected one upstream call, got {handler.calls}"
)
def test_transcript_cache_hits_within_ttl(app, patched_httpx, monkeypatch):
monkeypatch.setenv("OHM_DOCS_SESSIONS_CONTENT_TTL_SEC", "300")
fname = "SESSION-0001.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md"
handler = _UpstreamHandler({fname: (200, "# body\n")})
patched_httpx(handler)
with TestClient(app) as client:
r1 = client.get(f"/api/docs/sessions/0001/{fname}")
r2 = client.get(f"/api/docs/sessions/0001/{fname}")
assert r1.status_code == 200
assert r2.status_code == 200
assert len(handler.calls) == 1
def test_transcript_404_is_cached(app, patched_httpx, monkeypatch):
"""Negative caching: a 404 result is cached at the content TTL so a
deployment with no published transcripts doesn't hammer gitea on
every navigation. Documented in `docs_sessions.fetch_transcript`.
"""
monkeypatch.setenv("OHM_DOCS_SESSIONS_CONTENT_TTL_SEC", "300")
fname = "SESSION-9999.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md"
handler = _UpstreamHandler({}) # everything 404s
patched_httpx(handler)
with TestClient(app) as client:
r1 = client.get(f"/api/docs/sessions/9999/{fname}")
r2 = client.get(f"/api/docs/sessions/9999/{fname}")
assert r1.status_code == 404
assert r2.status_code == 404
assert len(handler.calls) == 1, "negative caching should suppress the 2nd call"