feat: triage agentic workflow (IMPL-0025) — WIP

backend/cliff/agents/runtime/deps.py

@@ -14,6 +14,27 @@
 from typing import Any
 
 
+@dataclass
+class ReadBudget:
+    """A mutable cap on the total bytes the read/grep tools may return within a
+    single agent run (ADR-0052). The Deep dive agents walk large real repos, so
+    without a cap the accumulated tool output overflows the model context window
+    (the 200K crash seen on the first live run). ``None`` on the deps (the
+    executor's case) means unlimited — current behaviour is unchanged."""
+
+    remaining: int  # bytes
+
+    def take(self, n: int) -> bool:
+        """Reserve *n* bytes. Returns False when the request would exceed the cap
+        (the caller then returns a short 'budget exhausted' marker instead of more
+        content). Rejecting an over-budget request — rather than allowing the first
+        one to go negative — is what actually prevents the context-window overflow."""
+        if n > self.remaining:
+            return False
+        self.remaining -= n
+        return True
+
+
 @dataclass(frozen=True)
 class WorkspaceDeps:
     """Per-run context handed to Pydantic AI as ``deps``."""
@@ -30,3 +51,6 @@ class WorkspaceDeps:
     # surface to prompt against. When True the ``ask`` tier auto-proceeds;
     # the ``deny`` tier (catastrophic commands) still denies.
     auto_approve: bool = False
+    # Cumulative read/grep byte cap for this run (ADR-0052 Deep dive). None =
+    # unlimited (the executor / repo-action path — unchanged).
+    read_budget: ReadBudget | None = None

backend/cliff/agents/runtime/model_tiers.py

@@ -0,0 +1,71 @@
+"""Deep dive model tiering (ADR-0052 §4, amends ADR-0037).
+
+ADR-0037 keeps one canonical provider + one model. The Deep dive needs three
+tiers — cheap (volume), strong (the moat: reachability + exploit planning), and
+judge (the adversarial challenge, which must out-rank the generator per
+ADR-0050's anti-self-preference rule). We *derive* the tier map from the
+configured provider's own lineup, so there's still one credential.
+
+Thin-lineup providers (ollama / custom) fall back to the single configured model
+for every tier — the judge is then not independent (the caller logs that).
+"""
+
+from __future__ import annotations
+
+#: (cheap, strong, judge) model ids per provider, without the provider prefix.
+_LINEUP: dict[str, tuple[str, str, str]] = {
+    "anthropic": ("claude-haiku-4-5", "claude-sonnet-4-6", "claude-opus-4-8"),
+    "openrouter": (
+        "anthropic/claude-haiku-4.5",
+        "anthropic/claude-sonnet-4.6",
+        "anthropic/claude-opus-4.8",
+    ),
+    "openai": ("gpt-5-mini", "gpt-5", "gpt-5"),
+    "google": ("gemini-2.5-flash", "gemini-2.5-pro", "gemini-2.5-pro"),
+}
+
+TIERS = ("cheap", "strong", "judge")
+
+
+def resolve_tier_model_ids(model_full_id: str) -> dict[str, str]:
+    """Derive ``{cheap, strong, judge}`` ``<provider>/<model>`` ids.
+
+    Falls back to the single configured model for every tier when the provider
+    has no known lineup (ollama / custom) or the id isn't ``provider/model``.
+    """
+    provider, sep, _ = model_full_id.partition("/")
+    lineup = _LINEUP.get(provider)
+    if not sep or lineup is None:
+        return {tier: model_full_id for tier in TIERS}
+    cheap, strong, judge = lineup
+    return {
+        "cheap": f"{provider}/{cheap}",
+        "strong": f"{provider}/{strong}",
+        "judge": f"{provider}/{judge}",
+    }
+
+
+def judge_is_independent(model_full_id: str) -> bool:
+    """True when the judge tier out-ranks the strong tier (a real second opinion)."""
+    ids = resolve_tier_model_ids(model_full_id)
+    return ids["judge"] != ids["strong"]
+
+
+def clearing_is_trusted(model_full_id: str) -> bool:
+    """True when the config has a known strong-judge lineup, so the Deep dive may
+    emit a DISMISSAL verdict (``unexploitable`` / ``false_positive``).
+
+    Clearing is the only verdict that can HIDE a real vuln, so it requires a
+    capable judge tier. A thin-lineup config (ollama / custom / unrecognized id)
+    collapses every tier to one possibly-weak model — there the deep dive may
+    DETECT (``real``) and FLAG (``needs_review``) on any tier, but must never
+    auto-dismiss (the flash-judge false-clears were exactly this weak-judge
+    failure). Known-lineup providers always derive a strong judge (opus / gpt-5 /
+    pro), so dismissal is trusted."""
+    provider, sep, model = model_full_id.partition("/")
+    # A malformed id (no '/', e.g. "openai") isn't a valid provider/model config —
+    # never trust dismissal on it.
+    return bool(sep and model) and provider in _LINEUP
+
+
+__all__ = ["TIERS", "clearing_is_trusted", "judge_is_independent", "resolve_tier_model_ids"]

