194 lines
11 KiB
Python
194 lines
11 KiB
Python
"""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'<h3>{side}</h3><pre>{html.escape(content)}</pre>')
|
|
sections.append(f'<section id="{row["blind_id"]}"><h2>{row["blind_id"]}</h2>'
|
|
+ ''.join(sides) + '</section>')
|
|
review_page = (
|
|
'<!doctype html><html lang="en"><meta charset="utf-8"><title>P0 semantic simplex schemas</title>'
|
|
'<style>body{font:16px/1.5 system-ui;max-width:1050px;margin:32px auto;padding:16px}'
|
|
'pre{white-space:pre-wrap;background:#f5f5f5;padding:16px}section{border-top:1px solid #ccc}</style>'
|
|
'<h1>65 semantic multi-role relation schemas</h1>'
|
|
'<p>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.</p>' + ''.join(sections) + '</html>')
|
|
(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()
|