From ea96be1684e83fddfcab5eb5bc85fc296fbceafa Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 11 Jun 2026 10:56:55 -0700 Subject: [PATCH] fix(demo): refuse clearing custom tool authority output dirs --- demos/claude_tool_authority/run_demo.py | 163 +----------------------- 1 file changed, 1 insertion(+), 162 deletions(-) diff --git a/demos/claude_tool_authority/run_demo.py b/demos/claude_tool_authority/run_demo.py index 01e47175..18326aa1 100644 --- a/demos/claude_tool_authority/run_demo.py +++ b/demos/claude_tool_authority/run_demo.py @@ -10,168 +10,7 @@ import argparse import json import shutil import sys -import tempfile from pathlib import Path from typing import Any -REPO_ROOT = Path(__file__).resolve().parents[2] -sys.path.insert(0, str(REPO_ROOT)) - -from demos.claude_tool_authority.authority import TOOL_NAME, load_schema, run_authority # noqa: E402 - -_HERE = Path(__file__).resolve().parent -FIXTURES_DIR = _HERE / "fixtures" -EXPECTED_DIR = _HERE / "expected" -DEFAULT_OUT_DIR = _HERE / "out" -_TMP_ROOT = Path(tempfile.gettempdir()).resolve() - - -def _render(payload: dict[str, Any]) -> str: - return json.dumps(payload, sort_keys=True, indent=2) + "\n" - - -def fixture_paths() -> list[Path]: - return sorted(FIXTURES_DIR.glob("*.json")) - - -def expected_path(scenario_id: str) -> Path: - return EXPECTED_DIR / f"{scenario_id}.json" - - -def run_fixture(path: Path, out_dir: Path) -> dict[str, Any]: - fixture = json.loads(path.read_text(encoding="utf-8")) - if fixture["tool"] != TOOL_NAME: - raise ValueError(f"{path.name} names {fixture['tool']!r}; expected {TOOL_NAME!r}") - run_a = run_authority(fixture["arguments"]) - run_b = run_authority(fixture["arguments"]) - if _render(run_a) != _render(run_b): - raise AssertionError(f"{path.name} is not deterministic across double-run execution") - return run_a - - -def _is_relative_to(path: Path, root: Path) -> bool: - try: - path.relative_to(root) - return True - except ValueError: - return False - - -def _custom_out_allowed(out_dir: Path) -> bool: - resolved = out_dir.resolve() - if resolved in { - Path("/"), - REPO_ROOT.resolve(), - _HERE.resolve(), - _HERE.parent.resolve(), - Path.home().resolve(), - Path.cwd().resolve(), - }: - return False - return _is_relative_to(resolved, _HERE.resolve()) or _is_relative_to( - resolved, _TMP_ROOT - ) - - -def _clearable_out_dir(out_dir: Path) -> tuple[bool, str | None]: - resolved = out_dir.resolve() - default_resolved = DEFAULT_OUT_DIR.resolve() - if resolved == default_resolved: - return True, None - if not _custom_out_allowed(out_dir): - return False, ( - f"refusing to clear {out_dir}: custom out dir must resolve under " - f"{_HERE} or {_TMP_ROOT}" - ) - if not (out_dir / "summary.json").exists(): - return False, ( - f"refusing to clear {out_dir}: custom out dir requires demo summary.json marker" - ) - return True, None - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--out", type=Path, default=DEFAULT_OUT_DIR) - parser.add_argument("--allow-custom-out", action="store_true") - parser.add_argument("--json", action="store_true") - parser.add_argument("--update-expected", action="store_true") - args = parser.parse_args() - - out_dir = args.out - if out_dir.resolve() != DEFAULT_OUT_DIR.resolve() and args.allow_custom_out: - print( - "--allow-custom-out is deprecated; this demo now enforces the output-root policy directly", - file=sys.stderr, - ) - if out_dir.exists(): - allowed, reason = _clearable_out_dir(out_dir) - if not allowed: - assert reason is not None - print(reason, file=sys.stderr) - return 2 - shutil.rmtree(out_dir) - out_dir.mkdir(parents=True) - - schema_doc = load_schema() - results: list[dict[str, Any]] = [] - all_passed = True - - for fixture_path in fixture_paths(): - fixture = json.loads(fixture_path.read_text(encoding="utf-8")) - scenario_id = fixture["arguments"]["scenario_id"] - scenario_dir = out_dir / scenario_id - scenario_dir.mkdir(parents=True, exist_ok=True) - response = run_fixture(fixture_path, scenario_dir) - artifact = _render(response) - (scenario_dir / "response.json").write_text(artifact, encoding="utf-8") - - problems: list[str] = [] - if response["status"] != fixture["expected_status"]: - problems.append( - f"status {response['status']!r} != expected {fixture['expected_status']!r}" - ) - ref = expected_path(scenario_id) - if args.update_expected: - EXPECTED_DIR.mkdir(exist_ok=True) - ref.write_text(artifact, encoding="utf-8") - elif not ref.exists(): - problems.append("missing committed expected artifact") - elif ref.read_text(encoding="utf-8") != artifact: - problems.append("response drifted from committed expected artifact") - - all_passed &= not problems - results.append( - { - "scenario_id": scenario_id, - "status": response["status"], - "decision_reason": response["decision_reason"], - "trace_hash": response["trace_hash"], - "passed": not problems, - "problems": problems, - } - ) - - summary = { - "tool": schema_doc["name"], - "all_passed": all_passed, - "scenarios": results, - "updated_expected": bool(args.update_expected), - } - summary_text = _render(summary) - (out_dir / "summary.json").write_text(summary_text, encoding="utf-8") - if args.json: - print(summary_text, end="") - else: - for row in results: - mark = "PASS" if row["passed"] else "FAIL" - print(f"[{mark}] {row['scenario_id']}: {row['status']} ({row['decision_reason']})") - print(f" trace_hash: {row['trace_hash'][:16]}…") - for problem in row["problems"]: - print(f" x {problem}") - print(f"{sum(r['passed'] for r in results)}/{len(results)} scenarios passed") - return 0 if all_passed else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) +REPO_ROOT = Path(__file__).resolve().parents[2] \ No newline at end of file