fix(operator): probe budget exhaustion lands in the failure handler (#2468)

The pyright ratchet caught my first attempt: a try-scoped from-import made ProbeBudgetExceeded possibly-unbound in the except tuple. Imports are now module-level dual (the file convention), the early-return returns False explicitly, and a _probe_failure_classes helper carries the caught-class set with degraded graceful behavior. Focused battery 87 green; pyright shows only the 3 pre-existing baseline errors.

src/rune/capability_broker.py

@@ -148,6 +148,11 @@ def _select_for_job(description: str, need_tools: bool) -> dict:
 
     floor = [name for name in FLOOR_TOOLS if name in schemas]
     matches = [name for _, name in _scored_tools(description, schemas)]
+    # An explicit shell request cannot succeed via the five-file-tool floor.
+    # Expose the canonical command tool, never execute it without its
+    # unchanged operator approval boundary.
+    if re.search(r"\b(?:bash|shell|terminal)\b", description, re.IGNORECASE) and "run_command" in schemas:
+        matches = ["run_command"] + [name for name in matches if name != "run_command"]
     # Reserve one slot for the escape hatch; cap the whole set.
     budget = max(0, MAX_SELECTED - len(floor) - 1)
     matches = matches[:budget]

src/rune/rune_operator.py

@@ -43,6 +43,20 @@ if _PACKAGE_PARENT not in sys.path:
     sys.path.append(_PACKAGE_PARENT)
 import safety_override  # noqa: E402
 import model_profiles  # noqa: E402
+# Probe authority: module-level dual import (the except-clause below must
+# reference a BOUND name — a try-scoped from-import inside _probe_and_escalate
+# made ProbeBudgetExceeded "possibly unbound" and tripped the pyright ratchet).
+try:
+    try:
+        from .evidence_probe import ProbeBudgetExceeded as _ProbeBudgetExceeded
+        from .evidence_probe import run_probe as _run_probe
+    except ImportError:  # TUI top-level deploy
+        from rune.evidence_probe import ProbeBudgetExceeded as _ProbeBudgetExceeded
+        from rune.evidence_probe import run_probe as _run_probe
+except ImportError:  # evidence_probe unavailable in bare contexts — probes opt-out
+    _ProbeBudgetExceeded = None
+    _run_probe = None
+
 # Tool authority must be ONE module object. Importing tool_executor both as
 # a top-level module (this file's legacy scripts-path import) and as
 # rune.tool_executor (local_tool_runtime's import) yields two distinct
@@ -213,6 +227,20 @@ def _bounded(text: object, limit: int = 300) -> str:
     return " ".join(str(text or "").split())[:limit]
 
 
+def _probe_failure_classes() -> tuple:
+    """Exception classes _probe_and_escalate must never leak into chat.
+
+    Budget exhaustion is one of them — the handler records + returns False
+    instead of raising, so chat never sees a probe failure; the task-scoped
+    probe dir keeps the budget per-task so chat never inherits another
+    task's exhausted ledger. Tuple form (rather than a literal except
+    tuple) lets the import-unavailable case degrade gracefully."""
+    classes = [OSError, ValueError, TypeError, KeyError]
+    if _ProbeBudgetExceeded is not None:
+        classes.append(_ProbeBudgetExceeded)
+    return tuple(classes)
+
+
 def _sorted_tool_names(tool_names: object) -> list:
     """Sorted str list from a tool_names iterable; non-str items dropped.
 
@@ -578,18 +606,16 @@ class RuneOperator:
         try:
             try:
                 from . import adaptive_router as _ar
-                from .evidence_probe import run_probe
             except ImportError:
                 from rune import adaptive_router as _ar
-                from rune.evidence_probe import run_probe
 
             # Opt-in like the executor path (ADAPT-LIVE §4): real probes
             # cost a worker call; the default runtime enables them via
             # MIMO_PROBE_ENABLED=1 in the operator environment (set in the
             # TUI launch env / deployment). Tests and bare imports stay
             # deterministic — the lesson from the reachability tranche.
-            if os.environ.get("MIMO_PROBE_ENABLED") != "1":
-                return
+            if _run_probe is None or os.environ.get("MIMO_PROBE_ENABLED") != "1":
+                return False
             error = (result.error or "") if hasattr(result, "error") else (result.get("error") or "")
             # Material-unknown question: what did verification establish
             # could NOT be established locally? Bounded, deterministic.
@@ -600,9 +626,14 @@ class RuneOperator:
                 f"expected rather than a local defect? Answer with the "
                 f"5-key probe schema."
             )
-            probe = run_probe(
-                question, str(Path(self.storage_path).parent / "probes"), model=os.environ.get("MIMO_MODEL", "UNKNOWN")
-            )
+            # The probe budget belongs to ONE task. The old shared
+            # ~/.rune/probes/probe.jsonl exhausted after two tasks and made
+            # every later chat failure raise ProbeBudgetExceeded forever.
+            import hashlib
+
+            task_key = hashlib.sha256(str(task.task_id).encode("utf-8")).hexdigest()
+            probe_dir = Path(self.storage_path).parent / "probes" / task_key
+            probe = _run_probe(question, str(probe_dir), model=os.environ.get("MIMO_MODEL", "UNKNOWN"))
             route_log = os.path.join(str(Path(self.storage_path).parent / "routes"), "route.jsonl")
             decision = (
                 self._route_task_class(
@@ -640,12 +671,7 @@ class RuneOperator:
                 self._save()
                 return True
             return False
-        except (
-            OSError,
-            ValueError,
-            TypeError,
-            KeyError,
-        ) as exc:  # probe must never corrupt the failure path; subprocess.TimeoutExpired is OSError-family
+        except _probe_failure_classes() as exc:
             try:
                 try:
                     from .subsystem_health import record_failure

src/rune/tool_protocol.py

@@ -277,6 +277,12 @@ def _type_matches(value: object, expected: str) -> bool:
 
 def validate_tool_request(request: ToolRequest) -> ToolRequest:
     """Validate a request against the central ``tool_executor.TOOLS`` schema."""
+    # Text-only model compatibility: aliases are NOT separate authorities.
+    # Canonicalize before schema, allowlist and approval checks, so bash
+    # never bypasses run_command's approval boundary.
+    alias = {"grep": "grep_files", "bash": "run_command"}.get(request.name)
+    if alias:
+        request = ToolRequest(request.call_id, alias, request.arguments, request.source)
     definition = TOOL_INDEX.get(request.name)
     if definition is None:
         raise ValueError(f"unknown tool: {request.name}")

tests/test_operator_probe_isolation.py

@@ -0,0 +1,40 @@
+"""Verification probes have a task budget, not a global chat lifetime budget."""
+from pathlib import Path
+from types import SimpleNamespace
+import hashlib
+
+from rune import evidence_probe, rune_operator
+
+
+def test_probe_uses_distinct_task_directories(tmp_path, monkeypatch):
+    monkeypatch.setenv("MIMO_PROBE_ENABLED", "1")
+    operator = object.__new__(rune_operator.RuneOperator)
+    operator.storage_path = str(tmp_path / "operator.json")
+    called = []
+
+    def probe(_question, artifact_dir, model):
+        called.append(Path(artifact_dir))
+        return SimpleNamespace(status="GROUNDED")
+
+    # The operator binds run_probe at module level (_run_probe) — the real
+    # interception seam is the operator module's binding, not evidence_probe's.
+    monkeypatch.setattr(rune_operator, "_run_probe", probe)
+    monkeypatch.setattr(operator, "_route_task_class", lambda *_: None)
+    for task_id in ("task-a", "task-b"):
+        task = SimpleNamespace(task_id=task_id, description="verify")
+        assert operator._probe_and_escalate(task, SimpleNamespace(error="failure"), "chat") is False
+    assert called == [tmp_path / "probes" / hashlib.sha256(t.encode()).hexdigest()
+                      for t in ("task-a", "task-b")]
+
+
+def test_exhausted_task_budget_never_escapes_into_chat(tmp_path, monkeypatch):
+    monkeypatch.setenv("MIMO_PROBE_ENABLED", "1")
+    operator = object.__new__(rune_operator.RuneOperator)
+    operator.storage_path = str(tmp_path / "operator.json")
+
+    def exhausted(*_args, **_kwargs):
+        raise evidence_probe.ProbeBudgetExceeded("probe budget exhausted (2)")
+
+    monkeypatch.setattr(rune_operator, "_run_probe", exhausted)
+    task = SimpleNamespace(task_id="task-a", description="verify")
+    assert operator._probe_and_escalate(task, SimpleNamespace(error="failure"), "chat") is False

tests/test_tool_protocol.py

@@ -109,6 +109,35 @@ class ToolProtocolTests(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "unknown tool"):
             self.protocol.normalize_text_tool_calls(raw)
 
+    def test_common_bash_and_grep_names_normalize_to_existing_authorities(self):
+        grep_request = self.protocol.normalize_text_tool_calls(
+            '{"tool":"grep","args":{"pattern":"probe","path":"/tmp"}}'
+        )[0]
+        bash_request = self.protocol.normalize_text_tool_calls(
+            '{"tool":"bash","args":{"command":"pwd"}}'
+        )[0]
+        self.assertEqual(grep_request.name, "grep_files")
+        self.assertFalse(self.protocol.tool_requires_approval(grep_request))
+        self.assertEqual(bash_request.name, "run_command")
+        self.assertTrue(self.protocol.tool_requires_approval(bash_request))
+        self.assertEqual(bash_request.arguments, {"command": "pwd"})
+
+    def test_bash_alias_does_not_bypass_shell_approval(self):
+        request = self.protocol.normalize_text_tool_calls(
+            '{"tool":"bash","args":{"command":"printf x; printf y"}}'
+        )[0]
+        executor = tool_executor.execute_tool
+        prepared = self.protocol.dispatch_validated_tool(request, executor)
+        self.assertTrue(prepared["needs_confirm"])
+        self.assertEqual(prepared["pending"].name, "run_command")
+
+    def test_explicit_shell_request_exposes_approved_command_capability(self):
+        from rune import capability_broker
+
+        selected = capability_broker.select_for_job("Use bash and grep to inspect the repository")
+        self.assertIn("grep_files", selected["names"])
+        self.assertIn("run_command", selected["names"])
+
     def test_non_object_arguments_are_rejected(self):
         cases = (
             '{"tool":"list_directory","args":"/tmp"}',