From 3398874f7330bf37c46dbf2cf193e06cc32e6a20 Mon Sep 17 00:00:00 2001 From: samuel <13449963+samuelGrahame@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:07:47 +1000 Subject: [PATCH 1/3] Avoid speculative type requirements in reconstruction prompts --- src/re_agent/agents/prompts/checker_system.md | 2 +- src/re_agent/agents/prompts/reverser_task.md | 4 ++-- tests/test_agents/test_prompt_contract.py | 18 ++++++++++++++++++ 3 files changed, 21 insertions(+), 3 deletions(-) create mode 100644 tests/test_agents/test_prompt_contract.py diff --git a/src/re_agent/agents/prompts/checker_system.md b/src/re_agent/agents/prompts/checker_system.md index 8936d453..3287894c 100644 --- a/src/re_agent/agents/prompts/checker_system.md +++ b/src/re_agent/agents/prompts/checker_system.md @@ -2,7 +2,7 @@ You are a reverse engineering quality checker. Your job is to verify that revers Verification standards: - Every line of Ghidra logic must have corresponding source code -- Every struct offset must map to a named member +- Use supplied type evidence to verify named members. When no layout is available, accept explicit byte offsets that preserve access widths and semantics; do not fail code solely for lacking member names or demand invented structs. - Every function call must be identified and matched - Expression order must match exactly (floating point is order-sensitive) - No missing branches, conditions, or edge cases diff --git a/src/re_agent/agents/prompts/reverser_task.md b/src/re_agent/agents/prompts/reverser_task.md index 891fcb86..c0e714d1 100644 --- a/src/re_agent/agents/prompts/reverser_task.md +++ b/src/re_agent/agents/prompts/reverser_task.md @@ -24,8 +24,8 @@ ${project_rules} Requirements: 1. Match every branch and call from the decompile -2. Map all offsets (e.g. param_1 + 0x88) to real member names +2. Use member names only when supported by supplied type evidence. Otherwise preserve the exact byte offset and access width, and document the unresolved layout; do not invent structs or member names. 3. Preserve exact expression/operand order 4. Use existing project patterns and naming conventions -5. Output the complete function implementation in a ```cpp block +5. Output exactly one complete function implementation in a ```cpp block. Do not wrap it in namespace/class definitions or add helper type/function definitions; report missing declarations outside the code block. 6. End with: REVERSED_FUNCTION: ${class_name}::${function_name} (${address}) diff --git a/tests/test_agents/test_prompt_contract.py b/tests/test_agents/test_prompt_contract.py new file mode 100644 index 00000000..03c06ed3 --- /dev/null +++ b/tests/test_agents/test_prompt_contract.py @@ -0,0 +1,18 @@ +"""Keep reconstruction/review instructions consistent with missing type evidence.""" +from re_agent.agents.checker import PROMPTS_DIR +from re_agent.utils.templates import render_template + + +def test_missing_layout_does_not_require_invented_members(): + reverser = render_template(PROMPTS_DIR / "reverser_task.md") + checker = render_template(PROMPTS_DIR / "checker_system.md") + assert "preserve the exact byte offset and access width" in reverser + assert "do not invent structs or member names" in reverser + assert "accept explicit byte offsets" in checker + assert "do not fail code solely for lacking member names" in checker + + +def test_requested_candidate_matches_single_body_overlay_contract(): + prompt = render_template(PROMPTS_DIR / "reverser_task.md") + assert "exactly one complete function implementation" in prompt + assert "report missing declarations outside the code block" in prompt From 13abe167613542eaeb913b4051a55e578101face Mon Sep 17 00:00:00 2001 From: samuel <13449963+samuelGrahame@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:07:47 +1000 Subject: [PATCH 2/3] Skip model reviews for candidates that fail local preflight --- src/re_agent/agents/loop.py | 23 +++++++++++++++-- tests/test_agents/test_preflight.py | 39 +++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) create mode 100644 tests/test_agents/test_preflight.py diff --git a/src/re_agent/agents/loop.py b/src/re_agent/agents/loop.py index 36a028b0..fc0d4487 100644 --- a/src/re_agent/agents/loop.py +++ b/src/re_agent/agents/loop.py @@ -25,6 +25,7 @@ from re_agent.llm.observed import CallBudget, ObservedProvider from re_agent.llm.protocol import LLMProvider from re_agent.parity.source_indexer import SourceIndexer +from re_agent.verification.candidate import extract_candidate_body from re_agent.verification.objective import verify_candidate @@ -128,8 +129,25 @@ def run_fix_loop( log_path = log_dir / f"round{round_num}-{timestamp}-reverser.json" log_path.write_text(json.dumps(log_entry, indent=2), encoding="utf-8") - # Check - verdict = checker.check(code, target) + # Avoid spending a reviewer call on output the configured candidate gate + # cannot consume. Keep the same repair/checkpoint path as other failures. + preflight_error = None + if candidate_gate is not None: + try: + extract_candidate_body(code) + except ValueError as exc: + preflight_error = str(exc) + if preflight_error: + checker.last_prompt = "" + checker.last_response = "" + verdict = CheckerVerdict( + verdict=Verdict.FAIL, + summary="Local candidate preflight failed; model review skipped", + issues=[preflight_error], + fix_instructions=["Return exactly one complete function definition without helper definitions."], + ) + else: + verdict = checker.check(code, target) last_verdict = verdict objective_verdict: ObjectiveVerdict | None = None @@ -148,6 +166,7 @@ def run_fix_loop( "round": round_num, "timestamp": timestamp, "phase": "check", + "checker_skipped": preflight_error is not None, "prompt": checker.last_prompt, "response": checker.last_response, "verdict": verdict.verdict.value, diff --git a/tests/test_agents/test_preflight.py b/tests/test_agents/test_preflight.py new file mode 100644 index 00000000..5dd0347d --- /dev/null +++ b/tests/test_agents/test_preflight.py @@ -0,0 +1,39 @@ +"""Reject unusable candidate shapes before spending reviewer calls.""" +import json + +import pytest + +from re_agent.agents.loop import run_fix_loop +from re_agent.backend.stub import StubBackend +from re_agent.core.models import FunctionTarget +from tests.test_agents.test_loop import MockLLM + + +@pytest.mark.parametrize("invalid", [ + "struct Layout { int field; }; int f(Layout* p) { return p->field; }", + "int f() { return 0;", + "int f() { return 0; } int g() { return 1; }", +]) +def test_invalid_candidate_skips_reviewer_then_repairs_within_budget(tmp_path, invalid): + reverser = MockLLM([f"```cpp\n{invalid}\n```", "```cpp\nint f() { return 0; }\n```"]) + checker = MockLLM(['{"verdict":"PASS","summary":"Matches","issues":[],"fix_instructions":[]}']) + result = run_fix_loop(FunctionTarget("100", "", "f"), StubBackend(), reverser, checker, + candidate_gate=lambda result: result, objective_verifier_enabled=False, + max_rounds=2, max_llm_calls=3, log_dir=tmp_path) + assert result.success + assert result.rounds_used == 2 + assert reverser._idx == 2 + assert checker._idx == 1 + log = next(tmp_path.glob("*/round1-*-checker.json")) + assert json.loads(log.read_text())["checker_skipped"] is True + + +def test_repeated_invalid_candidates_never_call_reviewer(tmp_path): + reverser = MockLLM(["```cpp\nint f() {\n```"]) + checker = MockLLM(["VERDICT: PASS"]) + result = run_fix_loop(FunctionTarget("100", "", "f"), StubBackend(), reverser, checker, + candidate_gate=lambda result: result, objective_verifier_enabled=False, + max_rounds=3, log_dir=tmp_path) + assert not result.success + assert checker._idx == 0 + assert result.error and "without progress" in result.error From 540de3e9547fcc0db39e337dc8ddfa97be27382f Mon Sep 17 00:00:00 2001 From: samuel <13449963+samuelGrahame@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:32:16 +1000 Subject: [PATCH 3/3] Add decoupled live progress monitor with tested worker controls --- README.md | 2 + docs/live-monitor.md | 44 ++++ pyproject.toml | 5 +- src/re_agent/cli/cmd_monitor.py | 22 ++ src/re_agent/cli/main.py | 14 ++ src/re_agent/monitor/__init__.py | 1 + src/re_agent/monitor/index.html | 393 +++++++++++++++++++++++++++++++ src/re_agent/monitor/server.py | 220 +++++++++++++++++ tests/test_monitor.py | 186 +++++++++++++++ 9 files changed, 885 insertions(+), 2 deletions(-) create mode 100644 docs/live-monitor.md create mode 100644 src/re_agent/cli/cmd_monitor.py create mode 100644 src/re_agent/monitor/__init__.py create mode 100644 src/re_agent/monitor/index.html create mode 100644 src/re_agent/monitor/server.py create mode 100644 tests/test_monitor.py diff --git a/README.md b/README.md index e30e268d..a320bf63 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # ReAgent +For a decoupled local progress window with optional worker controls, see the [live monitor guide](docs/live-monitor.md). + [![PyPI](https://img.shields.io/pypi/v/auto-re-agent)](https://pypi.org/project/auto-re-agent/) [![Python](https://img.shields.io/pypi/pyversions/auto-re-agent)](https://pypi.org/project/auto-re-agent/) [![CI](https://github.com/Dryxio/reagent/actions/workflows/ci.yml/badge.svg)](https://github.com/Dryxio/reagent/actions/workflows/ci.yml) diff --git a/docs/live-monitor.md b/docs/live-monitor.md new file mode 100644 index 00000000..75ca2cf5 --- /dev/null +++ b/docs/live-monitor.md @@ -0,0 +1,44 @@ +# Live progress monitor + +`re-agent monitor` serves a local dashboard independently of the reversal worker. It reads existing session JSON files and logs; it does not import model providers or start work automatically. + +```sh +re-agent monitor --work-dir /path/to/project --session-glob re-agent-progress.json --total 500 +``` + +Open the printed `http://127.0.0.1:8765` address. The page refreshes every two seconds and shows saved function results, review passes, failures, rounds, and recent log output. Omit `--total` when the planned count is unknown. A review pass is not a proof of compilation or binary equivalence. + +Session patterns are relative to `--work-dir` and can be repeated. For a batch runner: + +```sh +re-agent monitor --work-dir /path/to/project --session-glob 'reports/batch-*/progress.json' --log-glob 'reports/batch-*/stderr.log' --total 500 +``` + +The monitor deduplicates function addresses across matching files using their latest result timestamps. Select sessions for the same binary/version; unrelated projects can reuse addresses. Unchanged files are cached, partial JSON writes retain the last readable snapshot, and log output is bounded to the last 7,000 bytes. Counters describe saved final results, not in-flight rounds. + +## Optional worker controls + +Install the optional process-management dependency: + +```sh +pip install 'auto-re-agent[monitor]' +``` + +Supply an explicit argument array after `--worker`, which must be the last monitor option: + +```sh +re-agent monitor --work-dir /path/to/project --session-glob re-agent-progress.json --total 500 --worker re-agent --config re-agent.yaml reverse --manifest targets.json --max-functions 500 +``` + +On Windows, quote paths containing spaces and use an executable such as `python.exe` or `re-agent.exe`, rather than a shell script. Worker arguments are passed directly, without shell interpretation. + +- **Start / resume** launches the configured command. Resuming completed work depends on that command's checkpoint behavior. ReAgent uses its configured session and attempt limits; the monitor does not reset them. +- **Stop run** terminates the owned worker process tree. Saved session files remain intact. An interrupted model call may need to be repeated. +- Closing the browser or monitor host leaves the worker running. Restart the monitor with the same work directory, state directory and worker command to reconnect. +- Duplicate starts sharing a state directory are protected by a process lock. Adoption checks the command, working directory record, PID and process creation time, so stale PID records do not attach to unrelated processes. + +Control records and worker stdout/stderr live in `reports/monitor` by default; override with `--state-dir`. Treat these as local run artifacts, not source files. `--port` changes the port; zero selects an available port. + +## Local access + +The server binds only to IPv4 loopback. Host-header checks reject DNS-rebinding requests. Control endpoints require a random per-host token and reject cross-origin requests. HTTP clients cannot supply a different worker command or arbitrary file path. No external scripts, fonts, telemetry or services are used by the dashboard. This is a local tool, not a multi-user network service. diff --git a/pyproject.toml b/pyproject.toml index cd534874..d0a83f42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,10 +43,11 @@ dependencies = [ ] [project.optional-dependencies] +monitor = ["psutil>=7.0"] ghidra-bridge = ["ghidra-ai-bridge>=0.2.0"] headless = ["ghidra-ai-bridge[headless]>=0.2.0"] benchmark = ["pefile>=2024.8.26", "unicorn>=2.1"] -dev = ["pytest>=8.0", "pytest-cov>=5.0", "ruff>=0.8", "mypy>=1.13"] +dev = ["pytest>=8.0", "pytest-cov>=5.0", "ruff>=0.8", "mypy>=1.13", "psutil>=7.0"] [project.scripts] re-agent = "re_agent.cli.main:main" @@ -74,7 +75,7 @@ warn_return_any = true warn_unused_configs = true [[tool.mypy.overrides]] -module = ["anthropic.*", "openai.*"] +module = ["anthropic.*", "openai.*", "psutil.*"] ignore_missing_imports = true [tool.pytest.ini_options] diff --git a/src/re_agent/cli/cmd_monitor.py b/src/re_agent/cli/cmd_monitor.py new file mode 100644 index 00000000..67fea363 --- /dev/null +++ b/src/re_agent/cli/cmd_monitor.py @@ -0,0 +1,22 @@ +"""Serve a local progress dashboard without importing model providers.""" +from __future__ import annotations + +import argparse +import contextlib +from pathlib import Path + + +def cmd_monitor(args: argparse.Namespace) -> int: + from re_agent.monitor.server import Monitor, make_server + + if not 0 <= args.port <= 65535: + raise ValueError("Port must be between 0 and 65535") + root = Path(args.work_dir).resolve() + monitor = Monitor(root, root / args.state_dir, args.session_glob or ["re-agent-progress.json"], + args.log_glob, args.total, args.worker) + with make_server(monitor, args.port) as server: + print(f"Live monitor: http://127.0.0.1:{server.server_port}", flush=True) + print("Closing the monitor does not stop the worker. Use Stop run to terminate it.", flush=True) + with contextlib.suppress(KeyboardInterrupt): + server.serve_forever() + return 0 diff --git a/src/re_agent/cli/main.py b/src/re_agent/cli/main.py index d89a2a3b..5282bc8b 100644 --- a/src/re_agent/cli/main.py +++ b/src/re_agent/cli/main.py @@ -15,6 +15,15 @@ def build_parser() -> argparse.ArgumentParser: sub = parser.add_subparsers(dest="command", help="Available commands") + monitor_p = sub.add_parser("monitor", help="Host a local live progress dashboard") + monitor_p.add_argument("--work-dir", default=".", help="Directory containing run sessions") + monitor_p.add_argument("--state-dir", default="reports/monitor", help="Monitor control state and worker logs") + monitor_p.add_argument("--session-glob", action="append", help="Relative session pattern (repeatable)") + monitor_p.add_argument("--log-glob", help="Relative activity-log pattern; shows most recently modified log") + monitor_p.add_argument("--total", type=int, default=0, help="Planned function count; 0 means unknown") + monitor_p.add_argument("--port", type=int, default=8765) + monitor_p.add_argument("--worker", nargs=argparse.REMAINDER, help="Optional worker argv; must be the last option") + doctor_p = sub.add_parser("doctor", help="Check configuration and exported evidence without LLM calls") doctor_p.add_argument("--address", help="Check evidence for one function") benchmark_p = sub.add_parser("benchmark", help="Run a differential harness manifest") @@ -78,6 +87,11 @@ def _main(argv: list[str] | None = None) -> int: parser.print_help() return 0 + if args.command == "monitor": + from re_agent.cli.cmd_monitor import cmd_monitor + + return cmd_monitor(args) + if args.command == "evidence": from re_agent.cli.cmd_evidence import cmd_evidence diff --git a/src/re_agent/monitor/__init__.py b/src/re_agent/monitor/__init__.py new file mode 100644 index 00000000..c32bcb9a --- /dev/null +++ b/src/re_agent/monitor/__init__.py @@ -0,0 +1 @@ +"""Optional local progress dashboard; independent of the reversal worker.""" diff --git a/src/re_agent/monitor/index.html b/src/re_agent/monitor/index.html new file mode 100644 index 00000000..0ad65da4 --- /dev/null +++ b/src/re_agent/monitor/index.html @@ -0,0 +1,393 @@ + + + + + + Reconstruction · Live + + + +
+
+
+
ReAgent / local workspace
+

Reconstruction, live.

+

Session progress · generation and review

+
+
+ +
+
+
+
+
+
+ Connecting… +

Waiting for worker state

+
+
+ 0.00% +

No results yet

+
+
+
+

Loading live state…

+
+
+
+
FUNCTIONS ATTEMPTED
+
0
+
Saved final results
+
+
+
REVIEW PASSED
+
0
+
Not a proof of equivalence
+
+
+
NEEDS ATTENTION
+
0
+
Review or processing failed
+
+
+
REVIEW ROUNDS
+
0
+
Across saved results
+
+
+
+ Draft pass — not a verified build. +

+ Review passes do not establish compilation or binary equivalence. + Saved source remains separate from accepted implementations. +

+
+
+
+

Recent results

+ + + + + + + + + + + + + +
AddressOutcomeRounds
Results appear as functions finish.
+
+
+
+

Worker activity

+ Live +
+ 
Waiting for activity…
+
+
+ +
+ + + diff --git a/src/re_agent/monitor/server.py b/src/re_agent/monitor/server.py new file mode 100644 index 00000000..086767a9 --- /dev/null +++ b/src/re_agent/monitor/server.py @@ -0,0 +1,220 @@ +"""Loopback-only progress reporting and explicitly configured worker control.""" +from __future__ import annotations + +import contextlib +import json +import os +import secrets +import subprocess +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +from re_agent.utils.storage import atomic_json, file_lock + + +def read_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + return value if isinstance(value, dict) else {} + except (OSError, ValueError): + return {} + + +class Monitor: + """Read sessions without changing them; control only a configured child worker.""" + + def __init__(self, work_dir: Path, state_dir: Path, session_globs: list[str], + log_glob: str | None = None, total: int = 0, worker: list[str] | None = None) -> None: + if total < 0: + raise ValueError("Total function count must be nonnegative") + for pattern in [*session_globs, *([log_glob] if log_glob else [])]: + if Path(pattern).anchor or ".." in Path(pattern).parts: + raise ValueError("Monitor patterns must be relative to the working directory") + self.work_dir = work_dir.resolve() + self.state_dir = state_dir.resolve() + self.session_globs = session_globs + self.log_glob = log_glob + self.total = total + self.worker = worker or [] + self.record = self.state_dir / "worker.json" + self.process: Any = None + self.psutil: Any = None + if self.worker: + try: + import psutil + except ImportError as exc: + raise RuntimeError("Worker controls require: pip install 'auto-re-agent[monitor]'") from exc + self.psutil = psutil + self.lock = threading.RLock() + self.cache: dict[Path, tuple[int, list[dict[str, Any]]]] = {} + self._adopt() + + def _adopt(self) -> None: + if not self.worker: + return + saved = read_json(self.record) + try: + if saved.get("command") != self.worker or saved.get("cwd") != str(self.work_dir): + return + process = self.psutil.Process(saved["pid"]) + if (process.create_time() == saved["created"] + and process.cmdline() == saved.get("process_command", self.worker)): + self.process = process + except (self.psutil.Error, KeyError): + pass + + def active(self) -> bool: + if self.process is None: + return False + try: + if isinstance(self.process, self.psutil.Popen) and self.process.poll() is not None: + return False + return bool(self.process.is_running() and self.process.status() != self.psutil.STATUS_ZOMBIE) + except self.psutil.Error: + return False + + def start(self) -> dict[str, Any]: + if not self.worker: + raise ValueError("Read-only monitor: no worker command configured") + with self.lock, file_lock(self.record): + self._adopt() + if self.active(): + return {"message": "Already running", "pid": self.process.pid} + self.state_dir.mkdir(parents=True, exist_ok=True) + with (self.state_dir / "worker.stdout.log").open("ab") as out, \ + (self.state_dir / "worker.stderr.log").open("ab") as err: + self.process = self.psutil.Popen( + self.worker, cwd=self.work_dir, stdout=out, stderr=err, + creationflags=int(getattr(subprocess, "CREATE_NO_WINDOW", 0)) if os.name == "nt" else 0, + start_new_session=os.name != "nt", + ) + atomic_json(self.record, {"pid": self.process.pid, "created": self.process.create_time(), + "command": self.worker, "process_command": self.process.cmdline(), + "cwd": str(self.work_dir)}) + return {"message": "Worker started", "pid": self.process.pid} + + def stop(self) -> dict[str, str]: + if not self.worker: + raise ValueError("Read-only monitor: no worker command configured") + with self.lock, file_lock(self.record): + self._adopt() + if self.active(): + if os.name == "nt": + subprocess.run(["taskkill", "/PID", str(self.process.pid), "/T", "/F"], + capture_output=True, timeout=20, check=True) + else: + import signal + + os.killpg(self.process.pid, signal.SIGTERM) + try: + self.process.wait(timeout=5) + except self.psutil.TimeoutExpired: + os.killpg(self.process.pid, signal.SIGKILL) + # Descendants can ignore SIGTERM even when the parent exits. + with contextlib.suppress(ProcessLookupError): + os.killpg(self.process.pid, signal.SIGKILL) + self.process.wait(timeout=20) + return {"message": "Worker stopped; saved results retained"} + + def snapshot(self) -> dict[str, Any]: + with self.lock: + paths = {p for pattern in self.session_globs for p in self.work_dir.glob(pattern) if p.is_file()} + for stale in self.cache.keys() - paths: + del self.cache[stale] + functions: dict[str, dict[str, Any]] = {} + for path in sorted(paths): + try: + stamp = path.stat().st_mtime_ns + except OSError: + continue + if path not in self.cache or self.cache[path][0] != stamp: + data = read_json(path).get("functions") + if isinstance(data, dict): + rows = [{key: row.get(key) for key in + ("address", "function_name", "success", "rounds_used", "timestamp")} + for row in data.values() + if isinstance(row, dict) and isinstance(row.get("address"), str)] + self.cache[path] = (stamp, rows) + for row in self.cache.get(path, (0, []))[1]: + address = row["address"].lower().removeprefix("0x").lstrip("0") or "0" + prior = functions.get(address) + if prior is None or str(row.get("timestamp") or "") >= str(prior.get("timestamp") or ""): + functions[address] = row + recent = sorted(functions.values(), key=lambda r: str(r.get("timestamp") or ""), reverse=True) + passed = sum(row["success"] is True for row in recent) + rounds = sum(row["rounds_used"] for row in recent if type(row.get("rounds_used")) is int) + tail = "" + try: + logs = (list(self.work_dir.glob(self.log_glob)) if self.log_glob + else [self.state_dir / "worker.stderr.log"]) + log = max((p for p in logs if p.is_file()), key=lambda p: p.stat().st_mtime_ns, default=None) + if log: + with log.open("rb") as stream: + stream.seek(max(0, log.stat().st_size - 7000)) + tail = stream.read().decode("utf-8", errors="replace") + except OSError: + pass + return {"active": self.active(), "controls": bool(self.worker), + "phase": "running" if self.active() else "idle" if self.worker else "read-only", + "targets": self.total, "completed": len(recent), "passed": passed, + "failed": len(recent) - passed, "rounds": rounds, "recent": recent[:18], "log": tail, + "updated": time.strftime("%H:%M:%S"), "output": str(self.work_dir)} + + +def make_server(monitor: Monitor, port: int = 8765) -> ThreadingHTTPServer: + """Bind only loopback; never accept worker commands from HTTP clients.""" + token = secrets.token_urlsafe(32) + + class Handler(BaseHTTPRequestHandler): + server: ThreadingHTTPServer + + def reply(self, code: int, body: str, content_type: str = "application/json") -> None: + payload = body.encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", content_type + "; charset=utf-8") + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("Content-Security-Policy", "frame-ancestors 'none'") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def valid_host(self) -> bool: + return self.headers.get("Host") == f"127.0.0.1:{self.server.server_port}" + + def do_GET(self) -> None: + if not self.valid_host(): + self.reply(403, '{"error":"Invalid host"}') + elif self.path == "/": + html = Path(__file__).with_name("index.html").read_text(encoding="utf-8") + self.reply(200, html.replace("__TOKEN__", token), "text/html") + elif self.path == "/api/status": + self.reply(200, json.dumps(monitor.snapshot())) + else: + self.reply(404, '{"error":"Not found"}') + + def do_POST(self) -> None: + origin = f"http://127.0.0.1:{self.server.server_port}" + if (not self.valid_host() or self.headers.get("Origin") not in (None, origin) + or self.headers.get("X-Control-Token") != token): + self.reply(403, '{"error":"Control request rejected"}') + return + try: + if self.path == "/api/start": + result = monitor.start() + elif self.path == "/api/stop": + result = monitor.stop() + else: + self.reply(404, '{"error":"Not found"}') + return + self.reply(200, json.dumps(result)) + except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as exc: + self.reply(400, json.dumps({"error": str(exc)})) + + def log_message(self, format: str, *args: Any) -> None: + pass + + return ThreadingHTTPServer(("127.0.0.1", port), Handler) diff --git a/tests/test_monitor.py b/tests/test_monitor.py new file mode 100644 index 00000000..5e28c9af --- /dev/null +++ b/tests/test_monitor.py @@ -0,0 +1,186 @@ +"""Exercise monitor accounting, HTTP boundaries and real disposable worker trees.""" +from __future__ import annotations + +import json +import re +import sys +import threading +import time +import urllib.error +import urllib.request +from pathlib import Path + +import psutil +import pytest + +from re_agent.cli.main import build_parser +from re_agent.monitor.server import Monitor, make_server + + +def save(path, functions): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"functions": functions}), encoding="utf-8") + + +def test_sessions_count_latest_result_once_and_tolerate_partial_writes(tmp_path): + a, b = tmp_path / "a.json", tmp_path / "b.json" + save(a, {"f": {"address": "0x00100", "success": False, "rounds_used": 3, "timestamp": "1"}}) + save(b, {"f": {"address": "100", "success": True, "rounds_used": 1, "timestamp": "2"}, + "g": {"address": "200", "success": False, "rounds_used": 2}, "malformed": []}) + monitor = Monitor(tmp_path, tmp_path / "state", ["*.json", "a.json"]) + state = monitor.snapshot() + assert (state["completed"], state["passed"], state["failed"], state["rounds"]) == (2, 1, 1, 3) + assert state["phase"] == "read-only" + assert not state["controls"] + b.write_text('{"functions":', encoding="utf-8") + assert monitor.snapshot()["completed"] == 2 + b.unlink() + assert monitor.snapshot()["passed"] == 0 + assert monitor.snapshot()["completed"] == 1 + + +def test_log_tail_and_recent_results_are_bounded(tmp_path): + (tmp_path / "activity.log").write_bytes(b"x" * 20000 + b"END") + save(tmp_path / "session.json", {str(n): {"address": hex(n), "success": True} for n in range(30)}) + state = Monitor(tmp_path, tmp_path / "state", ["session.json"], "*.log").snapshot() + assert len(state["log"]) == 7000 + assert state["log"].endswith("END") + assert state["completed"] == 30 + assert len(state["recent"]) == 18 + + +@pytest.mark.parametrize("pattern", ["../*.json", "/absolute/session.json"]) +def test_patterns_cannot_escape_work_dir(tmp_path, pattern): + with pytest.raises(ValueError, match="relative"): + Monitor(tmp_path, tmp_path / "state", [pattern]) + + +def test_read_only_monitor_cannot_start_or_stop(tmp_path): + monitor = Monitor(tmp_path, tmp_path / "state", []) + for action in (monitor.start, monitor.stop): + with pytest.raises(ValueError, match="Read-only"): + action() + assert not (tmp_path / "state").exists() + + +def test_worker_tree_stop_duplicate_start_and_host_reconnect(tmp_path): + script = tmp_path / "worker with spaces.py" + script.write_text( + "import pathlib, subprocess, sys, time\n" + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(120)'])\n" + "pathlib.Path('child.pid').write_text(str(child.pid))\n" + "time.sleep(120)\n", encoding="utf-8") + command = [sys.executable, str(script)] + first = Monitor(tmp_path, tmp_path / "state", [], worker=command) + second = None + try: + first.start() + pid = first.process.pid + assert first.start()["pid"] == pid + second = Monitor(tmp_path, tmp_path / "state", [], worker=command) + assert second.active() + assert second.start()["pid"] == pid + deadline = time.monotonic() + 10 + while not (tmp_path / "child.pid").exists() and time.monotonic() < deadline: + time.sleep(.02) + child = psutil.Process(int((tmp_path / "child.pid").read_text())) + second.stop() + assert not second.active() + assert not first.active() + deadline = time.monotonic() + 5 + while child.is_running() and child.status() != psutil.STATUS_ZOMBIE and time.monotonic() < deadline: + time.sleep(.02) + assert not child.is_running() or child.status() == psutil.STATUS_ZOMBIE + first.start() + assert first.active() + first.stop() + finally: + first.stop() + if second: + second.stop() + + +def test_stale_identity_and_unrelated_process_are_not_adopted(tmp_path): + state = tmp_path / "state" + state.mkdir() + current = psutil.Process() + worker = [sys.executable, "not-the-running-worker.py"] + record = {"pid": current.pid, "created": current.create_time(), "command": worker, "cwd": str(tmp_path)} + (state / "worker.json").write_text(json.dumps(record)) + monitor = Monitor(tmp_path, state, [], worker=worker) + assert not monitor.active() + monitor.stop() + assert current.is_running() + record.update(created=0, command=current.cmdline()) + (state / "worker.json").write_text(json.dumps(record)) + assert not Monitor(tmp_path, state, [], worker=current.cmdline()).active() + + +@pytest.fixture +def http_monitor(tmp_path): + monitor = Monitor(tmp_path, tmp_path / "state", []) + server = make_server(monitor, 0) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield monitor, f"http://127.0.0.1:{server.server_port}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def test_http_serves_packaged_ui_and_read_only_status(http_monitor): + _, url = http_monitor + with urllib.request.urlopen(url) as response: + html = response.read().decode() + assert "Reconstruction, live." in html + assert "__TOKEN__" not in html + assert response.headers["Cache-Control"] == "no-store" + with urllib.request.urlopen(url + "/api/status") as response: + state = json.load(response) + assert not state["controls"] + + +@pytest.mark.parametrize("headers", [{}, {"X-Control-Token": "wrong"}, {"Host": "untrusted.example"}]) +def test_http_rejects_unauthorized_control(http_monitor, headers): + _, url = http_monitor + req = urllib.request.Request(url + "/api/start", method="POST", headers=headers) + with pytest.raises(urllib.error.HTTPError) as error: + urllib.request.urlopen(req) + assert error.value.code == 403 + + +def test_http_origin_rebinding_and_no_arbitrary_command(http_monitor, monkeypatch): + monitor, url = http_monitor + with urllib.request.urlopen(url) as response: + token = re.search(r"const token\s*=\s*[\"']([^\"']+)[\"']", response.read().decode()).group(1) + for path, headers in [("/", {"Host": "other.example"}), + ("/api/start", {"X-Control-Token": token, "Origin": "https://other.example"})]: + req = urllib.request.Request(url + path, method="GET" if path == "/" else "POST", headers=headers) + with pytest.raises(urllib.error.HTTPError) as error: + urllib.request.urlopen(req) + assert error.value.code == 403 + calls = [] + monkeypatch.setattr(monitor, "start", lambda: calls.append("configured worker") or {"message": "started"}) + req = urllib.request.Request(url + "/api/start", data=b'{"command":["untrusted"]}', + headers={"X-Control-Token": token, "Origin": url}, method="POST") + with urllib.request.urlopen(req) as response: + assert response.status == 200 + assert calls == ["configured worker"] + + +def test_cli_parses_worker_argv_without_shell_interpretation(): + args = build_parser().parse_args(["monitor", "--total", "10", "--session-glob", "batches/*/progress.json", + "--worker", "python", "script with spaces.py", "--flag", "a&b"]) + assert args.command == "monitor" + assert args.worker == ["python", "script with spaces.py", "--flag", "a&b"] + assert args.total == 10 + + +def test_packaged_html_uses_text_nodes_for_untrusted_results(): + import re_agent.monitor.server as server + + html = Path(server.__file__).with_name("index.html").read_text(encoding="utf-8") + assert "textContent" in html + assert "innerHTML" not in html