#!/usr/bin/env python3 """Approve a strict kb_stage proposal for guarded canonical apply. Stage 1 of the KB apply pipeline: REVIEW -> approve -> apply -> render. This tool does not create canonical rows. It only moves exactly one strict, pending, applyable proposal from ``pending_review`` to ``approved`` after a human/operator supplies a reviewer handle and verbatim review note. The apply step remains separate and is performed by ``scripts/apply_proposal.py``. The guardrails intentionally mirror the apply worker while keeping review and apply credentials separate: * proposal must be ``pending_review``; * proposal_type must be one of the worker-supported applyable types; * payload must contain a strict ``apply_payload`` contract; * the strict payload must build valid apply SQL before approval is recorded; * the reviewer connects as ``kb_review`` and can only execute the constrained ``kb_stage.approve_strict_proposal`` function; * ``kb_review`` has no direct proposal UPDATE or canonical write grant. ``--dry-run`` validates the proposal and prints the SQL without writing. """ 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 DEFAULT_REVIEW_SECRETS_FILE = "/home/teleo/.hermes/profiles/leoclean/secrets/kb-review.env" DEFAULT_REVIEW_ROLE = "kb_review" def load_review_password(secrets_file: str) -> str: path = Path(secrets_file) if not path.is_file(): raise SystemExit(f"review secrets file not found: {secrets_file}") for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") if key.strip() in ("KB_REVIEW_PASSWORD", "KB_REVIEW_DB_PASSWORD", "PGPASSWORD"): return value.strip().strip('"').strip("'") raise SystemExit( f"no KB_REVIEW_PASSWORD/KB_REVIEW_DB_PASSWORD/PGPASSWORD entry in {secrets_file}" ) def load_review_note(args: argparse.Namespace) -> str: if args.review_note and args.review_note_file: raise SystemExit("use either --review-note or --review-note-file, not both") note = Path(args.review_note_file).read_text(encoding="utf-8") if args.review_note_file else args.review_note or "" note = note.strip() if not note: raise SystemExit("review note is required and must be non-empty") return note def assert_approvable(proposal: dict[str, Any], reviewed_by: str, review_note: str) -> None: if not reviewed_by.strip(): raise SystemExit("--reviewed-by is required and must be non-empty") if not review_note.strip(): raise SystemExit("--review-note is required and must be non-empty") status = proposal.get("status") proposal_id = proposal.get("id") if status == "approved": raise SystemExit(f"proposal {proposal_id} is already approved") if status == "applied": raise SystemExit(f"proposal {proposal_id} is already applied") if status != "pending_review": raise SystemExit( f"proposal {proposal_id} has status {status!r}; only 'pending_review' proposals can be approved" ) ptype = proposal.get("proposal_type") if ptype not in ap.APPLYABLE_TYPES: raise SystemExit( f"proposal {proposal_id} has proposal_type {ptype!r}; only {ap.APPLYABLE_TYPES!r} are applyable" ) payload = proposal.get("payload") or {} if not isinstance(payload, dict) or payload.get("apply_payload") is None: raise SystemExit( f"proposal {proposal_id} has no payload.apply_payload; run/repair normalization before approval" ) # Validate the exact strict contract before recording approval. The builder # requires review metadata, so validate against a synthetic snapshot that # mirrors what the DB function will stamp. This builds SQL only. validation_snapshot = dict(proposal) validation_snapshot.update( { "status": "approved", "reviewed_by_handle": reviewed_by.strip(), "reviewed_by_agent_id": None, "reviewed_at": "2000-01-01T00:00:00+00:00", "review_note": review_note.strip(), } ) ap.build_apply_sql(validation_snapshot, ap.SERVICE_AGENT_HANDLE) def build_approve_sql( proposal: dict[str, Any], reviewed_by: str, review_note: str ) -> str: proposal_id = str(proposal["id"]) proposal_type = str(proposal["proposal_type"]) payload = proposal.get("payload") if not isinstance(payload, dict): raise ValueError("proposal payload must be an object") payload_sql = ap.sql_literal(json.dumps(payload, sort_keys=True)) + "::jsonb" return f""" begin; set local standard_conforming_strings = on; select jsonb_build_array(kb_stage.approve_strict_proposal( {ap.sql_literal(proposal_id)}::uuid, {ap.sql_literal(proposal_type)}, {payload_sql}, {ap.sql_literal(reviewed_by)}, {ap.sql_literal(review_note)} ))::text; commit; """.lstrip() def parse_reviewed_rows(output: str, proposal_id: str) -> list[dict[str, Any]]: rows = json.loads(output.strip() or "[]") if not isinstance(rows, list): raise SystemExit("approval update returned non-list JSON") if len(rows) != 1: raise SystemExit( f"approval update affected {len(rows)} rows for proposal {proposal_id}; expected exactly 1" ) return rows def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("proposal_id", help="UUID of the pending strict proposal to approve") parser.add_argument("--reviewed-by", required=True, help="reviewer handle to record") note = parser.add_mutually_exclusive_group(required=True) note.add_argument("--review-note", help="verbatim review note to record") note.add_argument("--review-note-file", help="path containing the verbatim review note") parser.add_argument("--dry-run", action="store_true", help="print the SQL, do not execute the approval") parser.add_argument("--secrets-file", default=DEFAULT_REVIEW_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=DEFAULT_REVIEW_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) review_note = load_review_note(args) password = load_review_password(args.secrets_file) proposal = ap.load_proposal(args, password) assert_approvable(proposal, args.reviewed_by, review_note) sql = build_approve_sql(proposal, args.reviewed_by.strip(), review_note) if args.dry_run: print(sql) return 0 out = ap.run_psql(args, sql, password) rows = parse_reviewed_rows(out, args.proposal_id) print(json.dumps(rows[0], sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())