| """Independent reference oracle for TOM Learner 0.1 promotion authority 0.5.2. |
| |
| This module is deliberately not the authority. The authoritative semantics are |
| ``examples/learner052/promotion_authority.formal.json`` evaluated inside the |
| seeded TOMAGI definition graph. The oracle is a separately written ordinary |
| Python reconstruction used to detect formal-program mistakes. |
| """ |
| from __future__ import annotations |
|
|
| import copy |
| import hashlib |
| import json |
| from typing import Any, Mapping, Sequence |
|
|
|
|
| class OracleError(ValueError): |
| pass |
|
|
|
|
| def canonical_bytes(value: Any) -> bytes: |
| return json.dumps( |
| value, |
| sort_keys=True, |
| ensure_ascii=False, |
| separators=(",", ":"), |
| allow_nan=False, |
| ).encode("utf-8") |
|
|
|
|
| def address(value: Any) -> str: |
| return "sha256:" + hashlib.sha256(canonical_bytes(value)).hexdigest() |
|
|
|
|
| def addressed(body: Mapping[str, Any]) -> dict[str, Any]: |
| result = copy.deepcopy(dict(body)) |
| if "content_hash" in result: |
| raise OracleError("addressed body must not already contain content_hash") |
| result["content_hash"] = address(result) |
| return result |
|
|
|
|
| def verify_addressed(record: Mapping[str, Any], label: str) -> dict[str, Any]: |
| if not isinstance(record, Mapping): |
| raise OracleError(f"{label} must be an object") |
| value = copy.deepcopy(dict(record)) |
| claimed = value.pop("content_hash", None) |
| if claimed != address(value): |
| raise OracleError(f"{label} content hash mismatch") |
| value["content_hash"] = claimed |
| return value |
|
|
|
|
| def _write(namespace: str, record: Mapping[str, Any]) -> dict[str, Any]: |
| return {"namespace": namespace, "record": copy.deepcopy(dict(record))} |
|
|
|
|
| def _unique(values: Sequence[str], label: str) -> list[str]: |
| result = list(values) |
| if len(result) != len(set(result)): |
| raise OracleError(f"{label} must be unique") |
| return result |
|
|
|
|
| def _dataset_map(datasets: Sequence[Mapping[str, Any]]) -> dict[str, dict[str, Any]]: |
| result: dict[str, dict[str, Any]] = {} |
| for index, dataset in enumerate(datasets): |
| checked = verify_addressed(dataset, f"dataset {index}") |
| ident = checked.get("id") |
| if not isinstance(ident, str) or not ident: |
| raise OracleError("dataset id must be a non-empty string") |
| if ident in result: |
| raise OracleError(f"duplicate dataset id {ident}") |
| result[ident] = checked |
| return result |
|
|
|
|
| def build_promotion_result( |
| promotion_program: Mapping[str, Any], |
| learner_program: Mapping[str, Any], |
| learner_execution: Mapping[str, Any], |
| datasets: Sequence[Mapping[str, Any]], |
| context: Mapping[str, Any], |
| corrective_handoff: Mapping[str, Any], |
| token_registry: Mapping[str, Any], |
| ) -> dict[str, Any]: |
| promotion_program = verify_addressed(promotion_program, "promotion program") |
| learner_program = verify_addressed(learner_program, "learner program") |
| learner_execution = verify_addressed(learner_execution, "learner execution") |
| learner_value = verify_addressed(learner_execution["value"], "learner value") |
| context = verify_addressed(context, "promotion context") |
| corrective_handoff = verify_addressed(corrective_handoff, "corrective handoff") |
| token_registry = verify_addressed(token_registry, "token registry") |
| datasets_by_id = _dataset_map(datasets) |
|
|
| if context.get("profile") != "TOM-LEARNER-0.1-PROMOTION-AUTHORITY": |
| raise OracleError("promotion context profile mismatch") |
| if context.get("canonical_seed_sha256") != "sha256:d1417a3136772c0cf3eddcd4962ce07d42cbf87616f7b5bae09fc652d9b807b5": |
| raise OracleError("promotion context seed mismatch") |
| if corrective_handoff["content_hash"] != context.get("corrective_handoff_content_hash"): |
| raise OracleError("corrective handoff mismatch") |
| if token_registry["content_hash"] != context.get("token_registry_content_hash"): |
| raise OracleError("token registry mismatch") |
| if learner_program["content_hash"] != context.get("expected_learner_program_content_hash"): |
| raise OracleError("learner program binding mismatch") |
| if learner_execution.get("program_hash") != learner_program["content_hash"]: |
| raise OracleError("learner execution program binding mismatch") |
| if learner_execution["content_hash"] != context.get("expected_learner_execution_content_hash"): |
| raise OracleError("learner execution identity mismatch") |
| if learner_value["content_hash"] != context.get("expected_learner_value_content_hash"): |
| raise OracleError("learner value identity mismatch") |
|
|
| rows = learner_value.get("results") |
| if not isinstance(rows, list): |
| raise OracleError("learner result rows must be an array") |
| if len(datasets) != context.get("expected_dataset_count") or len(rows) != len(datasets): |
| raise OracleError("dataset/result count mismatch") |
| bindings = [{"id": d["id"], "content_hash": d["content_hash"]} for d in datasets] |
| if bindings != learner_value.get("inputs"): |
| raise OracleError("literal dataset order differs from learner bindings") |
| if learner_value.get("accepted_count") != context.get("expected_accepted_count"): |
| raise OracleError("accepted count mismatch") |
| if learner_value.get("rejected_count") != context.get("expected_rejected_count"): |
| raise OracleError("rejected count mismatch") |
|
|
| shared_objects = [ |
| promotion_program, |
| learner_program, |
| learner_execution, |
| learner_value, |
| context, |
| corrective_handoff, |
| token_registry, |
| ] |
| shared_hashes = _unique([x["content_hash"] for x in shared_objects], "shared authority hashes") |
|
|
| descriptor = addressed({ |
| "schema": "TOMAGI-IMMUTABLE-STORE-DESCRIPTOR-1.0", |
| "profile": context["profile"], |
| "seed_sha256": context["canonical_seed_sha256"], |
| "base_world_hash": context["base_world_hash"], |
| "base_handoff_hash": context["base_handoff_hash"], |
| "corrective_handoff_hash": context["corrective_handoff_content_hash"], |
| "namespaces": copy.deepcopy(context["store_namespaces"]), |
| "head_namespace": context["head_namespace"], |
| "record_encoding": context["record_encoding"], |
| "publication_rule": context["publication_rule"], |
| }) |
| snapshot = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-SNAPSHOT-0.5.2", |
| "profile": context["profile"], |
| "base_world_hash": context["base_world_hash"], |
| "base_handoff_hash": context["base_handoff_hash"], |
| "corrective_handoff_hash": context["corrective_handoff_content_hash"], |
| "accepted_definitions": {}, |
| "sessions": [], |
| "accepted_sessions": [], |
| "rejected_sessions": [], |
| }) |
| genesis_transaction = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-GENESIS-TRANSACTION-0.5.2", |
| "profile": context["profile"], |
| "sequence": 0, |
| "parent_commit_hash": None, |
| "base_world_hash": context["base_world_hash"], |
| "base_handoff_hash": context["base_handoff_hash"], |
| "corrective_handoff_hash": context["corrective_handoff_content_hash"], |
| "seed_sha256": context["canonical_seed_sha256"], |
| "snapshot_hash": snapshot["content_hash"], |
| "authority_program_hash": promotion_program["content_hash"], |
| }) |
| head = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-COMMIT-0.5.2", |
| "profile": context["profile"], |
| "sequence": 0, |
| "parent_commit_hash": None, |
| "transaction_hash": genesis_transaction["content_hash"], |
| "snapshot_hash": snapshot["content_hash"], |
| "authority_program_hash": promotion_program["content_hash"], |
| }) |
| genesis_writes = [_write("objects", x) for x in shared_objects] + [ |
| _write("snapshots", snapshot), |
| _write("transactions", genesis_transaction), |
| _write("commits", head), |
| ] |
| genesis_required = shared_hashes + [ |
| snapshot["content_hash"], |
| genesis_transaction["content_hash"], |
| head["content_hash"], |
| ] |
| publications = [addressed({ |
| "schema": "TOMAGI-IMMUTABLE-PUBLICATION-1.0", |
| "profile": context["profile"], |
| "sequence": 0, |
| "expected_head": None, |
| "replacement_head": head["content_hash"], |
| "required_hashes": genesis_required, |
| "writes": genesis_writes, |
| })] |
|
|
| seen_sessions: set[str] = set() |
| for row_index, row_value in enumerate(rows): |
| row = verify_addressed(row_value, f"learner result row {row_index}") |
| dataset_id = row.get("dataset_id") |
| if dataset_id not in datasets_by_id: |
| raise OracleError(f"learner result row names unknown dataset {dataset_id!r}") |
| if dataset_id in seen_sessions: |
| raise OracleError(f"duplicate promotion session {dataset_id}") |
| seen_sessions.add(dataset_id) |
| dataset = datasets_by_id[dataset_id] |
| if row.get("dataset_content_hash") != dataset["content_hash"]: |
| raise OracleError("result row dataset hash differs from literal source") |
|
|
| split_policy = verify_addressed(dataset["split_policy"], "split policy") |
| hypothesis_family = verify_addressed(dataset["hypothesis_family"], "hypothesis family") |
| acceptance_policy = verify_addressed(dataset["acceptance_policy"], "acceptance policy") |
| observations = [ |
| verify_addressed(observation, f"observation {index}") |
| for index, observation in enumerate(dataset["observations"]) |
| ] |
| accepted = row.get("accepted") |
| if not isinstance(accepted, bool): |
| raise OracleError("accepted must be boolean") |
| relation_value = row.get("relation_definition") |
| if accepted: |
| if not isinstance(relation_value, Mapping): |
| raise OracleError("accepted row has no relation definition") |
| authority_object = verify_addressed(relation_value, "accepted relation") |
| rejection_lineage = None |
| else: |
| if relation_value is not None: |
| raise OracleError("rejected row emitted a relation definition") |
| rejection_lineage = addressed({ |
| "schema": "TOM-LEARNER-REJECTION-LINEAGE-0.5.2", |
| "profile": context["profile"], |
| "dataset_id": dataset_id, |
| "dataset_content_hash": dataset["content_hash"], |
| "result_row_hash": row["content_hash"], |
| "reasons": copy.deepcopy(row["acceptance_reasons"]), |
| "expected_parent_commit_hash": head["content_hash"], |
| "learner_result_hash": learner_execution["content_hash"], |
| "authority_program_hash": promotion_program["content_hash"], |
| }) |
| authority_object = rejection_lineage |
|
|
| decision = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-DECISION-0.5.2", |
| "profile": context["profile"], |
| "dataset_id": dataset_id, |
| "dataset_content_hash": dataset["content_hash"], |
| "result_row_hash": row["content_hash"], |
| "accepted": accepted, |
| "reasons": copy.deepcopy(row["acceptance_reasons"]), |
| "published_definition_hash": authority_object["content_hash"] if accepted else None, |
| "rejection_lineage_hash": None if accepted else authority_object["content_hash"], |
| "expected_parent_commit_hash": head["content_hash"], |
| "learner_program_hash": learner_program["content_hash"], |
| "learner_result_hash": learner_execution["content_hash"], |
| "authority_program_hash": promotion_program["content_hash"], |
| }) |
| dataset_objects = [dataset, split_policy, hypothesis_family, acceptance_policy, *observations] |
| pre_certificate_objects = [*dataset_objects, row, decision, authority_object] |
| pre_certificate_hashes = _unique( |
| [*shared_hashes, *[item["content_hash"] for item in pre_certificate_objects]], |
| "supporting evidence hashes", |
| ) |
| certificate = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-CERTIFICATE-0.5.2", |
| "profile": context["profile"], |
| "sequence": len(publications), |
| "dataset_id": dataset_id, |
| "dataset_content_hash": dataset["content_hash"], |
| "expected_parent_commit_hash": head["content_hash"], |
| "result_row_hash": row["content_hash"], |
| "decision_hash": decision["content_hash"], |
| "accepted": accepted, |
| "published_definition_hash": authority_object["content_hash"] if accepted else None, |
| "rejection_lineage_hash": None if accepted else authority_object["content_hash"], |
| "supporting_evidence_hashes": pre_certificate_hashes, |
| "authority_rule": context["authority_statement"], |
| }) |
| evidence_objects = [*pre_certificate_objects, certificate] |
| evidence_hashes = [*pre_certificate_hashes, certificate["content_hash"]] |
|
|
| accepted_definitions = copy.deepcopy(snapshot["accepted_definitions"]) |
| accepted_sessions = copy.deepcopy(snapshot["accepted_sessions"]) |
| rejected_sessions = copy.deepcopy(snapshot["rejected_sessions"]) |
| if accepted: |
| accepted_definitions[dataset_id] = authority_object["content_hash"] |
| accepted_sessions.append(certificate["content_hash"]) |
| else: |
| rejected_sessions.append(authority_object["content_hash"]) |
| new_snapshot = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-SNAPSHOT-0.5.2", |
| "profile": context["profile"], |
| "base_world_hash": context["base_world_hash"], |
| "base_handoff_hash": context["base_handoff_hash"], |
| "corrective_handoff_hash": context["corrective_handoff_content_hash"], |
| "accepted_definitions": accepted_definitions, |
| "sessions": [*snapshot["sessions"], dataset_id], |
| "accepted_sessions": accepted_sessions, |
| "rejected_sessions": rejected_sessions, |
| }) |
| transaction = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-TRANSACTION-0.5.2", |
| "profile": context["profile"], |
| "sequence": len(publications), |
| "expected_parent_commit_hash": head["content_hash"], |
| "parent_commit_hash": head["content_hash"], |
| "parent_snapshot_hash": snapshot["content_hash"], |
| "base_world_hash": context["base_world_hash"], |
| "base_handoff_hash": context["base_handoff_hash"], |
| "corrective_handoff_hash": context["corrective_handoff_content_hash"], |
| "observation_set_id": dataset_id, |
| "observation_set_hash": dataset["content_hash"], |
| "learner_program_hash": learner_program["content_hash"], |
| "learner_result_hash": learner_execution["content_hash"], |
| "result_row_hash": row["content_hash"], |
| "acceptance_decision_hash": decision["content_hash"], |
| "promotion_certificate_hash": certificate["content_hash"], |
| "evidence_record_hashes": evidence_hashes, |
| "accepted": accepted, |
| "published_definition_hash": authority_object["content_hash"] if accepted else None, |
| "rejection_lineage_hash": None if accepted else authority_object["content_hash"], |
| "new_snapshot_hash": new_snapshot["content_hash"], |
| "authority_program_hash": promotion_program["content_hash"], |
| "authority_rule": context["authority_statement"], |
| }) |
| new_head = addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-COMMIT-0.5.2", |
| "profile": context["profile"], |
| "sequence": len(publications), |
| "parent_commit_hash": head["content_hash"], |
| "transaction_hash": transaction["content_hash"], |
| "snapshot_hash": new_snapshot["content_hash"], |
| "authority_program_hash": promotion_program["content_hash"], |
| }) |
| writes = [_write("objects", item) for item in evidence_objects] + [ |
| _write("snapshots", new_snapshot), |
| _write("transactions", transaction), |
| _write("commits", new_head), |
| ] |
| required_hashes = _unique( |
| [*evidence_hashes, new_snapshot["content_hash"], transaction["content_hash"], new_head["content_hash"]], |
| "publication required hashes", |
| ) |
| publication = addressed({ |
| "schema": "TOMAGI-IMMUTABLE-PUBLICATION-1.0", |
| "profile": context["profile"], |
| "sequence": len(publications), |
| "expected_head": head["content_hash"], |
| "replacement_head": new_head["content_hash"], |
| "required_hashes": required_hashes, |
| "writes": writes, |
| }) |
| publications.append(publication) |
| snapshot = new_snapshot |
| head = new_head |
|
|
| plan = addressed({ |
| "schema": "TOMAGI-IMMUTABLE-PUBLICATION-PLAN-1.0", |
| "profile": context["profile"], |
| "store_descriptor": descriptor, |
| "publications": publications, |
| "terminal_head": head["content_hash"], |
| }) |
| return addressed({ |
| "schema": "TOM-LEARNER-PROMOTION-AUTHORITY-RESULT-0.5.2", |
| "profile": context["profile"], |
| "authority_program_hash": promotion_program["content_hash"], |
| "learner_program_hash": learner_program["content_hash"], |
| "learner_result_hash": learner_execution["content_hash"], |
| "learner_value_hash": learner_value["content_hash"], |
| "context_hash": context["content_hash"], |
| "corrective_handoff_hash": corrective_handoff["content_hash"], |
| "token_registry_hash": token_registry["content_hash"], |
| "dataset_count": len(datasets), |
| "accepted_count": learner_value["accepted_count"], |
| "rejected_count": learner_value["rejected_count"], |
| "publication_count": len(publications), |
| "terminal_head": head["content_hash"], |
| "terminal_snapshot_hash": snapshot["content_hash"], |
| "publication_plan": plan, |
| "claim_boundary": "exact finite affine promotion/evidence transaction authority only; no broader learner family or AGI claim", |
| }) |
|
|
|
|
| __all__ = [ |
| "OracleError", |
| "address", |
| "addressed", |
| "build_promotion_result", |
| "canonical_bytes", |
| "verify_addressed", |
| ] |
|
|
