J94/bit-vector-tensor-control-policy

1#!/usr/bin/env python32from __future__ import annotations3 4import argparse5import json6import subprocess7import sys8from datetime import datetime, timezone9from pathlib import Path10from typing import Any11 12from propose_self_improvement import ROOT, load_yaml, run_codex_proposal13 14DEFAULT_CONFIG = ROOT / "self_improve.yaml"15DEFAULT_SCHEMA = ROOT / "schemas" / "self_improve_proposal_v0.json"16 17 18def utc_stamp() -> str:19    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")20 21 22def run_command(command: str, *, cwd: Path, stdout_path: Path, stderr_path: Path) -> int:23    completed = subprocess.run(24        command,25        cwd=cwd,26        shell=True,27        text=True,28        stdout=stdout_path.open("w", encoding="utf-8"),29        stderr=stderr_path.open("w", encoding="utf-8"),30        check=False,31    )32    return completed.returncode33 34 35def main() -> int:36    parser = argparse.ArgumentParser(description="Propose and optionally apply one bounded self-improvement run.")37    parser.add_argument("--goal", required=True)38    parser.add_argument("--apply", action="store_true")39    parser.add_argument("--stage-on-pass", action="store_true")40    parser.add_argument("--config", default=str(DEFAULT_CONFIG))41    parser.add_argument("--schema", default=str(DEFAULT_SCHEMA))42    parser.add_argument("--output-dir", default="")43    args = parser.parse_args()44 45    config_path = Path(args.config)46    schema_path = Path(args.schema)47    config = load_yaml(config_path)48    run_dir = Path(args.output_dir) if args.output_dir else (ROOT / "runs" / "self_improve" / utc_stamp())49    run_dir.mkdir(parents=True, exist_ok=True)50 51    system_context_path = run_dir / "system_context.json"52    subprocess.run(53        [str(ROOT / "api" / "build_system_context.sh"), str(system_context_path)],54        cwd=ROOT,55        check=True,56        stdout=subprocess.DEVNULL,57    )58 59    proposal_path = run_dir / "proposal.json"60    manifest_path = run_dir / "manifest.json"61    brief_path = run_dir / "brief.md"62    proposal = run_codex_proposal(63        goal=args.goal,64        config_path=config_path,65        system_context_path=system_context_path,66        output_path=proposal_path,67        schema_path=schema_path,68    )69    manifest_path.write_text(json.dumps(proposal["manifest"], indent=2, sort_keys=True) + "\n", encoding="utf-8")70    brief_lines = [71        "# Self-Improve Brief",72        "",73        f"Clean product one-liner: {proposal['one_liner']}",74        "",75        f"- Goal: `{proposal['goal']}`",76        f"- Decision: {proposal['decision_brief']}",77        f"- Target files: `{', '.join(proposal['target_files'])}`",78        f"- Benchmark: `{proposal['benchmark']['command']}`",79        "",80        "## Change Summary",81        "",82    ]83    brief_lines.extend([f"- {item}" for item in proposal["change_summary"]])84    brief_path.write_text("\n".join(brief_lines) + "\n", encoding="utf-8")85 86    receipt_path = None87    runtime_summary_path = None88    benchmark_status = None89    benchmark_stdout = run_dir / "benchmark.stdout"90    benchmark_stderr = run_dir / "benchmark.stderr"91    staged_files: list[str] = []92 93    if args.apply:94        runtime = subprocess.run(95            [str(ROOT / "runtime" / "execute_manifest.sh"), str(manifest_path)],96            cwd=ROOT,97            check=True,98            text=True,99            capture_output=True,100        )101        runtime_lines = [line for line in runtime.stdout.splitlines() if line.strip()]102        if len(runtime_lines) >= 2:103            receipt_path = runtime_lines[0]104            runtime_summary_path = runtime_lines[1]105        benchmark_status = run_command(106            proposal["benchmark"]["command"],107            cwd=ROOT,108            stdout_path=benchmark_stdout,109            stderr_path=benchmark_stderr,110        )111        if benchmark_status == 0 and args.stage_on_pass:112            subprocess.run(["git", "add", *proposal["target_files"]], cwd=ROOT, check=True)113            staged_files = proposal["target_files"]114 115    summary = {116        "version": "self_improve_run_v0",117        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),118        "goal": args.goal,119        "config_path": str(config_path),120        "proposal_path": str(proposal_path),121        "manifest_path": str(manifest_path),122        "brief_path": str(brief_path),123        "applied": args.apply,124        "target_files": proposal["target_files"],125        "receipt_path": receipt_path,126        "runtime_summary_path": runtime_summary_path,127        "benchmark_command": proposal["benchmark"]["command"],128        "benchmark_status": benchmark_status,129        "benchmark_stdout": str(benchmark_stdout) if args.apply else None,130        "benchmark_stderr": str(benchmark_stderr) if args.apply else None,131        "staged_files": staged_files,132        "proposal": proposal,133    }134    summary_path = run_dir / "summary.json"135    summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8")136    json.dump(summary, sys.stdout, indent=2)137    sys.stdout.write("\n")138    if args.apply and config.get("proposal_contract", {}).get("fail_on_benchmark_error", False):139        if benchmark_status is None:140            return 1141        if benchmark_status != 0:142            return benchmark_status143    return 0144 145 146if __name__ == "__main__":147    raise SystemExit(main())148