teleo-infrastructure/diagnostics/kb_claim_routes.py

"""Read-only canonical KB claim routes for Argus.

These routes show the Postgres-backed claim graph that Leo uses through the
``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
from pathlib import Path
from typing import Any

from aiohttp import web
from shared_ui import render_page

ROOT = Path(__file__).resolve().parent.parent
SCRIPT_DIR_CANDIDATES = [
    ROOT / "scripts",
    Path(os.environ.get("TELEO_INFRA_REPO_DIR", "/opt/teleo-eval/workspaces/deploy-infra")) / "scripts",
]
for scripts_dir in SCRIPT_DIR_CANDIDATES:
    if str(scripts_dir) not in sys.path:
        sys.path.insert(0, str(scripts_dir))

import kb_proposal_review_packet as proposal_review  # noqa: E402

logger = logging.getLogger("argus.kb_claims")

KB_CLAIM_SELF_AUTH_PATHS = frozenset({"/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:
    return bool(UUID_RE.match(str(value or "")))


def claim_path(claim_id: str) -> str:
    return f"/kb/claims/{claim_id}"


def claim_link_html(claim_id: Any, *, label: str | None = None) -> str:
    value = str(claim_id or "")
    if not is_claim_id(value):
        return f"<code>{escape(value or 'unknown')}</code>"
    text = label or value
    return f'<a class="claim-link" href="{claim_path(value)}"><code>{escape(text)}</code></a>'


def short_claim_link_html(claim_id: Any) -> str:
    value = str(claim_id or "")
    if not is_claim_id(value):
        return f"<code>{escape(value or 'unknown')}</code>"
    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(
            "KB_CLAIM_REVIEW_SECRETS_FILE",
            os.environ.get(
                "KB_PROPOSAL_REVIEW_SECRETS_FILE",
                os.environ.get("KB_APPLY_SECRETS_FILE", proposal_review.ap.DEFAULT_SECRETS_FILE),
            ),
        ),
        container=os.environ.get("KB_CLAIM_REVIEW_CONTAINER", proposal_review.ap.DEFAULT_CONTAINER),
        db=os.environ.get("KB_CLAIM_REVIEW_DB", proposal_review.ap.DEFAULT_DB),
        host=os.environ.get("KB_CLAIM_REVIEW_HOST", proposal_review.ap.DEFAULT_HOST),
        role=os.environ.get("KB_CLAIM_REVIEW_ROLE", proposal_review.ap.DEFAULT_ROLE),
    )


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()
    sql = f"""
