diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index 8049ca1..6c4a489 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -111,11 +111,29 @@ receipt. Every accepted change after that epoch must carry a replayable strict apply payload and row-level postflight receipt. This prevents the historical gap from growing while the old import rules are reconstructed. -The existing local ingestion canary already proves one hash-bound document -fixture can become a normalized `pending_review` proposal without touching -canonical rows. The existing full-data clone canary proves a reviewed packet +The source compiler now turns one raw artifact, its strict UTF-8 extraction, +and a reviewed extraction manifest into a deterministic, hash-bound +`pending_review` proposal bundle: + +```bash +.venv/bin/python scripts/compile_kb_source_packet.py \ + --artifact fixtures/working-leo/document-ingestion-v1.json \ + --text fixtures/working-leo/document-ingestion-v1.json \ + --manifest fixtures/working-leo/source-compiler-manifest-v1.json \ + --output /tmp/working-leo-source-packet-v1.json +``` + +The compiler verifies artifact and extraction hashes, stable source identity, +current schema taxonomies, unique logical keys, and exact claim/evidence quotes. +It reuses the existing proposal normalizer and staging preflight, but it has no +database connection and executes neither staging nor apply. Its output is the +review packet, not canonical knowledge. + +The existing full-data clone canary separately proves that a reviewed packet can create source, claim, evidence, and edge rows and affect later reasoning. -The next capability is to join those into a corpus runner and replay ledger. +The next capability is to expose this compiler through Leo's bounded source +intake command, stage its output, and join accepted packets into a corpus runner +and replay ledger. ## Definition Of Working diff --git a/docs/reports/leo-working-state-20260709/source-document-compiler-canary-20260713.md b/docs/reports/leo-working-state-20260709/source-document-compiler-canary-20260713.md new file mode 100644 index 0000000..2061998 --- /dev/null +++ b/docs/reports/leo-working-state-20260709/source-document-compiler-canary-20260713.md @@ -0,0 +1,39 @@ +# Source Document Compiler Canary + +## Outcome + +A raw artifact plus its UTF-8 extraction and manifest can now become a +deterministic, hash-bound `pending_review` proposal bundle without connecting +to Postgres. This is the first concrete source-to-database build step; it does +not stage, approve, or apply the packet. + +## Command + +```bash +.venv/bin/python scripts/compile_kb_source_packet.py \ + --artifact fixtures/working-leo/document-ingestion-v1.json \ + --text fixtures/working-leo/document-ingestion-v1.json \ + --manifest fixtures/working-leo/source-compiler-manifest-v1.json \ + --output /tmp/working-leo-source-packet-v1.json +``` + +## Receipt + +- two independent CLI runs produced byte-identical JSON; +- bundle SHA-256: `8162cbd67cc8847d803b788874e0ab5cfcac9f4f859797958807f9622f683338`; +- output status: `pending_review`; +- strict child contract: `approve_claim` v2; +- declared artifact sources: `1`; +- `database_write_performed: false`; +- `stage_preview.executed: false`; +- focused compiler tests: `21 passed`; +- compiler plus normalizer, staging, and apply contract tests: `69 passed`; +- Ruff check and format check: passed. + +## Boundary + +This proves deterministic packet construction, not autonomous ingestion in the +live Leo profile and not a canonical database update. The next runtime slice is +to expose the compiler through a bounded Leo command, retain the generated +packet, stage it for review, and prove the same packet on a disposable database +before any production apply. diff --git a/fixtures/working-leo/source-compiler-manifest-v1.json b/fixtures/working-leo/source-compiler-manifest-v1.json new file mode 100644 index 0000000..f5bf1b4 --- /dev/null +++ b/fixtures/working-leo/source-compiler-manifest-v1.json @@ -0,0 +1,35 @@ +{ + "schema": "livingip.kbSourceExtractionManifest.v1", + "artifact_sha256": "3847ba59251cbcf63d78095c9bcd0c0824ff41685f84f693f1cbc7f4a823743e", + "extracted_text_sha256": "3847ba59251cbcf63d78095c9bcd0c0824ff41685f84f693f1cbc7f4a823743e", + "extractor": { + "name": "fixture-json-direct", + "version": "1" + }, + "source": { + "identity": "document:working-leo-review-gated-composition-v1", + "source_key": "working_leo_document_ingestion_v1", + "source_type": "article", + "title": "Working Leo review-gated composition note", + "locator": "fixture://working-leo/document-ingestion-v1" + }, + "claims": [ + { + "claim_key": "staged_proposal_remains_noncanonical", + "type": "structural", + "text": "A staged knowledge proposal remains non-canonical until review and guarded apply succeed.", + "quote": "A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds.", + "confidence": 0.99, + "tags": [ + "review", + "provenance" + ], + "evidence": [ + { + "quote": "The proposal must retain the source content hash, exact evidence excerpt, and source identity so reviewers can trace every planned row back to the captured artifact.", + "role": "grounds" + } + ] + } + ] +} diff --git a/scripts/compile_kb_source_packet.py b/scripts/compile_kb_source_packet.py new file mode 100644 index 0000000..e944f27 --- /dev/null +++ b/scripts/compile_kb_source_packet.py @@ -0,0 +1,533 @@ +#!/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" +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", +} +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"} + + +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 _validate_manifest( + manifest: dict[str, Any], artifact_sha256: str, text_sha256: str, extracted_text: str +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + _require_exact_keys(manifest, TOP_LEVEL_KEYS, "manifest") + if manifest.get("schema") != MANIFEST_SCHEMA: + raise CompilerError(f"manifest.schema must equal {MANIFEST_SCHEMA}") + + 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, + } + ) + + return ( + { + "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, + }, + 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"], + "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']}" + ) + 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": " ".join(claim["text"] for claim in manifest["claims"]), + "potential_existing_claim_ids": [], + "conflicts": [], + "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 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()) diff --git a/tests/test_compile_kb_source_packet.py b/tests/test_compile_kb_source_packet.py new file mode 100644 index 0000000..3675623 --- /dev/null +++ b/tests/test_compile_kb_source_packet.py @@ -0,0 +1,328 @@ +from __future__ import annotations + +import copy +import hashlib +import json +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "scripts")) + +import compile_kb_source_packet as compiler # noqa: E402, I001 + + +ARTIFACT_BYTES = b"%PDF-1.7\nsource compiler fixture\x00\xff\n%%EOF\n" +EXTRACTED_TEXT = ( + "Source packet compiler note.\n\n" + "A staged proposal remains non-canonical until a human review and guarded apply both succeed.\n\n" + "Every proposed claim must retain an exact quote from the extracted source text." +) +CLAIM_QUOTE = "A staged proposal remains non-canonical until a human review and guarded apply both succeed." +EVIDENCE_QUOTE = "Every proposed claim must retain an exact quote from the extracted source text." + + +def _sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _manifest() -> dict: + artifact_sha256 = _sha256(ARTIFACT_BYTES) + text_bytes = EXTRACTED_TEXT.encode("utf-8") + return { + "schema": compiler.MANIFEST_SCHEMA, + "artifact_sha256": artifact_sha256, + "extracted_text_sha256": _sha256(text_bytes), + "extractor": {"name": "fixture-text-extractor", "version": "1"}, + "source": { + "identity": f"document:sha256:{artifact_sha256}", + "source_key": "source_compiler_note", + "source_type": "article", + "title": "Source packet compiler note", + "locator": f"artifact://sha256/{artifact_sha256}", + }, + "claims": [ + { + "claim_key": "review_gate_precedes_canonical_apply", + "type": "structural", + "text": "A staged proposal is not canonical knowledge until review and guarded apply succeed.", + "quote": CLAIM_QUOTE, + "confidence": 0.99, + "tags": ["review", "provenance"], + "evidence": [{"quote": EVIDENCE_QUOTE, "role": "grounds"}], + } + ], + } + + +def _write_inputs(tmp_path: Path, manifest: dict | None = None, prefix: str = "input") -> tuple[Path, Path, Path]: + tmp_path.mkdir(parents=True, exist_ok=True) + artifact = tmp_path / f"{prefix}.bin" + text = tmp_path / f"{prefix}.txt" + manifest_path = tmp_path / f"{prefix}.json" + artifact.write_bytes(ARTIFACT_BYTES) + text.write_text(EXTRACTED_TEXT, encoding="utf-8") + manifest_path.write_text(json.dumps(manifest or _manifest(), indent=2), encoding="utf-8") + return artifact, text, manifest_path + + +def _compile(tmp_path: Path, manifest: dict | None = None) -> dict: + paths = _write_inputs(tmp_path, manifest) + return compiler.compile_source_packet(*paths) + + +def test_compiles_deterministic_hash_bound_pending_review_bundle_without_db_write( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + artifact, text, manifest_path = _write_inputs(tmp_path, prefix="first") + + def unexpected_db_call(*_args, **_kwargs): + raise AssertionError("compiler attempted a database write") + + monkeypatch.setattr(compiler.stage, "stage_normalized_child", unexpected_db_call) + monkeypatch.setattr(compiler.stage.bound, "_psql_json", unexpected_db_call) + first = compiler.compile_source_packet(artifact, text, manifest_path) + + copied = tmp_path / "copied" + copied.mkdir() + copied_paths = _write_inputs(copied, prefix="different-path") + second = compiler.compile_source_packet(*copied_paths) + + assert first == second + assert first["status"] == "pending_review" + assert first["build_only"] is True + assert first["database_write_performed"] is False + assert first["canonical_apply_performed"] is False + assert first["stage_preview"]["executed"] is False + assert len(first["stage_preview"]["stage_sql_sha256"]) == 64 + + parent = first["parent_proposal"] + child = first["strict_child_proposal"] + apply_payload = child["payload"]["apply_payload"] + assert parent["status"] == child["status"] == "pending_review" + assert len(parent["payload"]["source_candidates"]) == 1 + assert parent["payload"]["source_candidates"][0]["source_key"] == "source_compiler_note" + assert child["proposal_type"] == "approve_claim" + assert apply_payload["contract_version"] == 2 + assert apply_payload["source_proposal_id"] == parent["id"] + assert len(apply_payload["claims"]) == 1 + assert len(apply_payload["sources"]) == 2 + assert len(apply_payload["evidence"]) == 2 + assert first["hashes"]["artifact_sha256"] in {row["hash"] for row in apply_payload["sources"]} + assert first["compiler_contract"]["source_from_chat_labels"] is False + + without_bundle_hash = copy.deepcopy(first) + bundle_hash = without_bundle_hash["hashes"].pop("bundle_content_sha256") + assert bundle_hash == compiler.canonical_sha256(without_bundle_hash) + + +def test_cli_emits_structured_json_to_stdout_and_optional_output(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: + artifact, text, manifest_path = _write_inputs(tmp_path) + output = tmp_path / "compiled" / "packet.json" + + status = compiler.main( + [ + "--artifact", + str(artifact), + "--text", + str(text), + "--manifest", + str(manifest_path), + "--output", + str(output), + ] + ) + + stdout = capsys.readouterr().out + assert status == 0 + assert output.read_text(encoding="utf-8") == stdout + assert json.loads(stdout)["status"] == "pending_review" + + +@pytest.mark.parametrize( + ("field", "expected"), + [ + ("artifact_sha256", "artifact SHA-256 does not match"), + ("extracted_text_sha256", "extracted text SHA-256 does not match"), + ], +) +def test_rejects_hash_mismatches(tmp_path: Path, field: str, expected: str) -> None: + manifest = _manifest() + manifest[field] = "0" * 64 + + with pytest.raises(compiler.CompilerError, match=expected): + _compile(tmp_path, manifest) + + +@pytest.mark.parametrize( + ("target", "expected"), + [ + ("claim", "claims\\[0\\]\\.quote is not an exact substring"), + ("evidence", "evidence\\[0\\]\\.quote is not an exact substring"), + ], +) +def test_rejects_hallucinated_claim_or_evidence_quote(tmp_path: Path, target: str, expected: str) -> None: + manifest = _manifest() + if target == "claim": + manifest["claims"][0]["quote"] = "This quote was invented by the extractor." + else: + manifest["claims"][0]["evidence"][0]["quote"] = "This evidence was invented by the extractor." + + with pytest.raises(compiler.CompilerError, match=expected): + _compile(tmp_path, manifest) + + +def test_rejects_duplicate_claim_and_derived_source_keys(tmp_path: Path) -> None: + duplicate_claim = _manifest() + duplicate_claim["claims"].append(copy.deepcopy(duplicate_claim["claims"][0])) + with pytest.raises(compiler.CompilerError, match="duplicate claim_key"): + _compile(tmp_path / "claim", duplicate_claim) + + duplicate_source = _manifest() + duplicate_source["source"]["source_key"] = "review_gate_precedes_canonical_apply__proposal_body" + with pytest.raises(compiler.CompilerError, match="duplicate source_key after normalization"): + _compile(tmp_path / "source", duplicate_source) + + +@pytest.mark.parametrize( + ("target", "expected"), + [ + ("claim", "claims\\[0\\]\\.type must be one of"), + ("source", "source\\.source_type must be one of"), + ("evidence", "evidence\\[0\\]\\.role must be one of"), + ], +) +def test_rejects_types_outside_current_taxonomies(tmp_path: Path, target: str, expected: str) -> None: + manifest = _manifest() + if target == "claim": + manifest["claims"][0]["type"] = "fact" + elif target == "source": + manifest["source"]["source_type"] = "document" + else: + manifest["claims"][0]["evidence"][0]["role"] = "proves" + + with pytest.raises(compiler.CompilerError, match=expected): + _compile(tmp_path, manifest) + + +@pytest.mark.parametrize("field", ["identity", "locator"]) +def test_rejects_missing_or_unstable_source_provenance(tmp_path: Path, field: str) -> None: + manifest = _manifest() + manifest["source"][field] = "message from Alice" + + with pytest.raises(compiler.CompilerError, match="stable URI-like reference"): + _compile(tmp_path, manifest) + + +@pytest.mark.parametrize( + ("reference", "expected"), + [ + ("javascript:alert(1)", "unsafe URI scheme"), + ("https:/missing-host", "must include a host"), + ("https://user:password@example.com/source", "must not embed credentials"), + ("telegram:Alice", "stable chat or message identifier"), + ], +) +def test_rejects_unsafe_or_non_specific_source_references(tmp_path: Path, reference: str, expected: str) -> None: + manifest = _manifest() + manifest["source"]["locator"] = reference + + with pytest.raises(compiler.CompilerError, match=expected): + _compile(tmp_path, manifest) + + +def test_rejects_tags_with_ambiguous_whitespace(tmp_path: Path) -> None: + manifest = _manifest() + manifest["claims"][0]["tags"] = [" review"] + + with pytest.raises(compiler.CompilerError, match="tags must be a list of non-empty strings"): + _compile(tmp_path, manifest) + + +def test_rejects_path_gaps_invalid_utf8_and_output_overwrite(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: + artifact, text, manifest_path = _write_inputs(tmp_path) + with pytest.raises(compiler.CompilerError, match="artifact path is not a regular file"): + compiler.compile_source_packet(tmp_path / "missing.bin", text, manifest_path) + + text.write_bytes(b"\xff\xfe") + with pytest.raises(compiler.CompilerError, match="valid UTF-8"): + compiler.compile_source_packet(artifact, text, manifest_path) + + status = compiler.main( + [ + "--artifact", + str(artifact), + "--text", + str(text), + "--manifest", + str(manifest_path), + "--output", + str(artifact), + ] + ) + rejection = json.loads(capsys.readouterr().out) + assert status == 2 + assert rejection["status"] == "rejected" + assert rejection["database_write_performed"] is False + assert "must not overwrite" in rejection["error"] + + +def test_cli_reports_output_write_failure_as_structured_rejection( + tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + artifact, text, manifest_path = _write_inputs(tmp_path) + output = tmp_path / "packet.json" + original_write_text = Path.write_text + + def guarded_write_text(path: Path, data: str, *args, **kwargs): + if path == output: + raise PermissionError("fixture output denied") + return original_write_text(path, data, *args, **kwargs) + + monkeypatch.setattr(Path, "write_text", guarded_write_text) + status = compiler.main( + [ + "--artifact", + str(artifact), + "--text", + str(text), + "--manifest", + str(manifest_path), + "--output", + str(output), + ] + ) + + rejection = json.loads(capsys.readouterr().out) + assert status == 2 + assert rejection["status"] == "rejected" + assert rejection["database_write_performed"] is False + assert "could not write output" in rejection["error"] + + +def test_rejects_attempt_to_invent_source_from_chat_label(tmp_path: Path) -> None: + manifest = _manifest() + manifest["claims"][0]["evidence"][0]["source_label"] = "Alice in Telegram" + + with pytest.raises(compiler.CompilerError, match="unsupported keys: source_label"): + _compile(tmp_path, manifest) + + +def test_rejects_quotes_that_cannot_fit_the_reviewable_canonical_excerpt(tmp_path: Path) -> None: + long_quote = "x" * 8100 + text = f"prefix {long_quote} suffix" + artifact = tmp_path / "long.bin" + text_path = tmp_path / "long.txt" + manifest_path = tmp_path / "long.json" + artifact.write_bytes(ARTIFACT_BYTES) + text_path.write_text(text, encoding="utf-8") + manifest = _manifest() + manifest["extracted_text_sha256"] = _sha256(text.encode("utf-8")) + manifest["claims"][0]["quote"] = long_quote + manifest["claims"][0]["evidence"] = [{"quote": long_quote, "role": "grounds"}] + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + + with pytest.raises(compiler.CompilerError, match=r"truncated|excerpt capacity"): + compiler.compile_source_packet(artifact, text_path, manifest_path)