Add protected canonical claims read endpoint for the Observatory (#139)

* Add protected canonical claims read API

* Normalize claim loader exits in worker thread

* Allow file-gated claims reader defaults

diagnostics/app.py

@@ -23,10 +23,16 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "pipeline"))
 
 from aiohttp import web
 from daily_digest_routes import register_daily_digest_routes
-from kb_claim_routes import KB_CLAIM_PUBLIC_PATHS, KB_CLAIM_PUBLIC_PREFIXES, register_kb_claim_routes
+from kb_claim_routes import (
+    KB_CLAIM_GLOBAL_AUTH_PREFIXES,
+    KB_CLAIM_LIST_API_KEY,
+    KB_CLAIM_SELF_AUTH_PATHS,
+    load_claim_list_api_key,
+    register_kb_claim_routes,
+)
 from kb_proposal_routes import KB_PROPOSAL_PUBLIC_PATHS, register_kb_proposal_routes
 from leaderboard_routes import LEADERBOARD_PUBLIC_PATHS, register_leaderboard_routes
-from lib.search import search as kb_search, embed_query, search_qdrant
+from lib.search import embed_query, search as kb_search, search_qdrant
 from response_audit_routes import RESPONSE_AUDIT_PUBLIC_PATHS, register_response_audit_routes
 from review_queue_routes import register_review_queue_routes
 
@@ -516,13 +522,14 @@ async def auth_middleware(request, handler):
         or request.path in RESPONSE_AUDIT_PUBLIC_PATHS
         or request.path in LEADERBOARD_PUBLIC_PATHS
         or request.path in KB_PROPOSAL_PUBLIC_PATHS
-        or request.path in KB_CLAIM_PUBLIC_PATHS
+        or (request.method == "GET" and request.path in KB_CLAIM_SELF_AUTH_PATHS)
-        or request.path.startswith(KB_CLAIM_PUBLIC_PREFIXES)
         or request.path.startswith("/api/response-audit/")
     ):
         return await handler(request)
     expected = request.app.get("api_key")
     if not expected:
+        if request.path.startswith(KB_CLAIM_GLOBAL_AUTH_PREFIXES):
+            return web.json_response({"error": "argus_auth_unconfigured"}, status=503)
         # No key configured — all endpoints open (development mode)
         return await handler(request)
     provided = request.headers.get("X-Api-Key", "")
@@ -2338,10 +2345,16 @@ def create_app() -> web.Application:
     app = web.Application(middlewares=[auth_middleware])
     app["db"] = _get_db()
     app["api_key"] = _load_secret(API_KEY_FILE)
+    canonical_claims_api_key = load_claim_list_api_key()
+    if canonical_claims_api_key:
+        app[KB_CLAIM_LIST_API_KEY] = canonical_claims_api_key
+        logger.info("Canonical claims API auth enabled")
+    else:
+        logger.warning("Canonical claims API auth is not configured; list endpoint fails closed")
     if app["api_key"]:
         logger.info("API key auth enabled (protected endpoints require X-Api-Key)")
     else:
-        logger.info("No API key configured — all endpoints open")
+        logger.info("No global API key configured — canonical claim routes still fail closed")
     # Root redirects to /ops (legacy dashboard still at /legacy)
     app.router.add_get("/", handle_root_redirect)
     app.router.add_get("/prs", handle_prs_page)

diagnostics/kb_claim_routes.py

@@ -1,16 +1,23 @@
 """Read-only canonical KB claim routes for Argus.
 
 These routes show the Postgres-backed claim graph that Leo uses through the
-``teleo-kb`` bridge: one claim, its evidence rows, and its graph edges.
+``teleo-kb`` bridge: a protected canonical summary list plus one claim, its
+evidence rows, and its graph edges.
 """
 
 from __future__ import annotations
 
 import argparse
+import asyncio
+import base64
+import binascii
+import hmac
 import json
 import logging
 import os
 import re
+import shutil
+import subprocess
 import sys
 from datetime import datetime, timezone
 from html import escape
@@ -33,13 +40,84 @@ import kb_proposal_review_packet as proposal_review  # noqa: E402
 
 logger = logging.getLogger("argus.kb_claims")
 
-KB_CLAIM_PUBLIC_PATHS = frozenset()
+KB_CLAIM_SELF_AUTH_PATHS = frozenset({"/api/kb/claims"})
-KB_CLAIM_PUBLIC_PREFIXES = ("/kb/claims/", "/api/kb/claims/")
+KB_CLAIM_GLOBAL_AUTH_PREFIXES = ("/api/kb/claims/", "/kb/claims/")
 KB_CLAIM_LOADER_KEY = web.AppKey("kb_claim_loader", object)
+KB_CLAIM_LIST_LOADER_KEY = web.AppKey("kb_claim_list_loader", object)
+KB_CLAIM_LIST_API_KEY = web.AppKey("kb_claim_list_api_key", str)
+
+CLAIM_LIST_SCHEMA = "livingip.canonical-claims.v1"
+CLAIM_LIST_DEFAULT_LIMIT = 25
+CLAIM_LIST_MAX_LIMIT = 100
+CLAIM_LIST_MAX_RESPONSE_BYTES = 500_000
+CLAIM_LIST_MAX_QUERY_LENGTH = 200
+CLAIM_LIST_MAX_FILTER_LENGTH = 64
+CLAIM_LIST_REQUIRED_ROLE = "kb_observatory_read"
+CLAIM_LIST_DEFAULT_API_KEY_FILE = "/opt/teleo-eval/secrets/kb-observatory-api-key"
+CLAIM_LIST_DEFAULT_SECRETS_FILE = "/opt/teleo-eval/secrets/kb-observatory-read-password"
+CLAIM_LIST_HANDLER_TIMEOUT_SECONDS = 10.0
+CLAIM_LIST_PROCESS_TIMEOUT_SECONDS = 8.0
+CLAIM_LIST_CONNECT_TIMEOUT_SECONDS = 3
+CLAIM_LIST_STATEMENT_TIMEOUT_MS = 5_000
 
 UUID_RE = re.compile(
     r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
 )
