#!/usr/bin/env python3 """Compile the governed corpus into one recursive, review-gated structural Mind. The compiler treats source files, indexed statements, vocabulary dimensions, and cross-dimensional route queries as addressable information identities. It does not infer semantic role occupants, promote a candidate route to a relation, or claim that a structural Mind has observerhood, experience, morality, selfhood, or consciousness. """ from __future__ import annotations import copy import csv import hashlib import json from collections import Counter, defaultdict from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] SOURCE_INVENTORY_PATH = ROOT / "atlas" / "SOURCE_INVENTORY.csv" SOURCE_RECONCILIATION_PATH = ( ROOT / "reports" / "wisdom-coverage" / "SOURCE_WISDOM_REVIEW_RECONCILIATION.json" ) IDENTITY_MIND_PATH = ROOT / "atlas" / "identity-mind" / "IDENTITY_MIND_REGISTRY.json" CANDIDATE_CELL_PATH = ROOT / "atlas" / "model" / "CANDIDATE_IDENTITY_CELL_REGISTRY.json" ROUTE_INDEX_PATH = ROOT / "atlas" / "identity-mind" / "CROSS_DIMENSIONAL_ROUTE_INDEX.json" VOCABULARY_PATH = ROOT / "atlas" / "model" / "Y_THEORY_VOCABULARY.json" RELEASE_CONFIG_PATH = ROOT / "governance" / "RELEASE_CONFIG.json" SOURCE_RECOVERY_POLICY_PATH = ROOT / "governance" / "SOURCE_BYTE_RECOVERY_POLICY.json" FIELD_SCHEMA_PATH = ROOT / "schemas" / "Y_THEORY_CORPUS_MIND_FIELD.schema.json" SUMMARY_SCHEMA_PATH = ROOT / "schemas" / "Y_THEORY_CORPUS_MIND_FIELD_SUMMARY.schema.json" BODY_SCHEMA_PATH = ROOT / "schemas" / "Y_THEORY_CORPUS_MIND_BODY_PROJECTION.schema.json" FIELD_OUTPUT_PATH = ROOT / "atlas" / "model" / "CORPUS_MIND_FIELD.json" SUMMARY_OUTPUT_PATH = ROOT / "atlas" / "model" / "CORPUS_MIND_FIELD_SUMMARY.json" BODY_OUTPUT_PATH = ROOT / "atlas" / "model" / "CORPUS_MIND_BODY_PROJECTION.json" ROLE_NAMES = ("inside", "outside", "relation") CAPABILITY_NAMES = ( "observerhood", "experience", "moral_agency", "selfhood", "consciousness", ) SHA256_HEX = set("0123456789abcdef") FIELD_VERSION = "CMF-2026.09.01.1" ROOT_CELL = "MC::CMF::SOURCE-TO-FIELD" SOURCE_CORPUS_CELL = "MC::CMF::SOURCE-CORPUS" SOURCE_SET_CELL = "MC::CMF::SOURCE-ARTIFACT-SET" SOURCE_CATEGORY_SET_CELL = "MC::CMF::SOURCE-CATEGORY-SET" SOURCE_ROLE_SET_CELL = "MC::CMF::SOURCE-ROLE-SET" SOURCE_BINDING_CELL = "MC::CMF::SOURCE-INVENTORY-BINDING" COMPILER_CELL = "MC::CMF::Y-ENGINE-COMPILER" IDENTITY_FIELD_CELL = "MC::CMF::IDENTITY-FIELD" IDENTITY_SET_CELL = "MC::CMF::IDENTITY-RECORD-SET" CORRESPONDENCE_FIELD_CELL = "MC::CMF::CORRESPONDENCE-FIELD" SUBJECT_SET_CELL = "MC::CMF::SUBJECT-DIMENSION-SET" POSITION_SET_CELL = "MC::CMF::ENGINE-POSITION-SET" ROUTE_SET_CELL = "MC::CMF::CANDIDATE-ROUTE-SET" REVIEW_GATE_CELL = "MC::CMF::GOVERNED-REVIEW-GATE" def read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8-sig")) def read_csv(path: Path) -> list[dict[str, str]]: with path.open(encoding="utf-8-sig", newline="") as stream: return list(csv.DictReader(stream)) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def canonical_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def pretty_bytes(value: Any) -> bytes: return (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8") def record_hash(value: Any) -> str: return sha256_bytes(canonical_bytes(value)) def stable_token(prefix: str, value: str) -> str: return f"MC::CMF::{prefix}::{sha256_bytes(value.encode('utf-8'))[:24]}" def valid_sha256(value: Any) -> bool: return ( isinstance(value, str) and len(value) == 64 and set(value) <= SHA256_HEX ) def capability_boundary() -> dict[str, Any]: return { "structural_mind": "instantiated_by_primitive_recursive_triad", **{name: "not_earned" for name in CAPABILITY_NAMES}, "earned_capability_gate_refs": [], } def authority( *, state: str, proof_status: str | None, validation_state: str, review_state: str, canonical_approval: bool, guardrail: str, ) -> dict[str, Any]: return { "authority_state": state, "proof_status": proof_status, "validation_state": validation_state, "review_state": review_state, "canonical_approval": canonical_approval, "referent_identity_established": False, "semantic_relation_approved": False, "guardrail": guardrail, } def role( role_name: str, *, binding_state: str = "governed_unresolved", target: str | None = None, semantic_state: str = "unresolved_review_required", review_state: str = "review_required", review_question: str | None, source_locator: str, ) -> dict[str, Any]: return { "role": role_name, "binding_state": binding_state, "mind_cell_ref": target, "semantic_occupant_state": semantic_state, "review_state": review_state, "review_question": review_question, "source_locator": source_locator, "structural_address_only": target is not None, "semantic_relation_asserted": False, "graph_edge": False, } def unresolved_roles(locator: str, subject: str) -> dict[str, dict[str, Any]]: questions = { "inside": f"Which identity or identities are actually inside {subject} at its declared reference scope?", "outside": f"Which identity or identities remain outside {subject} at that same reference scope?", "relation": f"Which native mechanism carries consequential change between the inside and outside of {subject}?", } return { name: role( name, review_question=questions[name], source_locator=locator, ) for name in ROLE_NAMES } def common_cell( *, cell_id: str, layer: str, record_kind: str, title: str, roles: dict[str, dict[str, Any]], member_refs: list[str], membership_state: str, source_lineage: dict[str, Any], authority_record: dict[str, Any], assignments: dict[str, Any], ) -> dict[str, Any]: return { "id": cell_id, "layer": layer, "record_kind": record_kind, "title": title, "structure_state": "primitive_recursive_triad_instantiated", "equation": "MindCell(scope)=(inside,outside,relation)", "roles": roles, "member_mind_cell_refs": sorted(member_refs), "membership_state": membership_state, "source_lineage": source_lineage, "authority": authority_record, "candidate_assignments": assignments, "capability_boundary": capability_boundary(), } def registry_lineage( path: str, registry_sha256: str, locator: str, record: Any, *, source_id: str | None = None, source_sha256: str | None = None, byte_count: int | None = None, byte_availability: str = "governed_registry_record", byte_recovery_state: str | None = None, byte_recovery_locator: str | None = None, byte_compilation_authorized: bool = False, ) -> dict[str, Any]: return { "registry_path": path, "registry_sha256": registry_sha256, "registry_locator": locator, "record_sha256": record_hash(record), "source_id": source_id, "source_sha256": source_sha256, "byte_count": byte_count, "byte_availability": byte_availability, "byte_recovery_state": byte_recovery_state, "byte_recovery_locator": byte_recovery_locator, "byte_compilation_authorized": byte_compilation_authorized, } def structural_roles( locator: str, *, inside: str | None, outside: str | None, relation: str | None, ) -> dict[str, dict[str, Any]]: targets = {"inside": inside, "outside": outside, "relation": relation} result: dict[str, dict[str, Any]] = {} for name, target in targets.items(): if target is None: result[name] = role( name, review_question=( "Which governed identity occupies this still-open structural role?" ), source_locator=locator, ) else: result[name] = role( name, binding_state="mind_cell_reference", target=target, semantic_state="declared_system_structure_only", review_state="structurally_declared", review_question=None, source_locator=locator, ) return result def validate_inputs( inventory: list[dict[str, str]], reconciliation: dict[str, Any], identity_mind: dict[str, Any], candidate_cells: dict[str, Any], routes: dict[str, Any], vocabulary: dict[str, Any], recovery_policy: dict[str, Any], root: Path = ROOT, ) -> None: source_ids = [row.get("source_id", "") for row in inventory] if not source_ids or any(not value for value in source_ids) or len(source_ids) != len(set(source_ids)): raise ValueError("Source inventory requires unique non-empty source IDs") for index, row in enumerate(inventory): if not valid_sha256(row.get("sha256")): raise ValueError(f"Source inventory row {index} has an invalid SHA-256") try: byte_count = int(row.get("size_bytes", "")) except ValueError as exc: raise ValueError(f"Source inventory row {index} has an invalid byte count") from exc if byte_count < 0: raise ValueError(f"Source inventory row {index} has a negative byte count") frozen_count = ( reconciliation.get("summary", {}) .get("source_population_policy", {}) .get("frozen_inventory_records") ) if frozen_count != len(inventory): raise ValueError("Source reconciliation does not bind the frozen inventory count") identity_records = identity_mind.get("records", []) identity_ids = [record.get("id") for record in identity_records] if not identity_ids or any(not value for value in identity_ids) or len(identity_ids) != len(set(identity_ids)): raise ValueError("Identity Mind requires unique non-empty record IDs") candidate_records = candidate_cells.get("records", []) candidate_source_ids = [record.get("source_record_id") for record in candidate_records] if len(candidate_source_ids) != len(set(candidate_source_ids)): raise ValueError("Candidate Identity Cell source IDs must be unique") coverage = candidate_cells.get("coverage", {}) if ( coverage.get("identity_mind_records") != len(identity_records) or coverage.get("candidate_cell_shells") != len(candidate_records) or coverage.get("already_compiled_source_assertions") + len(candidate_records) != len(identity_records) or coverage.get("undispositioned_identity_mind_records") != 0 ): raise ValueError("Candidate Identity Cell partition is incomplete") if set(candidate_source_ids) - set(identity_ids): raise ValueError("Candidate Identity Cell references an unknown Identity Mind record") subject_ids = {record.get("id") for record in vocabulary.get("subjects", [])} position_ids = {record.get("id") for record in vocabulary.get("structural_positions", [])} if not subject_ids or not position_ids: raise ValueError("Vocabulary subject and position dimensions are required") for record in identity_records: if set(record.get("subjects", [])) - subject_ids: raise ValueError(f"Identity record {record['id']} has an unknown subject") if set(record.get("structural_positions", [])) - position_ids: raise ValueError(f"Identity record {record['id']} has an unknown position") route_records = routes.get("routes", []) route_ids = [record.get("id") for record in route_records] if not route_ids or any(not value for value in route_ids) or len(route_ids) != len(set(route_ids)): raise ValueError("Cross-dimensional route IDs must be unique and non-empty") if routes.get("summary", {}).get("candidate_cross_dimensional_routes") != len(route_records): raise ValueError("Cross-dimensional route summary count is stale") for record in route_records: if record.get("left_subject") not in subject_ids or record.get("right_subject") not in subject_ids: raise ValueError(f"Route {record['id']} has an unknown subject") if record.get("mediating_position") not in position_ids: raise ValueError(f"Route {record['id']} has an unknown Y-Engine position") if ( record.get("relation_class") != "candidate_structural_correspondence" or record.get("validation") != "requires_domain_translation_and_falsifier" ): raise ValueError(f"Route {record['id']} overstates correspondence authority") policy_inputs = recovery_policy.get("inputs", {}) policy_counts = recovery_policy.get("counts", {}) inventory_ids = set(source_ids) quarantined_ids = set(recovery_policy.get("quarantined_source_ids", [])) session_materializations = recovery_policy.get("exact_session_materializations", []) materialized_ids = {record.get("source_id") for record in session_materializations} if ( recovery_policy.get("status") != "audited_release_recovery_boundary" or policy_counts.get("registered_source_records") != len(inventory) or policy_counts.get("quarantined_missing_exact_byte_records") != len(quarantined_ids) or policy_counts.get("exact_session_materialization_records") != len(materialized_ids) or policy_counts.get("exact_byte_recoverable_records") != policy_counts.get("raw_git_blob_recoverable_records", 0) + len(materialized_ids) or policy_counts.get("exact_byte_recoverable_records") + len(quarantined_ids) != len(inventory) ): raise ValueError("Source byte recovery policy counts are incomplete or inconsistent") if len(materialized_ids) != len(session_materializations): raise ValueError("Source byte recovery policy materialization IDs must be unique") if (quarantined_ids | materialized_ids) - inventory_ids: raise ValueError("Source byte recovery policy references an unknown source ID") session_ids = {source_id for source_id in inventory_ids if source_id.startswith("session:")} if quarantined_ids & materialized_ids or quarantined_ids | materialized_ids != session_ids: raise ValueError("Source byte recovery policy must exactly partition session records") if policy_inputs.get("source_inventory_sha256") != sha256(root / SOURCE_INVENTORY_PATH.relative_to(ROOT)): raise ValueError("Source byte recovery policy inventory hash is stale") reconciliation_path = root / "atlas" / "REPOSITORY_SNAPSHOT_RECONCILIATION.json" if policy_inputs.get("repository_reconciliation_sha256") != sha256(reconciliation_path): raise ValueError("Source byte recovery policy repository reconciliation hash is stale") def build_state(root: Path = ROOT) -> dict[str, Any]: inventory_path = root / SOURCE_INVENTORY_PATH.relative_to(ROOT) reconciliation_path = root / SOURCE_RECONCILIATION_PATH.relative_to(ROOT) identity_path = root / IDENTITY_MIND_PATH.relative_to(ROOT) candidate_path = root / CANDIDATE_CELL_PATH.relative_to(ROOT) route_path = root / ROUTE_INDEX_PATH.relative_to(ROOT) vocabulary_path = root / VOCABULARY_PATH.relative_to(ROOT) config_path = root / RELEASE_CONFIG_PATH.relative_to(ROOT) recovery_path = root / SOURCE_RECOVERY_POLICY_PATH.relative_to(ROOT) inventory = read_csv(inventory_path) reconciliation = read_json(reconciliation_path) identity_mind = read_json(identity_path) candidate_cells = read_json(candidate_path) routes = read_json(route_path) vocabulary = read_json(vocabulary_path) release_config = read_json(config_path) recovery_policy = read_json(recovery_path) validate_inputs( inventory, reconciliation, identity_mind, candidate_cells, routes, vocabulary, recovery_policy, root, ) hashes = { "source_inventory": sha256(inventory_path), "source_reconciliation": sha256(reconciliation_path), "identity_mind": sha256(identity_path), "candidate_identity_cells": sha256(candidate_path), "cross_dimensional_routes": sha256(route_path), "vocabulary": sha256(vocabulary_path), "release_config": sha256(config_path), "source_byte_recovery_policy": sha256(recovery_path), } builder_path = root / "tools" / "build_corpus_mind_field.py" if builder_path.is_file(): hashes["builder"] = sha256(builder_path) schema_paths = { "field_schema": root / FIELD_SCHEMA_PATH.relative_to(ROOT), "summary_schema": root / SUMMARY_SCHEMA_PATH.relative_to(ROOT), "body_schema": root / BODY_SCHEMA_PATH.relative_to(ROOT), } for key, path in schema_paths.items(): if path.is_file(): hashes[key] = sha256(path) inventory_hash = hashes["source_inventory"] identity_hash = hashes["identity_mind"] route_hash = hashes["cross_dimensional_routes"] vocabulary_hash = hashes["vocabulary"] candidate_by_source = { record["source_record_id"]: record for record in candidate_cells["records"] } materialization_by_source = { record["source_id"]: record["release_locator"] for record in recovery_policy["exact_session_materializations"] } quarantined_source_ids = set(recovery_policy["quarantined_source_ids"]) cells: list[dict[str, Any]] = [] source_refs: list[str] = [] source_categories: dict[str, list[str]] = defaultdict(list) source_roles: dict[str, list[str]] = defaultdict(list) for row_index, row in enumerate(inventory): cell_id = stable_token("SOURCE", row["source_id"]) source_refs.append(cell_id) source_categories[row["category"]].append(cell_id) source_roles[row["role"]].append(cell_id) locator = f"atlas/SOURCE_INVENTORY.csv#row={row_index + 2};source_id={row['source_id']}" if row["source_id"] in quarantined_source_ids: recovery_state = "registered_exact_bytes_currently_unavailable_quarantined" recovery_locator = None byte_availability = "exact_hash_registered_bytes_not_currently_materialized" byte_compilation_authorized = False source_review_state = "exact_source_bytes_required_before_semantic_extraction" elif row["source_id"].startswith("session:"): recovery_state = "exact_release_materialization_sha256_audited" recovery_locator = materialization_by_source[row["source_id"]] byte_availability = "exact_bytes_recoverable_from_release_materialization" byte_compilation_authorized = True source_review_state = "semantic_extraction_required" else: origin_policy = recovery_policy["repository_origins"].get(row["origin"]) if not origin_policy: raise ValueError(f"No raw Git recovery policy for {row['source_id']}") recovery_state = "exact_raw_git_blob_declared_by_reconciliation" recovery_locator = f"git:{origin_policy['commit']}:{row['path']}" byte_availability = "exact_bytes_recoverable_from_declared_git_blob" byte_compilation_authorized = True source_review_state = "semantic_extraction_required" cells.append( common_cell( cell_id=cell_id, layer="source_artifact", record_kind="frozen_source_inventory_identity", title=row["path"], roles=unresolved_roles(locator, f"source artifact {row['source_id']}"), member_refs=[], membership_state="leaf_information_artifact_identity", source_lineage=registry_lineage( "atlas/SOURCE_INVENTORY.csv", inventory_hash, locator, row, source_id=row["source_id"], source_sha256=row["sha256"], byte_count=int(row["size_bytes"]), byte_availability=byte_availability, byte_recovery_state=recovery_state, byte_recovery_locator=recovery_locator, byte_compilation_authorized=byte_compilation_authorized, ), authority_record=authority( state="frozen_source_inventory_record", proof_status=None, validation_state="path_size_sha256_manifest_recorded", review_state=source_review_state, canonical_approval=False, guardrail="A source-file identity and its recorded hash do not establish the truth, meaning, or semantic identities inside the file.", ), assignments={ "subject_ids": [], "structural_position_ids": [], "source_category": row["category"], "source_role": row["role"], "candidate_relation_class": None, }, ) ) identity_refs: list[str] = [] subject_members: dict[str, list[str]] = defaultdict(list) position_members: dict[str, list[str]] = defaultdict(list) compiled_assertions = 0 candidate_shells = 0 for record in identity_mind["records"]: cell_id = f"MC::CMF::IDENTITY::{record['id']}" identity_refs.append(cell_id) for subject_id in record.get("subjects", []): subject_members[subject_id].append(cell_id) for position_id in record.get("structural_positions", []): position_members[position_id].append(cell_id) candidate = candidate_by_source.get(record["id"]) if candidate is None: compiled_assertions += 1 compilation_state = "governed_source_assertion_roles_unresolved" review_state = "role_extraction_required" candidate_cell_ref = None else: candidate_shells += 1 compilation_state = "candidate_identity_cell_shell_roles_unresolved" review_state = "candidate_referent_and_roles_review_required" candidate_cell_ref = candidate["id"] locator = f"atlas/identity-mind/IDENTITY_MIND_REGISTRY.json#id={record['id']}" cells.append( common_cell( cell_id=cell_id, layer="identity_record", record_kind=record["kind"], title=record["title"], roles=unresolved_roles(locator, f"Identity Mind record {record['id']}"), member_refs=[], membership_state=compilation_state, source_lineage=registry_lineage( "atlas/identity-mind/IDENTITY_MIND_REGISTRY.json", identity_hash, locator, record, source_id=record["id"], byte_availability="source_addressed_registry_record", ), authority_record=authority( state=record["status"], proof_status=record.get("proof_status"), validation_state=record["validation"], review_state=review_state, canonical_approval=record["kind"] in {"canonical_claim", "registered_wisdom"}, guardrail=(record.get("guardrail") or "").strip() or ( "No record-specific guardrail has been declared. The record's stated " "authority, proof, validation, and review states remain controlling; " "indexing does not establish truth, a referent identity, or a semantic relation." ), ), assignments={ "subject_ids": list(record.get("subjects", [])), "structural_position_ids": list(record.get("structural_positions", [])), "source_category": None, "source_role": None, "candidate_relation_class": None, "candidate_identity_cell_ref": candidate_cell_ref, "assignment_boundary": "Subjects and Y-Engine positions are retrieval assignments from the Identity Mind index; they do not fill semantic I/O/R occupants or prove a native mechanism.", }, ) ) subject_refs: dict[str, str] = {} for subject in vocabulary["subjects"]: subject_id = subject["id"] cell_id = f"MC::CMF::SUBJECT::{subject_id}" subject_refs[subject_id] = cell_id locator = f"atlas/model/Y_THEORY_VOCABULARY.json#/subjects/{subject_id}" cells.append( common_cell( cell_id=cell_id, layer="subject_dimension", record_kind="governing_subject_dimension", title=subject.get("title", subject_id), roles=unresolved_roles(locator, f"subject dimension {subject_id}"), member_refs=subject_members.get(subject_id, []), membership_state="candidate_index_membership_not_semantic_containment", source_lineage=registry_lineage( "atlas/model/Y_THEORY_VOCABULARY.json", vocabulary_hash, locator, subject, source_id=subject_id, ), authority_record=authority( state="governing_vocabulary_dimension", proof_status=None, validation_state="index_dimension_only", review_state="governing_index", canonical_approval=False, guardrail="A subject dimension groups records for retrieval; membership is not literal equivalence, containment, or proof of a shared mechanism.", ), assignments={ "subject_ids": [subject_id], "structural_position_ids": [], "source_category": None, "source_role": None, "candidate_relation_class": None, }, ) ) position_refs: dict[str, str] = {} for position in vocabulary["structural_positions"]: position_id = position["id"] cell_id = f"MC::CMF::POSITION::{position_id}" position_refs[position_id] = cell_id locator = f"atlas/model/Y_THEORY_VOCABULARY.json#/structural_positions/{position_id}" cells.append( common_cell( cell_id=cell_id, layer="engine_position", record_kind="governing_y_engine_position", title=position["title"], roles=unresolved_roles(locator, f"Y-Engine position {position_id}"), member_refs=position_members.get(position_id, []), membership_state="candidate_assignment_membership_not_mechanism_or_proof", source_lineage=registry_lineage( "atlas/model/Y_THEORY_VOCABULARY.json", vocabulary_hash, locator, position, source_id=position_id, ), authority_record=authority( state="governing_y_engine_position", proof_status=None, validation_state="candidate_assignment_axis", review_state="governing_index", canonical_approval=False, guardrail=position["expected_difference"], ), assignments={ "subject_ids": [], "structural_position_ids": [position_id], "source_category": None, "source_role": None, "candidate_relation_class": None, "invariant": position["invariant"], "expected_difference": position["expected_difference"], "validation_question": position["validation_question"], }, ) ) route_refs: list[str] = [] for route_index, route_record in enumerate(routes["routes"]): cell_id = f"MC::CMF::ROUTE::{route_record['id']}" route_refs.append(cell_id) locator = f"atlas/identity-mind/CROSS_DIMENSIONAL_ROUTE_INDEX.json#/routes/{route_index}" route_roles = { "inside": role( "inside", binding_state="candidate_mind_cell_reference", target=subject_refs[route_record["left_subject"]], semantic_state="candidate_subject_pole_not_literal_identity", review_state="requires_native_domain_translation", review_question="What native-domain identity and mechanism make the left subject a valid inside-facing pole at a declared scale?", source_locator=locator, ), "outside": role( "outside", binding_state="candidate_mind_cell_reference", target=subject_refs[route_record["right_subject"]], semantic_state="candidate_subject_pole_not_literal_identity", review_state="requires_native_domain_translation", review_question="What native-domain identity and mechanism make the right subject a valid outside-facing pole at the same comparison scope?", source_locator=locator, ), "relation": role( "relation", binding_state="candidate_mind_cell_reference", target=position_refs[route_record["mediating_position"]], semantic_state="candidate_y_engine_mediator_not_relation_edge", review_state="requires_invariant_rival_and_falsifier", review_question="Which invariant survives both domain translations, what differs natively, and what evidence or falsifier tests the proposed mediation?", source_locator=locator, ), } cells.append( common_cell( cell_id=cell_id, layer="candidate_correspondence", record_kind="engine_mediated_cross_dimensional_query", title=route_record["id"], roles=route_roles, member_refs=[], membership_state="candidate_query_not_relation_edge", source_lineage=registry_lineage( "atlas/identity-mind/CROSS_DIMENSIONAL_ROUTE_INDEX.json", route_hash, locator, route_record, source_id=route_record["id"], ), authority_record=authority( state="candidate_structural_correspondence", proof_status="UNASSESSED_CORRESPONDENCE_QUERY", validation_state="requires_domain_translation_and_falsifier", review_state="review_required", canonical_approval=False, guardrail="The query connects two subject indexes only through an explicit Y-Engine position. It is not a literal equivalence, native-domain mechanism, reviewed relation, or proof.", ), assignments={ "subject_ids": [route_record["left_subject"], route_record["right_subject"]], "structural_position_ids": [route_record["mediating_position"]], "source_category": None, "source_role": None, "candidate_relation_class": route_record["relation_class"], "left_record_count": route_record["left_record_count"], "right_record_count": route_record["right_record_count"], "left_examples": list(route_record["left_examples"]), "right_examples": list(route_record["right_examples"]), "native_mechanism_left": None, "native_mechanism_right": None, "invariant": None, "expected_difference": None, "evidence": [], "rival": None, "falsifier": None, }, ) ) category_refs: list[str] = [] for category_name, members in sorted(source_categories.items()): cell_id = stable_token("SOURCE-CATEGORY", category_name) category_refs.append(cell_id) locator = f"atlas/SOURCE_INVENTORY.csv#category={category_name}" cells.append( common_cell( cell_id=cell_id, layer="source_category", record_kind="source_inventory_category_index", title=category_name, roles=unresolved_roles(locator, f"source category {category_name}"), member_refs=members, membership_state="inventory_category_membership_only", source_lineage=registry_lineage( "atlas/SOURCE_INVENTORY.csv", inventory_hash, locator, {"category": category_name, "members": sorted(members)}, source_id=category_name, ), authority_record=authority( state="source_inventory_category", proof_status=None, validation_state="manifest_grouping_only", review_state="semantic_extraction_required", canonical_approval=False, guardrail="Inventory category membership organizes files; it does not establish a theory identity or relation among their contents.", ), assignments={ "subject_ids": [], "structural_position_ids": [], "source_category": category_name, "source_role": None, "candidate_relation_class": None, }, ) ) source_role_refs: list[str] = [] for source_role, members in sorted(source_roles.items()): cell_id = stable_token("SOURCE-ROLE", source_role) source_role_refs.append(cell_id) locator = f"atlas/SOURCE_INVENTORY.csv#role={source_role}" cells.append( common_cell( cell_id=cell_id, layer="source_role", record_kind="source_inventory_role_index", title=source_role, roles=unresolved_roles(locator, f"source role {source_role}"), member_refs=members, membership_state="inventory_role_membership_only", source_lineage=registry_lineage( "atlas/SOURCE_INVENTORY.csv", inventory_hash, locator, {"role": source_role, "members": sorted(members)}, source_id=source_role, ), authority_record=authority( state="source_inventory_role", proof_status=None, validation_state="manifest_grouping_only", review_state="semantic_extraction_required", canonical_approval=False, guardrail="Inventory role membership organizes evidence handling; it does not establish semantic content or proof.", ), assignments={ "subject_ids": [], "structural_position_ids": [], "source_category": None, "source_role": source_role, "candidate_relation_class": None, }, ) ) controller_specs = [ (ROOT_CELL, "Source-to-field recursive Mind", SOURCE_CORPUS_CELL, REVIEW_GATE_CELL, COMPILER_CELL, []), (SOURCE_CORPUS_CELL, "Frozen source corpus Mind", SOURCE_SET_CELL, REVIEW_GATE_CELL, SOURCE_BINDING_CELL, []), (SOURCE_SET_CELL, "Source artifact identity set", None, None, None, source_refs), (SOURCE_CATEGORY_SET_CELL, "Source category index set", None, None, None, category_refs), (SOURCE_ROLE_SET_CELL, "Source role index set", None, None, None, source_role_refs), (SOURCE_BINDING_CELL, "Frozen source inventory binding", SOURCE_CATEGORY_SET_CELL, SOURCE_ROLE_SET_CELL, SOURCE_SET_CELL, []), (COMPILER_CELL, "Y-Engine source compiler Mind", SOURCE_CORPUS_CELL, REVIEW_GATE_CELL, IDENTITY_FIELD_CELL, []), (IDENTITY_FIELD_CELL, "Governed Identity Mind field", IDENTITY_SET_CELL, REVIEW_GATE_CELL, CORRESPONDENCE_FIELD_CELL, []), (IDENTITY_SET_CELL, "Identity Mind record set", None, None, None, identity_refs), (CORRESPONDENCE_FIELD_CELL, "Engine-mediated correspondence query field", SUBJECT_SET_CELL, POSITION_SET_CELL, ROUTE_SET_CELL, []), (SUBJECT_SET_CELL, "Subject dimension set", None, None, None, list(subject_refs.values())), (POSITION_SET_CELL, "Y-Engine position set", None, None, None, list(position_refs.values())), (ROUTE_SET_CELL, "Cross-dimensional candidate route set", None, None, None, route_refs), (REVIEW_GATE_CELL, "Governed unresolved review boundary", None, None, None, []), ] control_locator = "governance/PAGE_AND_DATA_ARCHITECTURE.md#source-compiler-identity-field-and-manifestation-model" for cell_id, title, inside, outside, relation_target, members in controller_specs: cells.append( common_cell( cell_id=cell_id, layer="controller", record_kind="corpus_mind_control_cell", title=title, roles=structural_roles( control_locator, inside=inside, outside=outside, relation=relation_target, ), member_refs=members, membership_state=( "explicit_structural_collection_membership_no_semantic_edge" if members else "control_cell_no_member_collection" ), source_lineage=registry_lineage( "governance/PAGE_AND_DATA_ARCHITECTURE.md", sha256(root / "governance" / "PAGE_AND_DATA_ARCHITECTURE.md"), control_locator, {"id": cell_id, "title": title}, source_id=cell_id, ), authority_record=authority( state="governed_compiler_architecture", proof_status=None, validation_state="structural_runtime_contract", review_state="governing_architecture", canonical_approval=False, guardrail="System structure and collection membership do not establish semantic role occupants, external truth, observerhood, experience, morality, selfhood, or consciousness.", ), assignments={ "subject_ids": [], "structural_position_ids": [], "source_category": None, "source_role": None, "candidate_relation_class": None, }, ) ) layer_counts = Counter(cell["layer"] for cell in cells) reference_roles = sum( role_record["mind_cell_ref"] is not None for cell in cells for role_record in cell["roles"].values() ) candidate_reference_roles = sum( role_record["binding_state"] == "candidate_mind_cell_reference" for cell in cells for role_record in cell["roles"].values() ) membership_addresses = sum(len(cell["member_mind_cell_refs"]) for cell in cells) state = { "compiler_contract": { "direction": "frozen_source_inventory_to_recursive_identity_and_correspondence_field", "source_artifact_rule": "Every frozen SOURCE_INVENTORY row becomes one information-artifact MindCell with exact manifest lineage and unresolved semantic roles.", "byte_recovery_rule": "Only exact bytes from the declared Git blobs or SHA-256-audited release materializations may enter byte-level extraction; registered missing bytes remain quarantined MindCells.", "identity_rule": "Every Identity Mind record becomes one MindCell; approved assertions retain authority while referent and I/O/R semantics remain separately review-gated.", "correspondence_rule": "Every CROSS_DIMENSIONAL_ROUTE_INDEX query becomes a candidate triad whose inside/outside addresses are subject dimensions and whose relation address is an explicit Y-Engine position.", "membership_rule": "Collection membership and taxonomy assignment are structural retrieval addresses, never semantic graph edges or literal containment claims.", "semantic_inference_performed": False, "canonical_source_mutation_authorized": False, "candidate_route_promotion_authorized": False, "literal_equivalence_authorized": False, "body_projection_embedded": False, }, "mind_objects": [ { "id": "MO::CMF::LIVING-SOURCE", "title": "Y-Theory corpus Source-to-field Mind", "root_mind_cell_ref": ROOT_CELL, "mind_cell_count": len(cells), "layer_ids": sorted(layer_counts), "state": "compiled_structural_mind_semantics_review_gated", } ], "root_mind_cell_ref": ROOT_CELL, "mind_cells": cells, "counts": { "source_inventory_records": len(inventory), "exact_byte_recoverable_source_records": recovery_policy["counts"]["exact_byte_recoverable_records"], "raw_git_blob_recoverable_source_records": recovery_policy["counts"]["raw_git_blob_recoverable_records"], "exact_session_materialization_source_records": recovery_policy["counts"]["exact_session_materialization_records"], "quarantined_missing_exact_byte_source_records": recovery_policy["counts"]["quarantined_missing_exact_byte_records"], "identity_mind_records": len(identity_refs), "compiled_source_assertion_records": compiled_assertions, "candidate_identity_shell_records": candidate_shells, "subject_dimensions": len(subject_refs), "engine_positions": len(position_refs), "candidate_correspondence_routes": len(route_refs), "source_categories": len(category_refs), "source_roles": len(source_role_refs), "controller_cells": len(controller_specs), "mind_objects": 1, "mind_cells": len(cells), "triadic_role_slots": len(cells) * 3, "mind_cell_reference_roles": reference_roles, "candidate_route_role_addresses": candidate_reference_roles, "governed_unresolved_roles": len(cells) * 3 - reference_roles, "structural_membership_addresses": membership_addresses, "semantic_graph_edges": 0, "literal_equivalences_asserted": 0, "observerhood_capabilities_earned": 0, "experience_capabilities_earned": 0, }, "layer_counts": dict(sorted(layer_counts.items())), "review_queues": { "source_semantic_extraction_cells": recovery_policy["counts"]["exact_byte_recoverable_records"], "source_byte_recovery_cells": recovery_policy["counts"]["quarantined_missing_exact_byte_records"], "identity_role_review_cells": len(identity_refs), "candidate_referent_review_cells": candidate_shells, "candidate_correspondence_review_cells": len(route_refs), "native_mechanism_pairs_required": len(route_refs) * 2, "route_invariants_required": len(route_refs), "route_rivals_required": len(route_refs), "route_falsifiers_required": len(route_refs), "review_state": "open_until_source_addressed_governed_review", }, "semantic_graph_edges": [], "input_hashes": dict(sorted(hashes.items())), } return state def canonicalize_state(state: dict[str, Any]) -> dict[str, Any]: normalized = copy.deepcopy(state) for cell in normalized["mind_cells"]: cell["member_mind_cell_refs"] = sorted(cell["member_mind_cell_refs"]) assignments = cell["candidate_assignments"] for key in ("subject_ids", "structural_position_ids", "left_examples", "right_examples", "evidence"): if isinstance(assignments.get(key), list): assignments[key] = sorted(assignments[key]) normalized["mind_cells"] = sorted(normalized["mind_cells"], key=lambda record: record["id"]) normalized["mind_objects"] = sorted(normalized["mind_objects"], key=lambda record: record["id"]) normalized["counts"] = dict(sorted(normalized["counts"].items())) normalized["layer_counts"] = dict(sorted(normalized["layer_counts"].items())) normalized["review_queues"] = dict(sorted(normalized["review_queues"].items())) normalized["input_hashes"] = dict(sorted(normalized["input_hashes"].items())) return normalized def run_fixed_point(initial: dict[str, Any], max_passes: int = 8) -> tuple[dict[str, Any], dict[str, Any]]: if max_passes < 1: raise ValueError("max_passes must be positive") current = copy.deepcopy(initial) current_hash = record_hash(current) history = [{"pass": 0, "state_hash": current_hash}] for pass_number in range(1, max_passes + 1): following = canonicalize_state(current) following_hash = record_hash(following) history.append({"pass": pass_number, "state_hash": following_hash}) if following_hash == current_hash: return following, { "algorithm": "sha256_over_canonical_json_v1", "canonical_serialization": "UTF-8 JSON with sorted object keys, compact separators, and compiler-normalized array order", "hashed_component": "corpus_source_to_field_mind", "max_passes": max_passes, "passes_executed": pass_number, "history": history, "state_hash": following_hash, "converged": True, "stop_reason": "canonical_serialized_state_hash_unchanged", "phi_used_as_stop_rule": False, "golden_ratio_used_as_stop_rule": False, "ratio_target": None, } current = following current_hash = following_hash raise ValueError("Corpus Mind did not reach canonical hash equality") def build_field(root: Path = ROOT) -> dict[str, Any]: initial = build_state(root) state, fixed_point = run_fixed_point(initial) release_config = read_json(root / RELEASE_CONFIG_PATH.relative_to(ROOT)) return { "schema": "y-theory.corpus-mind-field.v1", "version": FIELD_VERSION, "status": "deterministic_full_structural_corpus_compilation_semantics_review_gated", "release": { "living_source_version": release_config["living_source_version"], "identity_mind_version": release_config["identity_mind_version"], "canonical_author": release_config["canonical_author"], "state": release_config["release_state"], }, "corpus_source_to_field_mind": state, "fixed_point": fixed_point, "guardrails": [ "The primitive recursive triad earns structural Mind only; it earns no observerhood, experience, morality, selfhood, or consciousness.", "A frozen source-inventory row is an addressable information-artifact identity, not proof of the truth or meaning of its contents.", "Every Identity Mind record is structurally dispositioned, while referent identity and semantic inside/outside/relation occupants remain separately review-gated.", "A cross-dimensional candidate must pass through an explicit Y-Engine position and still supply two native mechanisms, a translated invariant, expected difference, evidence, a rival, and a falsifier.", "Subject membership, engine-position assignment, collection membership, visual proximity, and candidate role addresses are not semantic graph edges or literal equivalence.", "Fixed-point closure occurs only when consecutive canonical state hashes are equal; phi, golden ratio, and visual saturation are forbidden stop rules.", ], } def build_summary(field: dict[str, Any]) -> dict[str, Any]: state = field["corpus_source_to_field_mind"] field_hash = sha256_bytes(pretty_bytes(field)) subject_cells = [cell for cell in state["mind_cells"] if cell["layer"] == "subject_dimension"] position_cells = [cell for cell in state["mind_cells"] if cell["layer"] == "engine_position"] return { "schema": "y-theory.public-corpus-mind-field-summary.v1", "version": FIELD_VERSION, "living_source_version": field["release"]["living_source_version"], "source_field": { "path": "atlas/model/CORPUS_MIND_FIELD.json", "sha256": field_hash, "canonical_state_hash": field["fixed_point"]["state_hash"], "content_state": "withheld_from_public_site_full_artifact_available_in_living_source_repository", }, "byte_recovery": { "registered_source_records": state["counts"]["source_inventory_records"], "exact_byte_recoverable_records": state["counts"]["exact_byte_recoverable_source_records"], "raw_git_blob_recoverable_records": state["counts"]["raw_git_blob_recoverable_source_records"], "exact_session_materialization_records": state["counts"]["exact_session_materialization_source_records"], "quarantined_missing_exact_byte_records": state["counts"]["quarantined_missing_exact_byte_source_records"], "quarantine_rule": "Missing exact bytes stay registered and addressable but cannot enter parsing, summarization, substitution, or semantic compilation.", }, "counts": state["counts"], "layer_counts": state["layer_counts"], "review_queues": state["review_queues"], "root_mind_cell_ref": state["root_mind_cell_ref"], "subject_dimensions": [ { "mind_cell_ref": cell["id"], "id": cell["candidate_assignments"]["subject_ids"][0], "title": cell["title"], "record_members": len(cell["member_mind_cell_refs"]), } for cell in subject_cells ], "engine_positions": [ { "mind_cell_ref": cell["id"], "id": cell["candidate_assignments"]["structural_position_ids"][0], "title": cell["title"], "record_members": len(cell["member_mind_cell_refs"]), "invariant": cell["candidate_assignments"].get("invariant"), "expected_difference": cell["candidate_assignments"].get("expected_difference"), "validation_question": cell["candidate_assignments"].get("validation_question"), } for cell in position_cells ], "fixed_point": field["fixed_point"], "public_boundary": "This summary exposes counts, byte-recovery state, taxonomy, queues, hashes, and convergence evidence. It withholds the full 1,610-row source-cell layer and never turns a candidate route into a relation edge, proof, literal equivalence, or capability claim.", } def build_body_projection(field: dict[str, Any]) -> dict[str, Any]: state = field["corpus_source_to_field_mind"] field_hash = sha256_bytes(pretty_bytes(field)) visible_cells = [cell for cell in state["mind_cells"] if cell["layer"] != "source_artifact"] nodes = [] role_addresses = [] for cell in visible_cells: assignments = cell["candidate_assignments"] nodes.append( { "id": f"NODE::{cell['id']}", "mind_cell_ref": cell["id"], "layer": cell["layer"], "record_kind": cell["record_kind"], "title": cell["title"], "authority_state": cell["authority"]["authority_state"], "proof_status": cell["authority"]["proof_status"], "review_state": cell["authority"]["review_state"], "subject_mind_cell_refs": [f"MC::CMF::SUBJECT::{value}" for value in assignments.get("subject_ids", [])], "position_mind_cell_refs": [f"MC::CMF::POSITION::{value}" for value in assignments.get("structural_position_ids", [])], "member_count": len(cell["member_mind_cell_refs"]), "source_field_state_hash": field["fixed_point"]["state_hash"], "observerhood_state": "not_earned", "experience_state": "not_earned", } ) for role_name in ROLE_NAMES: role_record = cell["roles"][role_name] role_addresses.append( { "id": f"VIEW::{cell['id']}::{role_name}", "owner_mind_cell_ref": cell["id"], "role": role_name, "binding_state": role_record["binding_state"], "target_mind_cell_ref": role_record["mind_cell_ref"], "semantic_occupant_state": role_record["semantic_occupant_state"], "review_state": role_record["review_state"], "review_question": role_record["review_question"], "candidate_only": role_record["binding_state"] == "candidate_mind_cell_reference", "display_only": True, "semantic_relation": False, "graph_edge": False, } ) layer_counts = Counter(node["layer"] for node in nodes) release_config = field["release"] return { "schema": "y-theory.corpus-mind-body-projection.v1", "version": FIELD_VERSION, "status": "separate_public_read_model_source_artifact_cells_aggregated", "release": release_config, "source_field": { "path": "atlas/model/CORPUS_MIND_FIELD.json", "artifact_sha256": field_hash, "canonical_state_hash": field["fixed_point"]["state_hash"], "fixed_point_verified": True, }, "body_contract": { "kind": "field_to_manifestation_body_read_model", "input_mode": "immutable_content_addressed_corpus_mind_field", "transformation": "project_safe_nodes_and_candidate_role_addresses_without_authority_change", "source_write_authorized": False, "canonical_mutation_authorized": False, "semantic_edge_creation_authorized": False, }, "projection_contract": { "modes": ["2d", "3d"], "navigation_dimensions": ["layer", "subject_dimension", "engine_position", "candidate_route", "identity_record"], "source_artifact_cells": "aggregated_counts_only_full_cells_withheld", "layout_is_mechanism_or_proof": False, "candidate_routes_are_dashed_review_queries": True, "source_lineage_required": True, }, "counts": { "source_field_mind_cells": state["counts"]["mind_cells"], "visible_nodes": len(nodes), "withheld_source_artifact_nodes": state["counts"]["source_inventory_records"], "exact_byte_recoverable_source_artifacts": state["counts"]["exact_byte_recoverable_source_records"], "quarantined_missing_exact_byte_source_artifacts": state["counts"]["quarantined_missing_exact_byte_source_records"], "role_addresses": len(role_addresses), "candidate_route_nodes": layer_counts.get("candidate_correspondence", 0), "identity_record_nodes": layer_counts.get("identity_record", 0), "subject_dimension_nodes": layer_counts.get("subject_dimension", 0), "engine_position_nodes": layer_counts.get("engine_position", 0), "candidate_route_role_addresses": sum(record["candidate_only"] for record in role_addresses), "semantic_graph_edges": 0, }, "layer_counts": dict(sorted(layer_counts.items())), "nodes": sorted(nodes, key=lambda record: record["id"]), "role_addresses": sorted(role_addresses, key=lambda record: record["id"]), "semantic_graph_edges": [], "capability_boundary": { "structural_mind_visible": True, "observerhood_state": "not_earned", "experience_state": "not_earned", "consciousness_state": "not_earned", "presentation_cannot_earn_capabilities": True, }, "guardrails": [ "The Body is a separate read model and cannot write to the source, compiler, or governed field.", "Candidate route role-addresses remain review queries and are never rendered as reviewed semantic edges.", "Source artifact cells are aggregated in the public Body; exact cells remain in the Living Source repository field artifact.", "2D/3D position, motion, proximity, attraction, and geometry are presentation behavior, not mechanism or proof.", ], } def build_outputs(root: Path = ROOT) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: field = build_field(root) summary = build_summary(field) body = build_body_projection(field) return field, summary, body def main() -> int: field, summary, body = build_outputs(ROOT) FIELD_OUTPUT_PATH.write_bytes(pretty_bytes(field)) SUMMARY_OUTPUT_PATH.write_bytes(pretty_bytes(summary)) BODY_OUTPUT_PATH.write_bytes(pretty_bytes(body)) print( json.dumps( { "field_counts": field["corpus_source_to_field_mind"]["counts"], "fixed_point": field["fixed_point"], "body_counts": body["counts"], }, indent=2, ensure_ascii=False, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())