| """Command-line interface for TOM Learner 0.1 / WQK 0.5.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
| import shutil |
| from typing import Any, Mapping |
|
|
| from tom_world03.canonical import canonical_bytes |
|
|
| from .baseline import trusted_affine_learning_baseline |
| from .handoff import verify_corrective_handoff, verify_literal_handoff |
| from .io import load_observation_set |
| from .learner import learn_observation_set |
| from .store import LearnerStore |
|
|
|
|
| def _emit(value: Mapping[str, Any] | list[Any], output: str | None = None) -> None: |
| data = canonical_bytes(value) + b"\n" |
| if output: |
| path = Path(output) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_bytes(data) |
| print(json.dumps(value, indent=2, sort_keys=True)) |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| parser = argparse.ArgumentParser( |
| prog="tom-learner05", |
| description=( |
| "Reference affine oracle/evidence-store tools and corrective seeded-authority verification" |
| ), |
| ) |
| sub = parser.add_subparsers(dest="command", required=True) |
|
|
| p_handoff = sub.add_parser("verify-handoff", help="verify every 0.4.2 authoritative literal file") |
| p_handoff.add_argument("root", nargs="?", default=".") |
| p_handoff.add_argument("--handoff") |
| p_handoff.add_argument("--output") |
|
|
| p_corrective_handoff = sub.add_parser( |
| "verify-corrective-handoff", |
| help="verify the 0.4.2 base plus the explicit 0.5.1 corrective overlay", |
| ) |
| p_corrective_handoff.add_argument("root", nargs="?", default=".") |
| p_corrective_handoff.add_argument("--corrective") |
| p_corrective_handoff.add_argument("--output") |
|
|
| p_validate = sub.add_parser("validate-dataset", help="verify and typecheck one observation set") |
| p_validate.add_argument("dataset") |
| p_validate.add_argument("--output") |
|
|
| p_split = sub.add_parser("split", help="produce the deterministic ID-only train/validation/holdout split") |
| p_split.add_argument("dataset") |
| p_split.add_argument("--output") |
|
|
| p_learn = sub.add_parser( |
| "learn", help="run the non-authoritative host reference for one exact affine hypothesis" |
| ) |
| p_learn.add_argument("dataset") |
| p_learn.add_argument("--output") |
| p_learn.add_argument("--records-dir") |
|
|
| p_baseline = sub.add_parser("baseline", help="run the independent fractions.Fraction baseline") |
| p_baseline.add_argument("dataset") |
| p_baseline.add_argument("--output") |
|
|
| p_init = sub.add_parser("init-store", help="initialize an empty parent-bound learner overlay store") |
| p_init.add_argument("store") |
| p_init.add_argument("--seed", default="TOM_seed_genome_2026-09-01.txt") |
| p_init.add_argument("--replace", action="store_true") |
| p_init.add_argument("--output") |
|
|
| p_promote = sub.add_parser("promote", help="learn and commit one accepted or rejected session") |
| p_promote.add_argument("store") |
| p_promote.add_argument("dataset") |
| p_promote.add_argument("--expected-parent", required=True) |
| p_promote.add_argument("--output") |
|
|
| p_audit = sub.add_parser("audit", help="audit all immutable learner evidence and commit ancestry") |
| p_audit.add_argument("store") |
| p_audit.add_argument("--allow-orphans", action="store_true") |
| p_audit.add_argument("--output") |
|
|
| p_reconstruct = sub.add_parser("reconstruct", help="reconstruct the authoritative learner overlay") |
| p_reconstruct.add_argument("store") |
| p_reconstruct.add_argument("--output") |
|
|
| args = parser.parse_args(argv) |
|
|
| if args.command == "verify-handoff": |
| record = verify_literal_handoff(args.root, args.handoff) |
| _emit(record, args.output) |
| return 0 if record["valid"] else 1 |
|
|
| if args.command == "verify-corrective-handoff": |
| record = verify_corrective_handoff(args.root, args.corrective) |
| _emit(record, args.output) |
| return 0 if record["valid"] else 1 |
|
|
| if args.command == "validate-dataset": |
| raw, dataset = load_observation_set(args.dataset) |
| record = { |
| "schema": "TOM-LEARNER-DATASET-VALIDATION-0.1", |
| "status": "valid", |
| "id": dataset.id, |
| "content_hash": dataset.content_hash, |
| "observation_count": len(dataset.observations), |
| "literal_bytes": len(canonical_bytes(raw)) + 1, |
| "base_world_hash": dataset.base_world_hash, |
| "base_handoff_hash": dataset.base_handoff_hash, |
| } |
| _emit(record, args.output) |
| return 0 |
|
|
| if args.command in {"split", "learn", "baseline", "promote"}: |
| raw, dataset = load_observation_set(args.dataset) |
|
|
| if args.command == "split": |
| from .split import deterministic_split |
| _emit(deterministic_split(dataset), args.output) |
| return 0 |
|
|
| if args.command == "learn": |
| run = learn_observation_set(dataset) |
| if args.records_dir: |
| directory = Path(args.records_dir) |
| directory.mkdir(parents=True, exist_ok=True) |
| for index, record in enumerate(run.all_records()): |
| schema = str(record.get("schema", "record")).lower().replace("tom-", "").replace("-", "_") |
| path = directory / f"{index:04d}_{schema}_{record['content_hash'][7:19]}.json" |
| path.write_bytes(canonical_bytes(record) + b"\n") |
| _emit(run.summary(), args.output) |
| return 0 if run.accepted else 3 |
|
|
| if args.command == "baseline": |
| _emit(trusted_affine_learning_baseline(raw), args.output) |
| return 0 |
|
|
| if args.command == "init-store": |
| target = Path(args.store) |
| if target.exists() and args.replace: |
| shutil.rmtree(target) |
| store = LearnerStore.initialize(target, Path(args.seed).read_bytes()) |
| record = { |
| "schema": "TOM-LEARNER-STORE-INITIALIZATION-0.1", |
| "store": str(target), |
| "head": store.head(), |
| "descriptor": store.descriptor(), |
| } |
| _emit(record, args.output) |
| return 0 |
|
|
| if args.command == "promote": |
| store = LearnerStore(args.store) |
| run = learn_observation_set(dataset) |
| commit = store.commit_learning(run, expected_parent=args.expected_parent) |
| record = { |
| "schema": "TOM-LEARNER-PROMOTION-RESULT-0.1", |
| "accepted": run.accepted, |
| "summary": run.summary(), |
| "commit": commit, |
| "new_head": store.head(), |
| } |
| _emit(record, args.output) |
| return 0 |
|
|
| if args.command == "audit": |
| record = LearnerStore(args.store).audit(require_no_orphans=not args.allow_orphans) |
| _emit(record, args.output) |
| return 0 if record["valid"] else 1 |
|
|
| if args.command == "reconstruct": |
| _emit(LearnerStore(args.store).reconstruct(), args.output) |
| return 0 |
|
|
| return 2 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
