265 lines
10 KiB
Python
Executable file
265 lines
10 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Render KB proposals as admin-review packets.
|
|
|
|
This is the read-only surface between Leo's Telegram reasoning and the apply
|
|
worker. It makes the distinction explicit:
|
|
|
|
* ``approved`` means a reviewer accepted the proposal intent.
|
|
* ``applyable`` means the proposal carries a strict ``apply_payload`` contract
|
|
that the narrow ``kb_apply`` worker can execute.
|
|
* ``applied`` means canonical ``public.*`` rows have changed and the proposal
|
|
ledger was stamped.
|
|
|
|
The script performs no writes. It can read live proposals through the same
|
|
restricted secret path used by the apply worker, or classify proposal JSON from
|
|
a file for tests and UI development.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(HERE))
|
|
|
|
import apply_proposal as ap # noqa: E402
|
|
import apply_worker as aw # noqa: E402
|
|
|
|
|
|
def _payload_list(payload: dict[str, Any], key: str) -> list[Any]:
|
|
value = payload.get(key)
|
|
return value if isinstance(value, list) else []
|
|
|
|
|
|
def _payload_dict(payload: dict[str, Any], key: str) -> dict[str, Any]:
|
|
value = payload.get(key)
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def classify_proposal(proposal: dict[str, Any]) -> dict[str, Any]:
|
|
payload = proposal.get("payload") or {}
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
|
|
status = proposal.get("status")
|
|
proposal_type = proposal.get("proposal_type")
|
|
has_apply_payload = "apply_payload" in payload
|
|
worker_supported_type = proposal_type in aw.WORKER_TYPES
|
|
worker_applyable = (
|
|
status == "approved" and worker_supported_type and has_apply_payload
|
|
)
|
|
|
|
if status == "applied":
|
|
review_state = "applied"
|
|
elif status == "pending_review":
|
|
review_state = "needs_human_review"
|
|
elif status == "approved" and worker_applyable:
|
|
review_state = "approved_applyable"
|
|
elif status == "approved" and not has_apply_payload:
|
|
review_state = "approved_needs_apply_payload"
|
|
elif not worker_supported_type:
|
|
review_state = "unsupported_by_apply_worker"
|
|
else:
|
|
review_state = "not_ready"
|
|
|
|
claim_candidates = _payload_list(payload, "claim_candidates")
|
|
source_candidates = _payload_list(payload, "source_candidates")
|
|
supersession_edges = _payload_list(payload, "proposed_supersession_edges")
|
|
concept_map = _payload_dict(payload, "proposed_concept_map_dependency")
|
|
governance_standard = _payload_dict(payload, "proposed_governance_standard")
|
|
approval_request = _payload_dict(payload, "approval_request")
|
|
review_request = _payload_dict(payload, "review_request")
|
|
|
|
missing_contract = []
|
|
if not has_apply_payload:
|
|
missing_contract.append("strict payload.apply_payload")
|
|
if claim_candidates and not has_apply_payload:
|
|
missing_contract.append("claim-row normalization")
|
|
if source_candidates and not has_apply_payload:
|
|
missing_contract.append("source/evidence normalization")
|
|
if supersession_edges and not has_apply_payload:
|
|
missing_contract.append("edge/supersession normalization")
|
|
if concept_map and not has_apply_payload:
|
|
missing_contract.append("concept-map storage decision")
|
|
|
|
if review_state == "approved_needs_apply_payload":
|
|
next_admin_action = (
|
|
"Show the approved intent, candidate rows, sources, dependencies, "
|
|
"and schema gaps; then normalize into one or more strict apply_payload "
|
|
"proposals before enabling the apply worker."
|
|
)
|
|
elif review_state == "approved_applyable":
|
|
next_admin_action = (
|
|
"Show dry-run SQL and before/after row preview; require operator apply."
|
|
)
|
|
elif review_state == "needs_human_review":
|
|
next_admin_action = (
|
|
"Reviewer should approve, reject, or request edits; no apply is possible yet."
|
|
)
|
|
elif review_state == "applied":
|
|
next_admin_action = "Show canonical readback and applied_at/applied_by stamp."
|
|
else:
|
|
next_admin_action = "Split or rewrite into a supported proposal contract."
|
|
|
|
return {
|
|
"proposal_id": proposal.get("id"),
|
|
"status": status,
|
|
"proposal_type": proposal_type,
|
|
"review_state": review_state,
|
|
"worker_supported_type": worker_supported_type,
|
|
"worker_applyable": worker_applyable,
|
|
"has_apply_payload": has_apply_payload,
|
|
"identity_and_authorization": {
|
|
"proposed_by_handle": proposal.get("proposed_by_handle"),
|
|
"reviewed_by_handle": proposal.get("reviewed_by_handle"),
|
|
"reviewed_at": proposal.get("reviewed_at"),
|
|
"review_note": proposal.get("review_note"),
|
|
"applied_by_handle": proposal.get("applied_by_handle"),
|
|
"applied_at": proposal.get("applied_at"),
|
|
"source_ref": proposal.get("source_ref") or payload.get("source_ref"),
|
|
"channel": proposal.get("channel") or payload.get("channel"),
|
|
},
|
|
"payload_summary": {
|
|
"proposal_kind": payload.get("proposal_kind"),
|
|
"old_claim_id": _payload_dict(payload, "old_claim").get("id"),
|
|
"claim_candidate_count": len(claim_candidates),
|
|
"source_candidate_count": len(source_candidates),
|
|
"supersession_edge_count": len(supersession_edges),
|
|
"has_concept_map_dependency": bool(concept_map),
|
|
"has_governance_standard": bool(governance_standard),
|
|
"schema_gap_note": payload.get("schema_gap_note"),
|
|
"approve_now": approval_request.get("approve_now", []),
|
|
"do_not_approve_yet": approval_request.get("do_not_approve_yet", []),
|
|
"requested_decision": review_request.get("requested_decision"),
|
|
"not_requested": review_request.get("not_requested"),
|
|
},
|
|
"applyability": {
|
|
"missing_contract": missing_contract,
|
|
"apply_worker_types": list(aw.WORKER_TYPES),
|
|
"strict_apply_payload_required": True,
|
|
},
|
|
"admin_panel_sections": [
|
|
"identity_and_authorization",
|
|
"payload_summary",
|
|
"candidate_rows_and_sources",
|
|
"dependencies_and_schema_gaps",
|
|
"dry_run_sql_only_after_apply_payload_exists",
|
|
],
|
|
"next_admin_action": next_admin_action,
|
|
}
|
|
|
|
|
|
def load_from_input(path: str) -> list[dict[str, Any]]:
|
|
data = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
if isinstance(data, list):
|
|
return data
|
|
if isinstance(data, dict) and isinstance(data.get("proposals"), list):
|
|
return data["proposals"]
|
|
if isinstance(data, dict):
|
|
return [data]
|
|
raise SystemExit(f"unsupported proposal JSON shape in {path}")
|
|
|
|
|
|
def _psql_args(args: argparse.Namespace) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
container=args.container, db=args.db, host=args.host, role=args.role
|
|
)
|
|
|
|
|
|
def load_from_db(args: argparse.Namespace) -> list[dict[str, Any]]:
|
|
password = ap.load_password(args.secrets_file)
|
|
where = "true"
|
|
if args.proposal_id:
|
|
where = f"id = {ap.sql_literal(args.proposal_id)}::uuid"
|
|
elif args.status:
|
|
where = f"status = {ap.sql_literal(args.status)}"
|
|
|
|
sql = f"""
|
|
select jsonb_build_object(
|
|
'id', id::text,
|
|
'proposal_type', proposal_type,
|
|
'status', status,
|
|
'proposed_by_handle', proposed_by_handle,
|
|
'channel', channel,
|
|
'source_ref', source_ref,
|
|
'rationale', rationale,
|
|
'payload', payload,
|
|
'reviewed_by_handle', reviewed_by_handle,
|
|
'reviewed_at', reviewed_at,
|
|
'review_note', review_note,
|
|
'applied_by_handle', applied_by_handle,
|
|
'applied_at', applied_at,
|
|
'created_at', created_at,
|
|
'updated_at', updated_at
|
|
)::text
|
|
from kb_stage.kb_proposals
|
|
where {where}
|
|
order by created_at desc
|
|
limit {int(args.limit)};
|
|
"""
|
|
out = ap.run_psql(_psql_args(args), sql, password)
|
|
rows = []
|
|
for line in out.splitlines():
|
|
line = line.strip()
|
|
if line:
|
|
rows.append(json.loads(line))
|
|
return rows
|
|
|
|
|
|
def render_markdown(packets: list[dict[str, Any]]) -> str:
|
|
chunks = []
|
|
for p in packets:
|
|
auth = p["identity_and_authorization"]
|
|
summary = p["payload_summary"]
|
|
applyability = p["applyability"]
|
|
chunks.append(
|
|
"\n".join(
|
|
[
|
|
f"## {p['proposal_id']}",
|
|
f"- Status: {p['status']}",
|
|
f"- Review state: {p['review_state']}",
|
|
f"- Worker applyable: {p['worker_applyable']}",
|
|
f"- Reviewed by: {auth.get('reviewed_by_handle') or 'unreviewed'}",
|
|
f"- Proposal kind: {summary.get('proposal_kind') or 'unknown'}",
|
|
f"- Claim candidates: {summary['claim_candidate_count']}",
|
|
f"- Source candidates: {summary['source_candidate_count']}",
|
|
f"- Missing contract: {', '.join(applyability['missing_contract']) or 'none'}",
|
|
f"- Next admin action: {p['next_admin_action']}",
|
|
]
|
|
)
|
|
)
|
|
return "\n\n".join(chunks) + ("\n" if chunks else "")
|
|
|
|
|
|
def parse_args(argv: list[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--input-json", help="classify proposal JSON from a file")
|
|
parser.add_argument("--proposal-id", help="load one live proposal from Postgres")
|
|
parser.add_argument("--status", default="approved", help="load live proposals by status")
|
|
parser.add_argument("--limit", type=int, default=20)
|
|
parser.add_argument("--format", choices=["json", "markdown"], default="json")
|
|
parser.add_argument("--secrets-file", default=ap.DEFAULT_SECRETS_FILE)
|
|
parser.add_argument("--container", default=ap.DEFAULT_CONTAINER)
|
|
parser.add_argument("--db", default=ap.DEFAULT_DB)
|
|
parser.add_argument("--host", default=ap.DEFAULT_HOST)
|
|
parser.add_argument("--role", default=ap.DEFAULT_ROLE)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(sys.argv[1:] if argv is None else argv)
|
|
proposals = load_from_input(args.input_json) if args.input_json else load_from_db(args)
|
|
packets = [classify_proposal(proposal) for proposal in proposals]
|
|
if args.format == "markdown":
|
|
print(render_markdown(packets), end="")
|
|
else:
|
|
print(json.dumps({"packets": packets}, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|