"""Materialize scoped multi-role relation schemas from the frozen 65-item review. The schema is the relation joining its slots. No geometric faces, inferred proof links or cross-theorem object identities are constructed. """ from __future__ import annotations import argparse import hashlib import html import json import re from collections import Counter from pathlib import Path from leanbench.project import project from leanbench.schema import KIND_OBJECT, Role, Simplex, role_from_dict from leanbench.schema_registry import SchemaRegistry, slot_key from leanbench.srcsimplex import parse_source_simplex def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def source_schema(name, payload): source = parse_source_simplex(payload) symbol_slots = {} for role in source.roles: if role.kind == KIND_OBJECT: for symbol in role.symbols: symbol_slots.setdefault(symbol, []).append(role.alias) roles = {} for role in source.roles: references = sorted({alias for symbol in role.symbols for alias in symbol_slots.get(symbol, []) if alias != role.alias and len(symbol_slots[symbol]) == 1}) roles[role.alias] = Role( role.symbols[0] if len(role.symbols) == 1 else role.alias, role.kind, context=role.description, attrs={"symbols": role.symbols, "keywords": role.keywords, "introduced": role.introduced, "quantifier_status": "unreviewed", "references": references, "reference_status": "explicit_symbol_candidates" if references else "unknown", "origin": "frozen_current_chat_annotation"}) return Simplex(name, roles, attrs={"side": "source", "goal_has_exists": source.goal_has_exists, "reference_completeness": "partial"}) def lean_schema(name, code): projection = project(code) if projection.simplex is None: return None, projection originals = projection.simplex.roles by_name = {} for alias, role in originals.items(): by_name.setdefault(role.name, []).append(alias) roles = {} for alias, role in originals.items(): identifiers = set(re.findall(r"[^\W\d]\w*'?", role.type_text, flags=re.UNICODE)) references = sorted({target for symbol in identifiers for target in by_name.get(symbol, []) if len(by_name[symbol]) == 1 and target != alias}) roles[alias] = Role(role.name, role.kind, type_text=role.type_text, context=role.context, choices=role.choices, attrs={**role.attrs, "references": references, "reference_status": "syntactic_candidates", "origin": "deterministic_signature_projection"}) return Simplex(name, roles, attrs={"side": "lean", "reference_completeness": "syntactic_only"}), projection def build(items, assessment, key, registry, namespace="p0-v1"): by_id = {item["id"]: item for item in items} records = {row["blind_id"]: row for row in assessment["records"]} if (not key or len(by_id) != len(items) or len(records) != len(assessment["records"]) or len(set(key.values())) != len(key) or set(by_id) != set(key.values()) or set(records) != set(key)): raise ValueError("Items, assessments and blind key must match exactly once") index, snapshots = [], [] for blind, item_id in key.items(): record = records[blind] if record["id"] != item_id: raise ValueError("Assessment identity mismatch") source_key = f"schema:{namespace}:{item_id}:source:v1" lean_key = f"schema:{namespace}:{item_id}:lean:v1" source = source_schema(source_key, record["source"]) lean, projection = lean_schema(lean_key, by_id[item_id]["lean_code"]) registry.set_simplex(source, source_key) if lean is not None: registry.set_simplex(lean, lean_key) source_restored = registry.get_simplex(source_key) if source_restored.as_dict() != source.as_dict(): raise ValueError("Source schema persistence roundtrip failed") if lean is not None and registry.get_simplex(lean_key).as_dict() != lean.as_dict(): raise ValueError("Lean schema persistence roundtrip failed") index.append({"blind_id": blind, "source_schema": source_key, "source_status": "unreviewed_annotation_seed", "source_slots": len(source.roles), "lean_schema": lean_key if lean is not None else None, "lean_slots": len(projection.roles), "lean_status": "materialized" if lean is not None else "insufficient_projected_roles", "source_role_counts": dict(Counter(r.kind for r in source.roles.values())), "lean_role_counts": dict(Counter(r.kind for r in projection.roles)), "source_slots_with_reference_candidates": sum(bool(r.attrs["references"]) for r in source.roles.values())}) snapshots.append({"blind_id": blind, "source": source.as_dict(), "lean": lean.as_dict() if lean is not None else None, "lean_projection": projection.as_dict(), "source_slot_keys": [slot_key(source_key, alias) for alias in source.roles]}) return index, snapshots def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--items", required=True, type=Path) parser.add_argument("--assessments", required=True, type=Path) parser.add_argument("--key", required=True, type=Path) parser.add_argument("--private-out", required=True, type=Path) parser.add_argument("--index", required=True, type=Path) parser.add_argument("--summary", required=True, type=Path) parser.add_argument("--review-page", type=Path, help="Also export the owner-approved schema-only HTML review into the project") args = parser.parse_args() if "data" not in args.private_out.parts: parser.error("Private schemas must remain in the gitignored data/ directory") assessment = json.loads(args.assessments.read_text()) key = json.loads(args.key.read_text())["key"] if (assessment["provenance"]["items_sha256"] != digest(args.items) or assessment["provenance"]["key_sha256"] != digest(args.key)): parser.error("Frozen assessment hashes do not match inputs/key") items = [json.loads(line) for line in args.items.read_text().splitlines() if line.strip()] args.private_out.mkdir(parents=True, exist_ok=True) identity = {"items": digest(args.items), "assessments": digest(args.assessments), "key": digest(args.key)} manifest = args.private_out / "schema-inputs.json" if manifest.exists() and json.loads(manifest.read_text()) != identity: parser.error("Existing schema output belongs to different inputs; choose a new directory") temporary = args.private_out / "registry.sqlite.tmp" if temporary.exists(): temporary.unlink() with SchemaRegistry(temporary) as registry: index, snapshots = build(items, assessment, key, registry, namespace=f"p0-{identity['assessments'][:12]}") info = registry.info() temporary.replace(args.private_out / "registry.sqlite") manifest.write_text(json.dumps(identity, indent=2) + "\n") (args.private_out / "schemas.json").write_text(json.dumps(snapshots, indent=2, ensure_ascii=False) + "\n") sections = [] for row in snapshots: sides = [] for side in ["source", "lean"]: value = row[side] content = (json.dumps(value, indent=2, ensure_ascii=False) if value is not None else "Insufficient projected roles; no object roles were invented.") sides.append(f'

