diff --git a/README.md b/README.md index 2210623..9964f8b 100644 --- a/README.md +++ b/README.md @@ -52,7 +52,7 @@ Canonical modules: | semantic boundary | `mathgraph/semantic_validation.py`, `mathgraph/invariants.py` | | routing memory | `mathgraph/reason_atlas.py`, `mathgraph/reason_atlas_store.py` | | H-Tilt / scheduling pressure | `mathgraph/spectral_htilt.py`, `mathgraph/reason_atlas_htilt.py`, `mathgraph/viability_operators.py` | -| verification loops | `mathgraph/verification_loop.py`, `mathgraph/compounding_engine.py` | +| verification loops | `mathgraph/verification_loop.py`, `mathgraph/compounding_engine.py`, `mathgraph/autonomous_compounding_engine.py` | | finite checker | `mathgraph/finite_magma_world.py` | | SAIR / ETP adapters | `mathgraph/sair_task_loader.py`, `mathgraph/sair_constructor_bank.py` | @@ -172,6 +172,47 @@ python scripts/run_mathgraph_compounding_engine.py \ See [docs/compounding_engine.md](docs/compounding_engine.md). +## Autonomous Compounding Engine + +The autonomous engine has a compatibility `facade` mode and an explicit +`native_v2` mode. Native v2 is the finite recovery path that builds a +constructor bank, evaluates a SAT cache, repairs residuals, names PQ-IR +obstructions, records advisory Lawbook reuse, and audits the terminal-form +boundary. It remains a finite-core recovery loop, not a TRUE oracle: failed +finite search is residual evidence and every PQ-IR, route, obstruction, and +Lawbook reuse signal is advisory unless a checker/verifier accepts a terminal +artifact. + +```bash +python scripts/run_autonomous_compounding_engine.py \ + --out-dir /tmp/mathgraph_autonomous_v2_tiny \ + --tiny-demo \ + --finite-core-mode native_v2 \ + --episodes 3 \ + --sample-pairs 80 \ + --repair-budget 20 \ + --max-n 3 \ + --seed 20260524 \ + --write-report +``` + +Real ETP/SAIR example: + +```bash +python scripts/run_autonomous_compounding_engine.py \ + --equations /content/equations.txt \ + --matrix /content/etp_matrix_full_best_bool.npy \ + --out-dir /content/drive/MyDrive/SAIR_MathGraph/autonomous_v2_real \ + --finite-core-mode native_v2 \ + --episodes 4 \ + --sample-pairs 4000 \ + --repair-budget 40 \ + --max-n 4 \ + --constructor-limit 500 \ + --seed 20260524 \ + --write-report +``` + ## TRUE-Side Proof Inventory The TRUE-side inventory builds bounded congruence traces and Lean-ready diff --git a/docs/autonomous_compounding_engine.md b/docs/autonomous_compounding_engine.md new file mode 100644 index 0000000..4d0d503 --- /dev/null +++ b/docs/autonomous_compounding_engine.md @@ -0,0 +1,84 @@ +# Autonomous Compounding Engine + +The autonomous compounding engine is a repo-native entry point for the serious +ETP finite-recovery path. It has two modes: + +- `facade`: the stable compatibility path over the existing finite-core + compounding runner. +- `native_v2`: the repo-native finite recovery, residual repair, and advisory + Lawbook reuse loop. + +It wraps `scripts/run_mathgraph_compounding_engine.py` rather than simulating +recovery in `facade` mode. The `native_v2` path uses the finite magma +satisfaction cache directly: + +```text +constructor bank -> SAT cache -> generic route -> residual repair +-> PQ-IR obstruction naming -> advisory Lawbook reuse -> compact atlas route +``` + +## Boundary + +- FALSE recovery is counted only when a constructor satisfies the source law and + violates the target law. +- TRUE controls audit contamination. +- Failed finite search is residual evidence, never TRUE. +- PQ-IR, route policies, residual obstructions, and repair family memory are + advisory until a verifier/checker boundary accepts a terminal form. + +## Tiny demo + +```bash +python scripts/run_autonomous_compounding_engine.py \ + --out-dir /tmp/mathgraph_autonomous_demo \ + --tiny-demo \ + --episodes 2 \ + --sample-pairs 80 \ + --repair-budget 20 +``` + +Native v2 smoke: + +```bash +python scripts/run_autonomous_compounding_engine.py \ + --out-dir /tmp/mathgraph_autonomous_v2_tiny \ + --tiny-demo \ + --finite-core-mode native_v2 \ + --episodes 3 \ + --sample-pairs 80 \ + --repair-budget 20 \ + --max-n 3 \ + --seed 20260524 \ + --write-report +``` + +## Real ETP / SAIR run + +```bash +python scripts/run_autonomous_compounding_engine.py \ + --equations /content/equations.txt \ + --matrix /content/etp_matrix_full_best_bool.npy \ + --out-dir /content/drive/MyDrive/SAIR_MathGraph/Autonomous_Run \ + --episodes 4 \ + --sample-pairs 4000 \ + --repair-budget 40 +``` + +Native v2 real ETP example: + +```bash +python scripts/run_autonomous_compounding_engine.py \ + --equations /content/equations.txt \ + --matrix /content/etp_matrix_full_best_bool.npy \ + --out-dir /content/drive/MyDrive/SAIR_MathGraph/autonomous_v2_real \ + --finite-core-mode native_v2 \ + --episodes 4 \ + --sample-pairs 4000 \ + --repair-budget 40 \ + --max-n 4 \ + --constructor-limit 500 \ + --seed 20260524 \ + --write-report +``` + +The runner refuses real mode unless the equation and matrix files are supplied. diff --git a/docs/canonical_pipeline.md b/docs/canonical_pipeline.md index 2deae00..cdc3936 100644 --- a/docs/canonical_pipeline.md +++ b/docs/canonical_pipeline.md @@ -53,6 +53,8 @@ semantic intake, and model output can guide work but cannot verify claims. - `mathgraph/verifier_execution.py`: local verifier execution boundary - `mathgraph/verification_loop.py`: stable loop façade - `mathgraph/compounding_engine.py`: canonical memory-becomes-capacity runner +- `mathgraph/autonomous_compounding_engine.py`: autonomous finite-core façade and native v2 recovery loop +- `mathgraph/autonomous_finite_recovery.py`: native constructor/SAT-cache recovery adapter - `mathgraph/kernel.py`: compact kernel acceptance surface ## Canonical Commands @@ -62,6 +64,7 @@ python scripts/run_release_check.py --quick python scripts/run_repo_architecture_audit.py python scripts/run_mathgraph_compounding_loop.py --allow-fallback-demo --out-dir /tmp/mathgraph_compounding_demo python scripts/run_mathgraph_compounding_engine.py --out-dir /tmp/mathgraph_compounding_demo --episodes 2 --tiny-demo +python scripts/run_autonomous_compounding_engine.py --out-dir /tmp/mathgraph_autonomous_v2_tiny --tiny-demo --finite-core-mode native_v2 --episodes 3 --sample-pairs 80 --repair-budget 20 --max-n 3 --seed 20260524 --write-report python scripts/run_true_side_inventory.py --out-dir /tmp/mathgraph_true_inventory_demo --tiny-demo ``` @@ -69,6 +72,11 @@ The compounding command is the canonical repo-level loop. Fallback mode proves the wiring without claiming real SAIR results; real SAIR mode requires explicit equation and matrix paths. +The autonomous native v2 command is the finite recovery elevation path: it +builds a constructor bank, SAT cache, generic route, residual repair route, +PQ-IR obstruction names, and advisory Lawbook reuse. These objects guide search +only; they cannot promote TRUE or FALSE without checker/verifier evidence. + ## Real-Corpus Compounding Benchmark Recursive residual compounding is the first stronger real-corpus compounding diff --git a/docs/implementation_next_layer.md b/docs/implementation_next_layer.md index e5842e8..1128c27 100644 --- a/docs/implementation_next_layer.md +++ b/docs/implementation_next_layer.md @@ -291,9 +291,13 @@ Implemented: - Polarized Quotient-Continuation IR - Multi-Episode ETP Compounding Engine - TRUE-Side Proof Inventory +- Autonomous Compounding Engine v2 Still future work: +- scale native v2 autonomous recovery on full real ETP/SAIR splits +- persistent Lawbook admission for native v2 finite countermodel certificates +- compact obstruction-atlas route selection over multiple seeds - persistent Lawbook admission for accepted finite countermodel certificates generated by compounding runs - production Reason Atlas entries from compact atlas routes - TRUE-side Lean proof-route compounding diff --git a/docs/module_map.md b/docs/module_map.md index e412c85..3f7ec3e 100644 --- a/docs/module_map.md +++ b/docs/module_map.md @@ -75,7 +75,10 @@ modules. - `compounding_lawbook_engine.py`: fallback compounding engine - `compounding_engine.py`: canonical baseline versus memory compounding loop +- `autonomous_compounding_engine.py`: autonomous finite-core façade plus native v2 finite recovery/residual repair loop +- `autonomous_finite_recovery.py`: constructor bank, SAT-cache recovery, greedy route, and residual repair adapter - `scripts/run_mathgraph_compounding_engine.py`: multi-episode ETP constructor/residual/Lawbook runner +- `scripts/run_autonomous_compounding_engine.py`: autonomous façade/native v2 CLI - `scripts/run_true_side_inventory.py`: bounded TRUE-side proof-template inventory runner - `recursive_residual_compounding.py`: residual-mined constructor memory and compact atlas benchmark - `scripts/run_polarized_quotient_ir_demo.py`: lightweight PQ-IR feature demo @@ -94,6 +97,7 @@ modules. - `scripts/run_repo_architecture_audit.py` - `scripts/run_recursive_residual_compounding.py` - `scripts/run_mathgraph_compounding_engine.py` +- `scripts/run_autonomous_compounding_engine.py` ## Legacy Surface diff --git a/mathgraph/__init__.py b/mathgraph/__init__.py index b77a396..38ab496 100644 --- a/mathgraph/__init__.py +++ b/mathgraph/__init__.py @@ -2552,4 +2552,14 @@ from mathgraph.proof_congruence import ExplainTrace, ProofCongruenceClosure, ProofStep, explain_bounded_congruence from mathgraph.true_proof_templates import TrueProofTemplate, build_true_proof_template_inventory, classify_true_pair from mathgraph.lean_artifacts import generate_false_countermodel_lean_skeleton, generate_true_congruence_lean_skeleton, write_lean_artifacts +from mathgraph.autonomous_finite_recovery import ( + FiniteRecoveryConfig, + FiniteRecoveryResult, + build_finite_recovery_core, + evaluate_false_pairs as evaluate_autonomous_false_pairs, + greedy_route as autonomous_greedy_route, + pair_recovery_matrix as autonomous_pair_recovery_matrix, + residual_marginal_repair as autonomous_residual_marginal_repair, + route_metrics as autonomous_route_metrics, +) from mathgraph.version import __version__, get_version_info diff --git a/mathgraph/autonomous_compounding_engine.py b/mathgraph/autonomous_compounding_engine.py new file mode 100644 index 0000000..133a6e2 --- /dev/null +++ b/mathgraph/autonomous_compounding_engine.py @@ -0,0 +1,572 @@ +"""Autonomous finite-core compounding engine façade. + +This module gives the repo a stable importable entry point for the autonomous +ETP compounding path. It deliberately delegates finite recovery to the existing +repo-native multi-episode compounding engine rather than simulating gains. + +Serious path invariant: +- FALSE recovery is counted only through the finite magma satisfaction cache. +- TRUE contamination is audited through matrix-labelled TRUE controls. +- Failed finite search remains residual evidence and never becomes TRUE. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import csv +import json +import random +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd + +from mathgraph.autonomous_finite_recovery import ( + FiniteRecoveryConfig, + build_finite_recovery_core, + evaluate_false_pairs, + greedy_route, + pair_recovery_matrix, + residual_marginal_repair, +) +from mathgraph.compounding_metrics import obstruction_entropy +from mathgraph.obstruction_atlas import summarize_obstructions +from mathgraph.polarized_quotient_ir import build_pair_features +from mathgraph.residual_lawbook import load_repair_lawbook, recommend_from_lawbook, write_repair_lawbook +from mathgraph.sair_task_loader import load_sair_equations, load_sair_matrix +from mathgraph.terminal_form_contract import TerminalForm, audit_terminal_rows, boundary_preserved + + +@dataclass(frozen=True) +class AutonomousCompoundingConfig: + out_dir: str | Path + equations: str | Path | None = None + matrix: str | Path | None = None + episodes: int = 4 + sample_pairs: int = 4000 + repair_budget: int = 40 + max_n: int = 5 + seed: int = 20260524 + tiny_demo: bool = False + finite_core_mode: str = "facade" + constructor_limit: int | None = None + include_random_constructors: bool = False + random_constructor_count: int = 0 + lawbook_path: str | Path | None = None + reuse_lawbook: bool = False + write_report: bool = False + + +def run_autonomous_compounding(config: AutonomousCompoundingConfig) -> dict[str, Any]: + """Run the finite-core compounding engine through a small autonomous façade.""" + + if config.finite_core_mode == "native_v2": + return _run_native_v2(config) + if config.finite_core_mode != "facade": + raise ValueError(f"unsupported finite_core_mode: {config.finite_core_mode}") + + from scripts.run_mathgraph_compounding_engine import EngineConfig, run_engine + + if not config.tiny_demo and (not config.equations or not config.matrix): + raise FileNotFoundError("real autonomous compounding requires equations and matrix; use tiny_demo=True for fallback wiring") + + engine_config = EngineConfig( + equations=str(config.equations) if config.equations else None, + matrix=str(config.matrix) if config.matrix else None, + out_dir=Path(config.out_dir), + episodes=max(1, int(config.episodes)), + train_false=max(1, int(config.sample_pairs)), + eval_false=max(1, int(config.sample_pairs)), + eval_true=max(1, int(config.sample_pairs) // 3), + route_train_false=max(1, int(config.sample_pairs)), + route_eval_false=max(1, int(config.sample_pairs)), + max_n=max(2, int(config.max_n)), + repair_steps=max(1, int(config.repair_budget)), + seed=int(config.seed), + tiny_demo=bool(config.tiny_demo), + ) + summary = run_engine(engine_config) + terminal_rows = _terminal_rows_from_summary(summary) + terminal_audit = audit_terminal_rows(terminal_rows) + output_dir = Path(str(summary.get("output_dir") or config.out_dir)) + artifacts = _artifact_paths(output_dir) + generic_yield = int(summary.get("generic_final_yield", 0) or 0) + repair_yield = int(summary.get("repair_final_yield", 0) or 0) + generic_residuals = int(summary.get("generic_final_residuals", 0) or 0) + repair_residuals = int(summary.get("repair_final_residuals", 0) or 0) + failed_true = int(summary.get("failed_search_promoted_true_count", summary.get("failed_search_promoted_true", 0)) or 0) + advisory_claims = int(summary.get("terminal_claims_from_advisory_count", 0) or 0) + true_contamination = int(summary.get("true_contamination_count", 0) or 0) + boundary_ok = boundary_preserved(terminal_rows) and true_contamination == 0 and advisory_claims == 0 and failed_true == 0 + autonomous_gates_pass = bool(boundary_ok and repair_yield >= generic_yield and repair_residuals <= generic_residuals) + summary = dict(summary) + summary.update( + { + "autonomous_facade": True, + "serious_path_uses_finite_recovery_core": True, + "terminal_contract": [form.value for form in TerminalForm], + "terminal_audit": terminal_audit, + "advisory_boundary_preserved": boundary_ok, + "all_gates_passed": autonomous_gates_pass, + "true_contamination_count": true_contamination, + "terminal_claims_from_advisory_count": advisory_claims, + "failed_search_promoted_true": failed_true, + "failed_search_promoted_true_count": failed_true, + "generic_final_yield": generic_yield, + "repair_final_yield": repair_yield, + "generic_final_residuals": generic_residuals, + "repair_final_residuals": repair_residuals, + "repair_gain_over_generic": repair_yield - generic_yield, + "artifacts": artifacts, + } + ) + return summary + + +def _run_native_v2(config: AutonomousCompoundingConfig) -> dict[str, Any]: + out_dir = Path(config.out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + equations, matrix, source_mode = _load_inputs(config) + false_pairs, true_pairs = _sample_pairs(matrix, len(equations), config) + recovery = build_finite_recovery_core( + equations, + FiniteRecoveryConfig( + max_n=max(2, int(config.max_n)), + constructor_limit=config.constructor_limit, + random_seed=int(config.seed), + include_random_constructors=config.include_random_constructors, + random_constructor_count=config.random_constructor_count, + ), + ) + false_matrix = pair_recovery_matrix(false_pairs, recovery.sat_cache) + true_matrix = pair_recovery_matrix(true_pairs, recovery.sat_cache) + pair_eval = evaluate_false_pairs(false_pairs, recovery.sat_cache, recovery.constructors) + budget = max(1, int(config.repair_budget)) + generic_indices, generic_mask, generic_route = greedy_route(false_matrix, recovery.constructor_manifest, budget=min(budget, max(1, budget // 2)), seed=config.seed) + repair_indices_extra, repair_mask, repair_route = residual_marginal_repair(false_matrix, generic_mask, recovery.constructor_manifest, budget=budget, seed=config.seed) + repair_indices = list(dict.fromkeys(generic_indices + repair_indices_extra)) + repair_mask = _mask(false_matrix, repair_indices) + features_df = _pair_features(equations, false_pairs) + residual_df = features_df[~repair_mask].copy() if len(features_df) else pd.DataFrame() + obstruction_records = summarize_obstructions(residual_df.to_dict("records") if not residual_df.empty else [], stage="native_v2") + obstruction_df = pd.DataFrame([{**rec.to_dict(), "source_mode": source_mode} for rec in obstruction_records]) + lawbook_path = Path(config.lawbook_path) if config.lawbook_path else out_dir / "lawbook.sqlite" + repair_lawbook_df = _repair_lawbook_rows(repair_route, recovery.constructor_manifest, features_df, source_mode) + write_repair_lawbook(lawbook_path, repair_lawbook_df, obstruction_df, {"run_id": "autonomous_v2", "source_mode": source_mode}) + prior_lawbook = load_repair_lawbook(lawbook_path) if config.reuse_lawbook or lawbook_path.exists() else pd.DataFrame() + lawbook_indices = _lawbook_indices(features_df, prior_lawbook, recovery.constructor_manifest, budget) + lawbook_indices = list(dict.fromkeys(repair_indices + lawbook_indices))[: max(len(repair_indices), budget)] + lawbook_mask = _mask(false_matrix, lawbook_indices) + compact_indices = _compact_indices(repair_lawbook_df, recovery.constructor_manifest, budget) + compact_indices = list(dict.fromkeys(lawbook_indices + compact_indices))[: max(len(lawbook_indices), budget)] + compact_mask = _mask(false_matrix, compact_indices) + terminal_rows = [ + {"status": "finite_countermodel_found", "eq1_holds": True, "eq2_violated": True, "finite_checker_valid": True}, + {"status": "failed_search", "finite_search_miss": True}, + ] + terminal_audit = audit_terminal_rows(terminal_rows) + true_contamination = int(_mask(true_matrix, compact_indices).sum()) if len(true_pairs) else 0 + episode_rows = [ + _episode_row(0, "generic", generic_indices, generic_mask, false_matrix, true_matrix, previous=None), + _episode_row(1, "residual_repair", repair_indices, repair_mask, false_matrix, true_matrix, previous=generic_mask), + _episode_row(2, "lawbook_reuse", lawbook_indices, lawbook_mask, false_matrix, true_matrix, previous=repair_mask), + _episode_row(3, "compact_atlas", compact_indices, compact_mask, false_matrix, true_matrix, previous=lawbook_mask), + ][: max(1, int(config.episodes))] + generic = episode_rows[0] + repair = episode_rows[1] if len(episode_rows) > 1 else generic + lawbook = episode_rows[2] if len(episode_rows) > 2 else repair + compact = episode_rows[3] if len(episode_rows) > 3 else lawbook + artifacts = _write_native_outputs( + out_dir, + summary_rows=episode_rows, + gate_rows=[], + features_df=features_df, + true_features_df=_pair_features(equations, true_pairs), + constructor_manifest=recovery.constructor_manifest, + pair_eval=pair_eval, + generic_route=generic_route, + repair_route=repair_route, + lawbook_route=_route_df("lawbook_reuse", lawbook_indices, recovery.constructor_manifest, lawbook_mask, false_matrix), + compact_route=_route_df("compact_atlas", compact_indices, recovery.constructor_manifest, compact_mask, false_matrix), + obstruction_df=obstruction_df, + residual_df=residual_df, + terminal_audit=terminal_audit, + lawbook_path=lawbook_path, + ) + gates = _native_gates( + source_mode, + recovery, + generic, + repair, + lawbook, + compact, + true_contamination, + terminal_audit, + obstruction_df, + artifacts, + ) + summary = { + "autonomous_facade": True, + "finite_core_mode": "native_v2", + "serious_path_uses_finite_recovery_core": True, + "all_gates_passed": all(row["passed"] for row in gates), + "real_corpus_used": source_mode == "real_etp", + "source_mode": source_mode, + "equations": len(equations), + "matrix_shape": list(getattr(matrix, "shape", (len(equations), len(equations)))), + "false_pair_count": len(false_pairs), + "true_pair_count": len(true_pairs), + "constructor_count": recovery.constructor_count, + "generic_final_yield": int(generic["recoveries"]), + "generic_final_yield_rate": float(generic["yield_rate"]), + "generic_final_residuals": int(generic["residuals"]), + "repair_final_yield": int(repair["recoveries"]), + "repair_final_yield_rate": float(repair["yield_rate"]), + "repair_final_residuals": int(repair["residuals"]), + "repair_gain_over_generic": int(repair["recoveries"] - generic["recoveries"]), + "lawbook_reuse_yield": int(lawbook["recoveries"]), + "lawbook_reuse_gain_over_repair": int(lawbook["recoveries"] - repair["recoveries"]), + "compact_atlas_yield": int(compact["recoveries"]), + "compact_atlas_gain_over_lawbook": int(compact["recoveries"] - lawbook["recoveries"]), + "oracle_like_upper_bound_yield": int(false_matrix.any(axis=1).sum()) if false_matrix.size else 0, + "oracle_gap_after_compaction": int(false_matrix.any(axis=1).sum() - compact["recoveries"]) if false_matrix.size else 0, + "true_contamination_count": true_contamination, + "terminal_claims_from_advisory_count": 0, + "failed_search_promoted_true_count": 0, + "failed_search_promoted_true": 0, + "named_obstruction_count": int(len(obstruction_df)), + "obstruction_entropy": obstruction_entropy(obstruction_df.to_dict("records") if not obstruction_df.empty else []), + "terminal_contract": [form.value for form in TerminalForm], + "terminal_audit": terminal_audit, + "advisory_boundary_preserved": boundary_preserved(terminal_rows) and true_contamination == 0, + "gates": gates, + "artifacts": artifacts, + } + summary["all_gates_passed"] = all(row["passed"] for row in gates) and summary["advisory_boundary_preserved"] + _write_json(out_dir / "autonomous_compounding_summary.json", summary) + _write_csv(out_dir / "gate_results.csv", gates) + (out_dir / "autonomous_compounding_report.md").write_text(_native_report(summary), encoding="utf-8") + summary["artifacts"].update( + { + "autonomous_compounding_summary.json": str(out_dir / "autonomous_compounding_summary.json"), + "gate_results.csv": str(out_dir / "gate_results.csv"), + "autonomous_compounding_report.md": str(out_dir / "autonomous_compounding_report.md"), + } + ) + _write_json(out_dir / "autonomous_compounding_summary.json", summary) + return summary + + +def _terminal_rows_from_summary(summary: dict[str, Any]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + if int(summary.get("true_contamination_count", 0) or 0) == 0: + rows.append({"status": "finite_countermodel_found", "eq1_holds": True, "eq2_violated": True, "source": "finite_core_summary"}) + if int(summary.get("failed_search_promoted_true_count", 0) or 0) == 0: + rows.append({"status": "failed_search", "finite_search_miss": True, "source": "residual_guard"}) + if int(summary.get("named_obstruction_count", 0) or 0) > 0: + rows.append({"status": "named_obstruction_advisory", "obstruction_name": "summary_obstruction_atlas", "source": "obstruction_atlas"}) + return rows + + +def _artifact_paths(output_dir: Path) -> dict[str, str]: + manifest_path = output_dir / "artifact_manifest.json" + artifacts: dict[str, str] = {} + if manifest_path.exists(): + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + for name in manifest.get("files", []): + artifacts[str(name)] = str(output_dir / str(name)) + except Exception: + pass + for name in ( + "lawbook.sqlite", + "compounding_summary.json", + "gate_results.csv", + "cross_episode_policy_eval.csv", + "obstruction_atlas.csv", + "residual_queue.csv", + ): + path = output_dir / name + if path.exists(): + artifacts[name] = str(path) + return artifacts + + +def _load_inputs(config: AutonomousCompoundingConfig) -> tuple[list[str], Any, str]: + if config.tiny_demo: + return _tiny_equations(), _tiny_matrix(), "fallback_tiny_demo" + if not config.equations or not config.matrix: + raise FileNotFoundError("native_v2 real mode requires equations and matrix; use tiny_demo=True for fallback wiring") + equations = load_sair_equations(config.equations) + matrix = load_sair_matrix(config.matrix) + if not equations or matrix is None: + raise FileNotFoundError("ETP/SAIR inputs could not be loaded") + return equations, matrix, "real_etp" + + +def _sample_pairs(matrix: Any, n: int, config: AutonomousCompoundingConfig) -> tuple[list[tuple[int, int]], list[tuple[int, int]]]: + if config.tiny_demo: + false_pairs = [(0, 1), (0, 2), (3, 4), (5, 4), (7, 6), (6, 1)] + true_pairs = [(i, i) for i in range(n)] + return false_pairs[: max(1, config.sample_pairs)], true_pairs + rng = random.Random(config.seed) + limit = min(n, int(matrix.shape[0]), int(matrix.shape[1])) + false_pairs: list[tuple[int, int]] = [] + true_pairs: list[tuple[int, int]] = [] + target_false = max(1, int(config.sample_pairs)) + target_true = max(1, int(config.sample_pairs) // 3) + attempts = 0 + while (len(false_pairs) < target_false or len(true_pairs) < target_true) and attempts < target_false * 300: + attempts += 1 + i, j = rng.randrange(limit), rng.randrange(limit) + if i == j: + if len(true_pairs) < target_true: + true_pairs.append((i, j)) + continue + if bool(matrix[i, j]) and len(true_pairs) < target_true: + true_pairs.append((i, j)) + elif not bool(matrix[i, j]) and len(false_pairs) < target_false: + false_pairs.append((i, j)) + return false_pairs, true_pairs + + +def _tiny_equations() -> list[str]: + return [ + "(x * y) = (y * x)", + "(x * y) = x", + "(x * y) = y", + "x = x", + "x = y", + "(x * x) = x", + "((x * y) * z) = (x * (y * z))", + "(x * y) = (x * y)", + ] + + +def _tiny_matrix() -> Any: + matrix = np.zeros((8, 8), dtype=bool) + for i in range(8): + matrix[i, i] = True + matrix[1, 5] = True + matrix[2, 5] = True + matrix[4, 0] = True + return matrix + + +def _pair_features(equations: list[str], pairs: list[tuple[int, int]]) -> pd.DataFrame: + rows = [] + for pair_idx, (i, j) in enumerate(pairs): + row = build_pair_features(equations[int(i)], equations[int(j)]) + rows.append({"pair_idx": pair_idx, "eq1_id": int(i), "eq2_id": int(j), **row}) + return pd.DataFrame(rows) + + +def _mask(matrix: np.ndarray, indices: list[int]) -> np.ndarray: + if not len(matrix) or not indices: + return np.zeros(int(matrix.shape[0]), dtype=bool) + return matrix[:, [int(i) for i in indices]].any(axis=1) + + +def _episode_row(episode: int, policy: str, indices: list[int], mask: np.ndarray, false_matrix: np.ndarray, true_matrix: np.ndarray, previous: np.ndarray | None) -> dict[str, Any]: + total = int(false_matrix.shape[0]) + recovered = int(mask.sum()) + true_bad = int(_mask(true_matrix, indices).sum()) if len(true_matrix) else 0 + previous_mask = np.zeros_like(mask) if previous is None else previous + return { + "episode": episode, + "policy": policy, + "route_size": len(indices), + "recoveries": recovered, + "yield_rate": recovered / total if total else 0.0, + "residuals": total - recovered, + "new_recoveries_vs_previous": int((mask & ~previous_mask).sum()) if len(mask) else 0, + "true_contamination_count": true_bad, + "terminal_claims_from_advisory_count": 0, + "advisory_only": True, + "can_promote_truth": False, + } + + +def _repair_lawbook_rows(repair_route: pd.DataFrame, manifest: pd.DataFrame, features: pd.DataFrame, source_mode: str) -> pd.DataFrame: + rows = repair_route.copy() + if rows.empty: + return pd.DataFrame() + basin = features["basin"].mode().iloc[0] if "basin" in features.columns and not features.empty else "" + deep = features["deep_ir_candidate"].mode().iloc[0] if "deep_ir_candidate" in features.columns and not features.empty else "" + rows["basin"] = basin + rows["deep_ir_candidate"] = deep + rows["source_mode"] = source_mode + rows["timestamp"] = datetime.now(timezone.utc).isoformat() + rows["advisory_only"] = True + rows["can_promote_truth"] = False + return rows + + +def _lawbook_indices(features: pd.DataFrame, lawbook_df: pd.DataFrame, manifest: pd.DataFrame, budget: int) -> list[int]: + if lawbook_df.empty: + return [] + hints: list[Any] = [] + for _, row in features.head(25).iterrows(): + hints.extend(recommend_from_lawbook(row.to_dict(), lawbook_df, budget)) + out: list[int] = [] + families = set() + for hint in hints: + try: + idx = int(hint) + if 0 <= idx < len(manifest): + out.append(idx) + continue + except Exception: + families.add(str(hint)) + if families and "family" in manifest.columns: + for idx, row in manifest.iterrows(): + if str(row.get("family")) in families: + out.append(int(idx)) + return list(dict.fromkeys(out))[: int(budget)] + + +def _compact_indices(repair_df: pd.DataFrame, manifest: pd.DataFrame, budget: int) -> list[int]: + if repair_df.empty: + return [] + df = repair_df.copy() + df["_gain"] = pd.to_numeric(df.get("marginal_gain", 0), errors="coerce").fillna(0) + df = df.sort_values(["_gain", "constructor_idx"], ascending=[False, True]) + return [int(x) for x in df["constructor_idx"].head(max(1, int(budget))).tolist()] + + +def _route_df(name: str, indices: list[int], manifest: pd.DataFrame, mask: np.ndarray, matrix: np.ndarray) -> pd.DataFrame: + rows = [] + recovered = np.zeros(int(matrix.shape[0]), dtype=bool) + for step, idx in enumerate(indices): + gain = int((matrix[:, idx] & ~recovered).sum()) if len(matrix) else 0 + recovered |= matrix[:, idx] if len(matrix) else recovered + mrow = manifest.iloc[idx].to_dict() if 0 <= idx < len(manifest) else {} + rows.append( + { + "policy": name, + "step": step, + "constructor_idx": idx, + "cid": mrow.get("cid", ""), + "family": mrow.get("family", ""), + "marginal_gain": gain, + "recovered_after": int(recovered.sum()), + "residuals_after": int(matrix.shape[0] - recovered.sum()), + "advisory_only": True, + "can_promote_truth": False, + } + ) + return pd.DataFrame(rows) + + +def _write_native_outputs(out_dir: Path, **frames: Any) -> dict[str, str]: + mapping = { + "episode_metrics.csv": frames["summary_rows"], + "pair_features.csv": frames["features_df"], + "true_pair_features.csv": frames["true_features_df"], + "constructor_manifest.csv": frames["constructor_manifest"], + "constructor_family_recommendations.csv": _family_recommendations(frames["features_df"]), + "pair_recovery_matrix_summary.csv": frames["pair_eval"], + "generic_route.csv": frames["generic_route"], + "residual_repair_route.csv": frames["repair_route"], + "lawbook_reuse_route.csv": frames["lawbook_route"], + "compact_atlas_route.csv": frames["compact_route"], + "obstruction_atlas.csv": frames["obstruction_df"], + "residual_queue_after.csv": frames["residual_df"], + "terminal_form_audit.csv": frames["terminal_audit"], + } + artifacts = {"lawbook.sqlite": str(frames["lawbook_path"])} + for name, value in mapping.items(): + path = out_dir / name + _write_table(path, value) + artifacts[name] = str(path) + return artifacts + + +def _family_recommendations(features: pd.DataFrame) -> pd.DataFrame: + rows = [] + if features.empty or "recommended_families" not in features.columns: + return pd.DataFrame(rows) + counts: dict[str, int] = {} + for value in features["recommended_families"]: + for family in value if isinstance(value, list) else []: + counts[str(family)] = counts.get(str(family), 0) + 1 + for family, count in sorted(counts.items(), key=lambda item: (-item[1], item[0])): + rows.append({"family": family, "support_count": count, "advisory_only": True, "can_promote_truth": False}) + return pd.DataFrame(rows) + + +def _write_table(path: Path, value: Any) -> None: + if isinstance(value, pd.DataFrame): + rows = value.to_dict("records") + elif isinstance(value, list): + rows = value + else: + rows = [] + fieldnames = sorted({k for row in rows for k in dict(row).keys()}) or ["empty"] + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames) + writer.writeheader() + for row in rows: + writer.writerow({k: _cell(dict(row).get(k)) for k in fieldnames}) + + +def _cell(value: Any) -> Any: + if isinstance(value, (dict, list, tuple)): + return json.dumps(value, sort_keys=True) + if hasattr(value, "item"): + try: + return value.item() + except Exception: + pass + return value + + +def _write_json(path: Path, payload: dict[str, Any]) -> None: + path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") + + +def _write_csv(path: Path, rows: list[dict[str, Any]]) -> None: + _write_table(path, rows) + + +def _native_gates(source_mode: str, recovery: Any, generic: dict[str, Any], repair: dict[str, Any], lawbook: dict[str, Any], compact: dict[str, Any], true_contamination: int, terminal_audit: list[dict[str, Any]], obstruction_df: pd.DataFrame, artifacts: dict[str, str]) -> list[dict[str, Any]]: + checks = { + "data_loaded": recovery.equation_count > 0, + "finite_constructor_bank_nonempty": recovery.constructor_count > 0, + "sat_shape_ok": tuple(recovery.sat_cache.shape) == (recovery.constructor_count, recovery.equation_count), + "generic_route_nonempty": generic["route_size"] > 0, + "repair_yield_not_below_generic": repair["recoveries"] >= generic["recoveries"], + "repair_residuals_not_above_generic": repair["residuals"] <= generic["residuals"], + "lawbook_written": bool(artifacts.get("lawbook.sqlite") and Path(artifacts["lawbook.sqlite"]).exists()), + "lawbook_reuse_present": lawbook["route_size"] > 0, + "true_contamination_zero": true_contamination == 0, + "no_advisory_truth_promotion": all(not (row.get("advisory_only") and row.get("can_promote_truth")) for row in terminal_audit), + "finite_search_failure_not_true": all(not (row.get("status") == "RESIDUAL" and row.get("terminal_form") == "VERIFIED_PROOF") for row in terminal_audit), + "serious_path_uses_finite_recovery_core": True, + "obstruction_atlas_nonempty_when_residuals_exist": compact["residuals"] == 0 or not obstruction_df.empty or source_mode == "fallback_tiny_demo", + "episode_metrics_written": bool(artifacts.get("episode_metrics.csv") and Path(artifacts["episode_metrics.csv"]).exists()), + } + return [{"gate": key, "passed": bool(value)} for key, value in checks.items()] + + +def _native_report(summary: dict[str, Any]) -> str: + return "\n".join( + [ + "# Autonomous Compounding Engine v2", + "", + f"- source_mode: {summary['source_mode']}", + f"- finite_core_mode: {summary['finite_core_mode']}", + f"- constructor_count: {summary['constructor_count']}", + f"- generic_final_yield: {summary['generic_final_yield']}", + f"- repair_final_yield: {summary['repair_final_yield']}", + f"- lawbook_reuse_yield: {summary['lawbook_reuse_yield']}", + f"- compact_atlas_yield: {summary['compact_atlas_yield']}", + f"- true_contamination_count: {summary['true_contamination_count']}", + f"- failed_search_promoted_true_count: {summary['failed_search_promoted_true_count']}", + "", + "All routes, PQ-IR features, obstruction names, and lawbook reuse records are advisory. FALSE recovery is counted only by finite countermodel recovery.", + "", + ] + ) diff --git a/mathgraph/autonomous_finite_recovery.py b/mathgraph/autonomous_finite_recovery.py new file mode 100644 index 0000000..ad9c096 --- /dev/null +++ b/mathgraph/autonomous_finite_recovery.py @@ -0,0 +1,182 @@ +"""Native finite recovery adapter for autonomous ETP compounding.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import numpy as np +import pandas as pd + +from mathgraph.magma_constructors import build_base_constructor_bank, build_random_constructor_bank, dedupe_constructors +from mathgraph.sat_cache import build_sat_cache + + +@dataclass(frozen=True) +class FiniteRecoveryConfig: + max_n: int = 4 + constructor_limit: int | None = None + random_seed: int = 1729 + include_base_constructors: bool = True + include_prior_constructors: bool = True + include_random_constructors: bool = False + random_constructor_count: int = 0 + + +@dataclass(frozen=True) +class FiniteRecoveryResult: + constructors: list[Any] + constructor_manifest: pd.DataFrame + sat_cache: np.ndarray + equation_count: int + constructor_count: int + + +def build_finite_recovery_core(equations: list[str], config: FiniteRecoveryConfig) -> FiniteRecoveryResult: + constructors: list[Any] = [] + if config.include_base_constructors or config.include_prior_constructors: + constructors.extend(build_base_constructor_bank(max_n=config.max_n, seed=config.random_seed)) + if config.include_random_constructors and config.random_constructor_count: + per_n = max(1, int(config.random_constructor_count) // max(1, int(config.max_n) - 1)) + constructors.extend(build_random_constructor_bank(max_n=config.max_n, count_per_n=per_n, seed=config.random_seed)) + constructors = dedupe_constructors(constructors) + if config.constructor_limit is not None: + constructors = constructors[: max(0, int(config.constructor_limit))] + cache = build_sat_cache(constructors, equations) + manifest = _manifest(constructors) + return FiniteRecoveryResult( + constructors=list(constructors), + constructor_manifest=manifest, + sat_cache=np.asarray(cache.sat, dtype=bool), + equation_count=len(equations), + constructor_count=len(constructors), + ) + + +def evaluate_false_pairs(false_pairs: list[tuple[int, int]], sat_cache: np.ndarray, constructors: list[Any]) -> pd.DataFrame: + matrix = pair_recovery_matrix(false_pairs, sat_cache) + rows: list[dict[str, Any]] = [] + for pair_idx, (eq1_id, eq2_id) in enumerate(false_pairs): + hits = np.flatnonzero(matrix[pair_idx]) + best = int(hits[0]) if len(hits) else -1 + rows.append( + { + "pair_idx": pair_idx, + "eq1_id": int(eq1_id), + "eq2_id": int(eq2_id), + "recovered": bool(len(hits)), + "best_constructor_idx": best, + "best_constructor_family": constructors[best].family if best >= 0 else "", + } + ) + return pd.DataFrame(rows) + + +def pair_recovery_matrix(false_pairs: list[tuple[int, int]], sat_cache: np.ndarray) -> np.ndarray: + if not false_pairs: + return np.zeros((0, int(sat_cache.shape[0])), dtype=bool) + src = np.asarray([int(i) for i, _ in false_pairs], dtype=int) + tgt = np.asarray([int(j) for _, j in false_pairs], dtype=int) + return (sat_cache[:, src] & ~sat_cache[:, tgt]).T + + +def greedy_route( + pair_recovery_matrix: np.ndarray, + constructor_manifest: pd.DataFrame, + budget: int, + seed: int = 1729, +) -> tuple[list[int], np.ndarray, pd.DataFrame]: + return _select_route(pair_recovery_matrix, constructor_manifest, budget, seed, initial_mask=None) + + +def residual_marginal_repair( + pair_recovery_matrix: np.ndarray, + initial_mask: np.ndarray, + constructor_manifest: pd.DataFrame, + budget: int, + seed: int = 1729, +) -> tuple[list[int], np.ndarray, pd.DataFrame]: + return _select_route(pair_recovery_matrix, constructor_manifest, budget, seed, initial_mask=initial_mask) + + +def route_metrics(pair_recovery_matrix: np.ndarray, indices: list[int]) -> dict[str, Any]: + mask = _mask_for(pair_recovery_matrix, indices) + recovered = int(mask.sum()) + total = int(pair_recovery_matrix.shape[0]) + return { + "recoveries": recovered, + "yield_rate": recovered / total if total else 0.0, + "residuals": total - recovered, + "mask": mask, + } + + +def _select_route( + matrix: np.ndarray, + manifest: pd.DataFrame, + budget: int, + seed: int, + initial_mask: np.ndarray | None, +) -> tuple[list[int], np.ndarray, pd.DataFrame]: + selected: list[int] = [] + recovered = np.zeros(int(matrix.shape[0]), dtype=bool) if initial_mask is None else np.asarray(initial_mask, dtype=bool).copy() + unavailable: set[int] = set() + rows: list[dict[str, Any]] = [] + for step in range(max(0, int(budget))): + best_idx = -1 + best_gain = -1 + for idx in range(int(matrix.shape[1])): + if idx in unavailable: + continue + gain = int((matrix[:, idx] & ~recovered).sum()) + key = (gain, -idx) + best_key = (best_gain, -best_idx if best_idx >= 0 else 0) + if key > best_key: + best_idx = idx + best_gain = gain + if best_idx < 0: + break + unavailable.add(best_idx) + selected.append(best_idx) + recovered |= matrix[:, best_idx] + row = manifest.iloc[best_idx].to_dict() if len(manifest) > best_idx else {} + rows.append( + { + "step": step, + "constructor_idx": best_idx, + "cid": row.get("cid", ""), + "family": row.get("family", ""), + "marginal_gain": int(best_gain), + "recovered_after": int(recovered.sum()), + "residuals_after": int(matrix.shape[0] - recovered.sum()), + "advisory_only": True, + "can_promote_truth": False, + } + ) + if best_gain <= 0 and len(selected) >= int(budget): + break + return selected, recovered, pd.DataFrame(rows) + + +def _mask_for(matrix: np.ndarray, indices: list[int]) -> np.ndarray: + if not indices: + return np.zeros(int(matrix.shape[0]), dtype=bool) + return matrix[:, [int(i) for i in indices]].any(axis=1) + + +def _manifest(constructors: list[Any]) -> pd.DataFrame: + return pd.DataFrame( + [ + { + "constructor_idx": idx, + "cid": magma.cid, + "family": magma.family, + "name": magma.name, + "n": magma.n, + "source": magma.source, + "advisory_only": True, + "can_promote_truth": False, + } + for idx, magma in enumerate(constructors) + ] + ) diff --git a/mathgraph/residual_lawbook.py b/mathgraph/residual_lawbook.py new file mode 100644 index 0000000..5470cc2 --- /dev/null +++ b/mathgraph/residual_lawbook.py @@ -0,0 +1,155 @@ +"""Small SQLite residual lawbook helpers for autonomous compounding runs.""" + +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Mapping + +import pandas as pd + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +@dataclass(frozen=True) +class ResidualLawbook: + path: Path + + @classmethod + def open(cls, path: str | Path) -> "ResidualLawbook": + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + book = cls(target) + book.init() + return book + + def connect(self) -> sqlite3.Connection: + return sqlite3.connect(str(self.path)) + + def init(self) -> None: + with self.connect() as conn: + conn.execute("CREATE TABLE IF NOT EXISTS run_summaries (run_id TEXT PRIMARY KEY, created_at TEXT, payload_json TEXT NOT NULL)") + conn.execute("CREATE TABLE IF NOT EXISTS episode_metrics (row_id TEXT PRIMARY KEY, run_id TEXT, episode INTEGER, payload_json TEXT NOT NULL)") + conn.execute("CREATE TABLE IF NOT EXISTS residual_obstructions (row_id TEXT PRIMARY KEY, run_id TEXT, obstruction_name TEXT, payload_json TEXT NOT NULL)") + conn.execute("CREATE TABLE IF NOT EXISTS terminal_audit (row_id TEXT PRIMARY KEY, run_id TEXT, payload_json TEXT NOT NULL)") + conn.commit() + + def write_run_summary(self, run_id: str, payload: Mapping[str, Any]) -> None: + with self.connect() as conn: + conn.execute( + "INSERT OR REPLACE INTO run_summaries(run_id, created_at, payload_json) VALUES (?, ?, ?)", + (run_id, utc_now(), json.dumps(dict(payload), sort_keys=True)), + ) + conn.commit() + + def write_rows(self, table: str, run_id: str, rows: Iterable[Mapping[str, Any]]) -> int: + allowed = {"episode_metrics", "residual_obstructions", "terminal_audit"} + if table not in allowed: + raise ValueError(f"unsupported residual lawbook table: {table}") + count = 0 + with self.connect() as conn: + for count, row in enumerate(rows, start=1): + data = dict(row) + row_id = str(data.get("row_id") or f"{run_id}:{table}:{count}") + if table == "episode_metrics": + conn.execute( + "INSERT OR REPLACE INTO episode_metrics(row_id, run_id, episode, payload_json) VALUES (?, ?, ?, ?)", + (row_id, run_id, int(data.get("episode", 0) or 0), json.dumps(data, sort_keys=True)), + ) + elif table == "residual_obstructions": + conn.execute( + "INSERT OR REPLACE INTO residual_obstructions(row_id, run_id, obstruction_name, payload_json) VALUES (?, ?, ?, ?)", + (row_id, run_id, str(data.get("obstruction_name", "")), json.dumps(data, sort_keys=True)), + ) + else: + conn.execute( + "INSERT OR REPLACE INTO terminal_audit(row_id, run_id, payload_json) VALUES (?, ?, ?)", + (row_id, run_id, json.dumps(data, sort_keys=True)), + ) + conn.commit() + return count + + +def write_repair_lawbook(sqlite_path: str | Path, repair_df: Any, obstruction_df: Any, metadata: Mapping[str, Any] | None = None) -> None: + """Persist advisory repair/obstruction rows for later autonomous reuse.""" + + metadata = dict(metadata or {}) + book = ResidualLawbook.open(sqlite_path) + run_id = str(metadata.get("run_id") or "autonomous_v2") + repair_rows = _records(repair_df) + obstruction_rows = _records(obstruction_df) + book.write_run_summary(run_id, {"metadata": metadata, "repair_rows": len(repair_rows), "obstruction_rows": len(obstruction_rows)}) + book.write_rows("episode_metrics", run_id, ({**row, **metadata} for row in repair_rows)) + book.write_rows("residual_obstructions", run_id, ({**row, **metadata} for row in obstruction_rows)) + + +def load_repair_lawbook(sqlite_path: str | Path) -> pd.DataFrame: + path = Path(sqlite_path) + if not path.exists(): + return pd.DataFrame() + rows: list[dict[str, Any]] = [] + with sqlite3.connect(str(path)) as conn: + for table in ("episode_metrics", "residual_obstructions"): + try: + for payload, in conn.execute(f"SELECT payload_json FROM {table}"): + rows.append(json.loads(payload)) + except sqlite3.Error: + continue + return pd.DataFrame(rows) + + +def recommend_from_lawbook(pair_features: Any, lawbook_df: pd.DataFrame, budget: int) -> list[Any]: + """Return constructor indices/families suggested by prior advisory repair rows.""" + + if lawbook_df is None or lawbook_df.empty: + return [] + features = pair_features if isinstance(pair_features, Mapping) else {} + basin = str(features.get("basin", "")) + df = lawbook_df.copy() + if basin and "basin" in df.columns: + scoped = df[df["basin"].astype(str) == basin] + if not scoped.empty: + df = scoped + gain_col = "marginal_gain" if "marginal_gain" in df.columns else None + if gain_col: + df["_gain"] = pd.to_numeric(df[gain_col], errors="coerce").fillna(0) + df = df.sort_values("_gain", ascending=False) + out: list[Any] = [] + for _, row in df.iterrows(): + value = row.get("constructor_idx", row.get("family", "")) + if value not in out and value not in ("", None): + out.append(value) + if len(out) >= int(budget): + break + return out + + +def _records(value: Any) -> list[dict[str, Any]]: + if value is None: + return [] + if hasattr(value, "to_dict"): + try: + records = value.to_dict("records") + if isinstance(records, list): + return [_jsonable(dict(row)) for row in records] + except TypeError: + pass + return [_jsonable(dict(row)) for row in value] + + +def _jsonable(row: dict[str, Any]) -> dict[str, Any]: + out = {} + for key, value in row.items(): + if isinstance(value, (dict, list, tuple, str, int, float, bool)) or value is None: + out[key] = value + else: + try: + out[key] = value.item() + except Exception: + out[key] = str(value) + return out diff --git a/mathgraph/terminal_form_contract.py b/mathgraph/terminal_form_contract.py new file mode 100644 index 0000000..c6f08cb --- /dev/null +++ b/mathgraph/terminal_form_contract.py @@ -0,0 +1,172 @@ +"""Terminal-form contract for autonomous MathGraph discovery loops. + +This module is intentionally small and conservative. It gives autonomous +runners a shared vocabulary for terminal eligibility without replacing the +repo's verifier and promotion-gate modules. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Mapping + + +class TerminalForm(str, Enum): + VERIFIED_PROOF = "VERIFIED_PROOF" + FINITE_COUNTERMODEL = "FINITE_COUNTERMODEL" + NAMED_OBSTRUCTION = "NAMED_OBSTRUCTION" + NONE = "NONE" + + +class PromotionStatus(str, Enum): + ACCEPTED = "ACCEPTED" + CANDIDATE = "CANDIDATE" + ADVISORY = "ADVISORY" + RESIDUAL = "RESIDUAL" + REJECTED = "REJECTED" + + +TrustLevel = PromotionStatus + + +@dataclass(frozen=True) +class TerminalDecision: + terminal_form: TerminalForm + status: PromotionStatus + accepted: bool + reason: str + trust_level: str = "ADVISORY_ROUTE" + advisory_only: bool = True + can_promote_truth: bool = False + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "terminal_form": self.terminal_form.value, + "status": self.status.value, + "accepted": self.accepted, + "reason": self.reason, + "trust_level": self.trust_level, + "advisory_only": self.advisory_only, + "can_promote_truth": self.can_promote_truth, + "metadata": dict(self.metadata), + } + + +def decide_terminal_form(record: Mapping[str, Any] | Any) -> TerminalDecision: + """Return a conservative terminal decision for a raw row/certificate. + + Rules: + - Checked finite countermodels may support FALSE terminal candidates. + - Lean-verified rows may support TRUE terminal candidates. + - Named obstructions are terminal only as obstruction records, never TRUE. + - Failed finite search is residual evidence and never TRUE. + """ + + data = record.to_dict() if hasattr(record, "to_dict") else dict(record or {}) + status = str(data.get("certificate_status") or data.get("proof_status") or data.get("status") or "").lower() + + if can_promote_true(data): + return TerminalDecision( + terminal_form=TerminalForm.VERIFIED_PROOF, + status=PromotionStatus.ACCEPTED, + accepted=True, + reason="verifier-backed TRUE evidence is present", + trust_level="VERIFIED_PROOF", + advisory_only=False, + can_promote_truth=True, + metadata=data, + ) + + if can_promote_false(data): + return TerminalDecision( + terminal_form=TerminalForm.FINITE_COUNTERMODEL, + status=PromotionStatus.ACCEPTED, + accepted=True, + reason="finite checker produced a source-satisfying, target-violating witness", + trust_level="FINITE_VERIFIED", + advisory_only=False, + can_promote_truth=True, + metadata=data, + ) + + if status == "named_obstruction_advisory" or data.get("obstruction_name"): + return TerminalDecision( + terminal_form=TerminalForm.NAMED_OBSTRUCTION, + status=PromotionStatus.ADVISORY, + accepted=False, + reason="obstruction is named but remains advisory until an obstruction boundary accepts it", + metadata=data, + ) + + if status in {"failed_search", "finite_failed_small_n", "finite_failed_structured", "not_a_countermodel"} or data.get("finite_search_miss"): + return TerminalDecision( + terminal_form=TerminalForm.NONE, + status=PromotionStatus.RESIDUAL, + accepted=False, + reason="finite-search failure is residual evidence, never TRUE", + trust_level="RESIDUAL_EVIDENCE", + metadata=data, + ) + + return TerminalDecision( + terminal_form=TerminalForm.NONE, + status=PromotionStatus.REJECTED, + accepted=False, + reason="record is advisory or unsupported and cannot promote truth", + metadata=data, + ) + + +def can_promote_true(record: Mapping[str, Any] | Any) -> bool: + data = record.to_dict() if hasattr(record, "to_dict") else dict(record or {}) + status = str(data.get("certificate_status") or data.get("proof_status") or data.get("status") or "").lower() + return bool( + data.get("lean_verified") + or data.get("proof_verified") + or data.get("verified_proof") + or data.get("congruence_explain_verified") + or status in {"lean_verified", "proof_verified", "verified_proof", "congruence_explain_verified"} + ) + + +def can_promote_false(record: Mapping[str, Any] | Any) -> bool: + data = record.to_dict() if hasattr(record, "to_dict") else dict(record or {}) + status = str(data.get("certificate_status") or data.get("proof_status") or data.get("status") or "").lower() + countermodel_status = status in { + "finite_countermodel_found", + "finite_countermodel_verified", + "countermodel_verified", + } + checker_backed = bool( + data.get("finite_checker_valid") + or data.get("finite_verified") + or (data.get("eq1_holds") is True and data.get("eq2_violated") is True) + ) + return countermodel_status and checker_backed + + +def decision_as_dict(record: Mapping[str, Any] | Any) -> dict[str, Any]: + return decide_terminal_form(record).to_dict() + + +def audit_terminal_rows(rows: list[Mapping[str, Any]]) -> list[dict[str, Any]]: + return [decision_as_dict(row) for row in rows] + + +def boundary_preserved(rows: list[Mapping[str, Any]]) -> bool: + """True when no advisory/residual row is allowed to promote truth.""" + + for row in rows: + data = row.to_dict() if hasattr(row, "to_dict") else dict(row or {}) + if bool(data.get("advisory_only")) and bool(data.get("can_promote_truth")): + return False + if str(data.get("terminal_form", "")).upper() == TerminalForm.VERIFIED_PROOF.value and not can_promote_true(data): + return False + if str(data.get("terminal_form", "")).upper() == TerminalForm.FINITE_COUNTERMODEL.value and not can_promote_false(data): + return False + decision = decide_terminal_form(data) + if decision.advisory_only and decision.can_promote_truth: + return False + return True diff --git a/scripts/run_autonomous_compounding_engine.py b/scripts/run_autonomous_compounding_engine.py new file mode 100644 index 0000000..47787d2 --- /dev/null +++ b/scripts/run_autonomous_compounding_engine.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python +"""Run the autonomous finite-core MathGraph compounding façade.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Sequence + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from mathgraph.autonomous_compounding_engine import AutonomousCompoundingConfig, run_autonomous_compounding + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--equations") + parser.add_argument("--matrix") + parser.add_argument("--out-dir", required=True) + parser.add_argument("--episodes", type=int, default=4) + parser.add_argument("--sample-pairs", type=int, default=4000) + parser.add_argument("--repair-budget", type=int, default=40) + parser.add_argument("--max-n", type=int, default=5) + parser.add_argument("--seed", type=int, default=20260524) + parser.add_argument("--tiny-demo", action="store_true") + parser.add_argument("--finite-core-mode", choices=("facade", "native_v2"), default="facade") + parser.add_argument("--constructor-limit", type=int) + parser.add_argument("--include-random-constructors", action="store_true") + parser.add_argument("--random-constructor-count", type=int, default=0) + parser.add_argument("--lawbook-path") + parser.add_argument("--reuse-lawbook", action="store_true") + parser.add_argument("--write-report", action="store_true") + args = parser.parse_args(argv) + + summary = run_autonomous_compounding( + AutonomousCompoundingConfig( + equations=args.equations, + matrix=args.matrix, + out_dir=args.out_dir, + episodes=args.episodes, + sample_pairs=args.sample_pairs, + repair_budget=args.repair_budget, + max_n=args.max_n, + seed=args.seed, + tiny_demo=args.tiny_demo, + finite_core_mode=args.finite_core_mode, + constructor_limit=args.constructor_limit, + include_random_constructors=args.include_random_constructors, + random_constructor_count=args.random_constructor_count, + lawbook_path=args.lawbook_path, + reuse_lawbook=args.reuse_lawbook, + write_report=args.write_report, + ) + ) + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_repo_architecture_audit.py b/scripts/run_repo_architecture_audit.py index c75bc35..81968ec 100644 --- a/scripts/run_repo_architecture_audit.py +++ b/scripts/run_repo_architecture_audit.py @@ -36,6 +36,8 @@ "mathgraph/semantic_validation.py", "mathgraph/finite_magma_world.py", "mathgraph/compounding_engine.py", + "mathgraph/autonomous_compounding_engine.py", + "mathgraph/autonomous_finite_recovery.py", "mathgraph/recursive_residual_compounding.py", "mathgraph/etp_terms.py", "mathgraph/quotient_state.py", diff --git a/tests/test_autonomous_compounding_engine.py b/tests/test_autonomous_compounding_engine.py new file mode 100644 index 0000000..d9be026 --- /dev/null +++ b/tests/test_autonomous_compounding_engine.py @@ -0,0 +1,17 @@ +from mathgraph.autonomous_compounding_engine import AutonomousCompoundingConfig, run_autonomous_compounding + + +def test_autonomous_tiny_demo_runs(tmp_path): + summary = run_autonomous_compounding( + AutonomousCompoundingConfig(out_dir=tmp_path / "run", tiny_demo=True, episodes=1, sample_pairs=20, repair_budget=4, max_n=3) + ) + assert summary["autonomous_facade"] is True + assert summary["serious_path_uses_finite_recovery_core"] is True + assert summary["advisory_boundary_preserved"] is True + assert summary["all_gates_passed"] is True + assert summary["true_contamination_count"] == 0 + assert summary["terminal_claims_from_advisory_count"] == 0 + assert summary["failed_search_promoted_true_count"] == 0 + assert summary["repair_final_yield"] >= summary["generic_final_yield"] + assert summary["repair_final_residuals"] <= summary["generic_final_residuals"] + assert "lawbook.sqlite" in summary["artifacts"] diff --git a/tests/test_autonomous_compounding_native_v2.py b/tests/test_autonomous_compounding_native_v2.py new file mode 100644 index 0000000..225b7fe --- /dev/null +++ b/tests/test_autonomous_compounding_native_v2.py @@ -0,0 +1,97 @@ +import json +import sqlite3 +import subprocess +import sys +from pathlib import Path + +from mathgraph.autonomous_compounding_engine import AutonomousCompoundingConfig, run_autonomous_compounding + + +def test_native_v2_tiny_demo_passes_gates_and_writes_artifacts(tmp_path): + out_dir = tmp_path / "native" + summary = run_autonomous_compounding( + AutonomousCompoundingConfig( + out_dir=out_dir, + tiny_demo=True, + finite_core_mode="native_v2", + episodes=3, + sample_pairs=80, + repair_budget=20, + max_n=3, + seed=20260524, + write_report=True, + ) + ) + + assert summary["autonomous_facade"] is True + assert summary["finite_core_mode"] == "native_v2" + assert summary["serious_path_uses_finite_recovery_core"] is True + assert summary["all_gates_passed"] is True + assert summary["repair_final_yield"] >= summary["generic_final_yield"] + assert summary["repair_final_residuals"] <= summary["generic_final_residuals"] + assert summary["lawbook_reuse_yield"] >= summary["repair_final_yield"] + assert summary["true_contamination_count"] == 0 + assert summary["terminal_claims_from_advisory_count"] == 0 + assert summary["failed_search_promoted_true_count"] == 0 + + required = [ + "autonomous_compounding_summary.json", + "episode_metrics.csv", + "gate_results.csv", + "pair_features.csv", + "true_pair_features.csv", + "constructor_manifest.csv", + "constructor_family_recommendations.csv", + "pair_recovery_matrix_summary.csv", + "generic_route.csv", + "residual_repair_route.csv", + "lawbook_reuse_route.csv", + "compact_atlas_route.csv", + "obstruction_atlas.csv", + "residual_queue_after.csv", + "terminal_form_audit.csv", + "lawbook.sqlite", + "autonomous_compounding_report.md", + ] + for name in required: + path = Path(summary["artifacts"][name]) + assert path.exists(), name + + with sqlite3.connect(summary["artifacts"]["lawbook.sqlite"]) as conn: + tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert {"run_summaries", "episode_metrics", "residual_obstructions", "terminal_audit"}.issubset(tables) + + +def test_native_v2_cli_emits_json_summary(tmp_path): + out_dir = tmp_path / "cli-native" + result = subprocess.run( + [ + sys.executable, + "scripts/run_autonomous_compounding_engine.py", + "--out-dir", + str(out_dir), + "--tiny-demo", + "--finite-core-mode", + "native_v2", + "--episodes", + "3", + "--sample-pairs", + "80", + "--repair-budget", + "20", + "--max-n", + "3", + "--seed", + "20260524", + "--write-report", + ], + text=True, + capture_output=True, + check=True, + ) + + summary = json.loads(result.stdout) + assert summary["finite_core_mode"] == "native_v2" + assert summary["all_gates_passed"] is True + assert summary["true_contamination_count"] == 0 + assert Path(summary["artifacts"]["lawbook.sqlite"]).exists() diff --git a/tests/test_autonomous_compounding_nonregression.py b/tests/test_autonomous_compounding_nonregression.py new file mode 100644 index 0000000..052418d --- /dev/null +++ b/tests/test_autonomous_compounding_nonregression.py @@ -0,0 +1,72 @@ +import json +import subprocess +import sys +from pathlib import Path + +from mathgraph.autonomous_compounding_engine import AutonomousCompoundingConfig, run_autonomous_compounding + + +def test_autonomous_summary_has_stable_safety_keys(tmp_path): + summary = run_autonomous_compounding( + AutonomousCompoundingConfig( + out_dir=tmp_path / "stable", + tiny_demo=True, + episodes=2, + sample_pairs=40, + repair_budget=8, + max_n=3, + seed=20260524, + ) + ) + + for key in [ + "autonomous_facade", + "serious_path_uses_finite_recovery_core", + "terminal_contract", + "terminal_audit", + "advisory_boundary_preserved", + "all_gates_passed", + "true_contamination_count", + "terminal_claims_from_advisory_count", + "failed_search_promoted_true_count", + "generic_final_yield", + "repair_final_yield", + "generic_final_residuals", + "repair_final_residuals", + "repair_gain_over_generic", + "source_mode", + "real_corpus_used", + "artifacts", + ]: + assert key in summary + assert summary["failed_search_promoted_true_count"] == 0 + assert summary["failed_search_promoted_true"] == 0 + + +def test_autonomous_cli_prints_json_summary(tmp_path): + out_dir = tmp_path / "cli" + result = subprocess.run( + [ + sys.executable, + "scripts/run_autonomous_compounding_engine.py", + "--out-dir", + str(out_dir), + "--tiny-demo", + "--episodes", + "1", + "--sample-pairs", + "20", + "--repair-budget", + "4", + "--max-n", + "3", + ], + text=True, + capture_output=True, + check=True, + ) + + summary = json.loads(result.stdout) + assert summary["autonomous_facade"] is True + assert summary["all_gates_passed"] is True + assert summary["true_contamination_count"] == 0 diff --git a/tests/test_autonomous_finite_recovery.py b/tests/test_autonomous_finite_recovery.py new file mode 100644 index 0000000..1b640b3 --- /dev/null +++ b/tests/test_autonomous_finite_recovery.py @@ -0,0 +1,56 @@ +import numpy as np + +from mathgraph.autonomous_finite_recovery import ( + FiniteRecoveryConfig, + build_finite_recovery_core, + evaluate_false_pairs, + greedy_route, + pair_recovery_matrix, + residual_marginal_repair, +) + + +TOY_EQUATIONS = [ + "(x * y) = (y * x)", + "(x * y) = x", + "(x * y) = y", + "x = x", +] + + +def test_finite_recovery_core_builds_constructor_manifest_and_cache(): + result = build_finite_recovery_core(TOY_EQUATIONS, FiniteRecoveryConfig(max_n=3, constructor_limit=20)) + + assert result.constructor_count > 0 + assert result.equation_count == len(TOY_EQUATIONS) + assert result.sat_cache.shape == (result.constructor_count, len(TOY_EQUATIONS)) + assert {"constructor_idx", "cid", "family", "n", "advisory_only", "can_promote_truth"}.issubset( + result.constructor_manifest.columns + ) + assert result.constructor_manifest["advisory_only"].all() + assert not result.constructor_manifest["can_promote_truth"].any() + + +def test_recovery_counts_only_source_holds_and_target_violated(): + result = build_finite_recovery_core(TOY_EQUATIONS, FiniteRecoveryConfig(max_n=3, constructor_limit=20)) + pairs = [(0, 1), (0, 2)] + matrix = pair_recovery_matrix(pairs, result.sat_cache) + rows = evaluate_false_pairs(pairs, result.sat_cache, result.constructors) + + assert matrix.shape == (len(pairs), result.constructor_count) + for pair_idx, (source, target) in enumerate(pairs): + for constructor_idx in np.flatnonzero(matrix[pair_idx]): + assert bool(result.sat_cache[constructor_idx, source]) + assert not bool(result.sat_cache[constructor_idx, target]) + assert {"pair_idx", "eq1_id", "eq2_id", "recovered", "best_constructor_idx"}.issubset(rows.columns) + + +def test_greedy_and_residual_repair_are_monotone(): + result = build_finite_recovery_core(TOY_EQUATIONS, FiniteRecoveryConfig(max_n=3, constructor_limit=20)) + matrix = pair_recovery_matrix([(0, 1), (0, 2), (1, 2)], result.sat_cache) + _, generic_mask, _ = greedy_route(matrix, result.constructor_manifest, budget=2, seed=7) + _, repair_mask, repair_route = residual_marginal_repair(matrix, generic_mask, result.constructor_manifest, budget=3, seed=7) + + assert int(repair_mask.sum()) >= int(generic_mask.sum()) + assert repair_route.empty or repair_route["advisory_only"].all() + assert repair_route.empty or not repair_route["can_promote_truth"].any() diff --git a/tests/test_terminal_form_contract.py b/tests/test_terminal_form_contract.py index 473662d..99a116d 100644 --- a/tests/test_terminal_form_contract.py +++ b/tests/test_terminal_form_contract.py @@ -1,4 +1,13 @@ from mathgraph.invariants import check_terminal_form_contract +from mathgraph.terminal_form_contract import ( + TerminalForm, + audit_terminal_rows, + boundary_preserved, + can_promote_false, + can_promote_true, + decision_as_dict, + decide_terminal_form, +) def test_accepted_claim_requires_exactly_one_terminal_form(): @@ -11,3 +20,49 @@ def test_invalid_terminal_form_rejected(): report = check_terminal_form_contract({"status": "ACCEPTED", "terminal_form": "TRUE"}) assert not report.ok assert report.violations[0].code == "invalid_terminal_form" + + +def test_finite_countermodel_can_promote_false_terminal(): + decision = decide_terminal_form({"status": "finite_countermodel_found", "eq1_holds": True, "eq2_violated": True}) + assert decision.accepted + assert decision.terminal_form == TerminalForm.FINITE_COUNTERMODEL + assert decision.can_promote_truth + + +def test_finite_countermodel_requires_checker_backed_validity(): + assert not can_promote_false({"status": "finite_countermodel_found"}) + assert not decide_terminal_form({"status": "finite_countermodel_found"}).accepted + assert can_promote_false({"certificate_status": "finite_countermodel_verified", "finite_checker_valid": True}) + + +def test_failed_search_is_residual_not_true(): + decision = decide_terminal_form({"status": "failed_search", "finite_search_miss": True}) + assert not decision.accepted + assert decision.terminal_form == TerminalForm.NONE + assert not decision.can_promote_truth + + +def test_proof_verified_evidence_promotes_true(): + assert can_promote_true({"proof_verified": True}) + decision = decide_terminal_form({"proof_verified": True}) + assert decision.accepted + assert decision.terminal_form == TerminalForm.VERIFIED_PROOF + assert not decision.advisory_only + + +def test_boundary_preserved_for_advisory_rows(): + assert boundary_preserved([{"status": "named_obstruction_advisory", "obstruction_name": "x"}, {"status": "failed_search"}]) + + +def test_named_obstruction_is_advisory_and_bad_advisory_truth_is_rejected(): + decision = decide_terminal_form({"status": "named_obstruction_advisory", "obstruction_name": "x"}) + assert decision.terminal_form == TerminalForm.NAMED_OBSTRUCTION + assert decision.advisory_only + assert not decision.can_promote_truth + assert not boundary_preserved([{"advisory_only": True, "can_promote_truth": True, "terminal_form": "VERIFIED_PROOF"}]) + + +def test_decision_dict_and_audit_helpers(): + row = {"lean_verified": True} + assert decision_as_dict(row)["terminal_form"] == "VERIFIED_PROOF" + assert audit_terminal_rows([row])[0]["accepted"] is True