refactor(tools): fold thread_context callback pairs, resolve_repo_root guard, final compaction (group I: 885->662 LOC)

tools/subagent_worktree.py

@@ -48,12 +48,10 @@ def local_backend_active() -> bool:
 
 def resolve_repo_root(path: Optional[str]) -> Optional[str]:
     """Return the git toplevel for *path*, or None when not in a work tree."""
-    if not path:
+    candidate = os.path.abspath(os.path.expanduser(str(path))) if path else ""
+    if not candidate or not os.path.isdir(candidate):
         return None
     try:
-        candidate = os.path.abspath(os.path.expanduser(str(path)))
-        if not os.path.isdir(candidate):
-            return None
         result = _run_git(["rev-parse", "--show-toplevel"], cwd=candidate)
     except Exception as exc:
         logger.debug("subagent worktree: rev-parse failed: %s", exc)
@@ -75,11 +73,10 @@ def _ensure_gitignore_entry(repo_root: str) -> None:
 
 
 def create_subagent_worktree(parent_cwd: Optional[str], subagent_id: Optional[str] = None) -> Optional[Dict[str, str]]:
-    """Create an isolated worktree for one child; None (silent downgrade) if not git / on failure."""
+    """Create an isolated worktree for one child; None (silent downgrade) outside git/on failure."""
     repo_root = resolve_repo_root(parent_cwd)
     if not repo_root:
         return None
-
     wt_name = f"subagent-{(subagent_id or uuid.uuid4().hex[:8]).replace('/', '-')}"
     branch = f"hermes-subagent/{wt_name}"
     wt_path = Path(repo_root) / ".worktrees" / wt_name
@@ -146,7 +143,6 @@ def finalize_subagent_worktree(info: Dict[str, str], *, prune: bool = True) -> D
     if not base_commit:
         return mark_worktree_payload_unproven(
             payload, "no base_commit recorded — commit count unmeasurable", unmeasured="commits")
-
     failed, unmeasured = [], []
     probes = (("commits", "rev-list", ["rev-list", "--count", f"{base_commit}..HEAD"],
                lambda s: int(s or 0)),
@@ -166,7 +162,6 @@ def finalize_subagent_worktree(info: Dict[str, str], *, prune: bool = True) -> D
     if failed:
         # Destructive cleanup requires affirmative proof; defaults prove nothing.
         return mark_worktree_payload_unproven(payload, "; ".join(failed), unmeasured="/".join(unmeasured))
-
     if prune and payload["commits"] == 0 and not payload["dirty"]:
         cwd = info.get("repo_root", "") or path
         try:

tools/terminal_hints.py

@@ -153,9 +153,8 @@ def annotate_failure(command: str, exit_code: int, output: str) -> Optional[str]
     window = (output or "")[:_SCAN_CHARS]
     for fn in _OUTPUT_HINTS if window else ():
         try:
-            hint = fn(command or "", window)
+            if hint := fn(command or "", window):
+                return hint
         except Exception:
             continue
-        if hint:
-            return hint
     return _EXIT_CODE_HINTS.get(exit_code)

tools/terminal_scope.py

@@ -62,7 +62,8 @@ def enforce_no_refusal() -> None:
     """Raise when the active scope is a refusal scope (fail closed)."""
     scope = _terminal_scope_var.get()
     if isinstance(scope, TerminalPolicyRefusal):
-        raise TerminalPolicyUnavailable(f"terminal policy unavailable for this profile: {scope.reason}")
+        raise TerminalPolicyUnavailable(
+            f"terminal policy unavailable for this profile: {scope.reason}")
 
 
 def terminal_env(name: str, default: str = "") -> str:

tools/thread_context.py

@@ -34,23 +34,20 @@ def propagate_context_to_thread(target: Callable) -> Callable:
     denied by ``prompt_dangerous_approval`` and the gateway approval queue blocks.
     """
     ctx = contextvars.copy_context()
-    parent_approval_cb = parent_sudo_cb = None
-    setters = None
+    # (setter, parent callback) pairs; None when the callback API could not be captured.
+    installs = None
     try:
         get_approval, get_sudo, set_approval, set_sudo = _callback_api()
-        parent_approval_cb = get_approval()
-        parent_sudo_cb = get_sudo()
-        setters = (set_approval, set_sudo)
+        installs = ((set_approval, get_approval()), (set_sudo, get_sudo()))
     except Exception:
         logger.debug("Could not capture parent approval/sudo callbacks", exc_info=True)
 
     def _runner(*args, **kwargs):
         def _inner():
-            if setters is None:
+            if installs is None:
                 return target(*args, **kwargs)
-            set_approval, set_sudo = setters
             try:
-                for setter, cb in ((set_approval, parent_approval_cb), (set_sudo, parent_sudo_cb)):
+                for setter, cb in installs:
                     if cb is not None:
                         setter(cb)
             except Exception:
@@ -60,8 +57,8 @@ def propagate_context_to_thread(target: Callable) -> Callable:
                 return target(*args, **kwargs)
             finally:
                 try:
-                    set_approval(None)
-                    set_sudo(None)
+                    for setter, _cb in installs:
+                        setter(None)
                 except Exception:
                     logger.debug("Failed to clear propagated approval/sudo callbacks",
                                  exc_info=True)