{side}

{html.escape(content)}
') sections.append(f'

{row["blind_id"]}

' + ''.join(sides) + '
') review_page = ( 'P0 semantic simplex schemas' '' '

65 semantic multi-role relation schemas

' '

Schema annotation seeds, not independently validated mathematical statements. ' 'A simplex joins the objects, assumptions and conclusion in its slots. ' 'References are partial symbol/syntax candidates. ' 'This review contains schema annotations and Lean signature text, not textbook passages ' 'or full assistant judgment explanations.

' + ''.join(sections) + '') (args.private_out / "schemas.html").write_text(review_page, encoding="utf-8") if args.review_page is not None: args.review_page.parent.mkdir(parents=True, exist_ok=True) args.review_page.write_text(review_page, encoding="utf-8") # JSON-only reload is independently checked; it does not use pickle. for row in snapshots: for side in ["source", "lean"]: if row[side] is not None: if role_from_dict(row[side]).as_dict() != row[side]: raise ValueError("JSON schema roundtrip failed") summary = {"representation": "semantic_multi_role_relation_schema_v1", "items": len(items), "source_schemas": len(index), "lean_schemas": sum(r["lean_schema"] is not None for r in index), "lean_incomplete": sum(r["lean_schema"] is None for r in index), "registry": info, "hashes": {"items": digest(args.items), "assessments": digest(args.assessments), "key": digest(args.key), "builder": digest(Path(__file__)), "schema_module": digest(Path(__file__).resolve().parents[2] / "leanbench/schema.py"), "registry_module": digest(Path(__file__).resolve().parents[2] / "leanbench/schema_registry.py")}, "limitations": ["Source annotations come from a non-independent current-chat diagnostic.", "Source quantifiers are preserved as unreviewed; existing given tags are not repaired.", "References are symbol/syntax candidates, not complete semantic dependencies.", "No cross-theorem identity or proof dependency is inferred.", "Fewer than two projected Lean roles is a projection limitation, not proof that objects are absent."]} for path, value in [(args.index, index), (args.summary, summary)]: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") print(json.dumps({k: summary[k] for k in ["items", "source_schemas", "lean_schemas", "lean_incomplete", "registry"]})) if __name__ == "__main__": main()