with target as (
  select c.id,
         c.type,
         c.text,
         c.status,
         c.confidence,
         c.tags,
         c.superseded_by,
         c.created_at,
         c.updated_at
    from public.claims c
   where c.id = {proposal_review.ap.sql_literal(claim_id)}::uuid
)
select jsonb_build_object(
  'claim', jsonb_build_object(
    'id', target.id::text,
    'type', target.type,
    'text', target.text,
    'status', target.status,
    'confidence', target.confidence,
    'tags', coalesce(to_jsonb(target.tags), '[]'::jsonb),
    'superseded_by', target.superseded_by::text,
    'created_at', target.created_at::text,
    'updated_at', target.updated_at::text
  ),
  'evidence', coalesce((
    select jsonb_agg(row_data order by rn)
      from (
        select row_number() over (
                 order by ce.role::text,
                          ce.weight desc nulls last,
                          s.storage_path nulls last,
                          s.url nulls last
               ) as rn,
               jsonb_build_object(
                 'role', ce.role::text,
                 'weight', ce.weight,
                 'source_type', s.source_type,
                 'url', s.url,
                 'storage_path', s.storage_path,
                 'excerpt', left(coalesce(s.excerpt, ''), 1200)
               ) as row_data
          from public.claim_evidence ce
          join public.sources s on s.id = ce.source_id
         where ce.claim_id = target.id
         limit 30
      ) evidence_rows
  ), '[]'::jsonb),
  'edges', coalesce((
    select jsonb_agg(row_data order by rn)
      from (
        select row_number() over (
                 order by e.edge_type::text,
                          other.text
               ) as rn,
               jsonb_build_object(
                 'direction', case when e.from_claim = target.id then 'outgoing' else 'incoming' end,
                 'edge_type', e.edge_type::text,
                 'connected_id', other.id::text,
                 'connected_text', other.text,
                 'connected_status', other.status,
                 'connected_confidence', other.confidence
               ) as row_data
          from public.claim_edges e
          join public.claims other
            on other.id = case when e.from_claim = target.id then e.to_claim else e.from_claim end
         where e.from_claim = target.id or e.to_claim = target.id
         limit 40
      ) edge_rows
  ), '[]'::jsonb)
)::text
from target;
"""
    out = proposal_review.ap.run_psql(args, sql, password).strip()
    return json.loads(out) if out else None


def _load_claim(request: web.Request, claim_id: str) -> dict[str, Any] | None:
    loader = request.app.get(KB_CLAIM_LOADER_KEY)
    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"))),
        ("Status", f"<code>{escape(str(claim.get('status') or 'unknown'))}</code>"),
        ("Type", f"<code>{escape(str(claim.get('type') or 'unknown'))}</code>"),
        ("Confidence", f"<code>{escape(str(claim.get('confidence') or 'unknown'))}</code>"),
        ("Tags", f"<code>{escape(', '.join(claim.get('tags') or []) or 'none')}</code>"),
        ("Created", f"<code>{escape(str(claim.get('created_at') or 'unknown'))}</code>"),
        ("Updated", f"<code>{escape(str(claim.get('updated_at') or 'unknown'))}</code>"),
    ]
    if claim.get("superseded_by"):
        rows.append(("Superseded by", claim_link_html(claim["superseded_by"])))
    return "\n".join(f"<tr><th>{escape(label)}</th><td>{value}</td></tr>" for label, value in rows)


def _render_evidence(evidence: list[dict[str, Any]]) -> str:
    if not evidence:
        return '<div class="empty-state">No evidence rows found for this claim.</div>'
    rows = []
    for row in evidence:
        source = row.get("url") or row.get("storage_path") or "no source pointer"
        source_html = escape(str(source))
        if row.get("url"):
            source_html = f'<a href="{escape(str(row["url"]))}">{source_html}</a>'
        rows.append(
            "<tr>"
            f"<td><code>{escape(str(row.get('role') or 'unknown'))}</code></td>"
            f"<td><code>{escape(str(row.get('weight') or ''))}</code></td>"
            f"<td><code>{escape(str(row.get('source_type') or 'unknown'))}</code></td>"
            f"<td>{source_html}</td>"
            f"<td>{escape(str(row.get('excerpt') or ''))}</td>"
            "</tr>"
        )
    return (
        "<table class=\"claim-table\"><thead><tr><th>Role</th><th>Weight</th>"
        "<th>Source type</th><th>Source</th><th>Excerpt</th></tr></thead>"
        f"<tbody>{''.join(rows)}</tbody></table>"
    )


def _render_edges(edges: list[dict[str, Any]]) -> str:
    if not edges:
        return '<div class="empty-state">No graph edges found for this claim.</div>'
    rows = []
    for row in edges:
        rows.append(
            "<tr>"
            f"<td><code>{escape(str(row.get('direction') or 'unknown'))}</code></td>"
            f"<td><code>{escape(str(row.get('edge_type') or 'unknown'))}</code></td>"
            f"<td>{short_claim_link_html(row.get('connected_id'))}</td>"
            f"<td>{escape(str(row.get('connected_text') or ''))}</td>"
            f"<td><code>{escape(str(row.get('connected_status') or ''))}</code></td>"
            "</tr>"
        )
    return (
        "<table class=\"claim-table\"><thead><tr><th>Direction</th><th>Edge</th>"
        "<th>Connected claim</th><th>Text</th><th>Status</th></tr></thead>"
        f"<tbody>{''.join(rows)}</tbody></table>"
    )


def render_kb_claim_page(data: dict[str, Any]) -> str:
    claim = data["claim"]
    generated_at = datetime.now(timezone.utc).isoformat()
    body = f"""
<div class="claim-card">
  <div class="label">Canonical claim</div>
  <p class="claim-text">{escape(str(claim.get("text") or ""))}</p>
  <table class="proposal-detail"><tbody>{_metadata_table(claim)}</tbody></table>
</div>
<div class="section">
  <div class="section-title">Evidence</div>
  {_render_evidence(data.get("evidence") or [])}
</div>
<div class="section">
  <div class="section-title">Edges</div>
  {_render_edges(data.get("edges") or [])}
</div>"""
    extra_css = """
  .claim-card { background: #161b22; border: 1px solid #30363d; border-radius: 8px; padding: 16px; }
  .claim-text { margin-top: 8px; color: #f0f6fc; font-size: 17px; line-height: 1.45; }
  .claim-link { color: #58a6ff; text-decoration: none; }
  .claim-link:hover { text-decoration: underline; }
  .claim-table td { vertical-align: top; line-height: 1.35; }
  .claim-table td:nth-child(4), .claim-table td:nth-child(5) { overflow-wrap: anywhere; }
  .proposal-detail { margin-top: 14px; }
  .proposal-detail th { width: 150px; vertical-align: top; }
  .empty-state { color: #8b949e; background: #161b22; border: 1px solid #30363d;
                 border-radius: 8px; padding: 14px; }
"""
    return render_page(
        "KB Claim",
        "Canonical claim, evidence rows, and graph edges",
        "/kb-proposals",
        body,
        extra_css=extra_css,
        timestamp=generated_at,
    )


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 _private_json_response({"error": "invalid_claim_id"}, status=400)
    try:
        data = _load_claim(request, claim_id)
    except (Exception, SystemExit) as exc:
        logger.error("KB claim API load failed (%s)", type(exc).__name__)
        return _private_json_response({"error": "kb_claim_load_failed"}, status=503)
    if not data:
        return _private_json_response({"error": "claim_not_found"}, status=404)
    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 _private_html_response("Invalid claim id", status=400)
    try:
        data = _load_claim(request, claim_id)
    except (Exception, SystemExit) as exc:
        logger.error("KB claim page load failed (%s)", type(exc).__name__)
        return _private_html_response(
            render_page(
                "KB Claim",
                "Canonical claim, evidence rows, and graph edges",
                "/kb-proposals",
                '<div class="alert-banner alert-critical">Canonical claim is temporarily unavailable.</div>',
            ),
            status=500,
        )
    if not data:
        return _private_html_response("Claim not found", status=404)
    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)