#!/usr/bin/env python3 """Compile one hash-bound source document into a pending-review KB proposal. The command is deliberately build-only. It validates a raw artifact, its UTF-8 text extraction, and an extraction manifest; then it reuses Leo's rich proposal normalizer and strict staging preflight to produce a schema-valid ``approve_claim`` child. It never connects to Postgres and never executes the generated staging SQL. """ from __future__ import annotations import argparse import hashlib import json import re import sys import uuid from pathlib import Path from typing import Any from urllib.parse import urlsplit HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import apply_proposal as ap # noqa: E402 import stage_normalized_proposal as stage # noqa: E402 MANIFEST_SCHEMA = "livingip.kbSourceExtractionManifest.v1" MANIFEST_SCHEMA_V2 = "livingip.kbSourceExtractionManifest.v2" MANIFEST_SCHEMAS = {MANIFEST_SCHEMA, MANIFEST_SCHEMA_V2} BUNDLE_SCHEMA = "livingip.kbSourceProposalBundle.v1" SHA256_RE = re.compile(r"^[0-9a-f]{64}$") KEY_RE = re.compile(r"^[a-z0-9][a-z0-9._:-]{0,127}$") URI_SCHEME_RE = re.compile(r"^[a-z][a-z0-9+.-]*:", re.IGNORECASE) UNSAFE_URI_SCHEMES = {"data", "javascript", "vbscript"} TOP_LEVEL_KEYS = { "schema", "artifact_sha256", "extracted_text_sha256", "extractor", "source", "claims", } TOP_LEVEL_KEYS_V2 = TOP_LEVEL_KEYS | {"dedupe"} EXTRACTOR_KEYS = {"name", "version"} SOURCE_KEYS = {"identity", "source_key", "source_type", "title", "locator"} CLAIM_KEYS = {"claim_key", "type", "text", "quote", "confidence", "tags", "evidence"} EVIDENCE_KEYS = {"quote", "role"} DEDUPE_KEYS = {"database_search_query", "candidate_claims", "duplicates"} DEDUPE_CANDIDATE_KEYS = {"id", "text", "score", "evidence_count", "edge_count"} DEDUPE_DUPLICATE_KEYS = {"claim_id", "source_quote", "reason"} class CompilerError(RuntimeError): """Raised when an input cannot safely become a reviewable proposal.""" def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def canonical_json_bytes(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") def canonical_sha256(value: Any) -> str: return sha256_bytes(canonical_json_bytes(value)) def stable_uuid(label: str) -> str: return str(uuid.uuid5(uuid.NAMESPACE_URL, f"livingip:kb-source-compiler:v1:{label}")) def _require_object(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict): raise CompilerError(f"{label} must be an object") return value def _require_exact_keys(value: dict[str, Any], allowed: set[str], label: str) -> None: unknown = sorted(set(value) - allowed) if unknown: raise CompilerError(f"{label} contains unsupported keys: {', '.join(unknown)}") def _require_nonempty_string(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise CompilerError(f"{label} must be a non-empty string") if value != value.strip(): raise CompilerError(f"{label} must not have leading or trailing whitespace") return value def _require_sha256(value: Any, label: str) -> str: text = _require_nonempty_string(value, label) if not SHA256_RE.fullmatch(text): raise CompilerError(f"{label} must be a lowercase SHA-256 digest") return text def _require_key(value: Any, label: str) -> str: text = _require_nonempty_string(value, label) if not KEY_RE.fullmatch(text): raise CompilerError(f"{label} must match {KEY_RE.pattern}") return text def _require_stable_reference(value: Any, label: str) -> str: text = _require_nonempty_string(value, label) if any(character.isspace() for character in text) or not URI_SCHEME_RE.match(text): raise CompilerError(f"{label} must be a stable URI-like reference with a scheme") parsed = urlsplit(text) if not parsed.scheme or text.endswith(":"): raise CompilerError(f"{label} must identify a specific source, not only a scheme or chat label") scheme = parsed.scheme.lower() if scheme in UNSAFE_URI_SCHEMES: raise CompilerError(f"{label} uses an unsafe URI scheme: {scheme}") if scheme in {"http", "https"}: if not parsed.netloc: raise CompilerError(f"{label} HTTP reference must include a host") if parsed.username is not None or parsed.password is not None: raise CompilerError(f"{label} must not embed credentials") if scheme in {"telegram", "tg"} and not re.search(r"\d", f"{parsed.netloc}{parsed.path}"): raise CompilerError(f"{label} Telegram reference must include a stable chat or message identifier") return text def _read_required_file(path: Path, label: str) -> bytes: try: if not path.is_file(): raise CompilerError(f"{label} path is not a regular file: {path}") value = path.read_bytes() except OSError as exc: raise CompilerError(f"could not read {label}: {exc}") from exc if not value: raise CompilerError(f"{label} must not be empty") return value def _quote_binding(text: str, quote: Any, label: str) -> dict[str, Any]: exact = _require_nonempty_string(quote, label) start = text.find(exact) if start < 0: raise CompilerError(f"{label} is not an exact substring of the extracted text") return { "quote": exact, "quote_sha256": sha256_bytes(exact.encode("utf-8")), "first_start": start, "first_end": start + len(exact), "occurrence_count": text.count(exact), } def _require_nonnegative_integer(value: Any, label: str) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise CompilerError(f"{label} must be a non-negative integer") return value def _validate_dedupe(value: Any, extracted_text: str) -> dict[str, Any]: dedupe = _require_object(value, "manifest.dedupe") _require_exact_keys(dedupe, DEDUPE_KEYS, "manifest.dedupe") query = _require_nonempty_string(dedupe.get("database_search_query"), "manifest.dedupe.database_search_query") raw_candidates = dedupe.get("candidate_claims") if not isinstance(raw_candidates, list) or len(raw_candidates) > 20: raise CompilerError("manifest.dedupe.candidate_claims must be a list with at most 20 rows") candidates: list[dict[str, Any]] = [] candidate_ids: set[str] = set() for index, raw_candidate in enumerate(raw_candidates): label = f"manifest.dedupe.candidate_claims[{index}]" candidate = _require_object(raw_candidate, label) _require_exact_keys(candidate, DEDUPE_CANDIDATE_KEYS, label) try: claim_id = str(uuid.UUID(_require_nonempty_string(candidate.get("id"), f"{label}.id"))) except ValueError as exc: raise CompilerError(f"{label}.id must be a full UUID") from exc if claim_id in candidate_ids: raise CompilerError(f"{label}.id duplicates an earlier canonical candidate") candidate_ids.add(claim_id) candidates.append( { "id": claim_id, "text": _require_nonempty_string(candidate.get("text"), f"{label}.text"), "score": _require_nonnegative_integer(candidate.get("score"), f"{label}.score"), "evidence_count": _require_nonnegative_integer( candidate.get("evidence_count"), f"{label}.evidence_count" ), "edge_count": _require_nonnegative_integer(candidate.get("edge_count"), f"{label}.edge_count"), } ) raw_duplicates = dedupe.get("duplicates") if not isinstance(raw_duplicates, list) or len(raw_duplicates) > len(candidates): raise CompilerError("manifest.dedupe.duplicates must not exceed the retrieved candidate count") duplicates: list[dict[str, Any]] = [] duplicate_ids: set[str] = set() for index, raw_duplicate in enumerate(raw_duplicates): label = f"manifest.dedupe.duplicates[{index}]" duplicate = _require_object(raw_duplicate, label) _require_exact_keys(duplicate, DEDUPE_DUPLICATE_KEYS, label) try: claim_id = str(uuid.UUID(_require_nonempty_string(duplicate.get("claim_id"), f"{label}.claim_id"))) except ValueError as exc: raise CompilerError(f"{label}.claim_id must be a full UUID") from exc if claim_id not in candidate_ids: raise CompilerError(f"{label}.claim_id was not present in the canonical retrieval candidates") if claim_id in duplicate_ids: raise CompilerError(f"{label}.claim_id duplicates an earlier duplicate judgment") duplicate_ids.add(claim_id) source_quote = _quote_binding(extracted_text, duplicate.get("source_quote"), f"{label}.source_quote") duplicates.append( { "claim_id": claim_id, "source_quote": source_quote["quote"], "reason": _require_nonempty_string(duplicate.get("reason"), f"{label}.reason"), } ) return { "database_search_query": query, "candidate_claims": candidates, "duplicates": duplicates, } def _validate_manifest( manifest: dict[str, Any], artifact_sha256: str, text_sha256: str, extracted_text: str ) -> tuple[dict[str, Any], list[dict[str, Any]]]: manifest_schema = manifest.get("schema") if manifest_schema not in MANIFEST_SCHEMAS: raise CompilerError(f"manifest.schema must be one of {sorted(MANIFEST_SCHEMAS)}") _require_exact_keys( manifest, TOP_LEVEL_KEYS_V2 if manifest_schema == MANIFEST_SCHEMA_V2 else TOP_LEVEL_KEYS, "manifest", ) declared_artifact_sha256 = _require_sha256(manifest.get("artifact_sha256"), "manifest.artifact_sha256") declared_text_sha256 = _require_sha256(manifest.get("extracted_text_sha256"), "manifest.extracted_text_sha256") if declared_artifact_sha256 != artifact_sha256: raise CompilerError("artifact SHA-256 does not match manifest.artifact_sha256") if declared_text_sha256 != text_sha256: raise CompilerError("extracted text SHA-256 does not match manifest.extracted_text_sha256") extractor = _require_object(manifest.get("extractor"), "manifest.extractor") _require_exact_keys(extractor, EXTRACTOR_KEYS, "manifest.extractor") extractor_name = _require_nonempty_string(extractor.get("name"), "manifest.extractor.name") extractor_version = _require_nonempty_string(extractor.get("version"), "manifest.extractor.version") source = _require_object(manifest.get("source"), "manifest.source") _require_exact_keys(source, SOURCE_KEYS, "manifest.source") source_identity = _require_stable_reference(source.get("identity"), "manifest.source.identity") source_locator = _require_stable_reference(source.get("locator"), "manifest.source.locator") source_key = _require_key(source.get("source_key"), "manifest.source.source_key") source_title = _require_nonempty_string(source.get("title"), "manifest.source.title") source_type = _require_nonempty_string(source.get("source_type"), "manifest.source.source_type") if source_type not in ap.SOURCE_TYPES: raise CompilerError(f"manifest.source.source_type must be one of {sorted(ap.SOURCE_TYPES)}") claims = manifest.get("claims") if not isinstance(claims, list) or not claims: raise CompilerError("manifest.claims must be a non-empty list of objects") if len(claims) > ap.MAX_BUNDLE_ROWS["claims"]: raise CompilerError(f"manifest.claims exceeds {ap.MAX_BUNDLE_ROWS['claims']} rows") validated_claims: list[dict[str, Any]] = [] quote_bindings: list[dict[str, Any]] = [] seen_claim_keys: set[str] = set() derived_source_keys = {source_key} for claim_index, raw_claim in enumerate(claims): label = f"manifest.claims[{claim_index}]" claim = _require_object(raw_claim, label) _require_exact_keys(claim, CLAIM_KEYS, label) claim_key = _require_key(claim.get("claim_key"), f"{label}.claim_key") if claim_key in seen_claim_keys: raise CompilerError(f"duplicate claim_key: {claim_key}") seen_claim_keys.add(claim_key) body_source_key = f"{claim_key}__proposal_body" if body_source_key in derived_source_keys: raise CompilerError(f"duplicate source_key after normalization: {body_source_key}") derived_source_keys.add(body_source_key) claim_type = _require_nonempty_string(claim.get("type"), f"{label}.type") if claim_type not in ap.CLAIM_TYPES: raise CompilerError(f"{label}.type must be one of {sorted(ap.CLAIM_TYPES)}") claim_text = _require_nonempty_string(claim.get("text"), f"{label}.text") claim_quote = _quote_binding(extracted_text, claim.get("quote"), f"{label}.quote") quote_bindings.append({"claim_key": claim_key, "kind": "claim", **claim_quote}) confidence = claim.get("confidence") if confidence is not None and ( isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1 ): raise CompilerError(f"{label}.confidence must be a number from 0 to 1 or null") tags = claim.get("tags", []) if not isinstance(tags, list) or any( not isinstance(tag, str) or not tag.strip() or tag != tag.strip() for tag in tags ): raise CompilerError(f"{label}.tags must be a list of non-empty strings") if len(tags) != len(set(tags)): raise CompilerError(f"{label}.tags contains duplicates") evidence = claim.get("evidence") if not isinstance(evidence, list) or not evidence: raise CompilerError(f"{label}.evidence must be a non-empty list of exact-quote objects") validated_evidence: list[dict[str, Any]] = [] seen_evidence: set[tuple[str, str]] = set() for evidence_index, raw_evidence in enumerate(evidence): evidence_label = f"{label}.evidence[{evidence_index}]" evidence_row = _require_object(raw_evidence, evidence_label) _require_exact_keys(evidence_row, EVIDENCE_KEYS, evidence_label) role = _require_nonempty_string(evidence_row.get("role"), f"{evidence_label}.role") if role not in ap.EVIDENCE_ROLES: raise CompilerError(f"{evidence_label}.role must be one of {sorted(ap.EVIDENCE_ROLES)}") binding = _quote_binding(extracted_text, evidence_row.get("quote"), f"{evidence_label}.quote") evidence_key = (binding["quote"], role) if evidence_key in seen_evidence: raise CompilerError(f"{evidence_label} duplicates an earlier evidence quote and role") seen_evidence.add(evidence_key) quote_bindings.append( { "claim_key": claim_key, "kind": "evidence", "evidence_role": role, **binding, } ) validated_evidence.append({"quote": binding["quote"], "role": role}) validated_claims.append( { "claim_key": claim_key, "type": claim_type, "text": claim_text, "quote": claim_quote["quote"], "confidence": confidence, "tags": tags, "evidence": validated_evidence, } ) validated_manifest = { "schema": manifest_schema, "artifact_sha256": artifact_sha256, "extracted_text_sha256": text_sha256, "extractor": {"name": extractor_name, "version": extractor_version}, "source": { "identity": source_identity, "source_key": source_key, "source_type": source_type, "title": source_title, "locator": source_locator, }, "claims": validated_claims, } if manifest_schema == MANIFEST_SCHEMA_V2: validated_manifest["dedupe"] = _validate_dedupe(manifest.get("dedupe"), extracted_text) return validated_manifest, quote_bindings def _locator_fields(locator: str) -> tuple[str | None, str | None]: scheme = urlsplit(locator).scheme.lower() if scheme in {"http", "https"}: return locator, None return None, locator def build_parent_proposal( manifest: dict[str, Any], manifest_sha256: str, quote_bindings: list[dict[str, Any]] ) -> dict[str, Any]: source = manifest["source"] parent_seed = ":".join((manifest["artifact_sha256"], manifest["extracted_text_sha256"], manifest_sha256)) parent_id = stable_uuid(f"parent:{parent_seed}") url, storage_path = _locator_fields(source["locator"]) excerpts: list[dict[str, Any]] = [] seen_quotes: set[str] = set() for binding in quote_bindings: quote = binding["quote"] if quote not in seen_quotes: seen_quotes.add(quote) excerpts.append({"text": quote}) claim_candidates = [] for claim in manifest["claims"]: claim_candidates.append( { "claim_key": claim["claim_key"], "type": claim["type"], "headline": claim["text"], "text": claim["text"], "body": claim["quote"], "confidence": claim["confidence"], "tags": claim["tags"], "evidence_tier": "hash_bound_exact_quote", "needs_research": False, "evidence": [{"source_key": source["source_key"], "role": item["role"]} for item in claim["evidence"]], "edges": [], } ) provenance_summary = ( f"identity={source['identity']}; locator={source['locator']}; " f"artifact_sha256={manifest['artifact_sha256']}; " f"extracted_text_sha256={manifest['extracted_text_sha256']}; " f"extractor={manifest['extractor']['name']}@{manifest['extractor']['version']}" ) dedupe = manifest.get("dedupe") or {} return { "id": parent_id, "proposal_type": "approve_claim", "status": "pending_review", "proposed_by_handle": "leo", "proposed_by_agent_id": None, "channel": "source_document_compiler", "source_ref": ( f"source-compiler:{manifest['artifact_sha256'][:16]}:" f"{manifest['extracted_text_sha256'][:16]}:{manifest_sha256[:16]}" ), "rationale": "Compile one hash-bound source artifact into exact-quote claim candidates for review.", "payload": { "source_document": { "manifest_schema": manifest["schema"], "manifest_sha256": manifest_sha256, "artifact_sha256": manifest["artifact_sha256"], "extracted_text_sha256": manifest["extracted_text_sha256"], "source_identity": source["identity"], "source_locator": source["locator"], "extractor": manifest["extractor"], }, "claim_candidates": claim_candidates, "source_candidates": [ { "source_key": source["source_key"], "source_type": source["source_type"], "title": source["title"], "url": url, "storage_path": storage_path, "source_quality": provenance_summary, "usage_limits": "Only the exact bound quotes may ground claims; human review is required.", "key_excerpts": excerpts, "content_sha256": manifest["artifact_sha256"], } ], "dedupe_conflict_assessment": { "database_search_query": dedupe.get("database_search_query") or " ".join(claim["text"] for claim in manifest["claims"]), "canonical_candidates": dedupe.get("candidate_claims", []), "potential_existing_claim_ids": [item["claim_id"] for item in dedupe.get("duplicates", [])], "conflicts": dedupe.get("duplicates", []), "review_required": True, }, }, } def _validate_compiled_child( child: dict[str, Any], manifest: dict[str, Any], quote_bindings: list[dict[str, Any]] ) -> None: if child.get("status") != "pending_review" or child.get("proposal_type") != "approve_claim": raise CompilerError("normalization did not produce one pending_review approve_claim child") apply_payload = (child.get("payload") or {}).get("apply_payload") if not isinstance(apply_payload, dict) or apply_payload.get("contract_version") != 2: raise CompilerError("strict child has no schema-valid approve_claim v2 apply_payload") claims = apply_payload.get("claims") or [] sources = apply_payload.get("sources") or [] if len(claims) != len(manifest["claims"]): raise CompilerError("strict child claim count differs from the extraction manifest") if len({row.get("id") for row in claims}) != len(claims): raise CompilerError("strict child contains duplicate claim ids") if [row.get("text") for row in claims] != [row["text"] for row in manifest["claims"]]: raise CompilerError("strict child claim text differs from the extracted atomic propositions") if len({row.get("id") for row in sources}) != len(sources): raise CompilerError("strict child contains duplicate source ids") source = manifest["source"] artifact_sources = [ row for row in sources if row.get("hash") == manifest["artifact_sha256"] and (row.get("url") == source["locator"] or row.get("storage_path") == source["locator"]) ] if len(artifact_sources) != 1: raise CompilerError("strict child must contain exactly one hash-bound declared artifact source") try: excerpt_packet = json.loads(artifact_sources[0]["excerpt"]) except (TypeError, json.JSONDecodeError) as exc: raise CompilerError("declared source excerpt was truncated or is not reviewable JSON") from exc retained_excerpt = excerpt_packet.get("excerpt") if isinstance(excerpt_packet, dict) else None if not isinstance(retained_excerpt, str): raise CompilerError("declared source excerpt does not retain exact quote text") missing_quotes = sorted( {binding["quote"] for binding in quote_bindings if binding["quote"] not in retained_excerpt} ) if missing_quotes: raise CompilerError("one or more exact quotes exceed the current canonical source excerpt capacity") def compile_source_packet(artifact_path: Path, text_path: Path, manifest_path: Path) -> dict[str, Any]: artifact_bytes = _read_required_file(artifact_path, "artifact") text_bytes = _read_required_file(text_path, "extracted text") manifest_bytes = _read_required_file(manifest_path, "extraction manifest") try: extracted_text = text_bytes.decode("utf-8", errors="strict") except UnicodeDecodeError as exc: raise CompilerError(f"extracted text must be valid UTF-8: {exc}") from exc try: manifest_raw = json.loads(manifest_bytes.decode("utf-8", errors="strict")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CompilerError(f"extraction manifest must be valid UTF-8 JSON: {exc}") from exc manifest_object = _require_object(manifest_raw, "manifest") artifact_sha256 = sha256_bytes(artifact_bytes) text_sha256 = sha256_bytes(text_bytes) manifest, quote_bindings = _validate_manifest(manifest_object, artifact_sha256, text_sha256, extracted_text) manifest_canonical_sha256 = canonical_sha256(manifest) parent = build_parent_proposal(manifest, manifest_canonical_sha256, quote_bindings) try: child = stage.prepare_normalized_child(parent) except (ValueError, stage.bound.CheckpointError) as exc: raise CompilerError(f"strict child preflight rejected the source packet: {exc}") from exc _validate_compiled_child(child, manifest, quote_bindings) stage_sql = stage.build_stage_sql(child) stable_bundle = { "schema": BUNDLE_SCHEMA, "status": "pending_review", "build_only": True, "database_write_performed": False, "canonical_apply_performed": False, "inputs": { "artifact_bytes": len(artifact_bytes), "extracted_text_bytes": len(text_bytes), "manifest_bytes": len(manifest_bytes), }, "validated_manifest": manifest, "quote_bindings": quote_bindings, "parent_proposal": parent, "strict_child_proposal": child, "stage_preview": { "target_table": "kb_stage.kb_proposals", "executed": False, "stage_sql_sha256": sha256_bytes(stage_sql.encode("utf-8")), }, "compiler_contract": { "normalizer": "scripts/kb_proposal_normalize.py", "strict_preflight": "scripts/stage_normalized_proposal.py", "declared_artifact_source_count": 1, "source_from_chat_labels": False, }, "hashes": { "artifact_sha256": artifact_sha256, "extracted_text_sha256": text_sha256, "manifest_file_sha256": sha256_bytes(manifest_bytes), "manifest_canonical_sha256": manifest_canonical_sha256, "parent_packet_sha256": child["parent_packet_sha256"], "strict_child_payload_sha256": child["payload_sha256"], }, } stable_bundle["hashes"]["bundle_content_sha256"] = canonical_sha256(stable_bundle) return stable_bundle def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--artifact", required=True, type=Path, help="raw source artifact; any binary format") parser.add_argument("--text", required=True, type=Path, help="strict UTF-8 extraction of the artifact") parser.add_argument("--manifest", required=True, type=Path, help="extraction manifest JSON") parser.add_argument("--output", type=Path, help="optional path for the same JSON emitted to stdout") return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: if args.output: output_path = args.output.resolve() input_paths = {args.artifact.resolve(), args.text.resolve(), args.manifest.resolve()} if output_path in input_paths: raise CompilerError("--output must not overwrite an input file") bundle = compile_source_packet(args.artifact, args.text, args.manifest) except (CompilerError, OSError, ValueError) as exc: print( json.dumps( { "schema": BUNDLE_SCHEMA, "status": "rejected", "database_write_performed": False, "error": str(exc), }, sort_keys=True, ) ) return 2 rendered = json.dumps(bundle, indent=2, sort_keys=True, ensure_ascii=True) + "\n" if args.output: try: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(rendered, encoding="utf-8") except OSError as exc: print( json.dumps( { "schema": BUNDLE_SCHEMA, "status": "rejected", "database_write_performed": False, "error": f"could not write output: {exc}", }, sort_keys=True, ) ) return 2 print(rendered, end="") return 0 if __name__ == "__main__": raise SystemExit(main())