| |
| """Session / progress tracker — adapted port of auto-re-agent's core/session.py, |
| reports/tracker.py, and reports/formatter.py. JSON-backed, append-only, atomic |
| writes. Mirrors the original's REPORT format so runs are resumable and progress |
| claims are auditable against recorded tool evidence (never the model's memory). |
| |
| A "function" record = one analysis target (a gate/feature/function) the loop has |
| attempted, keyed by normalized address. Record after EACH attempt so a crash or |
| new session resumes where it left off. |
| |
| Usage: |
| python progress.py record --file re-progress.json --address 0x401abc \ |
| --name "CLicense::Validate" --component license \ |
| --success true --rounds 2 --verdict PASS --parity GREEN \ |
| --objective PASS --note "jz at 0x401ad0 gates the dialog" |
| python progress.py summary --file re-progress.json |
| python progress.py table --file re-progress.json [--component license] |
| python progress.py attempted --file re-progress.json --address 0x401abc # exit 0 if attempted |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import sys |
| import time |
| from pathlib import Path |
|
|
|
|
| def norm_addr(a: str) -> str: |
| a = a.strip().lower() |
| if a.startswith("0x"): |
| try: |
| return hex(int(a, 16)) |
| except ValueError: |
| return a |
| return a |
|
|
|
|
| class Session: |
| def __init__(self, path: str | Path): |
| self.path = Path(path) |
| self.data = {"functions": {}, "runs": []} |
| if self.path.exists(): |
| try: |
| self.data = json.loads(self.path.read_text(encoding="utf-8")) |
| except (json.JSONDecodeError, OSError): |
| pass |
| self.data.setdefault("functions", {}) |
| self.data.setdefault("runs", []) |
|
|
| def save(self): |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| tmp = self.path.with_suffix(".tmp") |
| tmp.write_text(json.dumps(self.data, indent=2), encoding="utf-8") |
| tmp.replace(self.path) |
|
|
| def record(self, entry: dict): |
| addr = norm_addr(entry["address"]) |
| entry["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%S") |
| self.data["functions"][addr] = entry |
| self.data["runs"].append(entry) |
| self.save() |
|
|
| def is_attempted(self, address: str) -> bool: |
| return norm_addr(address) in self.data["functions"] |
|
|
| def summary(self) -> dict: |
| funcs = self.data["functions"].values() |
| total = len(self.data["functions"]) |
| passed = sum(1 for f in funcs if f.get("success")) |
| comps = {f.get("component", "") for f in funcs if f.get("component")} |
| return {"total": total, "passed": passed, "failed": total - passed, |
| "components": len(comps)} |
|
|
| def table(self, component: str | None): |
| rows = [] |
| for f in self.data["functions"].values(): |
| if component and f.get("component") != component: |
| continue |
| rows.append(f) |
| return rows |
|
|
|
|
| def _b(v: str | None) -> bool: |
| return str(v).lower() in {"1", "true", "yes", "pass", "y"} |
|
|
|
|
| def cmd_record(a): |
| s = Session(a.file) |
| s.record({ |
| "address": a.address, |
| "function_name": a.name or "", |
| "component": a.component or "", |
| "success": _b(a.success), |
| "rounds_used": a.rounds, |
| "verdict": a.verdict, |
| "objective_verdict": a.objective, |
| "parity_status": a.parity, |
| "note": a.note or "", |
| }) |
| print(f"recorded {norm_addr(a.address)} ({'PASS' if _b(a.success) else 'FAIL'})") |
| return 0 |
|
|
|
|
| def cmd_summary(a): |
| s = Session(a.file) |
| su = s.summary() |
| print("RE Progress Summary") |
| print("=" * 40) |
| print(f"Total targets: {su['total']}") |
| print(f"Passed: {su['passed']}") |
| print(f"Failed: {su['failed']}") |
| print(f"Components: {su['components']}") |
| return 0 |
|
|
|
|
| def cmd_table(a): |
| s = Session(a.file) |
| rows = s.table(a.component) |
| print("| Address | Function | Comp | Status | Rounds | Parity | Obj |") |
| print("|---------|----------|------|--------|--------|--------|-----|") |
| for f in rows: |
| st = "PASS" if f.get("success") else "FAIL" |
| print(f"| {f.get('address','')} | {f.get('function_name','')} | " |
| f"{f.get('component','')} | {st} | {f.get('rounds_used','')} | " |
| f"{f.get('parity_status') or '-'} | {f.get('objective_verdict') or '-'} |") |
| return 0 |
|
|
|
|
| def cmd_attempted(a): |
| return 0 if Session(a.file).is_attempted(a.address) else 1 |
|
|
|
|
| def main() -> int: |
| ap = argparse.ArgumentParser(description=__doc__, |
| formatter_class=argparse.RawDescriptionHelpFormatter) |
| sub = ap.add_subparsers(dest="cmd", required=True) |
|
|
| r = sub.add_parser("record") |
| r.add_argument("--file", required=True) |
| r.add_argument("--address", required=True) |
| r.add_argument("--name", default="") |
| r.add_argument("--component", default="") |
| r.add_argument("--success", default="false") |
| r.add_argument("--rounds", type=int, default=0) |
| r.add_argument("--verdict", default=None) |
| r.add_argument("--objective", default=None) |
| r.add_argument("--parity", default=None) |
| r.add_argument("--note", default="") |
| r.set_defaults(fn=cmd_record) |
|
|
| for name, fn in (("summary", cmd_summary), ("table", cmd_table)): |
| p = sub.add_parser(name) |
| p.add_argument("--file", required=True) |
| if name == "table": |
| p.add_argument("--component", default=None) |
| p.set_defaults(fn=fn) |
|
|
| at = sub.add_parser("attempted") |
| at.add_argument("--file", required=True) |
| at.add_argument("--address", required=True) |
| at.set_defaults(fn=cmd_attempted) |
|
|
| a = ap.parse_args() |
| return a.fn(a) |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
