#!/usr/bin/env python3 # ruff: noqa: E402,I001 """Preview the claim/source contract needed for rich approved KB proposals. The current apply pipeline intentionally supports only strict contracts that reference existing canonical IDs: add_edge, attach_evidence, and revise_strategy. Older Leo proposal packets can contain richer approved intent: new claims, new sources, old-claim supersession, evidence, and graph edges. This helper is read-only. It does not emit executable write SQL. It translates one rich proposal into a machine-readable contract preview that names: * claim rows that would need canonical IDs; * source rows that would need canonical IDs or deduplication; * edge/evidence fragments blocked until those IDs exist; * schema/role gaps that make direct apply unsafe today. """ 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 kb_proposal_review_packet as review_packet CLAIMS_COLUMNS = { "id", "type", "text", "status", "confidence", "tags", "created_by", "superseded_by", "created_at", "updated_at", } SOURCES_COLUMNS = { "id", "source_type", "url", "storage_path", "excerpt", "hash", "captured_at", "created_by", "created_at", } def _payload_dict(payload: dict[str, Any], key: str) -> dict[str, Any]: value = payload.get(key) return value if isinstance(value, dict) else {} def _payload_list(payload: dict[str, Any], key: str) -> list[Any]: value = payload.get(key) return value if isinstance(value, list) else [] def _compact(value: Any, max_chars: int = 420) -> Any: if isinstance(value, str) and len(value) > max_chars: return value[: max_chars - 3] + "..." return value def _claim_row_preview(candidate: dict[str, Any], source_ref: str | None) -> dict[str, Any]: headline = str(candidate.get("headline") or "").strip() body = str(candidate.get("body") or "").strip() claim_text = headline or body omitted_fields = [] if body and body != claim_text: omitted_fields.append("body") if candidate.get("review_note"): omitted_fields.append("review_note") if candidate.get("evidence_tier"): omitted_fields.append("evidence_tier") if candidate.get("needs_research"): omitted_fields.append("needs_research") return { "claim_key": candidate.get("claim_key"), "requires_canonical_id": True, "suggested_public_claims_fields": { "type": candidate.get("type") or candidate.get("node_kind") or "structural", "text": claim_text, "status": "open", "confidence": candidate.get("confidence"), "tags": candidate.get("tags") or [], }, "full_body": body, "source_ref": source_ref, "fields_not_represented_in_public_claims": omitted_fields, "body_storage_recommendation": ( "Store the full body/review detail as a source excerpt or add a reviewed schema field before apply." if omitted_fields else "No extra body storage required." ), } def _source_row_preview(source: dict[str, Any]) -> dict[str, Any]: excerpts = source.get("key_excerpts") if isinstance(excerpts, list): excerpt_text = "\n\n".join( item.get("text", "") if isinstance(item, dict) else str(item) for item in excerpts ).strip() else: excerpt_text = str(excerpts or "").strip() storage_path = source.get("local_text") or source.get("local_snapshot") or source.get("storage_path") url = source.get("url") source_type = source.get("source_type") or ("url" if url else "document" if storage_path else "note") omitted_fields = [] for key in ("source_quality", "usage_limits", "source_key"): if source.get(key): omitted_fields.append(key) return { "source_key": source.get("source_key"), "requires_canonical_id": True, "requires_deduplication": True, "suggested_public_sources_fields": { "source_type": source_type, "url": url, "storage_path": storage_path, "excerpt": _compact(excerpt_text), }, "fields_not_represented_in_public_sources": omitted_fields, } def _edge_preview(edge: dict[str, Any], *, default_from_key: str | None = None) -> dict[str, Any]: return { "edge_type": edge.get("edge_type"), "from_claim_key": edge.get("from") or edge.get("from_claim_key") or default_from_key, "from_claim_id": edge.get("from_claim") or edge.get("from_claim_id"), "to_claim_key": edge.get("target") if not str(edge.get("target", "")).startswith("rio:") else None, "to_claim_id": edge.get("to_claim") or edge.get("to_claim_id") or edge.get("target_claim_id"), "external_or_context_target": ( edge.get("target") if edge.get("target") and not edge.get("target_claim_id") else None ), "blocked_until": [ item for item, missing in ( ("canonical from_claim id", not (edge.get("from_claim") or edge.get("from_claim_id") or default_from_key)), ( "canonical to_claim id or supported external target mapping", not (edge.get("to_claim") or edge.get("to_claim_id") or edge.get("target_claim_id")), ), ) if missing ], } def _evidence_preview(evidence: dict[str, Any], *, default_claim_key: str | None = None) -> dict[str, Any]: return { "claim_key": evidence.get("claim_key") or default_claim_key, "claim_id": evidence.get("claim_id") or evidence.get("target_claim_id"), "source_key": evidence.get("source_key"), "source_id": evidence.get("source_id"), "excerpt_anchor": evidence.get("excerpt_anchor"), "blocked_until": [ item for item, missing in ( ("canonical claim_id", not (evidence.get("claim_id") or evidence.get("target_claim_id") or default_claim_key)), ("canonical source_id", not evidence.get("source_id")), ) if missing ], } def preview_proposal(proposal: dict[str, Any]) -> dict[str, Any]: payload = proposal.get("payload") or {} if not isinstance(payload, dict): payload = {} claim_candidates = [ _claim_row_preview(candidate, proposal.get("source_ref") or payload.get("source_ref")) for candidate in _payload_list(payload, "claim_candidates") if isinstance(candidate, dict) ] source_candidates = [ _source_row_preview(source) for source in _payload_list(payload, "source_candidates") if isinstance(source, dict) ] edge_fragments: list[dict[str, Any]] = [] for edge in _payload_list(payload, "proposed_supersession_edges"): if isinstance(edge, dict): edge_fragments.append(_edge_preview(edge)) for candidate in _payload_list(payload, "claim_candidates"): if not isinstance(candidate, dict): continue claim_key = candidate.get("claim_key") for edge in _payload_list(candidate, "edges"): if isinstance(edge, dict): edge_fragments.append(_edge_preview(edge, default_from_key=claim_key)) evidence_fragments: list[dict[str, Any]] = [] for candidate in _payload_list(payload, "claim_candidates"): if not isinstance(candidate, dict): continue claim_key = candidate.get("claim_key") for evidence in _payload_list(candidate, "evidence"): if isinstance(evidence, dict): evidence_fragments.append(_evidence_preview(evidence, default_claim_key=claim_key)) old_claim = _payload_dict(payload, "old_claim") concept_map = _payload_dict(payload, "proposed_concept_map_dependency") claim_body_gap = any(row["fields_not_represented_in_public_claims"] for row in claim_candidates) source_metadata_gap = any(row["fields_not_represented_in_public_sources"] for row in source_candidates) contract_gaps = [ { "gap": "current_apply_role_cannot_insert_claims_or_sources", "reason": ( "kb_apply currently writes only strategies, strategy_nodes, claim_evidence, claim_edges, " "and the proposal ledger." ), "required_before_apply": "A separate reviewed role/contract for public.claims and public.sources creation.", }, { "gap": "claim_candidates_need_canonical_ids", "reason": "Edges and evidence cannot become strict apply_payload rows until every new claim has an id.", "required_before_apply": "Create or deduplicate claim rows, then rewrite edges/evidence using UUIDs.", }, ] if claim_body_gap: contract_gaps.append( { "gap": "public_claims_has_no_body_or_review_note_columns", "reason": "The approved proposal includes full bodies, review notes, evidence tiers, and research caveats.", "required_before_apply": "Store those details in public.sources excerpts or add reviewed schema support.", } ) if source_metadata_gap: contract_gaps.append( { "gap": "public_sources_has_limited_metadata_columns", "reason": "The approved proposal includes source keys, quality, and usage limits not directly represented.", "required_before_apply": "Deduplicate sources and decide where provenance metadata is retained.", } ) if old_claim.get("id") and len(claim_candidates) > 1: contract_gaps.append( { "gap": "old_claim_supersession_is_many_successors", "reason": "public.claims.superseded_by stores one UUID, but this proposal splits one old claim into multiple successors.", "required_before_apply": "Use typed claim_edges for supersession components or introduce an aggregate successor node.", } ) if concept_map: contract_gaps.append( { "gap": "concept_map_dependency_has_no_first_class_storage", "reason": "The proposal explicitly approves a concept-map dependency, not just claim rows.", "required_before_apply": "Create first-class concept-map storage or keep the concept map as source-backed context.", } ) return { "proposal_id": proposal.get("id"), "status": proposal.get("status"), "proposal_type": proposal.get("proposal_type"), "reviewed_by_handle": proposal.get("reviewed_by_handle"), "review_note": proposal.get("review_note"), "read_only": True, "direct_apply_ready": False, "claim_candidates": claim_candidates, "source_candidates": source_candidates, "edge_fragments": edge_fragments, "evidence_fragments": evidence_fragments, "contract_gaps": contract_gaps, "next_action": ( "Do not apply directly. Review a claim/source creation contract in a clone, assign/deduplicate canonical IDs, " "then rerun strict normalization to produce apply_payload child proposals." ), } def load_from_input(path: str) -> list[dict[str, Any]]: return review_packet.load_from_input(path) def load_from_db(args: argparse.Namespace) -> list[dict[str, Any]]: return review_packet.load_from_db(args) def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input-json", help="preview 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("--secrets-file", default=review_packet.ap.DEFAULT_SECRETS_FILE) parser.add_argument("--container", default=review_packet.ap.DEFAULT_CONTAINER) parser.add_argument("--db", default=review_packet.ap.DEFAULT_DB) parser.add_argument("--host", default=review_packet.ap.DEFAULT_HOST) parser.add_argument("--role", default=review_packet.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) previews = [preview_proposal(proposal) for proposal in proposals] print(json.dumps({"contract_previews": previews}, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())