#!/usr/bin/env python3 import hashlib import json import os import re import unicodedata from collections import defaultdict from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MANIFEST = ROOT / "script/project-manifest.json" TOKEN = re.compile(r"\w+", re.UNICODE) NGRAM = 12 def sha256_bytes(value): return hashlib.sha256(value).hexdigest() def sha256(path): return sha256_bytes(path.read_bytes()) def normalize_tokens(text): text = unicodedata.normalize("NFC", text).casefold() return TOKEN.findall(text) def choose_extraction(entry): path = Path(entry["artifact_path"]) raw = path.read_bytes() if sha256_bytes(raw) != entry["artifact_sha256"]: raise RuntimeError(f"corpus artifact drift: {entry['project']}") mode = entry["extraction_mode"] if mode in {"utf8_text_raw_canon", "utf8_text"}: source = raw.decode("utf-8") variants = [source, source.rstrip("\n")] else: data = json.loads(raw.decode("utf-8")) if mode == "parts_narration_lf2": variants = ["\n\n".join(row["narration"] for row in data["parts"])] elif mode == "top_level_narration": variants = [data["narration"]] elif mode == "parts_paragraph_text_lf2": flat = [] per_part = [] for part in data["parts"]: values = [row if isinstance(row, str) else row["text"] for row in part["paragraphs"]] flat.extend(values) per_part.append("\n\n".join(values)) variants = ["\n\n".join(flat), "\n\n".join(per_part)] else: raise RuntimeError(f"unknown corpus extraction mode: {mode}") unique_variants = {} for text in variants: raw_variant = text.encode("utf-8") unique_variants.setdefault(raw_variant, text) matches = [ text for raw_variant, text in unique_variants.items() if sha256_bytes(raw_variant) == entry["extracted_prose_sha256"] ] if len(matches) != 1: raise RuntimeError(f"corpus extraction recipe ambiguous/drifted: {entry['project']}") return matches[0] def ngram_positions(tokens, n): rows = defaultdict(list) for index in range(len(tokens) - n + 1): rows[tuple(tokens[index:index + n])].append(index) return rows def longest_exact_run(candidate, corpus, cap=200): if not candidate or not corpus: return 0 positions = defaultdict(list) for index, token in enumerate(corpus): positions[token].append(index) best = 0 for i, token in enumerate(candidate): for j in positions.get(token, []): length = 0 while i + length < len(candidate) and j + length < len(corpus) and candidate[i + length] == corpus[j + length] and length < cap: length += 1 if length > best: best = length return best def atomic_json(path, value): path.parent.mkdir(parents=True, exist_ok=True) part = path.with_suffix(path.suffix + ".part") data = (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8") with open(part, "xb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(part, path) def fail(message): raise RuntimeError(message) def main(): manifest = json.loads(MANIFEST.read_text(encoding="utf-8")) if manifest.get("active_run") != "run-002" or manifest.get("status") != "run-002_candidate_ready_for_originality": fail("run-002 candidate is not enabled for originality") active = manifest["active_paths"] candidate_path = ROOT / active["candidate"] validation_path = ROOT / active["candidate_validation"] output_path = ROOT / active["originality_receipt"] matrix_rel = active.get("postprose_semantic_matrix") if not isinstance(matrix_rel, str) or not matrix_rel: fail("active postprose semantic matrix path missing") matrix_path = ROOT / matrix_rel inventory_path = ROOT / "work/story/run-002/corpus-inventory.json" findings_path = ROOT / "work/story/run-002/corpus-fingerprint-findings.json" if output_path.exists(): fail("originality receipt must be virgin") for path in (candidate_path, validation_path, matrix_path, inventory_path, findings_path): if not path.exists(): fail(f"originality prerequisite missing: {path}") validation = json.loads(validation_path.read_text(encoding="utf-8")) if validation.get("status") != "PASS" or validation.get("verified") is not True: fail("candidate validation not terminal PASS") if validation.get("candidate_sha256") != sha256(candidate_path): fail("candidate-validation hash drift") inventory = json.loads(inventory_path.read_text(encoding="utf-8")) findings = json.loads(findings_path.read_text(encoding="utf-8")) matrix = json.loads(matrix_path.read_text(encoding="utf-8")) if findings.get("verified") is not True or findings.get("actual_unique_fingerprinted_project_count") != 17: fail("pre-premise semantic corpus coverage not terminal 17/17") if matrix.get("candidate_sha256") != sha256(candidate_path): fail("post-prose semantic matrix candidate drift") if matrix.get("corpus_inventory_sha256") != sha256(inventory_path): fail("post-prose semantic matrix corpus drift") if matrix.get("manual_candidate_full_read_verified") is not True: fail("post-prose semantic matrix lacks manual full read") rows = matrix.get("comparisons", []) projects = {entry["project"] for entry in inventory["entries"]} if len(rows) != 17 or {row.get("project") for row in rows} != projects: fail("post-prose semantic matrix does not cover exact 17 projects") for row in rows: if row.get("verdict") != "PASS" or row.get("causal_chain_same") is not False: fail(f"semantic originality failed against {row.get('project')}") if not isinstance(row.get("candidate_scene_refs"), list) or not row["candidate_scene_refs"]: fail(f"candidate scene refs missing against {row.get('project')}") if not isinstance(row.get("corpus_scene_refs"), list) or not row["corpus_scene_refs"]: fail(f"corpus scene refs missing against {row.get('project')}") if not isinstance(row.get("decisive_differences"), list) or len(row["decisive_differences"]) < 5: fail(f"fewer than five decisive differences against {row.get('project')}") if len(row.get("shared_specific_axes", [])) >= 3 and row.get("three_axis_chain_adjudication") != "different_causal_chain": fail(f"three-axis semantic collision not cleared against {row.get('project')}") candidate_text = candidate_path.read_text(encoding="utf-8") candidate_tokens = normalize_tokens(candidate_text) candidate_ngrams = ngram_positions(candidate_tokens, NGRAM) lexical_rows = [] total_matches = 0 maximum_run = 0 for entry in inventory["entries"]: corpus_text = choose_extraction(entry) corpus_tokens = normalize_tokens(corpus_text) corpus_ngrams = set(ngram_positions(corpus_tokens, NGRAM)) matches = [gram for gram in candidate_ngrams if gram in corpus_ngrams] longest = longest_exact_run(candidate_tokens, corpus_tokens) total_matches += len(matches) maximum_run = max(maximum_run, longest) lexical_rows.append({ "project": entry["project"], "artifact_path": entry["artifact_path"], "artifact_sha256": entry["artifact_sha256"], "extracted_prose_sha256": entry["extracted_prose_sha256"], "exact_12gram_match_count": len(matches), "longest_exact_token_run": longest, "sample_matches": [" ".join(gram) for gram in matches[:5]], }) if total_matches != 0: fail(f"exact 12-gram collision count is {total_matches}") now = datetime.now(timezone.utc).isoformat() receipt = { "status": "PASS", "verified": True, "run_id": "run-002", "candidate_path": active["candidate"], "candidate_sha256": sha256(candidate_path), "candidate_words": len(candidate_text.split()), "corpus_inventory_path": "work/story/run-002/corpus-inventory.json", "corpus_inventory_sha256": sha256(inventory_path), "corpus_fingerprint_findings_sha256": sha256(findings_path), "corpus_project_count": 17, "artifact_hashes_verified": 17, "extracted_prose_hashes_verified": 17, "lexical": { "tokenization": "NFC casefold Unicode word tokens", "ngram_size": NGRAM, "total_exact_12gram_matches": total_matches, "maximum_longest_exact_token_run": maximum_run, "projects": lexical_rows, }, "semantic": { "matrix_path": matrix_rel, "matrix_sha256": sha256(matrix_path), "manual_candidate_full_read_verified": True, "comparison_count": 17, "semantic_architecture_verdict": "PASS", "all_causal_chain_same_false": True, "minimum_decisive_differences_per_project": 5, }, "semantic_architecture_verdict": "PASS", "completed_at": now, } atomic_json(output_path, receipt) if sha256(candidate_path) != receipt["candidate_sha256"]: fail("candidate drift during originality write") print(json.dumps({ "status": "PASS", "candidate_sha256": receipt["candidate_sha256"], "corpus_project_count": 17, "exact_12gram_matches": total_matches, "maximum_longest_exact_token_run": maximum_run, "semantic_architecture_verdict": "PASS", }, ensure_ascii=False)) if __name__ == "__main__": main()