backend/cliff/agents/runtime/tools/__init__.py

@@ -14,6 +14,7 @@
 from cliff.agents.runtime.tools.bash import bash
 from cliff.agents.runtime.tools.edit import edit
 from cliff.agents.runtime.tools.gh import gh
+from cliff.agents.runtime.tools.grep import grep
 from cliff.agents.runtime.tools.permissions import (
     classify_tool_request,
     gate_tool_call,
@@ -32,6 +33,7 @@
     "edit",
     "gate_tool_call",
     "gh",
+    "grep",
     "read",
     "webfetch",
 ]

backend/cliff/agents/runtime/tools/grep.py

@@ -0,0 +1,88 @@
+"""``grep`` tool — read-only regex search over the workspace (ADR-0052 §3).
+
+The Deep dive's code-walkers (``trace_path`` / ``challenge``) need to find
+references without a shell. This is pure-Python regex search scoped to the
+workspace (the cached clone), no subprocess, so it's safe (read-only, can't
+escape) and deterministic (gradeable by the eval). Auto-tier, like ``read``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from pathlib import Path
+
+from pydantic_ai import RunContext
+
+from cliff.agents.runtime.deps import WorkspaceDeps
+from cliff.agents.runtime.tools.permissions import escapes_workspace
+
+_MAX_MATCHES = 100
+_MAX_FILE_BYTES = 1024 * 1024
+_SKIP_DIRS = frozenset(
+    {".git", "node_modules", ".venv", "venv", "__pycache__", "dist", "build", ".mypy_cache"}
+)
+
+
+def search_workspace(workspace_dir: str, pattern: str, path: str = ".") -> str:
+    """Search files under *workspace_dir*/*path* for *pattern* (regex).
+
+    Returns up to 100 ``relpath:line: text`` matches, or a bracketed status
+    ([refused] / [invalid regex] / [no matches]). Pure + synchronous so it's
+    unit-testable without a RunContext; the tool wraps it on a thread.
+    """
+    if escapes_workspace(workspace_dir, path):
+        return f"[refused: {path} resolves outside the workspace]"
+    try:
+        rx = re.compile(pattern)
+    except re.error as exc:
+        return f"[invalid regex: {exc}]"
+
+    # Resolve the root too: on macOS a temp dir like /var/folders/... canonicalizes
+    # to /private/var/folders/..., and base.resolve() below follows that symlink —
+    # so an unresolved root would break relative_to() on every match.
+    root = Path(workspace_dir).resolve()
+    base = (root / path).resolve()
+    if not base.exists():
+        return f"[path not found: {path}]"
+
+    matches: list[str] = []
+    candidates = [base] if base.is_file() else base.rglob("*")
+    for f in candidates:
+        if len(matches) >= _MAX_MATCHES:
+            break
+        if not f.is_file() or any(part in _SKIP_DIRS for part in f.parts):
+            continue
+        try:
+            if f.stat().st_size > _MAX_FILE_BYTES:
+                continue
+            with f.open("r", encoding="utf-8", errors="replace") as fh:
+                for lineno, line in enumerate(fh, 1):
+                    if rx.search(line):
+                        rel = f.relative_to(root)
+                        matches.append(f"{rel}:{lineno}: {line.rstrip()[:200]}")
+                        if len(matches) >= _MAX_MATCHES:
+                            break
+        except OSError:
+            continue
+
+    if not matches:
+        return f"[no matches for {pattern!r}]"
+    out = "\n".join(matches)
+    if len(matches) >= _MAX_MATCHES:
+        out += f"\n[... capped at {_MAX_MATCHES} matches ...]"
+    return out
+
+
+async def grep(ctx: RunContext[WorkspaceDeps], pattern: str, path: str = ".") -> str:
+    """Search the workspace for *pattern* (regex), under *path* (default: root)."""
+    out = await asyncio.to_thread(
+        search_workspace, ctx.deps.workspace_dir, pattern, path
+    )
+    budget = ctx.deps.read_budget
+    if budget is not None and not budget.take(len(out.encode("utf-8", errors="ignore"))):
+        return "[grep budget exhausted for this analysis]"
+    return out
+
+
+__all__ = ["grep", "search_workspace"]

backend/cliff/agents/runtime/tools/read.py

@@ -50,6 +50,9 @@ def _read() -> tuple[str, bool]:
         return (text, truncated)
 
     body, truncated = await asyncio.to_thread(_read)
+    budget = ctx.deps.read_budget
+    if budget is not None and not budget.take(len(body.encode("utf-8", errors="ignore"))):
+        return "[read budget exhausted for this analysis — reason from what you've already read]"
     if truncated:
         return body + f"\n[... truncated at {_MAX_READ_BYTES} bytes ...]"
     return body