Merge pull request #64 from living-ip/codex/claim-links-telegram-rendering-20260709

Add canonical KB claim links

diagnostics/app.py

@@ -22,12 +22,13 @@ from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "pipeline"))
 
 from aiohttp import web
-from review_queue_routes import register_review_queue_routes
 from daily_digest_routes import register_daily_digest_routes
-from response_audit_routes import register_response_audit_routes, RESPONSE_AUDIT_PUBLIC_PATHS
-from leaderboard_routes import register_leaderboard_routes, LEADERBOARD_PUBLIC_PATHS
+from kb_claim_routes import KB_CLAIM_PUBLIC_PATHS, KB_CLAIM_PUBLIC_PREFIXES, 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 response_audit_routes import RESPONSE_AUDIT_PUBLIC_PATHS, register_response_audit_routes
+from review_queue_routes import register_review_queue_routes
 
 logger = logging.getLogger("argus")
 
@@ -510,7 +511,15 @@ def _load_secret(path: Path) -> str | None:
 @web.middleware
 async def auth_middleware(request, handler):
     """API key check. Public paths skip auth. Protected paths require X-Api-Key header."""
-    if request.path in _PUBLIC_PATHS 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.startswith("/api/response-audit/"):
+    if (
+        request.path in _PUBLIC_PATHS
+        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.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:
@@ -2357,6 +2366,7 @@ def create_app() -> web.Application:
     register_dashboard_routes(app, lambda: _conn_from_app(app))
     register_review_queue_routes(app)
     register_daily_digest_routes(app, db_path=str(DB_PATH))
+    register_kb_claim_routes(app)
     register_kb_proposal_routes(app)
     # Portfolio
     from dashboard_portfolio import register_portfolio_routes

diagnostics/kb_claim_routes.py

@@ -0,0 +1,309 @@
+"""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.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import re
+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_PUBLIC_PATHS = frozenset()
+KB_CLAIM_PUBLIC_PREFIXES = ("/kb/claims/", "/api/kb/claims/")
+KB_CLAIM_LOADER_KEY = web.AppKey("kb_claim_loader", object)
+
+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}$"
+)
+
+
+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 _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 _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 _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 web.json_response({"error": "invalid_claim_id", "claim_id": claim_id}, status=400)
+    try:
+        data = _load_claim(request, claim_id)
+    except BaseException as exc:
+        logger.exception("KB claim API load failed")
+        return web.json_response({"error": "kb_claim_load_failed", "detail": str(exc)}, status=500)
+    if not data:
+        return web.json_response({"error": "claim_not_found", "claim_id": claim_id}, status=404)
+    return web.json_response(data)
+
+
+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)
+    try:
+        data = _load_claim(request, claim_id)
+    except BaseException as exc:
+        logger.exception("KB claim page load failed")
+        return web.Response(
+            text=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>',
+            ),
+            content_type="text/html",
+            status=500,
+        )
+    if not data:
+        return web.Response(text="Claim not found", status=404)
+    return web.Response(text=render_kb_claim_page(data), content_type="text/html")
+
+
+def register_kb_claim_routes(app: web.Application) -> None:
+    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)

diagnostics/kb_proposal_routes.py

@@ -33,6 +33,7 @@ for scripts_dir in SCRIPT_DIR_CANDIDATES:
 
 import kb_proposal_normalize as proposal_normalize  # noqa: E402
 import kb_proposal_review_packet as proposal_review  # noqa: E402
+from kb_claim_routes import claim_link_html  # noqa: E402
 
 logger = logging.getLogger("argus.kb_proposals")
 
@@ -130,6 +131,10 @@ def _join_values(values: list[Any]) -> str:
     return ", ".join(escape(str(value)) for value in values)
 
 
+def _code(value: Any) -> str:
+    return f"<code>{escape(str(value))}</code>"
+
+
 def _status_link(label: str, status: str, filters: dict[str, Any]) -> str:
     query = urlencode({"status": status, "limit": filters["limit"]})
     return f'<a class="filter-link" href="/kb-proposals?{query}">{escape(label)}</a>'
@@ -139,7 +144,8 @@ def _packet_card(packet: dict[str, Any]) -> str:
     auth = packet.get("identity_and_authorization") or {}
     summary = packet.get("payload_summary") or {}
     applyability = packet.get("applyability") or {}
-    proposal_id = escape(str(packet.get("proposal_id") or "unknown"))
+    proposal_id_raw = str(packet.get("proposal_id") or "unknown")
+    proposal_id = _code(proposal_id_raw)
     proposal_kind = escape(str(summary.get("proposal_kind") or "unknown"))
     next_action = escape(str(packet.get("next_admin_action") or "No next action recorded."))
     source_ref = escape(str(auth.get("source_ref") or "unknown"))
@@ -155,10 +161,10 @@ def _packet_card(packet: dict[str, Any]) -> str:
         ("Reviewed by", reviewed_by),
         ("Channel", channel),
         ("Source ref", source_ref),
-        ("Old claim id", escape(str(summary.get("old_claim_id") or "unknown"))),
-        ("Claim candidates", str(summary.get("claim_candidate_count", 0))),
-        ("Source candidates", str(summary.get("source_candidate_count", 0))),
-        ("Supersession edges", str(summary.get("supersession_edge_count", 0))),
+        ("Old claim id", claim_link_html(summary.get("old_claim_id") or "unknown")),
+        ("Claim candidates", _code(summary.get("claim_candidate_count", 0))),
+        ("Source candidates", _code(summary.get("source_candidate_count", 0))),
+        ("Supersession edges", _code(summary.get("supersession_edge_count", 0))),
         ("Missing contract", _join_values(applyability.get("missing_contract") or [])),
     ]
     normalization = packet.get("normalization_preview") or {}
@@ -166,8 +172,8 @@ def _packet_card(packet: dict[str, Any]) -> str:
         rows.extend(
             [
                 ("Normalization", _badge(normalization.get("normalization_state"))),
-                ("Strict child proposals", str(normalization.get("strict_child_count", 0))),
-                ("Blocked fragments", str(normalization.get("blocked_count", 0))),
+                ("Strict child proposals", _code(normalization.get("strict_child_count", 0))),
+                ("Blocked fragments", _code(normalization.get("blocked_count", 0))),
             ]
         )
     details = "\n".join(
@@ -235,6 +241,8 @@ def render_kb_proposals_page(data: dict[str, Any]) -> str:
   .filter-link { color: #58a6ff; background: #161b22; border: 1px solid #30363d;
                  border-radius: 6px; padding: 6px 10px; text-decoration: none; font-size: 13px; }
   .filter-link:hover { background: #21262d; }
+  .claim-link { color: #58a6ff; text-decoration: none; }
+  .claim-link:hover { text-decoration: underline; }
   .proposal-list { display: grid; gap: 16px; }
   .proposal-card { background: #161b22; border: 1px solid #30363d; border-radius: 8px; padding: 16px; }
   .proposal-card-header { display: flex; justify-content: space-between; gap: 12px; align-items: start; }

hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md

@@ -50,6 +50,21 @@ good default is:
 3. final answer with what is grounded, what is weak, and what evidence would
    improve it.
 
+## Telegram Rendering
+
+Make KB answers easy to scan in Telegram:
+
+- wrap claim IDs, proposal IDs, edge types, table names, statuses, counts, and
+  command names in backticks;
+- when citing a specific claim, include both the claim headline and the claim
+  ID, for example: `claim text` (`<claim_id>`);
+- when a dashboard URL is available, include the canonical claim page as
+  `https://<argus-host>/kb/claims/<claim_id>`; otherwise name the dashboard
+  path `/kb/claims/<claim_id>` so the operator can open the claim, body,
+  evidence, and edges;
+- prefer short sections such as `Claim`, `Body readback`, `Edges`,
+  `Evidence`, and `Proposal` instead of dense paragraphs.
+
 Do not browse the public web just because the KB has evidence gaps. If the user
 asks for current Twitter/X, current market activity, today's news, or another
 fresh external read, use the configured current-source path if it exists. If it

hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md

@@ -42,6 +42,21 @@ For KB questions, prefer the bridge over raw database access. A good default is:
 3. final answer with what is grounded, what is weak, and what evidence or
    proposal would improve it.
 
+## Telegram Rendering
+
+Make KB answers easy to scan in Telegram:
+
+- wrap claim IDs, proposal IDs, edge types, table names, statuses, counts, and
+  command names in backticks;
+- when citing a specific claim, include both the claim headline and the claim
+  ID, for example: `claim text` (`<claim_id>`);
+- when a dashboard URL is available, include the canonical claim page as
+  `https://<argus-host>/kb/claims/<claim_id>`; otherwise name the dashboard
+  path `/kb/claims/<claim_id>` so the operator can open the claim, body,
+  evidence, and edges;
+- prefer short sections such as `Claim`, `Body readback`, `Edges`,
+  `Evidence`, and `Proposal` instead of dense paragraphs.
+
 Use raw `docker exec ... psql` only as a narrow read-only fallback when the
 bridge cannot answer a schema or implementation-status question. If you use
 that fallback, say it was a read-only inspection. Do not present raw SQL as the

tests/test_hermes_leoclean_skill_surfaces.py

@@ -32,6 +32,15 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None:
     assert "Never end a normal Telegram answer by offering to run direct `INSERT`, `UPDATE`" in text
     assert "Next admin-panel action" in text
     assert "draft or refresh the admin review packet" in squashed
+    assert "/kb/claims/<claim_id>" in text
+    assert "wrap claim IDs, proposal IDs" in text
+
+
+def test_gcp_kb_skill_keeps_claim_links_and_backtick_rendering() -> None:
+    text = (SKILL_ROOT / "gcp" / "teleo-kb-bridge" / "SKILL.md").read_text()
+
+    assert "/kb/claims/<claim_id>" in text
+    assert "wrap claim IDs, proposal IDs" in text
 
 
 def test_vps_live_telegram_skill_uses_systemd_for_gateway_liveness() -> None:

tests/test_kb_claim_routes.py

@@ -0,0 +1,93 @@
+"""Tests for canonical KB claim routes."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("aiohttp")
+
+from aiohttp import web
+from aiohttp.test_utils import make_mocked_request
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "diagnostics"))
+
+import kb_claim_routes as routes  # noqa: E402
+
+
+CLAIM_ID = "d0000000-0000-0000-0000-000000000005"
+CONNECTED_ID = "e0000000-0000-0000-0000-000000000006"
+
+
+def _claim_data():
+    return {
+        "claim": {
+            "id": CLAIM_ID,
+            "type": "strategy_kernel",
+            "text": "Claims should be falsifiable enough to support review.",
+            "status": "open",
+            "confidence": "0.72",
+            "tags": ["claims", "review"],
+            "created_at": "2026-07-09T00:00:00Z",
+            "updated_at": "2026-07-09T00:00:00Z",
+        },
+        "evidence": [
+            {
+                "role": "grounds",
+                "weight": "0.8",
+                "source_type": "telegram",
+                "url": "https://example.test/source",
+                "storage_path": None,
+                "excerpt": "m3taversal asked Leo to split a broad claim into atomic claims.",
+            }
+        ],
+        "edges": [
+            {
+                "direction": "outgoing",
+                "edge_type": "supports",
+                "connected_id": CONNECTED_ID,
+                "connected_text": "Reviewable claims need supporting evidence and edges.",
+                "connected_status": "open",
+            }
+        ],
+    }
+
+
+def test_claim_link_html_links_valid_claim_ids() -> None:
+    html = routes.claim_link_html(CLAIM_ID)
+
+    assert f'href="/kb/claims/{CLAIM_ID}"' in html
+    assert f"<code>{CLAIM_ID}</code>" in html
+
+
+def test_render_kb_claim_page_shows_body_evidence_and_linked_edges() -> None:
+    html = routes.render_kb_claim_page(_claim_data())
+
+    assert "Claims should be falsifiable enough to support review." in html
+    assert "m3taversal asked Leo" in html
+    assert f'href="/kb/claims/{CONNECTED_ID}"' in html
+    assert "<code>supports</code>" in html
+    assert "<code>open</code>" in html
+
+
+def test_api_kb_claim_uses_loader_and_returns_claim_graph() -> None:
+    app = web.Application()
+    app[routes.KB_CLAIM_LOADER_KEY] = lambda claim_id: _claim_data() if claim_id == CLAIM_ID else None
+    req = make_mocked_request(
+        "GET",
+        f"/api/kb/claims/{CLAIM_ID}",
+        app=app,
+        match_info={"claim_id": CLAIM_ID},
+    )
+
+    response = asyncio.run(routes.handle_api_kb_claim(req))
+    data = json.loads(response.body.decode())
+
+    assert response.status == 200
+    assert data["claim"]["id"] == CLAIM_ID
+    assert data["edges"][0]["connected_id"] == CONNECTED_ID

tests/test_kb_proposal_routes.py

@@ -103,6 +103,8 @@ def test_page_renders_applyability_and_next_admin_action():
     assert "normalize into one or more strict apply_payload" in html
     assert "Normalization" in html
     assert "blocked_missing_canonical_ids" in html
+    assert 'href="/kb/claims/d0000000-0000-0000-0000-000000000005"' in html
+    assert "<code>d0000000-0000-0000-0000-000000000005</code>" in html
 
 
 def test_api_kb_proposals_supports_proposal_id_filter():