wip: close the six PRHunt delay defects

Coordinator lease so one hunt has one owner (#63). Checkpoint packet for
candidates that are ready before the quota is met (#64). Review budget
counted across resume-review calls (#65). Target assessments offloaded and
captured streams capped, which is what made one orchestration record 124 MB
(#66). Usage-limit and infrastructure stops classified apart from candidate
failures, with the exact resume command (#67). Narrow-first duplicate
discovery that stops on a confirmed duplicate (#68).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: wolfgang-aura

mailman/cli.py

@@ -122,7 +122,7 @@ def _build_parser() -> argparse.ArgumentParser:
         "action",
         choices=(
             "init", "add", "drop", "restore", "status", "finish", "list",
-            "escalate", "refresh-procedure",
+            "escalate", "refresh-procedure", "lease", "release",
         ),
     )
     hunt.add_argument("hunt_id", nargs="?")
@@ -132,6 +132,16 @@ def _build_parser() -> argparse.ArgumentParser:
         hunt.add_argument(f"--{role}-model")
     hunt.add_argument("--reason")
     hunt.add_argument("--evidence")
+    hunt.add_argument(
+        "--owner",
+        help="this coordinator's lease token, printed by hunt init. One hunt "
+        "has one owner; a second coordinator must take it over explicitly",
+    )
+    hunt.add_argument(
+        "--takeover",
+        action="store_true",
+        help="take a live lease from another coordinator; needs --reason",
+    )
     hunt.add_argument("--attempted")
     hunt.add_argument("--why-user")
     hunt.add_argument("--user-action")
@@ -288,6 +298,14 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     duplicate.add_argument("run_id")
     duplicate.add_argument("--query", required=True)
+    duplicate.add_argument(
+        "--symbol",
+        action="append",
+        dest="symbols",
+        default=[],
+        help="a function, class or file the change touches, repeatable. "
+        "Searched with the issue number before the broad listing",
+    )
     duplicate.add_argument("--limit", type=int, default=30)
     duplicate.add_argument("--executable")
     duplicate.add_argument("--timeout", type=float, default=60)
@@ -480,6 +498,13 @@ def _build_parser() -> argparse.ArgumentParser:
         help="how hard a Codex model is asked to think, recorded with the run",
     )
     orchestrate_parser.add_argument("--max-revisions", type=int, default=1)
+    orchestrate_parser.add_argument(
+        "--max-review-cycles",
+        type=int,
+        default=3,
+        help="reviewer passes this run may spend in total, counted across "
+        "every orchestrate and resume-review call",
+    )
     orchestrate_parser.add_argument(
         "--acknowledge-prior-attempts",
         action="store_true",
@@ -665,7 +690,8 @@ def _hunt(arguments: argparse.Namespace) -> int:
             raise ValueError("ask for primary and reviewer model IDs, then pass all four model flags")
         record = hunt.create_hunt(root, int(arguments.hunt_id), primary=arguments.primary,
                                   primary_model=arguments.primary_model, reviewer=arguments.reviewer,
-                                  reviewer_model=arguments.reviewer_model)
+                                  reviewer_model=arguments.reviewer_model,
+                                  owner=arguments.owner)
         print(json.dumps(record, indent=2))
         return 0
     if arguments.action == "refresh-procedure":
@@ -677,6 +703,19 @@ def _hunt(arguments: argparse.Namespace) -> int:
         record["procedure_sha256"] = hashlib.sha256(hunt.PROCEDURE.read_bytes()).hexdigest()
         hunt.save(path, record)
     record = hunt.load_hunt(root, arguments.hunt_id)
+    if arguments.action == "lease":
+        lease = hunt.acquire_lease(
+            root, record, owner=arguments.owner,
+            takeover_reason=arguments.reason if arguments.takeover else None,
+        )
+        print(json.dumps(lease, indent=2))
+        return 0
+    if arguments.action == "release":
+        hunt.release_lease(root, record, owner=arguments.owner)
+        print(json.dumps({"hunt_id": record["hunt_id"], "lease": None}, indent=2))
+        return 0
+    if arguments.action in ("add", "drop", "restore", "escalate", "finish"):
+        hunt.require_lease(record, arguments.owner)
     if arguments.action == "add":
         if not arguments.run_id:
             raise ValueError("provide the run ID to add")
@@ -901,12 +940,14 @@ def _duplicate_search(arguments: argparse.Namespace) -> int:
         executable=arguments.executable,
         timeout_seconds=arguments.timeout,
         limit=arguments.limit,
+        symbols=arguments.symbols,
     )
     print(
         json.dumps(
             {
                 "run_id": run.run_id,
                 "repository": record["repository"],
+                "decided_by": record.get("decided_by"),
                 "query": record["query"],
                 "success": record["success"],
                 "complete": record["complete"],
@@ -1409,6 +1450,7 @@ def _orchestrate(arguments: argparse.Namespace) -> int:
         agent_timeout_seconds=arguments.agent_timeout,
         verification_timeout_seconds=arguments.verification_timeout,
         max_revisions=arguments.max_revisions,
+        max_review_cycles=arguments.max_review_cycles,
         announce=_emit,
         acknowledge_prior_attempts=arguments.acknowledge_prior_attempts,
         acknowledge_claims=arguments.acknowledge_claims,

mailman/executor.py

@@ -27,7 +27,19 @@ class CommandResult:
     environment: dict[str, str]
 
     def to_dict(self) -> dict[str, object]:
-        return asdict(self)
+        """The recorded form of this result, with captured streams capped.
+
+        `to_dict` is only ever written to an evidence file; every reader that
+        needs the whole stream reads `stdout` or `stderr` directly. Capping
+        here stops one agent transcript from making a record unreadable. See
+        https://github.com/wolfgang-aura/Mailman/issues/66.
+        """
+        from mailman.limits import truncate_stream
+
+        record = asdict(self)
+        for key in ("stdout", "stderr"):
+            record[key] = truncate_stream(record[key])
+        return record
 
 
 def _environment_metadata() -> dict[str, str]:

mailman/health.py

@@ -0,0 +1,93 @@
+"""Run health states a coordinator must not mistake for a candidate failure.
+
+Both agent tasks in hunt `20260907T164341Z-1ca91a` stopped mid-stage when the
+shared model allowance ran out. Neither left a machine-readable resume point,
+so the next session reconstructed stage state from run directories by hand.
+
+See https://github.com/wolfgang-aura/Mailman/issues/67.
+"""
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+from mailman.models import utc_now
+
+HEALTH_FILENAME = "health.json"
+
+USAGE_LIMIT = "USAGE_LIMIT"
+INFRASTRUCTURE = "INFRASTRUCTURE"
+
+#: Phrases a provider CLI prints when the account, not the candidate, is what
+#: stopped the stage. Matched case-insensitively against the agent's stop
+#: reason and captured streams.
+_USAGE_PATTERNS = (
+    r"usage limit",
+    r"rate limit",
+    r"quota (?:exceeded|exhausted)",
+    r"insufficient (?:credit|quota)",
+    r"upgrade your plan",
+    r"try again at\b",
+    r"429\b",
+)
+
+#: The host, not the code under test, failed. A temporary directory that
+#: cannot be created is not a defect in the candidate.
+_INFRASTRUCTURE_PATTERNS = (
+    r"permission denied: .*(?:temp|tmp)",
+    r"\[Errno 13\]",
+    r"WinError 5\b",
+    r"no space left on device",
+    r"could not create cache",
+)
+
+_USAGE = re.compile("|".join(_USAGE_PATTERNS), re.IGNORECASE)
+_INFRASTRUCTURE = re.compile("|".join(_INFRASTRUCTURE_PATTERNS), re.IGNORECASE)
+
+
+def classify(*texts: str | None) -> str | None:
+    """Name the non-candidate cause in these texts, if there is one."""
+    joined = "\n".join(text for text in texts if text)
+    if not joined:
+        return None
+    if _USAGE.search(joined):
+        return USAGE_LIMIT
+    if _INFRASTRUCTURE.search(joined):
+        return INFRASTRUCTURE
+    return None
+
+
+def record(run_directory: Path, *, state: str, stage: str, resume_command: str,
+           detail: str) -> Path:
+    destination = run_directory / HEALTH_FILENAME
+    destination.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "state": state,
+                "stage": stage,
+                "resume_command": resume_command,
+                "detail": detail,
+                "at": utc_now(),
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    return destination
+
+
+def load(run_directory: Path) -> dict | None:
+    path = run_directory / HEALTH_FILENAME
+    if not path.is_file():
+        return None
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+def clear(run_directory: Path) -> None:
+    (run_directory / HEALTH_FILENAME).unlink(missing_ok=True)