309 lines
12 KiB
Python
309 lines
12 KiB
Python
"""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)
|