265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""Read-only KB proposal review routes for Argus.
|
|
|
|
This surface is the operator-visible bridge between Leo's Telegram KB reasoning
|
|
and the narrow apply worker. It intentionally performs no approve/reject/apply
|
|
mutation; it only renders the same packets produced by
|
|
``scripts/kb_proposal_review_packet.py``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
from html import escape
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlencode
|
|
|
|
from aiohttp import web
|
|
from shared_ui import render_page
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCRIPTS_DIR = ROOT / "scripts"
|
|
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_proposals")
|
|
|
|
KB_PROPOSAL_PUBLIC_PATHS = frozenset({"/kb-proposals", "/api/kb-proposals"})
|
|
KB_PROPOSAL_LOADER_KEY = web.AppKey("kb_proposal_loader", object)
|
|
|
|
DEFAULT_LIMIT = 20
|
|
MAX_LIMIT = 100
|
|
|
|
|
|
def _query_limit(value: str | None) -> int:
|
|
if not value:
|
|
return DEFAULT_LIMIT
|
|
try:
|
|
limit = int(value)
|
|
except ValueError:
|
|
return DEFAULT_LIMIT
|
|
return max(1, min(limit, MAX_LIMIT))
|
|
|
|
|
|
def _query_filters(request: web.Request) -> dict[str, Any]:
|
|
status = request.query.get("status", "approved").strip() or "approved"
|
|
proposal_id = (request.query.get("proposal_id") or "").strip()
|
|
if proposal_id:
|
|
status = ""
|
|
return {
|
|
"status": status,
|
|
"proposal_id": proposal_id,
|
|
"limit": _query_limit(request.query.get("limit")),
|
|
}
|
|
|
|
|
|
def _db_args(filters: dict[str, Any]) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
proposal_id=filters["proposal_id"] or None,
|
|
status=filters["status"] or None,
|
|
limit=filters["limit"],
|
|
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_PROPOSAL_REVIEW_CONTAINER", proposal_review.ap.DEFAULT_CONTAINER),
|
|
db=os.environ.get("KB_PROPOSAL_REVIEW_DB", proposal_review.ap.DEFAULT_DB),
|
|
host=os.environ.get("KB_PROPOSAL_REVIEW_HOST", proposal_review.ap.DEFAULT_HOST),
|
|
role=os.environ.get("KB_PROPOSAL_REVIEW_ROLE", proposal_review.ap.DEFAULT_ROLE),
|
|
)
|
|
|
|
|
|
def _load_packets(request: web.Request, filters: dict[str, Any]) -> list[dict[str, Any]]:
|
|
loader = request.app.get(KB_PROPOSAL_LOADER_KEY)
|
|
proposals = loader(_db_args(filters)) if loader else proposal_review.load_from_db(_db_args(filters))
|
|
return [proposal_review.classify_proposal(proposal) for proposal in proposals]
|
|
|
|
|
|
def _counts(packets: list[dict[str, Any]], key: str) -> dict[str, int]:
|
|
return dict(Counter(str(packet.get(key) or "unknown") for packet in packets))
|
|
|
|
|
|
def build_kb_proposal_response(packets: list[dict[str, Any]], filters: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"filters": filters,
|
|
"total": len(packets),
|
|
"status_counts": _counts(packets, "status"),
|
|
"review_state_counts": _counts(packets, "review_state"),
|
|
"worker_applyable_count": sum(1 for packet in packets if packet.get("worker_applyable")),
|
|
"packets": packets,
|
|
"read_only": True,
|
|
}
|
|
|
|
|
|
def _badge_class(value: str) -> str:
|
|
if value in {"applied", "approved_applyable"}:
|
|
return "badge-green"
|
|
if value in {"approved", "approved_needs_apply_payload", "needs_human_review", "pending_review"}:
|
|
return "badge-yellow"
|
|
if value in {"unsupported_by_apply_worker", "not_ready"}:
|
|
return "badge-red"
|
|
return "badge-blue"
|
|
|
|
|
|
def _badge(value: Any) -> str:
|
|
text = escape(str(value or "unknown"))
|
|
return f'<span class="badge {_badge_class(str(value or ""))}">{text}</span>'
|
|
|
|
|
|
def _join_values(values: list[Any]) -> str:
|
|
if not values:
|
|
return '<span class="muted">none</span>'
|
|
return ", ".join(escape(str(value)) for value in values)
|
|
|
|
|
|
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>'
|
|
|
|
|
|
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_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"))
|
|
channel = escape(str(auth.get("channel") or "unknown"))
|
|
reviewed_by = escape(str(auth.get("reviewed_by_handle") or "unreviewed"))
|
|
|
|
rows = [
|
|
("Status", _badge(packet.get("status"))),
|
|
("Review state", _badge(packet.get("review_state"))),
|
|
("Worker applyable", "yes" if packet.get("worker_applyable") else "no"),
|
|
("Has strict apply payload", "yes" if packet.get("has_apply_payload") else "no"),
|
|
("Proposal kind", proposal_kind),
|
|
("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))),
|
|
("Missing contract", _join_values(applyability.get("missing_contract") or [])),
|
|
]
|
|
details = "\n".join(
|
|
f"<tr><th>{escape(label)}</th><td>{value}</td></tr>" for label, value in rows
|
|
)
|
|
return f"""
|
|
<article class="proposal-card">
|
|
<div class="proposal-card-header">
|
|
<h2>{proposal_id}</h2>
|
|
</div>
|
|
<table class="proposal-detail">
|
|
<tbody>{details}</tbody>
|
|
</table>
|
|
<div class="next-action">
|
|
<div class="label">Next admin action</div>
|
|
<p>{next_action}</p>
|
|
</div>
|
|
</article>"""
|
|
|
|
|
|
def render_kb_proposals_page(data: dict[str, Any]) -> str:
|
|
filters = data["filters"]
|
|
packets = data["packets"]
|
|
generated_at = escape(str(data["generated_at"]))
|
|
cards = "\n".join(_packet_card(packet) for packet in packets)
|
|
if not cards:
|
|
cards = '<div class="alert-banner alert-info">No proposals matched this filter.</div>'
|
|
|
|
filter_html = " ".join(
|
|
[
|
|
_status_link("Approved", "approved", filters),
|
|
_status_link("Pending review", "pending_review", filters),
|
|
_status_link("Applied", "applied", filters),
|
|
]
|
|
)
|
|
body = f"""
|
|
<div class="alert-banner alert-info">
|
|
Read-only review surface. It shows proposal intent, applyability, missing contracts,
|
|
and the next admin action; it does not approve, reject, or apply proposals.
|
|
</div>
|
|
<div class="grid">
|
|
<div class="card"><div class="label">Matching proposals</div><div class="value">{data["total"]}</div></div>
|
|
<div class="card"><div class="label">Worker applyable</div><div class="value">{data["worker_applyable_count"]}</div></div>
|
|
<div class="card"><div class="label">Current status filter</div><div class="value small-value">{escape(filters.get("status") or "by id")}</div></div>
|
|
</div>
|
|
<div class="section">
|
|
<div class="section-title">Filters</div>
|
|
<div class="filters">{filter_html}</div>
|
|
</div>
|
|
<div class="section">
|
|
<div class="section-title">Proposal Review Packets</div>
|
|
<div class="proposal-list">{cards}</div>
|
|
</div>"""
|
|
extra_css = """
|
|
.small-value { font-size: 16px !important; overflow-wrap: anywhere; }
|
|
.filters { display: flex; flex-wrap: wrap; gap: 8px; }
|
|
.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; }
|
|
.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; }
|
|
.proposal-card h2 { color: #c9d1d9; font-size: 15px; overflow-wrap: anywhere; }
|
|
.proposal-detail { margin-top: 12px; }
|
|
.proposal-detail th { width: 210px; vertical-align: top; }
|
|
.proposal-detail td { overflow-wrap: anywhere; }
|
|
.next-action { margin-top: 14px; padding: 12px; border: 1px solid #30363d; border-radius: 6px; background: #0d1117; }
|
|
.next-action .label { color: #8b949e; font-size: 11px; text-transform: uppercase; margin-bottom: 6px; }
|
|
.next-action p { color: #c9d1d9; line-height: 1.4; }
|
|
.muted { color: #8b949e; }
|
|
"""
|
|
return render_page(
|
|
"KB Proposals",
|
|
"Review Leo KB proposal packets before any apply step",
|
|
"/kb-proposals",
|
|
body,
|
|
extra_css=extra_css,
|
|
timestamp=generated_at,
|
|
)
|
|
|
|
|
|
async def handle_api_kb_proposals(request: web.Request) -> web.Response:
|
|
filters = _query_filters(request)
|
|
try:
|
|
packets = _load_packets(request, filters)
|
|
except BaseException as exc:
|
|
logger.exception("KB proposal packet load failed")
|
|
return web.json_response({"error": "kb_proposal_load_failed", "detail": str(exc)}, status=500)
|
|
return web.json_response(build_kb_proposal_response(packets, filters))
|
|
|
|
|
|
async def handle_kb_proposals_page(request: web.Request) -> web.Response:
|
|
filters = _query_filters(request)
|
|
try:
|
|
packets = _load_packets(request, filters)
|
|
except BaseException as exc:
|
|
logger.exception("KB proposal packet page failed")
|
|
return web.Response(
|
|
text=render_page(
|
|
"KB Proposals",
|
|
"Review Leo KB proposal packets before any apply step",
|
|
"/kb-proposals",
|
|
f'<div class="alert-banner alert-critical">Failed to load KB proposals: {escape(str(exc))}</div>',
|
|
),
|
|
content_type="text/html",
|
|
status=500,
|
|
)
|
|
data = build_kb_proposal_response(packets, filters)
|
|
return web.Response(text=render_kb_proposals_page(data), content_type="text/html")
|
|
|
|
|
|
def register_kb_proposal_routes(app: web.Application) -> None:
|
|
app.router.add_get("/api/kb-proposals", handle_api_kb_proposals)
|
|
app.router.add_get("/kb-proposals", handle_kb_proposals_page)
|