+SIMPLE_FILTER_RE = re.compile(r"^[A-Za-z0-9_.:-]+$")
+
+
+class _ClaimListLoaderExited(RuntimeError):
+    """Keep loader exits inside the worker thread's normal exception channel."""
+
+
+def load_claim_list_api_key() -> str | None:
+    """Load the Observatory-only API token from its root-managed file."""
+    path = Path(
+        os.environ.get("KB_CLAIM_BROWSER_API_KEY_FILE", CLAIM_LIST_DEFAULT_API_KEY_FILE)
+    )
+    try:
+        if not path.is_file() or path.stat().st_mode & 0o007:
+            return None
+        token = path.read_text(encoding="utf-8").strip()
+    except OSError:
+        return None
+    if (
+        len(token) < 24
+        or len(token) > 512
+        or not token.isascii()
+        or any(character.isspace() for character in token)
+    ):
+        return None
+    return token
+
+
+def _load_claim_list_password(secrets_file: str) -> str:
+    """Parse one literal ``PGPASSWORD=...`` assignment from a dedicated file."""
+    path = Path(secrets_file)
+    try:
+        if not path.is_file():
+            raise RuntimeError("canonical claim browser secrets file is unavailable")
+        if path.stat().st_mode & 0o007:
+            raise RuntimeError("canonical claim browser secrets file is world-accessible")
+        lines = path.read_text(encoding="utf-8").splitlines()
+    except OSError as exc:
+        raise RuntimeError("canonical claim browser secrets file is unavailable") from exc
+
+    password: str | None = None
+    for raw_line in lines:
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+        key, separator, value = line.partition("=")
+        if not separator or key.strip() != "PGPASSWORD" or password is not None:
+            raise RuntimeError("canonical claim browser secrets file has invalid format")
+        password = value.strip()
+        if len(password) >= 2 and password[0] == password[-1] and password[0] in {"'", '"'}:
+            password = password[1:-1]
+
+    if not password or "\n" in password or "\r" in password:
+        raise RuntimeError("canonical claim browser secrets file has invalid format")
+    return password
 
 
 def is_claim_id(value: Any) -> bool:
@@ -65,6 +143,60 @@ def short_claim_link_html(claim_id: Any) -> str:
     return claim_link_html(value, label=value[:8])
 
 
