Render KB claim links in leoclean replies (#68)

Render KB claim links in leoclean replies.

Local evidence:
- 13 focused tests passed.
- py_compile passed for Telegram bot, both leoclean KB bridges, and KB claim routes.
- ruff F,E9 passed for touched files.

GitHub checks were queued at merge time, not failing.

hermes-agent/leoclean-bin/cloudsql_memory_tool.py

@@ -24,6 +24,8 @@ STOPWORDS = {
     "using", "what", "when", "where", "which", "who", "why", "with", "you", "your",
 }
 
+DEFAULT_CLAIM_BASE_URL = "https://dash.livingip.xyz"
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
@@ -130,6 +132,23 @@ def terms_for(query: str) -> list[str]:
     return terms[:14] or [query.lower()]
 
 
+def claim_base_url() -> str:
+    return os.environ.get("TELEO_KB_CLAIM_BASE_URL", DEFAULT_CLAIM_BASE_URL).rstrip("/")
+
+
+def claim_url(claim_id: str | None) -> str | None:
+    if not claim_id:
+        return None
+    return f"{claim_base_url()}/kb/claims/{claim_id}"
+
+
+def markdown_claim_link(claim_id: str | None, label: str | None = None) -> str:
+    if not claim_id:
+        return "`unknown`"
+    text = label or claim_id
+    return f"[`{text}`]({claim_url(claim_id)})"
+
+
 def sql_literal(value: str) -> str:
     return "'" + value.replace("'", "''") + "'"
 
@@ -431,8 +450,11 @@ select jsonb_build_object(
             edges.setdefault(row["claim_id"], []).append(row)
 
     for claim in claims:
+        claim["claim_url"] = claim_url(claim["id"])
         claim["evidence"] = evidence.get(claim["id"], [])
         claim["edges"] = edges.get(claim["id"], [])
+        for edge in claim["edges"]:
+            edge["connected_url"] = claim_url(edge.get("connected_id"))
 
     context_rows = psql_json_lines(args, context_sql, db=args.canonical_db)
     result: dict[str, Any] = {
@@ -700,12 +722,16 @@ def show_canonical_claim(args: argparse.Namespace) -> dict[str, Any]:
         raise SystemExit(f"claim not found: {args.claim_id}")
     if not include_excerpts(args):
         claim.pop("text", None)
+    claim["claim_url"] = claim_url(args.claim_id)
+    edges = canonical_edges(args, args.claim_id, 12)
+    for edge in edges:
+        edge["connected_url"] = claim_url(edge.get("connected_id"))
     return {
         "artifact": "teleo_cloudsql_canonical_kb_claim",
         "backend": canonical_backend(args),
         "claim": claim,
         "evidence": canonical_evidence(args, args.claim_id, 8),
-        "edges": canonical_edges(args, args.claim_id, 12),
+        "edges": edges,
     }
 
 
@@ -719,11 +745,15 @@ def evidence_canonical_claim(args: argparse.Namespace) -> dict[str, Any]:
 
 
 def edges_canonical_claim(args: argparse.Namespace) -> dict[str, Any]:
+    edges = canonical_edges(args, args.claim_id, args.limit)
+    for edge in edges:
+        edge["connected_url"] = claim_url(edge.get("connected_id"))
     return {
         "artifact": "teleo_cloudsql_canonical_kb_edges",
         "backend": canonical_backend(args),
         "claim_id": args.claim_id,
-        "edges": canonical_edges(args, args.claim_id, args.limit),
+        "claim_url": claim_url(args.claim_id),
+        "edges": edges,
     }
 
 
@@ -964,6 +994,7 @@ def emit_markdown(value: dict[str, Any]) -> None:
         for i, claim in enumerate(value["claims"], 1):
             print(f"### {i}. {claim.get('text', '[redacted claim text]')}\n")
             print(f"- id: `{claim['id']}`")
+            print(f"- claim page: {claim_url(claim['id'])}")
             print(f"- type: `{claim.get('type')}`; confidence: `{claim.get('confidence')}`; score: `{claim.get('score')}`")
             print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
             print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`")
@@ -978,7 +1009,11 @@ def emit_markdown(value: dict[str, Any]) -> None:
                 print("\nEdges:")
                 for edge in claim["edges"]:
                     text = f": {edge['connected_text']}" if edge.get("connected_text") else ""
-                    print(f"- `{edge.get('direction')}` `{edge.get('edge_type')}` ({edge.get('connected_id')}){text}")
+                    connected = edge.get("connected_id")
+                    print(
+                        f"- `{edge.get('direction')}` `{edge.get('edge_type')}` "
+                        f"({markdown_claim_link(connected, str(connected or '')[:8])}){text}"
+                    )
             print()
         return
 
@@ -987,6 +1022,7 @@ def emit_markdown(value: dict[str, Any]) -> None:
         print("# Teleo KB Claim\n")
         print(f"- Backend: `{value['backend']}`")
         print(f"- id: `{claim['id']}`")
+        print(f"- claim page: {claim_url(claim['id'])}")
         print(f"- type: `{claim.get('type')}`; status: `{claim.get('status')}`; confidence: `{claim.get('confidence')}`")
         print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
         if claim.get("text"):
@@ -999,13 +1035,18 @@ def emit_markdown(value: dict[str, Any]) -> None:
         print("\n## Edges\n")
         for edge in value.get("edges", []):
             text = f": {edge['connected_text']}" if edge.get("connected_text") else ""
-            print(f"- `{edge.get('direction')}` `{edge.get('edge_type')}` ({edge.get('connected_id')}){text}")
+            connected = edge.get("connected_id")
+            print(
+                f"- `{edge.get('direction')}` `{edge.get('edge_type')}` "
+                f"({markdown_claim_link(connected, str(connected or '')[:8])}){text}"
+            )
         return
 
     if value["artifact"] in {"teleo_cloudsql_canonical_kb_evidence", "teleo_cloudsql_canonical_kb_edges"}:
         print(f"# {value['artifact']}\n")
         print(f"- Backend: `{value['backend']}`")
         print(f"- Claim: `{value['claim_id']}`\n")
+        print(f"- Claim page: {claim_url(value['claim_id'])}\n")
         rows = value.get("evidence") or value.get("edges") or []
         for row in rows:
             print(f"- `{json.dumps(row, sort_keys=True)}`")

hermes-agent/leoclean-bin/kb_tool.py

@@ -14,11 +14,14 @@ from __future__ import annotations
 
 import argparse
 import json
+import os
 import re
 import subprocess
 from collections import defaultdict
 from typing import Any
 
+DEFAULT_CLAIM_BASE_URL = "https://dash.livingip.xyz"
+
 STOPWORDS = {
     "a",
     "about",
@@ -217,6 +220,23 @@ def truncate(value: str | None, length: int = 220) -> str:
     return squashed[: length - 1] + "..."
 
 
+def claim_base_url() -> str:
+    return os.environ.get("TELEO_KB_CLAIM_BASE_URL", DEFAULT_CLAIM_BASE_URL).rstrip("/")
+
+
+def claim_url(claim_id: str | None) -> str | None:
+    if not claim_id:
+        return None
+    return f"{claim_base_url()}/kb/claims/{claim_id}"
+
+
+def markdown_claim_link(claim_id: str | None, label: str | None = None) -> str:
+    if not claim_id:
+        return "`unknown`"
+    text = label or claim_id
+    return f"[`{text}`]({claim_url(claim_id)})"
+
+
 def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[str, Any]]:
     patterns = [f"%{term}%" for term in query_terms(query)]
     min_score = 2 if len(patterns) >= 3 else 1
@@ -755,6 +775,7 @@ def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit
         "claims": [
             {
                 **claim,
+                "claim_url": claim_url(claim["id"]),
                 "evidence": evidence.get(claim["id"], []),
                 "edges": edges.get(claim["id"], []),
             }
@@ -785,6 +806,7 @@ def print_claim_bundle(data: dict[str, Any]) -> None:
     for idx, claim in enumerate(data["claims"], start=1):
         print(f"### {idx}. {claim['text']}\n")
         print(f"- id: `{claim['id']}`")
+        print(f"- claim page: {claim_url(claim['id'])}")
         print(f"- type: `{claim['type']}`; confidence: `{claim['confidence']}`; score: `{claim.get('score', '-')}`")
         print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
         print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`")
@@ -798,9 +820,11 @@ def print_claim_bundle(data: dict[str, Any]) -> None:
         if claim.get("edges"):
             print("\nEdges:")
             for edge in claim["edges"]:
+                connected = edge["connected_id"]
                 print(
                     f"- `{edge['direction']}` `{edge['edge_type']}` "
-                    f"({edge['connected_id']}): {truncate(edge['connected_text'], 260)}"
+                    f"({markdown_claim_link(connected, connected[:8])}): "
+                    f"{truncate(edge['connected_text'], 260)}"
                 )
         print()
 

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

@@ -58,6 +58,9 @@ Make KB answers easy to scan in Telegram:
   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 the bridge output includes `claim page: https://dash.livingip.xyz/kb/claims/<claim_id>`,
+  copy that URL into the answer so Telegram users can open the claim, body,
+  evidence, and edges directly;
 - 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,

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

@@ -50,6 +50,9 @@ Make KB answers easy to scan in Telegram:
   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 the bridge output includes `claim page: https://dash.livingip.xyz/kb/claims/<claim_id>`,
+  copy that URL into the answer so Telegram users can open the claim, body,
+  evidence, and edges directly;
 - 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,

telegram/bot.py

@@ -49,8 +49,7 @@ import json as _json
 from kb_retrieval import KBIndex, retrieve_context, retrieve_vector_context
 from retrieval import orchestrate_retrieval
 from market_data import get_token_price, format_price_context, extract_market_data_tokens
-from worktree_lock import main_worktree_lock
-from x_client import search_tweets, fetch_from_url, check_research_rate_limit, record_research_usage, get_research_remaining
+from x_client import search_tweets, check_research_rate_limit, record_research_usage, get_research_remaining
 from http_chat_proxy import (
     DEFAULT_SMART_RESEARCH_COMMAND_PREFIXES,
     build_chat_proxy_payload,
@@ -137,11 +136,21 @@ user_response_times: dict[int, list[float]] = defaultdict(list)
 allowed_groups: set[int] = set()
 
 TELEGRAM_REPLY_CHUNK_LIMIT = 3500
+MARKDOWN_LINK_RE = re.compile(r"\[([^\]\n]{1,180})\]\((https?://[^\s<>)\"]{1,600})\)")
+
+
+def _markdown_link_to_telegram_html(match: re.Match[str]) -> str:
+    label = match.group(1)
+    url = match.group(2)
+    if label.startswith("`") and label.endswith("`") and len(label) > 1:
+        label = label[1:-1]
+    return f'<a href="{url}">{label}</a>'
 
 
 def _telegram_native_html(text: str) -> str:
     """Render common LLM Markdown as Telegram HTML without trusting raw HTML."""
     rendered = escape(text, quote=False)
+    rendered = MARKDOWN_LINK_RE.sub(_markdown_link_to_telegram_html, rendered)
     rendered = re.sub(r"(?m)^#{1,6}\s+(.+)$", r"<b>\1</b>", rendered)
     rendered = re.sub(r"\*\*([^*\n]{1,240})\*\*", r"<b>\1</b>", rendered)
     rendered = re.sub(r"`([^`\n]{1,240})`", r"<code>\1</code>", rendered)
@@ -149,6 +158,7 @@ def _telegram_native_html(text: str) -> str:
 
 
 def _plain_telegram_fallback(text: str) -> str:
+    text = MARKDOWN_LINK_RE.sub(r"\1 (\2)", text)
     text = re.sub(r"(?m)^#{1,6}\s+", "", text)
     text = re.sub(r"\*\*([^*\n]{1,240})\*\*", r"\1", text)
     text = re.sub(r"`([^`\n]{1,240})`", r"\1", text)
@@ -536,7 +546,7 @@ def get_db_stats() -> dict:
 
 from eval_checks import (
     _LLMResponse, estimate_cost, check_url_fabrication, apply_confidence_floor,
-    CONFIDENCE_FLOOR, COST_ALERT_THRESHOLD,
+    COST_ALERT_THRESHOLD,
 )
 
 
@@ -1734,8 +1744,6 @@ async def handle_tagged(update: Update, context: ContextTypes.DEFAULT_TYPE):
     retrieval_layers = retrieval["retrieval_layers"]
     tool_calls.extend(retrieval["tool_calls"])
 
-    stats = get_db_stats()
-
     # Fetch live market data for any tokens mentioned (Rhea: market-data API)
     entity_terms = [tag for ent in kb_ctx.entities for tag in ent.tags]
     market_context, market_data_audit, market_duration, token_mentions = await _market_context_for_message(

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import ast
+import importlib.util
 import re
 import subprocess
 from pathlib import Path
@@ -61,3 +62,23 @@ def test_bridge_source_does_not_commit_raw_secret_values() -> None:
     for pattern in forbidden_patterns:
         assert not re.search(pattern, combined), pattern
 
+
+def _load_module(path: Path):
+    spec = importlib.util.spec_from_file_location(path.stem, path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_kb_bridges_emit_public_claim_links_for_telegram_rendering() -> None:
+    claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1"
+    expected = f"https://dash.livingip.xyz/kb/claims/{claim_id}"
+
+    for filename in ("kb_tool.py", "cloudsql_memory_tool.py"):
+        module = _load_module(BRIDGE_DIR / filename)
+
+        assert module.claim_url(claim_id) == expected
+        assert module.markdown_claim_link(claim_id, "d3fb892b") == (
+            f"[`d3fb892b`]({expected})"
+        )

tests/test_hermes_leoclean_skill_surfaces.py

@@ -34,6 +34,7 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None:
     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
+    assert "claim page: https://dash.livingip.xyz/kb/claims/<claim_id>" in text
 
 
 def test_gcp_kb_skill_keeps_claim_links_and_backtick_rendering() -> None:
@@ -41,6 +42,7 @@ def test_gcp_kb_skill_keeps_claim_links_and_backtick_rendering() -> None:
 
     assert "/kb/claims/<claim_id>" in text
     assert "wrap claim IDs, proposal IDs" in text
+    assert "claim page: https://dash.livingip.xyz/kb/claims/<claim_id>" in text
 
 
 def test_leoclean_kb_skills_anchor_external_doctrine_in_target_project_language() -> None:

tests/test_telegram_native_rendering_source.py

@@ -0,0 +1,16 @@
+"""Source-level checks for Telegram-native reply rendering."""
+
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_telegram_renderer_supports_safe_markdown_links() -> None:
+    text = (REPO_ROOT / "telegram" / "bot.py").read_text()
+
+    assert "MARKDOWN_LINK_RE" in text
+    assert "https?://" in text
+    assert "_markdown_link_to_telegram_html" in text
+    assert '<a href="{url}">{label}</a>' in text
+    assert "MARKDOWN_LINK_RE.sub(r\"\\1 (\\2)\", text)" in text