feat: distribute agent detection manifests

3 files changed, 1012 insertions, 0 deletions

diff --git a/scripts/agent_detection_manifest_check.py b/scripts/agent_detection_manifest_check.py
new file mode 100644
index 00000000..46dcdf69
--- /dev/null
+++ b/scripts/agent_detection_manifest_check.py
@@ -0,0 +1,334 @@
+#!/usr/bin/env python3
+"""Validate bundled and published agent detection manifests."""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+import tomllib
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_BUNDLED_DIR = PROJECT_ROOT / "src" / "detect" / "manifests"
+DEFAULT_WEBSITE_DIR = PROJECT_ROOT / "website" / "agent-detection"
+ENGINE_SOURCE = PROJECT_ROOT / "src" / "detect" / "manifest_update.rs"
+
+MANIFEST_KEYS = {"id", "version", "min_engine_version", "updated_at", "aliases", "rules"}
+RULE_KEYS = {
+    "id",
+    "state",
+    "priority",
+    "region",
+    "visible_idle",
+    "visible_blocker",
+    "visible_working",
+    "skip_state_update",
+    "all",
+    "any",
+    "not",
+    "contains",
+    "regex",
+    "line_regex",
+}
+GATE_KEYS = {"all", "any", "not", "contains", "regex", "line_regex"}
+STATES = {"idle", "working", "blocked", "unknown"}
+REGION_RE = re.compile(
+    r"^(whole_recent|whole_recent_without_current_prompt_marker|after_last_prompt_marker|"
+    r"before_current_prompt_marker|current_prompt_block_marker|after_current_prompt_block_marker|"
+    r"prompt_box_body|above_prompt_box|last_non_empty_above_prompt_box|after_last_horizontal_rule|"
+    r"bottom_lines\([1-9][0-9]*\)|bottom_non_empty_lines\([1-9][0-9]*\))$"
+)
+VERSION_RE = re.compile(r"^[0-9]+(?:\.[0-9]+)*$")
+MAX_RULES_PER_MANIFEST = 128
+MAX_GATE_DEPTH = 8
+MAX_TOTAL_GATES = 512
+MAX_MATCHERS_PER_GATE = 32
+MAX_TOTAL_MATCHERS = 1024
+MAX_MATCHER_CHARS = 512
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--bundled-dir", type=Path, default=DEFAULT_BUNDLED_DIR)
+    parser.add_argument("--website-dir", type=Path, default=DEFAULT_WEBSITE_DIR)
+    parser.add_argument("--engine-version", type=int)
+    parser.add_argument(
+        "--require-website",
+        action="store_true",
+        help="fail if website agent-detection assets or catalog are missing",
+    )
+    return parser.parse_args()
+
+
+def read_engine_version(explicit: int | None) -> int:
+    if explicit is not None:
+        return explicit
+    content = ENGINE_SOURCE.read_text(encoding="utf-8")
+    match = re.search(r"MANIFEST_ENGINE_VERSION:\s*u32\s*=\s*([0-9]+)", content)
+    if not match:
+        raise CheckError(f"could not find MANIFEST_ENGINE_VERSION in {ENGINE_SOURCE}")
+    return int(match.group(1))
+
+
+class CheckError(Exception):
+    pass
+
+
+def load_toml(path: Path) -> dict:
+    try:
+        with path.open("rb") as fh:
+            value = tomllib.load(fh)
+    except tomllib.TOMLDecodeError as exc:
+        raise CheckError(f"{path}: invalid TOML: {exc}") from exc
+    if not isinstance(value, dict):
+        raise CheckError(f"{path}: TOML root must be a table")
+    return value
+
+
+def version_tuple(value: str, path: Path) -> tuple[int, ...]:
+    if not isinstance(value, str) or not VERSION_RE.fullmatch(value):
+        raise CheckError(f"{path}: version must be dotted numeric")
+    return tuple(int(part) for part in value.split("."))
+
+
+def compare_versions(left: str, right: str, path: Path) -> int:
+    left_parts = list(version_tuple(left, path))
+    right_parts = list(version_tuple(right, path))
+    width = max(len(left_parts), len(right_parts))
+    left_parts.extend([0] * (width - len(left_parts)))
+    right_parts.extend([0] * (width - len(right_parts)))
+    return (left_parts > right_parts) - (left_parts < right_parts)
+
+
+def validate_manifest(path: Path, engine_version: int) -> dict:
+    manifest = load_toml(path)
+    unknown = sorted(set(manifest) - MANIFEST_KEYS)
+    if unknown:
+        raise CheckError(f"{path}: unknown manifest field(s): {', '.join(unknown)}")
+
+    agent_id = manifest.get("id")
+    if not isinstance(agent_id, str) or not agent_id.strip():
+        raise CheckError(f"{path}: id must be a non-empty string")
+
+    version = manifest.get("version")
+    version_tuple(version, path)
+
+    min_engine = manifest.get("min_engine_version")
+    if not isinstance(min_engine, int):
+        raise CheckError(f"{path}: min_engine_version must be an integer")
+    if min_engine > engine_version:
+        raise CheckError(
+            f"{path}: min_engine_version {min_engine} exceeds engine {engine_version}"
+        )
+
+    aliases = manifest.get("aliases", [])
+    if not isinstance(aliases, list) or not all(isinstance(item, str) for item in aliases):
+        raise CheckError(f"{path}: aliases must be an array of strings")
+
+    rules = manifest.get("rules")
+    if not isinstance(rules, list) or not rules:
+        raise CheckError(f"{path}: rules must be a non-empty array")
+    if len(rules) > MAX_RULES_PER_MANIFEST:
+        raise CheckError(f"{path}: manifest exceeds max rule count {MAX_RULES_PER_MANIFEST}")
+    complexity = {"gates": 0, "matchers": 0}
+    for index, rule in enumerate(rules):
+        validate_rule(path, index, rule, complexity)
+
+    return manifest
+
+
+def validate_rule(path: Path, index: int, rule: object, complexity: dict[str, int]) -> None:
+    if not isinstance(rule, dict):
+        raise CheckError(f"{path}: rule {index} must be a table")
+    unknown = sorted(set(rule) - RULE_KEYS)
+    if unknown:
+        raise CheckError(f"{path}: rule {index} has unknown field(s): {', '.join(unknown)}")
+    rule_id = rule.get("id")
+    if not isinstance(rule_id, str) or not rule_id.strip():
+        raise CheckError(f"{path}: rule {index} id must be a non-empty string")
+    state = rule.get("state")
+    if state is not None and state not in STATES:
+        raise CheckError(f"{path}: rule {rule_id} has invalid state {state!r}")
+    region = rule.get("region", "whole_recent")
+    if not isinstance(region, str) or not REGION_RE.fullmatch(region):
+        raise CheckError(f"{path}: rule {rule_id} has invalid region {region!r}")
+    if rule.get("skip_state_update"):
+        if state != "unknown":
+            raise CheckError(f"{path}: rule {rule_id} skip_state_update requires state unknown")
+        if rule.get("visible_idle") or rule.get("visible_blocker") or rule.get("visible_working"):
+            raise CheckError(f"{path}: rule {rule_id} skip_state_update cannot set visible flags")
+    validate_gate(path, f"rule {rule_id}", rule, require_positive=True, depth=0, complexity=complexity)
+
+
+def validate_gate(
+    path: Path,
+    label: str,
+    gate: dict,
+    require_positive: bool,
+    depth: int,
+    complexity: dict[str, int],
+) -> None:
+    if depth > MAX_GATE_DEPTH:
+        raise CheckError(f"{path}: {label} exceeds max gate depth {MAX_GATE_DEPTH}")
+    complexity["gates"] += 1
+    if complexity["gates"] > MAX_TOTAL_GATES:
+        raise CheckError(f"{path}: manifest exceeds max gate count {MAX_TOTAL_GATES}")
+    unknown = sorted(set(gate) - (RULE_KEYS if label.startswith("rule ") else GATE_KEYS))
+    if unknown:
+        raise CheckError(f"{path}: {label} has unknown gate field(s): {', '.join(unknown)}")
+    matcher_count = 0
+    for key in ("contains", "regex", "line_regex"):
+        values = gate.get(key, [])
+        if not isinstance(values, list) or not all(isinstance(item, str) for item in values):
+            raise CheckError(f"{path}: {label} {key} must be an array of strings")
+        matcher_count += len(values)
+        for value in values:
+            if len(value) > MAX_MATCHER_CHARS:
+                raise CheckError(f"{path}: {label} matcher exceeds max length {MAX_MATCHER_CHARS}")
+    if matcher_count > MAX_MATCHERS_PER_GATE:
+        raise CheckError(f"{path}: {label} exceeds max direct matcher count {MAX_MATCHERS_PER_GATE}")
+    complexity["matchers"] += matcher_count
+    if complexity["matchers"] > MAX_TOTAL_MATCHERS:
+        raise CheckError(f"{path}: manifest exceeds max matcher count {MAX_TOTAL_MATCHERS}")
+    nested_any = gate.get("any", [])
+    nested_all = gate.get("all", [])
+    nested_not = gate.get("not", [])
+    for key, values in (("any", nested_any), ("all", nested_all), ("not", nested_not)):
+        if not isinstance(values, list):
+            raise CheckError(f"{path}: {label} {key} must be an array")
+    if require_positive and not has_positive_matcher(gate):
+        raise CheckError(f"{path}: {label} must contain a positive matcher")
+    for idx, nested in enumerate(nested_any):
+        validate_nested_gate(path, f"{label} any[{idx}]", nested, require_positive=True, depth=depth + 1, complexity=complexity)
+    for idx, nested in enumerate(nested_all):
+        validate_nested_gate(path, f"{label} all[{idx}]", nested, require_positive=True, depth=depth + 1, complexity=complexity)
+    for idx, nested in enumerate(nested_not):
+        validate_nested_gate(path, f"{label} not[{idx}]", nested, require_positive=False, depth=depth + 1, complexity=complexity)
+
+
+def validate_nested_gate(
+    path: Path,
+    label: str,
+    gate: object,
+    require_positive: bool,
+    depth: int,
+    complexity: dict[str, int],
+) -> None:
+    if not isinstance(gate, dict):
+        raise CheckError(f"{path}: {label} must be a table")
+    if not require_positive and not has_any_matcher(gate):
+        raise CheckError(f"{path}: {label} must contain a matcher")
+    validate_gate(path, label, gate, require_positive=require_positive, depth=depth, complexity=complexity)
+
+
+def has_positive_matcher(gate: dict) -> bool:
+    return bool(gate.get("contains") or gate.get("regex") or gate.get("line_regex") or gate.get("any") or gate.get("all"))
+
+
+def has_any_matcher(gate: dict) -> bool:
+    return bool(
+        gate.get("contains")
+        or gate.get("regex")
+        or gate.get("line_regex")
+        or gate.get("any")
+        or gate.get("all")
+        or gate.get("not")
+    )
+
+
+def load_manifest_dir(path: Path, engine_version: int) -> dict[str, tuple[Path, dict]]:
+    if not path.is_dir():
+        raise CheckError(f"{path}: manifest directory is missing")
+    manifests: dict[str, tuple[Path, dict]] = {}
+    for manifest_path in sorted(path.glob("*.toml")):
+        if manifest_path.name == "index.toml":
+            continue
+        manifest = validate_manifest(manifest_path, engine_version)
+        agent_id = manifest["id"]
+        if agent_id in manifests:
+            raise CheckError(
+                f"{manifest_path}: duplicate manifest id {agent_id!r}; already seen in {manifests[agent_id][0]}"
+            )
+        manifests[agent_id] = (manifest_path, manifest)
+    if not manifests:
+        raise CheckError(f"{path}: no manifests found")
+    return manifests
+
+
+def validate_catalog(
+    website_dir: Path,
+    bundled: dict[str, tuple[Path, dict]],
+    engine_version: int,
+) -> None:
+    catalog_path = website_dir / "index.toml"
+    catalog = load_toml(catalog_path)
+    if set(catalog) != {"schema_version", "agents"}:
+        raise CheckError(f"{catalog_path}: expected only schema_version and agents")
+    if catalog.get("schema_version") != 1:
+        raise CheckError(f"{catalog_path}: schema_version must be 1")
+    agents = catalog.get("agents")
+    if not isinstance(agents, list):
+        raise CheckError(f"{catalog_path}: agents must be an array")
+
+    seen: dict[str, str] = {}
+    for entry in agents:
+        if not isinstance(entry, dict) or set(entry) != {"id", "path"}:
+            raise CheckError(f"{catalog_path}: each agent entry must contain id and path")
+        agent_id = entry["id"]
+        rel_path = entry["path"]
+        if not isinstance(agent_id, str) or not isinstance(rel_path, str):
+            raise CheckError(f"{catalog_path}: agent id and path must be strings")
+        if agent_id in seen:
+            raise CheckError(f"{catalog_path}: duplicate catalog agent {agent_id}")
+        if "://" in rel_path or rel_path.startswith("/") or ".." in Path(rel_path).parts:
+            raise CheckError(f"{catalog_path}: unsafe path for {agent_id}: {rel_path}")
+        if agent_id not in bundled:
+            raise CheckError(f"{catalog_path}: unknown agent {agent_id}; binary cannot identify it")
+        manifest_path = website_dir / rel_path
+        manifest = validate_manifest(manifest_path, engine_version)
+        if manifest["id"] != agent_id:
+            raise CheckError(f"{manifest_path}: id {manifest['id']} does not match catalog {agent_id}")
+        seen[agent_id] = rel_path
+
+        bundled_path, bundled_manifest = bundled[agent_id]
+        cmp = compare_versions(manifest["version"], bundled_manifest["version"], manifest_path)
+        if cmp < 0:
+            raise CheckError(
+                f"{manifest_path}: website version {manifest['version']} is lower than bundled "
+                f"{bundled_manifest['version']} in {bundled_path}"
+            )
+        if cmp == 0 and manifest_path.read_text(encoding="utf-8") != bundled_path.read_text(encoding="utf-8"):
+            raise CheckError(
+                f"{manifest_path}: same version as bundled {bundled_manifest['version']} but content differs"
+            )
+
+    missing = sorted(set(bundled) - set(seen))
+    if missing:
+        raise CheckError(f"{catalog_path}: missing bundled agent(s): {', '.join(missing)}")
+
+    catalog_paths = set(seen.values()) | {"index.toml"}
+    extra = sorted(path.name for path in website_dir.glob("*.toml") if path.name not in catalog_paths)
+    if extra:
+        raise CheckError(f"{website_dir}: TOML file(s) not listed in catalog: {', '.join(extra)}")
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        engine_version = read_engine_version(args.engine_version)
+        bundled = load_manifest_dir(args.bundled_dir, engine_version)
+        if args.require_website or args.website_dir.exists():
+            if not args.website_dir.is_dir():
+                raise CheckError(f"{args.website_dir}: website manifest directory is missing")
+            validate_catalog(args.website_dir, bundled, engine_version)
+    except CheckError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+    print("agent detection manifests ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/capture_agent_screen.py b/scripts/capture_agent_screen.py
new file mode 100755
index 00000000..4694c35d
--- /dev/null
+++ b/scripts/capture_agent_screen.py
@@ -0,0 +1,572 @@
+#!/usr/bin/env python3
+"""Capture repeated Herdr pane reads for agent screen detection fixtures."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+DEFAULT_OUT_DIR = Path(".local/agent-screen-captures")
+DEFAULT_PANE = "harness-test"
+STATE_CHOICES = {
+    "i": "idle",
+    "idle": "idle",
+    "w": "working",
+    "working": "working",
+    "b": "blocked",
+    "blocked": "blocked",
+    "d": "done",
+    "done": "done",
+    "u": "unknown",
+    "unknown": "unknown",
+    "c": "custom",
+    "custom": "custom",
+}
+
+
+@dataclass
+class CommandResult:
+    code: int
+    stdout: bytes
+    stderr: bytes
+
+
+@dataclass
+class PaneMatch:
+    pane_id: str
+    label: str | None
+    name: str | None
+    agent: str | None
+    title: str | None
+    display_agent: str | None
+    raw: dict[str, Any] | None
+    agent_raw: dict[str, Any] | None = None
+
+
+def main() -> int:
+    args = parse_args()
+    run_dir = args.out / datetime.now().strftime("%Y%m%d-%H%M%S")
+    run_dir.mkdir(parents=True, exist_ok=True)
+
+    print(f"writing captures under {run_dir}")
+    print("state shortcuts: i=idle, w=working, b=blocked, d=done, u=unknown, c=custom, q=quit")
+
+    capture_index = 1
+    while True:
+        pane = resolve_target(args.herdr, args.pane, args.agent)
+        if pane is None:
+            if args.agent:
+                print(f"agent '{args.agent}' was not found by `herdr agent get`")
+            else:
+                print(f"pane '{args.pane}' was not found by `herdr pane list`")
+                print("pass a pane id with --pane, or rename the target pane to harness-test")
+            return 1
+
+        print(f"\npane found: {pane.pane_id}, agent: {agent_display(pane)}{format_pane_context(pane)}")
+        state = prompt_state()
+        if state is None:
+            break
+
+        name = prompt_name(pane, state)
+        if name is None:
+            break
+
+        capture_dir = run_dir / f"{capture_index:03d}-{slugify(name)}"
+        capture_dir.mkdir(parents=True, exist_ok=False)
+        print(f"capturing {args.samples} sample(s) into {capture_dir}")
+
+        started_at = iso_now()
+        failures = capture_case(args, pane, state, name, capture_dir)
+        write_metadata(
+            capture_dir,
+            args=args,
+            pane=pane,
+            state=state,
+            name=name,
+            started_at=started_at,
+            finished_at=iso_now(),
+            failures=failures,
+        )
+
+        if failures:
+            print(f"saved with {len(failures)} command failure(s); see metadata.toml")
+        else:
+            print("saved")
+
+        capture_index += 1
+        if args.once:
+            break
+
+    print("done")
+    return 0
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Interactively capture Herdr pane screen reads for agent detection fixture work."
+    )
+    parser.add_argument(
+        "--pane",
+        default=DEFAULT_PANE,
+        help=f"pane label/title/id to capture when --agent is not set (default: {DEFAULT_PANE})",
+    )
+    parser.add_argument(
+        "--agent",
+        help="agent target to capture; resolved with `herdr agent get`",
+    )
+    parser.add_argument(
+        "--out",
+        type=Path,
+        default=DEFAULT_OUT_DIR,
+        help=f"output directory (default: {DEFAULT_OUT_DIR})",
+    )
+    parser.add_argument(
+        "--samples",
+        type=positive_int,
+        default=5,
+        help="samples to capture per state (default: 5)",
+    )
+    parser.add_argument(
+        "--interval",
+        type=non_negative_float,
+        default=1.0,
+        help="seconds between samples (default: 1.0)",
+    )
+    parser.add_argument(
+        "--lines",
+        type=positive_int,
+        default=120,
+        help="recent-buffer lines to save per sample (default: 120)",
+    )
+    parser.add_argument(
+        "--herdr",
+        default="herdr",
+        help="Herdr CLI binary to call (default: herdr)",
+    )
+    parser.add_argument(
+        "--once",
+        action="store_true",
+        help="capture one state and exit",
+    )
+    return parser.parse_args()
+
+
+def positive_int(value: str) -> int:
+    parsed = int(value)
+    if parsed <= 0:
+        raise argparse.ArgumentTypeError("must be greater than zero")
+    return parsed
+
+
+def non_negative_float(value: str) -> float:
+    parsed = float(value)
+    if parsed < 0:
+        raise argparse.ArgumentTypeError("must be zero or greater")
+    return parsed
+
+
+def resolve_pane(herdr: str, pane_ref: str) -> PaneMatch | None:
+    result = run_command([herdr, "pane", "list"])
+    if result.code != 0:
+        pane = fallback_pane_id(pane_ref)
+        return enrich_pane_with_agent(herdr, pane) if pane else None
+
+    try:
+        response = json.loads(result.stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        pane = fallback_pane_id(pane_ref)
+        return enrich_pane_with_agent(herdr, pane) if pane else None
+
+    panes = response.get("result", {}).get("panes", [])
+    if not isinstance(panes, list):
+        pane = fallback_pane_id(pane_ref)
+        return enrich_pane_with_agent(herdr, pane) if pane else None
+
+    exact_matches = []
+    loose_matches = []
+    for pane in panes:
+        if not isinstance(pane, dict):
+            continue
+        pane_id = string_value(pane.get("pane_id"))
+        if pane_id is None:
+            continue
+        fields = pane_match_fields(pane)
+        if any(field == pane_ref for field in fields):
+            exact_matches.append(pane)
+        elif any(field and pane_ref in field for field in fields):
+            loose_matches.append(pane)
+
+    matches = exact_matches or loose_matches
+    if len(matches) == 1:
+        return enrich_pane_with_agent(herdr, pane_from_dict(matches[0]))
+    if len(matches) > 1:
+        print(f"pane ref '{pane_ref}' matched multiple panes:")
+        for pane in matches:
+            print(f"  {pane.get('pane_id')}{format_pane_context(pane_from_dict(pane))}")
+        return None
+
+    pane = fallback_pane_id(pane_ref)
+    return enrich_pane_with_agent(herdr, pane) if pane else None
+
+
+def resolve_target(herdr: str, pane_ref: str, agent_ref: str | None) -> PaneMatch | None:
+    if agent_ref:
+        agent = get_agent(herdr, agent_ref)
+        if agent is None:
+            return None
+        pane = pane_from_agent_dict(agent)
+        return enrich_pane_with_pane_list(herdr, pane)
+    return resolve_pane(herdr, pane_ref)
+
+
+def get_agent(herdr: str, agent_ref: str) -> dict[str, Any] | None:
+    result = run_command([herdr, "agent", "get", agent_ref])
+    if result.code != 0:
+        return None
+    try:
+        response = json.loads(result.stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None
+    agent = response.get("result", {}).get("agent")
+    return agent if isinstance(agent, dict) else None
+
+
+def enrich_pane_with_agent(herdr: str, pane: PaneMatch) -> PaneMatch:
+    result = run_command([herdr, "agent", "list"])
+    if result.code != 0:
+        return pane
+    try:
+        response = json.loads(result.stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return pane
+    agents = response.get("result", {}).get("agents", [])
+    if not isinstance(agents, list):
+        return pane
+    for agent in agents:
+        if isinstance(agent, dict) and agent.get("pane_id") == pane.pane_id:
+            return merge_agent_into_pane(pane, agent)
+    return pane
+
+
+def enrich_pane_with_pane_list(herdr: str, pane: PaneMatch) -> PaneMatch:
+    result = run_command([herdr, "pane", "list"])
+    if result.code != 0:
+        return pane
+    try:
+        response = json.loads(result.stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return pane
+    panes = response.get("result", {}).get("panes", [])
+    if not isinstance(panes, list):
+        return pane
+    for candidate in panes:
+        if isinstance(candidate, dict) and candidate.get("pane_id") == pane.pane_id:
+            pane_info = pane_from_dict(candidate)
+            pane_info.name = pane.name or pane_info.name
+            pane_info.agent = pane.agent or pane_info.agent
+            pane_info.display_agent = pane.display_agent or pane_info.display_agent
+            pane_info.agent_raw = pane.agent_raw
+            return pane_info
+    return pane
+
+
+def merge_agent_into_pane(pane: PaneMatch, agent: dict[str, Any]) -> PaneMatch:
+    pane.name = string_value(agent.get("name")) or pane.name
+    pane.agent = string_value(agent.get("agent")) or pane.agent
+    pane.display_agent = string_value(agent.get("display_agent")) or pane.display_agent
+    pane.title = string_value(agent.get("title")) or pane.title
+    pane.agent_raw = agent
+    return pane
+
+
+def fallback_pane_id(pane_ref: str) -> PaneMatch | None:
+    if re.fullmatch(r"p_(?:[A-Za-z0-9]+_)?\d+", pane_ref) or re.fullmatch(
+        r"[A-Za-z0-9_]+-\d+", pane_ref
+    ):
+        return PaneMatch(
+            pane_id=normalize_pane_id(pane_ref),
+            label=None,
+            name=None,
+            agent=None,
+            title=None,
+            display_agent=None,
+            raw=None,
+        )
+    return None
+
+
+def normalize_pane_id(pane_ref: str) -> str:
+    return pane_ref
+
+
+def pane_match_fields(pane: dict[str, Any]) -> list[str]:
+    fields = [
+        string_value(pane.get("pane_id")),
+        string_value(pane.get("terminal_id")),
+        string_value(pane.get("label")),
+        string_value(pane.get("title")),
+        string_value(pane.get("agent")),
+        string_value(pane.get("display_agent")),
+    ]
+    return [field for field in fields if field]
+
+
+def pane_from_dict(pane: dict[str, Any]) -> PaneMatch:
+    return PaneMatch(
+        pane_id=string_value(pane.get("pane_id")) or "",
+        label=string_value(pane.get("label")),
+        name=None,
+        agent=string_value(pane.get("agent")),
+        title=string_value(pane.get("title")),
+        display_agent=string_value(pane.get("display_agent")),
+        raw=pane,
+    )
+
+
+def pane_from_agent_dict(agent: dict[str, Any]) -> PaneMatch:
+    return PaneMatch(
+        pane_id=string_value(agent.get("pane_id")) or "",
+        label=None,
+        name=string_value(agent.get("name")),
+        agent=string_value(agent.get("agent")),
+        title=string_value(agent.get("title")),
+        display_agent=string_value(agent.get("display_agent")),
+        raw=None,
+        agent_raw=agent,
+    )
+
+
+def agent_display(pane: PaneMatch) -> str:
+    return pane.agent or pane.display_agent or "unknown"
+
+
+def string_value(value: Any) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def format_pane_context(pane: PaneMatch) -> str:
+    details = []
+    if pane.name:
+        details.append(f"name={pane.name}")
+    if pane.label:
+        details.append(f"label={pane.label}")
+    if pane.title:
+        details.append(f"title={pane.title}")
+    if not details:
+        return ""
+    return " (" + ", ".join(details) + ")"
+
+
+def prompt_state() -> str | None:
+    while True:
+        raw = input("state [idle/working/blocked/done/unknown/custom/q]: ").strip()
+        if raw.lower() in {"q", "quit", "exit"}:
+            return None
+        if not raw:
+            continue
+        state = STATE_CHOICES.get(raw.lower())
+        if state == "custom":
+            custom = input("custom state label: ").strip()
+            if custom:
+                return custom
+            continue
+        if state:
+            return state
+        return raw
+
+
+def prompt_name(pane: PaneMatch, state: str) -> str | None:
+    default_parts = [
+        pane.agent or pane.display_agent or pane.name or pane.label or "agent",
+        state,
+        datetime.now().strftime("%H%M%S"),
+    ]
+    default_name = "-".join(slugify(part) for part in default_parts if part)
+    raw = input(f"capture name [{default_name}]: ").strip()
+    if raw.lower() in {"q", "quit", "exit"}:
+        return None
+    return raw or default_name
+
+
+def capture_case(
+    args: argparse.Namespace,
+    pane: PaneMatch,
+    state: str,
+    name: str,
+    capture_dir: Path,
+) -> list[str]:
+    failures: list[str] = []
+    for index in range(1, args.samples + 1):
+        prefix = f"sample-{index:03d}"
+        print(f"  sample {index}/{args.samples}")
+
+        commands = [
+            (
+                f"{prefix}.detection.txt",
+                [args.herdr, "pane", "read", pane.pane_id, "--source", "detection", "--format", "text"],
+            ),
+            (
+                f"{prefix}.detection.ansi",
+                [args.herdr, "pane", "read", pane.pane_id, "--source", "detection", "--format", "ansi"],
+            ),
+            (
+                f"{prefix}.recent.txt",
+                [
+                    args.herdr,
+                    "pane",
+                    "read",
+                    pane.pane_id,
+                    "--source",
+                    "recent",
+                    "--lines",
+                    str(args.lines),
+                    "--format",
+                    "text",
+                ],
+            ),
+            (
+                f"{prefix}.recent.ansi",
+                [
+                    args.herdr,
+                    "pane",
+                    "read",
+                    pane.pane_id,
+                    "--source",
+                    "recent",
+                    "--lines",
+                    str(args.lines),
+                    "--format",
+                    "ansi",
+                ],
+            ),
+            (
+                f"{prefix}.explain.json",
+                [args.herdr, "agent", "explain", pane.pane_id, "--json"],
+            ),
+        ]
+
+        for filename, command in commands:
+            result = run_command(command)
+            output_path = capture_dir / filename
+            if result.code == 0:
+                output_path.write_bytes(result.stdout)
+            else:
+                failures.append(f"{filename}: exit {result.code}: {' '.join(command)}")
+                output_path.with_suffix(output_path.suffix + ".stderr").write_bytes(result.stderr)
+                output_path.write_bytes(result.stdout)
+
+        sample_meta = {
+            "captured_at": iso_now(),
+            "sample": index,
+            "state": state,
+            "name": name,
+        }
+        (capture_dir / f"{prefix}.json").write_text(
+            json.dumps(sample_meta, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+
+        if index < args.samples:
+            time.sleep(args.interval)
+    return failures
+
+
+def run_command(command: list[str]) -> CommandResult:
+    try:
+        completed = subprocess.run(command, capture_output=True, check=False)
+    except FileNotFoundError as err:
+        return CommandResult(code=127, stdout=b"", stderr=str(err).encode("utf-8"))
+    return CommandResult(
+        code=completed.returncode,
+        stdout=completed.stdout,
+        stderr=completed.stderr,
+    )
+
+
+def write_metadata(
+    capture_dir: Path,
+    *,
+    args: argparse.Namespace,
+    pane: PaneMatch,
+    state: str,
+    name: str,
+    started_at: str,
+    finished_at: str,
+    failures: list[str],
+) -> None:
+    lines = [
+        f"name = {toml_string(name)}",
+        f"state = {toml_string(state)}",
+        f"started_at = {toml_string(started_at)}",
+        f"finished_at = {toml_string(finished_at)}",
+        f"pane_ref = {toml_string(args.pane)}",
+        f"agent_ref = {toml_optional_string(args.agent)}",
+        f"pane_id = {toml_string(pane.pane_id)}",
+        f"samples = {args.samples}",
+        f"interval_seconds = {args.interval}",
+        f"recent_lines = {args.lines}",
+        f"herdr = {toml_string(args.herdr)}",
+        f"pane_label = {toml_optional_string(pane.label)}",
+        f"agent_name = {toml_optional_string(pane.name)}",
+        f"pane_agent = {toml_optional_string(pane.agent)}",
+        f"display_agent = {toml_optional_string(pane.display_agent)}",
+        f"pane_title = {toml_optional_string(pane.title)}",
+        "",
+        "[commands]",
+        'detection_text = "herdr pane read <pane> --source detection --format text"',
+        'detection_ansi = "herdr pane read <pane> --source detection --format ansi"',
+        'recent_text = "herdr pane read <pane> --source recent --lines <n> --format text"',
+        'recent_ansi = "herdr pane read <pane> --source recent --lines <n> --format ansi"',
+        'explain = "herdr agent explain <pane> --json"',
+    ]
+    if failures:
+        lines.append("")
+        lines.append("failures = [")
+        for failure in failures:
+            lines.append(f"  {toml_string(failure)},")
+        lines.append("]")
+
+    (capture_dir / "metadata.toml").write_text("\n".join(lines) + "\n", encoding="utf-8")
+    if pane.raw is not None:
+        (capture_dir / "pane.json").write_text(
+            json.dumps(pane.raw, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+    if pane.agent_raw is not None:
+        (capture_dir / "agent.json").write_text(
+            json.dumps(pane.agent_raw, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+
+
+def toml_optional_string(value: str | None) -> str:
+    if value is None:
+        return '""'
+    return toml_string(value)
+
+
+def toml_string(value: str) -> str:
+    return json.dumps(value)
+
+
+def slugify(value: str) -> str:
+    slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip().lower())
+    slug = re.sub(r"-+", "-", slug).strip("-")
+    return slug or "capture"
+
+
+def iso_now() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/scripts/test_agent_detection_manifest_check.py b/scripts/test_agent_detection_manifest_check.py
new file mode 100644
index 00000000..3ab69b55
--- /dev/null
+++ b/scripts/test_agent_detection_manifest_check.py
@@ -0,0 +1,106 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from scripts import agent_detection_manifest_check as check
+
+
+def manifest(agent_id: str, version: str, contains: str = "ready") -> str:
+    return f'''id = "{agent_id}"
+version = "{version}"
+min_engine_version = 1
+updated_at = "2026-06-10T00:00:00Z"
+
+[[rules]]
+id = "idle"
+state = "idle"
+contains = ["{contains}"]
+'''
+
+
+def catalog(agent_id: str = "codex", path: str = "codex.toml") -> str:
+    return f'''schema_version = 1
+
+[[agents]]
+id = "{agent_id}"
+path = "{path}"
+'''
+
+
+class AgentDetectionManifestCheckTests(unittest.TestCase):
+    def test_validates_bundled_and_matching_website_catalog(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bundled = root / "bundled"
+            website = root / "website"
+            bundled.mkdir()
+            website.mkdir()
+            content = manifest("codex", "2026.06.10.1")
+            (bundled / "codex.toml").write_text(content)
+            (website / "codex.toml").write_text(content)
+            (website / "index.toml").write_text(catalog())
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=1)
+            check.validate_catalog(website, bundled_manifests, engine_version=1)
+
+    def test_rejects_website_version_lower_than_bundled(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bundled = root / "bundled"
+            website = root / "website"
+            bundled.mkdir()
+            website.mkdir()
+            (bundled / "codex.toml").write_text(manifest("codex", "2026.06.10.2"))
+            (website / "codex.toml").write_text(manifest("codex", "2026.06.10.1"))
+            (website / "index.toml").write_text(catalog())
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=1)
+            with self.assertRaisesRegex(check.CheckError, "lower than bundled"):
+                check.validate_catalog(website, bundled_manifests, engine_version=1)
+
+    def test_rejects_same_version_content_drift(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bundled = root / "bundled"
+            website = root / "website"
+            bundled.mkdir()
+            website.mkdir()
+            (bundled / "codex.toml").write_text(manifest("codex", "2026.06.10.1", "ready"))
+            (website / "codex.toml").write_text(manifest("codex", "2026.06.10.1", "changed"))
+            (website / "index.toml").write_text(catalog())
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=1)
+            with self.assertRaisesRegex(check.CheckError, "same version"):
+                check.validate_catalog(website, bundled_manifests, engine_version=1)
+
+    def test_rejects_unknown_catalog_agent(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bundled = root / "bundled"
+            website = root / "website"
+            bundled.mkdir()
+            website.mkdir()
+            (bundled / "codex.toml").write_text(manifest("codex", "2026.06.10.1"))
+            (website / "newagent.toml").write_text(manifest("newagent", "2026.06.10.1"))
+            (website / "index.toml").write_text(catalog("newagent", "newagent.toml"))
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=1)
+            with self.assertRaisesRegex(check.CheckError, "unknown agent"):
+                check.validate_catalog(website, bundled_manifests, engine_version=1)
+
+    def test_rejects_manifest_requiring_newer_engine(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundled = Path(tmp) / "bundled"
+            bundled.mkdir()
+            (bundled / "codex.toml").write_text(
+                manifest("codex", "2026.06.10.1").replace(
+                    "min_engine_version = 1", "min_engine_version = 2"
+                )
+            )
+
+            with self.assertRaisesRegex(check.CheckError, "exceeds engine"):
+                check.load_manifest_dir(bundled, engine_version=1)
+
+
+if __name__ == "__main__":
+    unittest.main()