+def encode_claim_cursor(sort_timestamp: str, claim_id: str) -> str:
+    payload = json.dumps(
+        {"updated_at": sort_timestamp, "id": claim_id},
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
+
+
+def decode_claim_cursor(value: str) -> tuple[str, str]:
+    try:
+        padded = value + "=" * (-len(value) % 4)
+        payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8"))
+        timestamp = str(payload["updated_at"])
+        claim_id = str(payload["id"])
+        parsed_timestamp = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+        if parsed_timestamp.tzinfo is None:
+            raise ValueError("cursor timestamp must be timezone-aware")
+        if not is_claim_id(claim_id):
+            raise ValueError("invalid claim id")
+        return timestamp, claim_id.lower()
+    except (binascii.Error, KeyError, TypeError, ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError("invalid cursor") from exc
+
+
+def _parse_claim_list_request(request: web.Request) -> dict[str, Any]:
+    raw_limit = request.query.get("limit", str(CLAIM_LIST_DEFAULT_LIMIT))
+    try:
+        limit = max(1, min(int(raw_limit), CLAIM_LIST_MAX_LIMIT))
+    except ValueError as exc:
+        raise ValueError("invalid_limit") from exc
+
+    filters: dict[str, Any] = {
+        "q": request.query.get("q", "").strip(),
+        "status": request.query.get("status", "open").strip(),
+        "type": request.query.get("type", "").strip(),
+        "tag": request.query.get("tag", "").strip(),
+        "limit": limit,
+        "cursor": request.query.get("cursor", "").strip(),
+    }
+    if len(filters["q"]) > CLAIM_LIST_MAX_QUERY_LENGTH:
+        raise ValueError("query_too_long")
+    if len(filters["cursor"]) > 512:
+        raise ValueError("invalid_cursor")
+    for key in ("status", "type", "tag"):
+        value = filters[key]
+        if len(value) > CLAIM_LIST_MAX_FILTER_LENGTH or (value and not SIMPLE_FILTER_RE.match(value)):
+            raise ValueError(f"invalid_{key}")
+    if filters["cursor"]:
+        filters["cursor_values"] = decode_claim_cursor(filters["cursor"])
+    else:
+        filters["cursor_values"] = None
+    return filters
+
+
 def _db_args() -> argparse.Namespace:
     return argparse.Namespace(
         secrets_file=os.environ.get(
@@ -81,6 +213,66 @@ def _db_args() -> argparse.Namespace:
     )
 
 
+def _claim_list_db_args() -> argparse.Namespace:
+    role = os.environ.get("KB_CLAIM_BROWSER_ROLE", "").strip()
+    secrets_file = os.environ.get("KB_CLAIM_BROWSER_SECRETS_FILE", "").strip()
+    if not role and not secrets_file and Path(CLAIM_LIST_DEFAULT_SECRETS_FILE).is_file():
+        role = CLAIM_LIST_REQUIRED_ROLE
+        secrets_file = CLAIM_LIST_DEFAULT_SECRETS_FILE
+    if not role or not secrets_file:
+        raise RuntimeError("canonical claim browser read role is not configured")
+    if role != CLAIM_LIST_REQUIRED_ROLE:
+        raise RuntimeError(f"canonical claim browser requires role {CLAIM_LIST_REQUIRED_ROLE}")
+    return argparse.Namespace(
+        secrets_file=secrets_file,
+        container=os.environ.get("KB_CLAIM_BROWSER_CONTAINER", proposal_review.ap.DEFAULT_CONTAINER),
+        db=os.environ.get("KB_CLAIM_BROWSER_DB", proposal_review.ap.DEFAULT_DB),
+        host=os.environ.get("KB_CLAIM_BROWSER_HOST", proposal_review.ap.DEFAULT_HOST),
+        role=role,
+    )
+
+
+def _run_claim_list_psql(args: argparse.Namespace, sql: str, password: str) -> str:
+    """Run the bounded, read-only browser query without leaking process output."""
+    docker_binary = shutil.which("docker") or "docker"
+    command = [
+        docker_binary,
+        "exec",
+        "-e",
+        "PGPASSWORD",
+        "-e",
+        f"PGCONNECT_TIMEOUT={CLAIM_LIST_CONNECT_TIMEOUT_SECONDS}",
+        "-i",
+        args.container,
+        "psql",
+        "-U",
+        args.role,
+        "-h",
+        args.host,
+        "-d",
+        args.db,
+        "-v",
+        "ON_ERROR_STOP=1",
+        "-At",
+        "-q",
+    ]
+    try:
+        result = subprocess.run(
+            command,
+            input=sql,
+            text=True,
+            capture_output=True,
+            env={"PGPASSWORD": password, "PATH": "/usr/bin:/bin:/usr/local/bin"},
+            check=False,
+            timeout=CLAIM_LIST_PROCESS_TIMEOUT_SECONDS,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise TimeoutError("canonical claim browser database query timed out") from exc
+    if result.returncode != 0:
+        raise RuntimeError("canonical claim browser database query failed")
+    return result.stdout
+
+
 def _load_claim_from_db(claim_id: str) -> dict[str, Any] | None:
     password = proposal_review.ap.load_password(_db_args().secrets_file)
     args = _db_args()
@@ -167,6 +359,262 @@ def _load_claim(request: web.Request, claim_id: str) -> dict[str, Any] | None:
     return loader(claim_id) if loader else _load_claim_from_db(claim_id)
 
 
+def _load_claim_list_from_db(filters: dict[str, Any]) -> dict[str, Any]:
+    args = _claim_list_db_args()
+    password = _load_claim_list_password(args.secrets_file)
+    where_clauses: list[str] = []
+
+    if filters["q"]:
+        query_literal = proposal_review.ap.sql_literal(filters["q"].lower())
+        where_clauses.append(
+            "(strpos(lower(c.text), "
+            f"{query_literal}) > 0 or exists ("
+            "select 1 from unnest(coalesce(c.tags, '{}'::text[])) claim_tag "
+            f"where strpos(lower(claim_tag), {query_literal}) > 0))"
+        )
+    if filters["status"] and filters["status"].lower() != "all":
+        where_clauses.append(f"c.status::text = {proposal_review.ap.sql_literal(filters['status'])}")
+    if filters["type"]:
+        where_clauses.append(f"c.type::text = {proposal_review.ap.sql_literal(filters['type'])}")
+    if filters["tag"]:
+        where_clauses.append(
+            f"{proposal_review.ap.sql_literal(filters['tag'])} = any(coalesce(c.tags, '{{}}'::text[]))"
+        )
+
+    where_sql = f"where {' and '.join(where_clauses)}" if where_clauses else ""
+    cursor_sql = ""
+    if filters["cursor_values"]:
+        cursor_timestamp, cursor_id = filters["cursor_values"]
+        cursor_sql = (
+            "where (sort_timestamp, id) < ("
+            f"{proposal_review.ap.sql_literal(cursor_timestamp)}::timestamptz, "
+            f"{proposal_review.ap.sql_literal(cursor_id)}::uuid)"
+        )
+
+    fetch_limit = filters["limit"] + 1
+    sql = f"""
+set statement_timeout = '{CLAIM_LIST_STATEMENT_TIMEOUT_MS}ms';
+set lock_timeout = '1000ms';
+
+with filtered as (
+  select c.id,
+         c.type,
+         c.text,
+         c.status,
+         c.confidence,
+         c.tags,
+         c.superseded_by,
+         c.created_at,
+         c.updated_at,
+         coalesce(c.updated_at, c.created_at, '1970-01-01'::timestamptz) as sort_timestamp
+    from public.claims c
+   {where_sql}
+), page_rows as (
+  select *
+    from filtered
+   {cursor_sql}
+   order by sort_timestamp desc, id desc
+   limit {fetch_limit}
+), enriched_page_rows as (
+  select page_rows.*,
+         coalesce(evidence_counts.evidence_count, 0) as evidence_count,
+         coalesce(edge_counts.edge_count, 0) as edge_count
+    from page_rows
+    left join lateral (
+      select count(*) as evidence_count
+        from public.claim_evidence ce
+       where ce.claim_id = page_rows.id
+    ) evidence_counts on true
+    left join lateral (
+      select count(*) as edge_count
+        from public.claim_edges e
+       where e.from_claim = page_rows.id or e.to_claim = page_rows.id
+    ) edge_counts on true
+)
+select jsonb_build_object(
+  'session_read_only', current_setting('transaction_read_only'),
+  'total', (select count(*) from filtered),
+  'rows', coalesce((
+    select jsonb_agg(
+      jsonb_build_object(
+        'id', id::text,
+        'type', type::text,
+        'text', left(text, 1200),
+        'text_truncated', length(text) > 1200,
+        'status', status::text,
+        'confidence', confidence,
+        'tags', coalesce(to_jsonb(tags), '[]'::jsonb),
+        'superseded_by', superseded_by::text,
+        'created_at', created_at::text,
+        'updated_at', updated_at::text,
+        'evidence_count', evidence_count,
+        'edge_count', edge_count,
+        '_cursor_timestamp', sort_timestamp::text
+      ) order by sort_timestamp desc, id desc
+    ) from enriched_page_rows
+  ), '[]'::jsonb)
+)::text;
+"""
+    out = _run_claim_list_psql(args, sql, password).strip()
+    result = json.loads(out) if out else {"total": 0, "rows": []}
+    if result.pop("session_read_only", None) != "on":
+        raise RuntimeError("canonical claim browser database session is not read-only")
+    return result
+
+
+def _load_claim_list_sync(request: web.Request, filters: dict[str, Any]) -> dict[str, Any]:
+    loader = request.app.get(KB_CLAIM_LIST_LOADER_KEY)
+    try:
+        return loader(filters) if loader else _load_claim_list_from_db(filters)
+    except SystemExit as exc:
+        # Python 3.11 can propagate SystemExit from asyncio.to_thread before the
+        # awaiting handler gets a chance to sanitize it. Convert only that exit
+        # signal here; KeyboardInterrupt and other BaseException types retain
+        # their normal process-level semantics.
+        raise _ClaimListLoaderExited from exc
+
+
+async def _load_claim_list(request: web.Request, filters: dict[str, Any]) -> dict[str, Any]:
+    return await asyncio.wait_for(
+        asyncio.to_thread(_load_claim_list_sync, request, filters),
+        timeout=CLAIM_LIST_HANDLER_TIMEOUT_SECONDS,
+    )
+
+
+def _sanitize_claim_summary(row: dict[str, Any]) -> tuple[dict[str, Any], str]:
+    claim_id = str(row.get("id") or "").lower()
+    if not is_claim_id(claim_id):
+        raise ValueError("invalid canonical claim row")
+    cursor_timestamp = str(row.get("_cursor_timestamp") or row.get("updated_at") or row.get("created_at") or "")
+    parsed_timestamp = datetime.fromisoformat(cursor_timestamp.replace("Z", "+00:00"))
+    if parsed_timestamp.tzinfo is None:
+        raise ValueError("invalid canonical claim timestamp")
+    tags = [str(tag)[:128] for tag in (row.get("tags") or []) if isinstance(tag, (str, int, float))][:32]
+    summary = {
+        "id": claim_id,
+        "type": str(row.get("type") or "unknown")[:64],
+        "text": str(row.get("text") or "")[:1200],
+        "text_truncated": bool(row.get("text_truncated")),
+        "status": str(row.get("status") or "unknown")[:64],
+        "confidence": row.get("confidence"),
+        "tags": tags,
+        "superseded_by": str(row["superseded_by"]) if is_claim_id(row.get("superseded_by")) else None,
+        "created_at": str(row.get("created_at") or "") or None,
+        "updated_at": str(row.get("updated_at") or "") or None,
+        "evidence_count": max(0, int(row.get("evidence_count") or 0)),
+        "edge_count": max(0, int(row.get("edge_count") or 0)),
+    }
+    return summary, cursor_timestamp
+
+
+def _claim_list_payload(raw: dict[str, Any], filters: dict[str, Any]) -> dict[str, Any]:
+    raw_rows = list(raw.get("rows") or [])
+    if len(raw_rows) > filters["limit"] + 1:
+        raise ValueError("canonical claim page exceeded its requested bound")
+
+    sanitized_rows: list[tuple[dict[str, Any], str]] = []
+    seen_ids: set[str] = set()
+    previous_key: tuple[datetime, str] | None = None
+    request_cursor_key: tuple[datetime, str] | None = None
+    if filters["cursor_values"]:
+        request_cursor_timestamp, request_cursor_id = filters["cursor_values"]
+        request_cursor_key = (
+            datetime.fromisoformat(request_cursor_timestamp.replace("Z", "+00:00")),
+            request_cursor_id,
+        )
+
+    for row in raw_rows:
+        claim, cursor_timestamp = _sanitize_claim_summary(row)
+        if claim["id"] in seen_ids:
+            raise ValueError("canonical claim page contains duplicate ids")
+        seen_ids.add(claim["id"])
+        row_key = (
+            datetime.fromisoformat(cursor_timestamp.replace("Z", "+00:00")),
+            claim["id"],
+        )
+        if request_cursor_key is not None and row_key >= request_cursor_key:
+            raise ValueError("canonical claim page crossed its request cursor")
+        if previous_key is not None and row_key >= previous_key:
+            raise ValueError("canonical claim page is not strictly ordered")
+        previous_key = row_key
+        sanitized_rows.append((claim, cursor_timestamp))
+
+    visible_rows = sanitized_rows[: filters["limit"]]
+    total = max(0, int(raw.get("total") or 0))
+    if total < len(visible_rows):
+        raise ValueError("canonical claim total is smaller than its page")
+    generated_at = datetime.now(timezone.utc).isoformat()
+
+    def build_payload(page_rows: list[tuple[dict[str, Any], str]]) -> dict[str, Any]:
+        claims = [claim for claim, _cursor_timestamp in page_rows]
+        has_more = len(page_rows) < len(sanitized_rows)
+        last_cursor_timestamp = page_rows[-1][1] if page_rows else ""
+        next_cursor = (
+            encode_claim_cursor(last_cursor_timestamp, claims[-1]["id"])
+            if has_more and claims
+            else None
+        )
+        return {
+            "schema": CLAIM_LIST_SCHEMA,
+            "generated_at": generated_at,
+            "read_only": True,
+            "source": {
+                "store": "canonical_postgres",
+                "relation": "public.claims",
+                "receipt": "server-side read-only query",
+            },
+            "filters": {
+                "q": filters["q"] or None,
+                "status": filters["status"] or None,
+                "type": filters["type"] or None,
+                "tag": filters["tag"] or None,
+            },
+            "page": {
+                "limit": filters["limit"],
+                "returned": len(claims),
+                "total": total,
+                "has_more": has_more,
+                "next_cursor": next_cursor,
+            },
+            "claims": claims,
+        }
+
+    payload = build_payload(visible_rows)
+    if len(json.dumps(payload).encode("utf-8")) <= CLAIM_LIST_MAX_RESPONSE_BYTES:
+        return payload
+
+    fitted_payload: dict[str, Any] | None = None
+    low = 1
+    high = len(visible_rows) - 1
+    while low <= high:
+        midpoint = (low + high) // 2
+        candidate = build_payload(visible_rows[:midpoint])
+        if len(json.dumps(candidate).encode("utf-8")) <= CLAIM_LIST_MAX_RESPONSE_BYTES:
+            fitted_payload = candidate
+            low = midpoint + 1
+        else:
+            high = midpoint - 1
+    if fitted_payload is None:
+        raise ValueError("canonical claim response exceeded its byte bound")
+    return fitted_payload
+
+
+def _private_json_response(payload: dict[str, Any], *, status: int = 200) -> web.Response:
+    response = web.json_response(payload, status=status)
+    response.headers["Cache-Control"] = "private, no-store, max-age=0"
+    response.headers["Pragma"] = "no-cache"
+    response.headers["Vary"] = "X-Api-Key"
+    return response
+
+
+def _private_html_response(body: str, *, status: int = 200) -> web.Response:
+    response = web.Response(text=body, content_type="text/html", status=status)
+    response.headers["Cache-Control"] = "private, no-store, max-age=0"
+    response.headers["Pragma"] = "no-cache"
+    response.headers["Vary"] = "X-Api-Key"
+    return response
+
+
 def _metadata_table(claim: dict[str, Any]) -> str:
     rows = [
         ("ID", claim_link_html(claim.get("id"))),
@@ -270,40 +718,67 @@ def render_kb_claim_page(data: dict[str, Any]) -> str:
 async def handle_api_kb_claim(request: web.Request) -> web.Response:
     claim_id = request.match_info["claim_id"]
     if not is_claim_id(claim_id):
-        return web.json_response({"error": "invalid_claim_id", "claim_id": claim_id}, status=400)
+        return _private_json_response({"error": "invalid_claim_id"}, status=400)
     try:
         data = _load_claim(request, claim_id)
-    except BaseException as exc:
+    except (Exception, SystemExit) as exc:
-        logger.exception("KB claim API load failed")
+        logger.error("KB claim API load failed (%s)", type(exc).__name__)
-        return web.json_response({"error": "kb_claim_load_failed", "detail": str(exc)}, status=500)
+        return _private_json_response({"error": "kb_claim_load_failed"}, status=503)
     if not data:
-        return web.json_response({"error": "claim_not_found", "claim_id": claim_id}, status=404)
+        return _private_json_response({"error": "claim_not_found"}, status=404)
-    return web.json_response(data)
+    return _private_json_response(data)
+
+
+async def handle_api_kb_claims(request: web.Request) -> web.Response:
+    expected_api_key = request.app.get(KB_CLAIM_LIST_API_KEY)
+    if not expected_api_key:
+        return _private_json_response({"error": "canonical_claims_auth_unconfigured"}, status=503)
+    provided_api_key = request.headers.get("X-Api-Key", "")
+    try:
+        authenticated = hmac.compare_digest(
+            provided_api_key.encode("ascii"),
+            str(expected_api_key).encode("ascii"),
+        )
+    except UnicodeEncodeError:
+        authenticated = False
+    if not authenticated:
+        return _private_json_response({"error": "unauthorized"}, status=401)
+    try:
+        filters = _parse_claim_list_request(request)
+    except ValueError as exc:
+        return _private_json_response({"error": str(exc)}, status=400)
+    try:
+        raw = await _load_claim_list(request, filters)
+        payload = _claim_list_payload(raw, filters)
+    except (Exception, SystemExit) as exc:
+        logger.error("Canonical KB claim list load failed (%s)", type(exc).__name__)
+        return _private_json_response({"error": "canonical_claims_unavailable"}, status=503)
+    return _private_json_response(payload)
 
 
 async def handle_kb_claim_page(request: web.Request) -> web.Response:
     claim_id = request.match_info["claim_id"]
     if not is_claim_id(claim_id):
-        return web.Response(text="Invalid claim id", status=400)
+        return _private_html_response("Invalid claim id", status=400)
     try:
         data = _load_claim(request, claim_id)
-    except BaseException as exc:
+    except (Exception, SystemExit) as exc:
-        logger.exception("KB claim page load failed")
+        logger.error("KB claim page load failed (%s)", type(exc).__name__)
-        return web.Response(
+        return _private_html_response(
-            text=render_page(
+            render_page(
                 "KB Claim",
                 "Canonical claim, evidence rows, and graph edges",
                 "/kb-proposals",
-                f'<div class="alert-banner alert-critical">Failed to load KB claim: {escape(str(exc))}</div>',
+                '<div class="alert-banner alert-critical">Canonical claim is temporarily unavailable.</div>',
             ),
-            content_type="text/html",
             status=500,
         )
     if not data:
-        return web.Response(text="Claim not found", status=404)
+        return _private_html_response("Claim not found", status=404)
-    return web.Response(text=render_kb_claim_page(data), content_type="text/html")
+    return _private_html_response(render_kb_claim_page(data))
 
 
 def register_kb_claim_routes(app: web.Application) -> None:
+    app.router.add_get("/api/kb/claims", handle_api_kb_claims)
     app.router.add_get("/api/kb/claims/{claim_id}", handle_api_kb_claim)
     app.router.add_get("/kb/claims/{claim_id}", handle_kb_claim_page)

docs/canonical-claims-browser.md

@@ -0,0 +1,119 @@
+# Canonical Claims Browser Read Path
+
+Status: implementation-ready, not deployed
+
+The canonical claims browser exposes a protected, read-only summary of
+`public.claims` for the LivingIP Observatory. It is deliberately separate from
+the public Markdown/Qdrant knowledge browser.
+
+## Contract
+
+`GET /api/kb/claims`
+
+- protected by a route-scoped Observatory token; the browser never receives it
+- schema: `livingip.canonical-claims.v1`
+- filters: `q`, `status`, `type`, and `tag`
+- opaque keyset cursor over `(updated_at, id)`
+- default requested page size 25; maximum 100; the server may return fewer rows
+  while preserving `has_more` and `next_cursor` to keep the serialized response
+  at or below 500,000 bytes
+- allowlisted output only: canonical UUID, bounded claim text, type, status,
+  confidence, tags, timestamps, supersession pointer, evidence count, and edge
+  count
+- response headers: `Cache-Control: private, no-store, max-age=0`
+- no mutation methods, source excerpts, storage paths, credentials, internal
+  endpoints, or embeddings
+
+Only the exact `GET /api/kb/claims` path bypasses the global Argus middleware so
+its handler can authenticate the route-scoped token. The legacy
+`/api/kb/claims/{uuid}` and `/kb/claims/{uuid}` routes require the global Argus
+key and fail closed when that key is not configured. The Observatory does not
+receive, reuse, or expose the global Argus key.
+
+Those legacy detail routes are explicitly outside
+`livingip.canonical-claims.v1`: they include source-rich evidence and still use
+the pre-existing claim-review database credential path. The Observatory adapter
+must not call them. Migrating legacy detail reads to a separate role (including
+a deliberate `public.sources` grant) is residual hardening, not a prerequisite
+for this summary-only browser.
+
+Create a separate high-entropy token in a root-managed, service-readable file:
+
+```text
+/opt/teleo-eval/secrets/kb-observatory-api-key
+```
+
+The file contains only a 24-to-512-character ASCII token with no whitespace (a
+trailing newline is accepted), and it must not be world-accessible. Set
+`KB_CLAIM_BROWSER_API_KEY_FILE` only when using a different root-managed path.
+Configure the same value as the server-only
+`OBSERVATORY_CANONICAL_API_KEY` secret in Vercel; never use a `NEXT_PUBLIC_`
+variable.
+
+## Dedicated Database Role
+
+The handler accepts either both explicit `KB_CLAIM_BROWSER_ROLE` and
+`KB_CLAIM_BROWSER_SECRETS_FILE` settings, or the protected default password file
+at `/opt/teleo-eval/secrets/kb-observatory-read-password`. The default path is
+file-gated: if the file is absent, partial environment configuration is present,
+or the file fails its permission/format checks, the endpoint fails closed. The
+only accepted role name is `kb_observatory_read`, and every response is withheld
+unless Postgres reports that the current session has
+`transaction_read_only=on`.
+
+Create a dedicated login using a password supplied out of band, then grant only
+the relations needed by the list query:
+
+```sql
+create role kb_observatory_read login password :'observatory_read_password';
+alter role kb_observatory_read set default_transaction_read_only = on;
+grant connect on database teleo to kb_observatory_read;
+grant usage on schema public to kb_observatory_read;
+grant select on public.claims, public.claim_evidence, public.claim_edges
+  to kb_observatory_read;
+```
+
+Store the password in a dedicated root-managed file readable by the Argus
+service. Its format is intentionally narrow: blank lines and `#` comments are
+allowed, followed by exactly one literal assignment (no shell expansion):
+
+```text
+PGPASSWORD='the-observatory-read-role-password'
+```
+
+No `KB_APPLY_PASSWORD`, `KB_APPLY_DB_PASSWORD`, or additional assignments are
+accepted. The file must not be world-accessible; a suitable deployment mode is
+`root:teleo 0640`. Never place the password in Git, systemd unit text, browser
+configuration, or a `NEXT_PUBLIC_` environment variable.
+
+Optional explicit Argus overrides:
+
+```text
+KB_CLAIM_BROWSER_ROLE=kb_observatory_read
+KB_CLAIM_BROWSER_SECRETS_FILE=/opt/teleo-eval/secrets/kb-observatory-read-password
+```
+
+Optional connection overrides use the same prefix:
+`KB_CLAIM_BROWSER_CONTAINER`, `KB_CLAIM_BROWSER_DB`,
+`KB_CLAIM_BROWSER_HOST`.
+
+The query has three independent bounds: a three-second Postgres connection
+timeout, a five-second Postgres statement timeout, and an eight-second process
+timeout. The aiohttp handler runs the blocking database work in a worker thread
+and stops awaiting it after ten seconds, so a slow database cannot block the
+Argus event loop.
+
+## Promotion Checks
+
+1. Run `pytest tests/test_kb_claim_routes.py -q` and Ruff on the route and test.
+2. Deploy both dedicated secret files without changing the global Argus key.
+3. Verify an unauthenticated list request returns `401`.
+4. Verify an authenticated request returns the versioned contract and no-store
+   headers.
+5. Verify every response body is at most 500,000 bytes and two cursor pages are
+   disjoint and stable, including when the first page is shortened by the byte
+   cap.
+6. From the database session, verify `current_setting('transaction_read_only')`
+   is `on` and INSERT/UPDATE/DELETE all fail.
+7. Configure the Vercel adapter with a server-only URL and API key only after
+   preview Deployment Protection is enabled.

tests/test_kb_claim_routes.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import asyncio
 import json
 import sys
+import threading
+import time
 from pathlib import Path
 
 import pytest
@@ -17,11 +19,24 @@ from aiohttp.test_utils import make_mocked_request
 REPO_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(REPO_ROOT / "diagnostics"))
 
+import app as argus_app  # noqa: E402
 import kb_claim_routes as routes  # noqa: E402
 
-
 CLAIM_ID = "d0000000-0000-0000-0000-000000000005"
 CONNECTED_ID = "e0000000-0000-0000-0000-000000000006"
+SECOND_CLAIM_ID = "c0000000-0000-0000-0000-000000000004"
+THIRD_CLAIM_ID = "b0000000-0000-0000-0000-000000000003"
+API_KEY = "test-argus-api-key"
+
+
+def _claim_list_app() -> web.Application:
+    app = web.Application()
+    app[routes.KB_CLAIM_LIST_API_KEY] = API_KEY
+    return app
+
+
+def _claim_list_request(path: str, *, app: web.Application):
+    return make_mocked_request("GET", path, app=app, headers={"X-Api-Key": API_KEY})
 
 
 def _claim_data():
@@ -58,6 +73,24 @@ def _claim_data():
     }
 
 
+def _claim_summary(claim_id: str, *, timestamp: str, text: str = "Canonical claim"):
+    return {
+        "id": claim_id,
+        "type": "strategy_kernel",
+        "text": text,
+        "text_truncated": False,
+        "status": "open",
+        "confidence": "0.72",
+        "tags": ["claims", "review"],
+        "created_at": timestamp,
+        "updated_at": timestamp,
+        "evidence_count": 3,
+        "edge_count": 5,
+        "_cursor_timestamp": timestamp,
+        "internal_secret": "must-not-leak",
+    }
+
+
 def test_claim_link_html_links_valid_claim_ids() -> None:
     html = routes.claim_link_html(CLAIM_ID)
 
@@ -91,3 +124,417 @@ def test_api_kb_claim_uses_loader_and_returns_claim_graph() -> None:
     assert response.status == 200
     assert data["claim"]["id"] == CLAIM_ID
     assert data["edges"][0]["connected_id"] == CONNECTED_ID
+    assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
+
+
+def test_claim_cursor_round_trip() -> None:
+    timestamp = "2026-07-13T21:12:00+00:00"
+    cursor = routes.encode_claim_cursor(timestamp, CLAIM_ID)
+
+    assert routes.decode_claim_cursor(cursor) == (timestamp, CLAIM_ID)
+
+
+def test_canonical_claim_list_database_role_fails_closed(monkeypatch, tmp_path) -> None:
+    monkeypatch.delenv("KB_CLAIM_BROWSER_ROLE", raising=False)
+    monkeypatch.delenv("KB_CLAIM_BROWSER_SECRETS_FILE", raising=False)
+    default_secrets_file = tmp_path / "missing-kb-observatory-read-password"
+    monkeypatch.setattr(routes, "CLAIM_LIST_DEFAULT_SECRETS_FILE", str(default_secrets_file))
+
+    with pytest.raises(RuntimeError, match="read role is not configured"):
+        routes._claim_list_db_args()
+
+    default_secrets_file.write_text("PGPASSWORD=not-printed\n", encoding="utf-8")
+    default_secrets_file.chmod(0o600)
+    default_args = routes._claim_list_db_args()
+
+    assert default_args.role == routes.CLAIM_LIST_REQUIRED_ROLE
+    assert default_args.secrets_file == str(default_secrets_file)
+
+    monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", "kb_apply")
+    monkeypatch.setenv("KB_CLAIM_BROWSER_SECRETS_FILE", "/run/secrets/read-password")
+    with pytest.raises(RuntimeError, match=routes.CLAIM_LIST_REQUIRED_ROLE):
+        routes._claim_list_db_args()
+
+    monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", routes.CLAIM_LIST_REQUIRED_ROLE)
+    args = routes._claim_list_db_args()
+
+    assert args.role == routes.CLAIM_LIST_REQUIRED_ROLE
+    assert args.secrets_file == "/run/secrets/read-password"
+
+
+def test_canonical_claim_list_database_session_must_be_read_only(monkeypatch, tmp_path) -> None:
+    secrets_file = tmp_path / "kb-observatory-read.env"
+    secrets_file.write_text("PGPASSWORD=not-printed\n", encoding="utf-8")
+    secrets_file.chmod(0o640)
+    monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", routes.CLAIM_LIST_REQUIRED_ROLE)
+    monkeypatch.setenv("KB_CLAIM_BROWSER_SECRETS_FILE", str(secrets_file))
+    observed_sql = {}
+
+    def run_psql(_args, sql, _password):
+        observed_sql["value"] = sql
+        return json.dumps({"session_read_only": "off", "total": 0, "rows": []})
+
+    monkeypatch.setattr(routes, "_run_claim_list_psql", run_psql)
+    filters = {
+        "q": "",
+        "status": "all",
+        "type": "",
+        "tag": "",
+        "limit": 25,
+        "cursor": "",
+        "cursor_values": None,
+    }
+
+    with pytest.raises(RuntimeError, match="session is not read-only"):
+        routes._load_claim_list_from_db(filters)
+
+    assert "current_setting('transaction_read_only')" in observed_sql["value"]
+    assert "set statement_timeout = '5000ms'" in observed_sql["value"]
+    assert observed_sql["value"].index("page_rows as") < observed_sql["value"].index(
+        "from public.claim_evidence"
+    )
+
+
+def test_canonical_claim_list_secrets_file_accepts_only_pgpassword(tmp_path) -> None:
+    secrets_file = tmp_path / "kb-observatory-read.env"
+    secrets_file.write_text("# dedicated read role\nPGPASSWORD='read-only-secret'\n", encoding="utf-8")
+    secrets_file.chmod(0o640)
+
+    assert routes._load_claim_list_password(str(secrets_file)) == "read-only-secret"
+
+    secrets_file.write_text("KB_APPLY_PASSWORD=wrong-boundary\n", encoding="utf-8")
+    with pytest.raises(RuntimeError, match="invalid format"):
+        routes._load_claim_list_password(str(secrets_file))
+
+    secrets_file.write_text("PGPASSWORD=secret\n", encoding="utf-8")
+    secrets_file.chmod(0o644)
+    with pytest.raises(RuntimeError, match="world-accessible"):
+        routes._load_claim_list_password(str(secrets_file))
+
+
+def test_canonical_claim_list_process_timeout_is_enforced_and_sanitized(monkeypatch) -> None:
+    args = routes.argparse.Namespace(
+        container="postgres",
+        db="teleo",
+        host="127.0.0.1",
+        role=routes.CLAIM_LIST_REQUIRED_ROLE,
+    )
+    observed = {}
+
+    def timeout_run(command, **kwargs):
+        observed["command"] = command
+        observed.update(kwargs)
+        raise routes.subprocess.TimeoutExpired(command, kwargs["timeout"], stderr="secret")
+
+    monkeypatch.setattr(routes.subprocess, "run", timeout_run)
+
+    with pytest.raises(TimeoutError, match="database query timed out") as exc_info:
+        routes._run_claim_list_psql(args, "select 1", "must-not-leak")
+
+    assert "secret" not in str(exc_info.value)
+    assert observed["timeout"] == routes.CLAIM_LIST_PROCESS_TIMEOUT_SECONDS
+    assert f"PGCONNECT_TIMEOUT={routes.CLAIM_LIST_CONNECT_TIMEOUT_SECONDS}" in observed[
+        "command"
+    ]
+
+
+def test_canonical_claim_list_api_key_loads_only_from_dedicated_file(
+    monkeypatch, tmp_path
+) -> None:
+    api_key_file = tmp_path / "kb-observatory-api-key"
+    api_key_file.write_text("route-scoped-observatory-key\n", encoding="utf-8")
+    api_key_file.chmod(0o640)
+    monkeypatch.setenv("KB_CLAIM_BROWSER_API_KEY_FILE", str(api_key_file))
+
+    assert routes.load_claim_list_api_key() == "route-scoped-observatory-key"
+
+    api_key_file.write_text("two tokens\n", encoding="utf-8")
+    assert routes.load_claim_list_api_key() is None
+
+    api_key_file.write_text("é" * 24, encoding="utf-8")
+    assert routes.load_claim_list_api_key() is None
+
+    api_key_file.write_text("route-scoped-observatory-key\n", encoding="utf-8")
+    api_key_file.chmod(0o644)
+    assert routes.load_claim_list_api_key() is None
+
+
+def test_canonical_claim_list_is_versioned_paginated_and_sanitized() -> None:
+    app = _claim_list_app()
+    captured_filters = {}
+
+    def loader(filters):
+        captured_filters.update(filters)
+        return {
+            "total": 1837,
+            "rows": [
+                _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00"),
+                _claim_summary(SECOND_CLAIM_ID, timestamp="2026-07-13T20:12:00+00:00"),
+            ],
+        }
+
+    app[routes.KB_CLAIM_LIST_LOADER_KEY] = loader
+    req = _claim_list_request(
+        "/api/kb/claims?limit=1&q=falsifiable&type=strategy_kernel&tag=review",
+        app=app,
+    )
+
+    response = asyncio.run(routes.handle_api_kb_claims(req))
+    data = json.loads(response.body.decode())
+
+    assert response.status == 200
+    assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
+    assert response.headers["Vary"] == "X-Api-Key"
+    assert data["schema"] == routes.CLAIM_LIST_SCHEMA
+    assert data["read_only"] is True
+    assert data["source"]["store"] == "canonical_postgres"
+    assert data["page"] == {
+        "limit": 1,
+        "returned": 1,
+        "total": 1837,
+        "has_more": True,
+        "next_cursor": data["page"]["next_cursor"],
+    }
+    assert routes.decode_claim_cursor(data["page"]["next_cursor"])[1] == CLAIM_ID
+    assert data["claims"][0]["id"] == CLAIM_ID
+    assert "internal_secret" not in data["claims"][0]
+    assert captured_filters["q"] == "falsifiable"
+    assert captured_filters["type"] == "strategy_kernel"
+    assert captured_filters["tag"] == "review"
+
+
+def test_canonical_claim_list_caps_response_bytes_without_skipping_rows(monkeypatch) -> None:
+    app = _claim_list_app()
+    rows = [
+        _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00", text="🔥" * 1200),
+        _claim_summary(
+            SECOND_CLAIM_ID,
+            timestamp="2026-07-13T20:12:00+00:00",
+            text="🔥" * 1200,
+        ),
+        _claim_summary(
+            THIRD_CLAIM_ID,
+            timestamp="2026-07-13T19:12:00+00:00",
+            text="🔥" * 1200,
+        ),
+    ]
+    for row in rows:
+        row["tags"] = ["🔥" * 128] * 32
+    app[routes.KB_CLAIM_LIST_LOADER_KEY] = lambda _filters: {"total": 3, "rows": rows}
+    assert routes.CLAIM_LIST_MAX_RESPONSE_BYTES < 512_000
+    byte_limit = 100_000
+    monkeypatch.setattr(routes, "CLAIM_LIST_MAX_RESPONSE_BYTES", byte_limit)
+    request = _claim_list_request("/api/kb/claims?limit=3&status=all", app=app)
+
+    response = asyncio.run(routes.handle_api_kb_claims(request))
+    data = json.loads(response.body.decode())
+
+    assert response.status == 200
+    assert len(response.body) <= byte_limit
+    assert data["page"]["limit"] == 3
+    assert data["page"]["returned"] == 1
+    assert data["page"]["has_more"] is True
+    assert routes.decode_claim_cursor(data["page"]["next_cursor"])[1] == CLAIM_ID
+    assert [claim["id"] for claim in data["claims"]] == [CLAIM_ID]
+
+
+def test_canonical_claim_list_clamps_limit_and_rejects_bad_cursor() -> None:
+    app = _claim_list_app()
+    observed = {}
+    app[routes.KB_CLAIM_LIST_LOADER_KEY] = lambda filters: observed.update(filters) or {"total": 0, "rows": []}
+    large_req = _claim_list_request("/api/kb/claims?limit=999", app=app)
+
+    large_response = asyncio.run(routes.handle_api_kb_claims(large_req))
+    large_data = json.loads(large_response.body.decode())
+
+    assert large_response.status == 200
+    assert large_data["page"]["limit"] == routes.CLAIM_LIST_MAX_LIMIT
+    assert observed["limit"] == routes.CLAIM_LIST_MAX_LIMIT
+
+    bad_req = _claim_list_request("/api/kb/claims?cursor=not-a-cursor", app=app)
+    bad_response = asyncio.run(routes.handle_api_kb_claims(bad_req))
+    bad_data = json.loads(bad_response.body.decode())
+
+    assert bad_response.status == 400
+    assert bad_data == {"error": "invalid cursor"}
+
+    long_req = _claim_list_request(f"/api/kb/claims?cursor={'x' * 513}", app=app)
+    long_response = asyncio.run(routes.handle_api_kb_claims(long_req))
+
+    assert long_response.status == 400
+
+
+def test_canonical_claim_list_failure_is_sanitized() -> None:
+    app = _claim_list_app()
+
+    def fail(_filters):
+        raise SystemExit("database password would have appeared here")
+
+    app[routes.KB_CLAIM_LIST_LOADER_KEY] = fail
+    req = _claim_list_request("/api/kb/claims", app=app)
+
+    response = asyncio.run(routes.handle_api_kb_claims(req))
+    data = json.loads(response.body.decode())
+
+    assert response.status == 503
+    assert data == {"error": "canonical_claims_unavailable"}
+    assert b"password" not in response.body
+
+
+def test_canonical_claim_auth_boundaries_are_exact() -> None:
+    list_path = "/api/kb/claims"
+
+    assert list_path in routes.KB_CLAIM_SELF_AUTH_PATHS
+    assert f"{list_path}/{CLAIM_ID}" not in routes.KB_CLAIM_SELF_AUTH_PATHS
+    assert f"/api/kb/claims/{CLAIM_ID}".startswith(routes.KB_CLAIM_GLOBAL_AUTH_PREFIXES)
+    assert f"/kb/claims/{CLAIM_ID}".startswith(routes.KB_CLAIM_GLOBAL_AUTH_PREFIXES)
+
+
+def test_canonical_claim_list_authentication_fails_closed() -> None:
+    app_without_key = web.Application()
+    unconfigured_req = make_mocked_request("GET", "/api/kb/claims", app=app_without_key)
+    unconfigured_response = asyncio.run(routes.handle_api_kb_claims(unconfigured_req))
+
+    assert unconfigured_response.status == 503
+    assert json.loads(unconfigured_response.body.decode()) == {"error": "canonical_claims_auth_unconfigured"}
+
+    configured_app = _claim_list_app()
+    unauthorized_req = make_mocked_request("GET", "/api/kb/claims", app=configured_app)
+    unauthorized_response = asyncio.run(routes.handle_api_kb_claims(unauthorized_req))
+
+    assert unauthorized_response.status == 401
+    assert json.loads(unauthorized_response.body.decode()) == {"error": "unauthorized"}
+
+    non_ascii_request = make_mocked_request(
+        "GET",
+        "/api/kb/claims",
+        app=configured_app,
+        headers={"X-Api-Key": "é" * 24},
+    )
+    non_ascii_response = asyncio.run(routes.handle_api_kb_claims(non_ascii_request))
+
+    assert non_ascii_response.status == 401
+    assert json.loads(non_ascii_response.body.decode()) == {"error": "unauthorized"}
+
+
+def test_argus_middleware_only_bypasses_global_key_for_exact_list_path() -> None:
+    async def ok_handler(_request):
+        return web.Response(status=204)
+
+    app = web.Application()
+    list_request = make_mocked_request("GET", "/api/kb/claims", app=app)
+    detail_request = make_mocked_request(
+        "GET", f"/api/kb/claims/{CLAIM_ID}", app=app
+    )
+    html_request = make_mocked_request("GET", f"/kb/claims/{CLAIM_ID}", app=app)
+
+    list_response = asyncio.run(argus_app.auth_middleware(list_request, ok_handler))
+    detail_response = asyncio.run(argus_app.auth_middleware(detail_request, ok_handler))
+    html_response = asyncio.run(argus_app.auth_middleware(html_request, ok_handler))
+
+    assert list_response.status == 204
+    assert detail_response.status == 503
+    assert html_response.status == 503
+
+
+def test_canonical_claim_list_loader_runs_off_the_event_loop() -> None:
+    app = _claim_list_app()
+    event_loop_thread = threading.get_ident()
+    loader_thread = None
+
+    def loader(_filters):
+        nonlocal loader_thread
+        loader_thread = threading.get_ident()
+        return {"total": 0, "rows": []}
+
+    app[routes.KB_CLAIM_LIST_LOADER_KEY] = loader
+    request = _claim_list_request("/api/kb/claims", app=app)
+
+    response = asyncio.run(routes.handle_api_kb_claims(request))
+
+    assert response.status == 200
+    assert loader_thread is not None
+    assert loader_thread != event_loop_thread
+
+
+def test_canonical_claim_list_loader_timeout_is_sanitized(monkeypatch) -> None:
+    app = _claim_list_app()
+
+    def slow_loader(_filters):
+        time.sleep(0.03)
+        return {"total": 0, "rows": []}
+
+    app[routes.KB_CLAIM_LIST_LOADER_KEY] = slow_loader
+    monkeypatch.setattr(routes, "CLAIM_LIST_HANDLER_TIMEOUT_SECONDS", 0.001)
+    request = _claim_list_request("/api/kb/claims", app=app)
+
+    response = asyncio.run(routes.handle_api_kb_claims(request))
+
+    assert response.status == 503
+    assert json.loads(response.body.decode()) == {"error": "canonical_claims_unavailable"}
+
+
+def test_canonical_claim_list_rejects_duplicate_or_out_of_order_pages() -> None:
+    filters = {
+        "q": "",
+        "status": "all",
+        "type": "",
+        "tag": "",
+        "limit": 2,
+        "cursor": "",
+        "cursor_values": None,
+    }
+    newest = _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00")
+    older = _claim_summary(SECOND_CLAIM_ID, timestamp="2026-07-13T20:12:00+00:00")
+
+    with pytest.raises(ValueError, match="duplicate ids"):
+        routes._claim_list_payload({"total": 2, "rows": [newest, newest]}, filters)
+
+    with pytest.raises(ValueError, match="strictly ordered"):
+        routes._claim_list_payload({"total": 2, "rows": [older, newest]}, filters)
+
+
+def test_canonical_claim_list_rejects_rows_at_or_before_wrong_cursor_side() -> None:
+    cursor_timestamp = "2026-07-13T20:12:00+00:00"
+    filters = {
+        "q": "",
+        "status": "all",
+        "type": "",
+        "tag": "",
+        "limit": 1,
+        "cursor": routes.encode_claim_cursor(cursor_timestamp, SECOND_CLAIM_ID),
+        "cursor_values": (cursor_timestamp, SECOND_CLAIM_ID),
+    }
+    newer = _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00")
+
+    with pytest.raises(ValueError, match="crossed its request cursor"):
+        routes._claim_list_payload({"total": 2, "rows": [newer]}, filters)
+
+
+def test_legacy_claim_errors_do_not_expose_exception_text() -> None:
+    app = web.Application()
+
+    def fail(_claim_id):
+        raise RuntimeError("PGPASSWORD=must-not-leak")
+
+    app[routes.KB_CLAIM_LOADER_KEY] = fail
+    api_request = make_mocked_request(
+        "GET",
+        f"/api/kb/claims/{CLAIM_ID}",
+        app=app,
+        match_info={"claim_id": CLAIM_ID},
+    )
+    html_request = make_mocked_request(
+        "GET",
+        f"/kb/claims/{CLAIM_ID}",
+        app=app,
+        match_info={"claim_id": CLAIM_ID},
+    )
+
+    api_response = asyncio.run(routes.handle_api_kb_claim(api_request))
+    html_response = asyncio.run(routes.handle_kb_claim_page(html_request))
+
+    assert api_response.status == 503
+    assert html_response.status == 500
+    assert b"PGPASSWORD" not in api_response.body
+    assert b"PGPASSWORD" not in html_response.body