fix(run): terminalize interrupted and stale runs (#382)

Author: solomonneas

src/brigade/aboyeur.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import inspect
 import json
 import re
 from dataclasses import dataclass
@@ -76,7 +77,7 @@ def _status_payload(stdout: str) -> tuple[dict[str, Any] | None, str]:
     return payload, _safe_diagnostic(safe_stdout)
 
 
-def cursor_auth_status() -> CursorAuthStatus:
+def cursor_auth_status(*, process_registry: proc.ProcessRegistry | None = None) -> CursorAuthStatus:
     """Return a bounded, prompt-free diagnosis for the headless Cursor CLI."""
     if proc.which("cursor-agent") is None:
         return CursorAuthStatus(
@@ -86,10 +87,11 @@ def cursor_auth_status() -> CursorAuthStatus:
             "",
             127,
         )
-    result = proc.run(
-        ["cursor-agent", "status", "--format", "json"],
-        timeout=CURSOR_AUTH_TIMEOUT_SECONDS,
-    )
+    argv = ["cursor-agent", "status", "--format", "json"]
+    if process_registry is None:
+        result = proc.run(argv, timeout=CURSOR_AUTH_TIMEOUT_SECONDS)
+    else:
+        result = proc.run(argv, timeout=CURSOR_AUTH_TIMEOUT_SECONDS, process_registry=process_registry)
     payload, stdout = _status_payload(result.stdout)
     stderr = _safe_diagnostic(result.stderr)
     diagnostic = _diagnostic_line(stdout, stderr)
@@ -164,8 +166,11 @@ def build_argv(
     return argv
 
 
-def installed_version() -> tuple[str | None, str]:
-    result = proc.run(["acpx", "--version"], timeout=10.0)
+def installed_version(*, process_registry: proc.ProcessRegistry | None = None) -> tuple[str | None, str]:
+    if process_registry is None:
+        result = proc.run(["acpx", "--version"], timeout=10.0)
+    else:
+        result = proc.run(["acpx", "--version"], timeout=10.0, process_registry=process_registry)
     if result.code != 0:
         return None, result.stderr.strip() or result.stdout.strip() or f"exit {result.code}"
     match = re.search(r"\b(\d+\.\d+\.\d+)\b", result.stdout)
@@ -174,6 +179,17 @@ def installed_version() -> tuple[str | None, str]:
     return match.group(1), ""
 
 
+def _call_with_process_registry(function, *, process_registry: proc.ProcessRegistry | None):
+    parameters = inspect.signature(function).parameters.values()
+    accepts_registry = any(
+        parameter.name == "process_registry" or parameter.kind is inspect.Parameter.VAR_KEYWORD
+        for parameter in parameters
+    )
+    if accepts_registry:
+        return function(process_registry=process_registry)
+    return function()
+
+
 def _permission_prompt_diagnostic(error: object) -> dict[str, object] | None:
     if not isinstance(error, dict):
         return None
@@ -349,6 +365,7 @@ def run_cursor(
     version: str,
     read_only: bool,
     writable_worktree: bool = False,
+    process_registry: proc.ProcessRegistry | None = None,
 ) -> AgentResult:
     if version != SUPPORTED_VERSION:
         return AgentResult(
@@ -377,7 +394,10 @@ def run_cursor(
             failure_kind="missing-executable",
             transport="acpx",
         )
-    installed, version_error = installed_version()
+    installed, version_error = _call_with_process_registry(
+        installed_version,
+        process_registry=process_registry,
+    )
     if installed != version:
         found = installed or version_error
         return AgentResult(
@@ -389,7 +409,10 @@ def run_cursor(
             transport="acpx",
             acpx_version=installed,
         )
-    auth = cursor_auth_status()
+    auth = _call_with_process_registry(
+        cursor_auth_status,
+        process_registry=process_registry,
+    )
     auth_event: dict[str, object] = {
         "type": "provider_auth",
         "status": auth.state,
@@ -436,7 +459,12 @@ def run_cursor(
             acpx_version=installed,
             safe_events=(auth_event,),
         )
-    result = proc.run(argv, timeout=timeout + 5.0, cwd=cwd)
+    result = proc.run(
+        argv,
+        timeout=timeout + 5.0,
+        cwd=cwd,
+        process_registry=process_registry,
+    )
     if result.decode_failed:
         detail = (result.stderr.strip() or result.decode_failure_detail)[:200]
         return AgentResult(

src/brigade/acpx_adapter.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import functools
+import inspect
 import json
 import os
 import re
@@ -507,7 +508,20 @@ def detect(cli_ref: str) -> bool:
     return resolve_agent_executable(cli_ref).runnable
 
 
-def ollama_model_present(model: str, executable: proc.ExecutableIdentity | None = None) -> tuple[bool, str]:
+def _accepts_process_registry(function: Callable[..., object]) -> bool:
+    parameters = inspect.signature(function).parameters.values()
+    return any(
+        parameter.name == "process_registry" or parameter.kind is inspect.Parameter.VAR_KEYWORD
+        for parameter in parameters
+    )
+
+
+def ollama_model_present(
+    model: str,
+    executable: proc.ExecutableIdentity | None = None,
+    *,
+    process_registry: proc.ProcessRegistry | None = None,
+) -> tuple[bool, str]:
     """Check whether an ollama model is already pulled locally.
 
     `ollama run` on a missing model silently auto-pulls it (tens of GB for
@@ -517,7 +531,10 @@ def ollama_model_present(model: str, executable: proc.ExecutableIdentity | None
     ollama = executable or proc.resolve_executable("ollama")
     if not ollama.runnable or ollama.path is None:
         return False, "ollama is not installed"
-    listing = proc.run([ollama.path, "list"], timeout=15.0)
+    if process_registry is None:
+        listing = proc.run([ollama.path, "list"], timeout=15.0)
+    else:
+        listing = proc.run([ollama.path, "list"], timeout=15.0, process_registry=process_registry)
     if listing.code != 0:
         reason = listing.stderr.strip() or f"exit {listing.code}"
         return False, f"could not list local ollama models ({reason[:120]}); is the ollama server running?"
@@ -657,6 +674,7 @@ def run_agent(
     reasoning: str | None = None,
     env: dict[str, str] | None = None,
     resume_session_id: str | None = None,
+    process_registry: proc.ProcessRegistry | None = None,
 ) -> AgentResult:
     child_env: dict[str, str] | None = None
     resolved_overrides: dict[str, str] | None = None
@@ -719,12 +737,27 @@ def run_agent(
             )
         from . import codex_cloud
 
+        if process_registry is not None and _accepts_process_registry(codex_cloud.run_cloud_task):
+            return codex_cloud.run_cloud_task(
+                prompt,
+                env_id=env_id,
+                timeout=timeout,
+                cwd=cwd,
+                process_registry=process_registry,
+            )
         return codex_cloud.run_cloud_task(prompt, env_id=env_id, timeout=timeout, cwd=cwd)
 
     if cli_ref.startswith(_OLLAMA_PREFIX):
         ollama_model = cli_ref[len(_OLLAMA_PREFIX) :]
         if ollama_model:
-            present, missing_detail = ollama_model_present(ollama_model, executable)
+            if process_registry is not None and _accepts_process_registry(ollama_model_present):
+                present, missing_detail = ollama_model_present(
+                    ollama_model,
+                    executable,
+                    process_registry=process_registry,
+                )
+            else:
+                present, missing_detail = ollama_model_present(ollama_model, executable)
             if not present:
                 return AgentResult(text="", ok=False, detail=missing_detail)
     cursor_limitation = None
@@ -769,13 +802,15 @@ def run_agent(
             cwd=cwd,
             env=child_env,
             stdin=prompt.encode(),
+            process_registry=process_registry,
         )
     else:
         result = proc.run(
             argv,
             timeout=timeout,
             cwd=cwd,
             env=child_env,
+            process_registry=process_registry,
         )
 
     def scrub_detail(detail: str) -> str: