diff --git a/diagnostics/app.py b/diagnostics/app.py
index 0218cc8..47315a2 100644
--- a/diagnostics/app.py
+++ b/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
diff --git a/diagnostics/kb_claim_routes.py b/diagnostics/kb_claim_routes.py
new file mode 100644
index 0000000..268c207
--- /dev/null
+++ b/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"{escape(value or 'unknown')}"
+ text = label or value
+ return f'{escape(text)}'
+
+
+def short_claim_link_html(claim_id: Any) -> str:
+ value = str(claim_id or "")
+ if not is_claim_id(value):
+ return f"{escape(value or 'unknown')}"
+ 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"{escape(str(claim.get('status') or 'unknown'))}"),
+ ("Type", f"{escape(str(claim.get('type') or 'unknown'))}"),
+ ("Confidence", f"{escape(str(claim.get('confidence') or 'unknown'))}"),
+ ("Tags", f"{escape(', '.join(claim.get('tags') or []) or 'none')}"),
+ ("Created", f"{escape(str(claim.get('created_at') or 'unknown'))}"),
+ ("Updated", f"{escape(str(claim.get('updated_at') or 'unknown'))}"),
+ ]
+ if claim.get("superseded_by"):
+ rows.append(("Superseded by", claim_link_html(claim["superseded_by"])))
+ return "\n".join(f"| {escape(label)} | {value} |
|---|
" for label, value in rows)
+
+
+def _render_evidence(evidence: list[dict[str, Any]]) -> str:
+ if not evidence:
+ return 'No evidence rows found for this claim.
'
+ 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'{source_html}'
+ rows.append(
+ ""
+ f"{escape(str(row.get('role') or 'unknown'))} | "
+ f"{escape(str(row.get('weight') or ''))} | "
+ f"{escape(str(row.get('source_type') or 'unknown'))} | "
+ f"{source_html} | "
+ f"{escape(str(row.get('excerpt') or ''))} | "
+ "
"
+ )
+ return (
+ "| Role | Weight | "
+ "Source type | Source | Excerpt |
|---|
"
+ f"{''.join(rows)}
"
+ )
+
+
+def _render_edges(edges: list[dict[str, Any]]) -> str:
+ if not edges:
+ return 'No graph edges found for this claim.
'
+ rows = []
+ for row in edges:
+ rows.append(
+ ""
+ f"{escape(str(row.get('direction') or 'unknown'))} | "
+ f"{escape(str(row.get('edge_type') or 'unknown'))} | "
+ f"{short_claim_link_html(row.get('connected_id'))} | "
+ f"{escape(str(row.get('connected_text') or ''))} | "
+ f"{escape(str(row.get('connected_status') or ''))} | "
+ "
"
+ )
+ return (
+ "| Direction | Edge | "
+ "Connected claim | Text | Status |
|---|
"
+ f"{''.join(rows)}
"
+ )
+
+
+def render_kb_claim_page(data: dict[str, Any]) -> str:
+ claim = data["claim"]
+ generated_at = datetime.now(timezone.utc).isoformat()
+ body = f"""
+
+
Canonical claim
+
{escape(str(claim.get("text") or ""))}
+
+
+
Evidence
+ {_render_evidence(data.get("evidence") or [])}
+
+
Edges
+ {_render_edges(data.get("edges") or [])}
+
Failed to load KB claim: {escape(str(exc))}
',
+ ),
+ 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)
diff --git a/diagnostics/kb_proposal_routes.py b/diagnostics/kb_proposal_routes.py
index f209ee6..ef39dee 100644
--- a/diagnostics/kb_proposal_routes.py
+++ b/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"{escape(str(value))}"
+
+
def _status_link(label: str, status: str, filters: dict[str, Any]) -> str:
query = urlencode({"status": status, "limit": filters["limit"]})
return f'{escape(label)}'
@@ -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; }
diff --git a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md
index 1217ba4..921d1d7 100644
--- a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md
+++ b/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` (``);
+- when a dashboard URL is available, include the canonical claim page as
+ `https:///kb/claims/`; otherwise name the dashboard
+ path `/kb/claims/` 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
diff --git a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md
index 1d21307..b8188dd 100644
--- a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md
+++ b/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` (``);
+- when a dashboard URL is available, include the canonical claim page as
+ `https:///kb/claims/`; otherwise name the dashboard
+ path `/kb/claims/` 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
diff --git a/tests/test_hermes_leoclean_skill_surfaces.py b/tests/test_hermes_leoclean_skill_surfaces.py
index d99aa0c..61020b9 100644
--- a/tests/test_hermes_leoclean_skill_surfaces.py
+++ b/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/" 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/" in text
+ assert "wrap claim IDs, proposal IDs" in text
def test_vps_live_telegram_skill_uses_systemd_for_gateway_liveness() -> None:
diff --git a/tests/test_kb_claim_routes.py b/tests/test_kb_claim_routes.py
new file mode 100644
index 0000000..2fef627
--- /dev/null
+++ b/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"{CLAIM_ID}" 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 "supports" in html
+ assert "open" 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
diff --git a/tests/test_kb_proposal_routes.py b/tests/test_kb_proposal_routes.py
index ddf28f0..e5eda80 100644
--- a/tests/test_kb_proposal_routes.py
+++ b/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 "d0000000-0000-0000-0000-000000000005" in html
def test_api_kb_proposals_supports_proposal_id_filter():