refactor(acp): AST-neutral bracket packing pass 2 (118 cols) + docstring closer folds

acp_adapter/auth.py

@@ -14,8 +14,7 @@ def detect_provider() -> Optional[str]:
     A callable ``api_key`` (Azure Foundry Entra ID bearer-token provider, see
     :mod:`agent.azure_identity_adapter`) counts as a valid credential; otherwise
     Entra-configured Foundry deployments would default to ``"openrouter"`` and
-    the ACP auth handshake would reject the legitimate provider.
-    """
+    the ACP auth handshake would reject the legitimate provider."""
     try:
         from hermes_cli.runtime_provider import resolve_runtime_provider
         runtime = resolve_runtime_provider()
@@ -35,8 +34,7 @@ def build_auth_methods() -> list[Any]:
     The ACP registry requires at least one usable auth method in the initial
     handshake. A fresh Zed install may have no Hermes credentials yet, so the
     terminal setup method is always advertised; when credentials resolve, the
-    provider is also advertised as the default agent-managed runtime method.
-    """
+    provider is also advertised as the default agent-managed runtime method."""
     from acp.schema import AuthMethodAgent, TerminalAuthMethod
 
     methods: list[Any] = []

acp_adapter/commands.py

@@ -203,16 +203,14 @@ class SlashCommandsMixin:
                 else:
                     remaining = max(threshold_tokens - approx_tokens, 0)
                     lines.append(
-                        f"Compression: ~{remaining:,} tokens until threshold "
-                        f"(~{threshold_tokens:,}{pct_note})."
+                        f"Compression: ~{remaining:,} tokens until threshold (~{threshold_tokens:,}{pct_note})."
                     )
             else:
                 lines.append(f"Compression threshold: ~{threshold_tokens:,} tokens")
 
         if getattr(agent, "compression_enabled", True) is False:
             lines.append(
-                "Auto-compaction is disabled (compression.enabled: false); "
-                "/compress still compresses manually."
+                "Auto-compaction is disabled (compression.enabled: false); /compress still compresses manually."
             )
         else:
             lines.append("Tip: run /compress to compress manually before the threshold.")

acp_adapter/content.py

@@ -16,11 +16,7 @@ from acp.schema import (
 logger = logging.getLogger("acp_adapter.server")
 
 PromptBlock = (
-    TextContentBlock
-    | ImageContentBlock
-    | AudioContentBlock
-    | ResourceContentBlock
-    | EmbeddedResourceContentBlock
+    TextContentBlock | ImageContentBlock | AudioContentBlock | ResourceContentBlock | EmbeddedResourceContentBlock
 )
 
 _MAX_ACP_RESOURCE_BYTES = 512 * 1024

acp_adapter/edit_approval.py

@@ -144,8 +144,7 @@ def _is_sensitive_auto_approve_path(path: str) -> bool:
 def should_auto_approve_edit(proposal: EditProposal, policy: str, cwd: str | None = None) -> bool:
     """Return whether an ACP edit proposal may bypass the prompt for this session.
 
-    Session-scoped and conservative: sensitive paths still ask under autonomous policies.
-    """
+    Session-scoped and conservative: sensitive paths still ask under autonomous policies."""
     policy = str(policy or AUTO_APPROVE_ASK).strip()
     if policy == AUTO_APPROVE_ASK or _is_sensitive_auto_approve_path(proposal.path):
         return False
@@ -170,8 +169,7 @@ def maybe_require_edit_approval(tool_name: str, arguments: dict[str, Any]) -> st
     """Run ACP edit approval if bound.
 
     Returns a JSON tool-error string when the edit must be blocked, otherwise
-    ``None`` so dispatch can continue.  Requester exceptions deny by default.
-    """
+    ``None`` so dispatch can continue.  Requester exceptions deny by default."""
     requester = _EDIT_APPROVAL_REQUESTER.get()
     if requester is None:
         return None
@@ -195,9 +193,7 @@ def build_acp_edit_tool_call(proposal: EditProposal):
     import acp
 
     return acp.update_tool_call(
-        f"edit-approval-{next(_PERMISSION_REQUEST_IDS)}",
-        title=f"Approve edit: {proposal.path}",
-        kind="edit",
+        f"edit-approval-{next(_PERMISSION_REQUEST_IDS)}", title=f"Approve edit: {proposal.path}", kind="edit",
         status="pending",
         content=[acp.tool_diff_content(path=proposal.path, old_text=proposal.old_text, new_text=proposal.new_text)],
         raw_input={"tool": proposal.tool_name, "arguments": proposal.arguments},

acp_adapter/entry.py

@@ -89,8 +89,7 @@ def _load_env() -> None:
 def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(prog="hermes-acp", description="Run Hermes Agent as an ACP stdio server.")
     parser.add_argument("--version", action="store_true", help="Print Hermes version and exit")
-    parser.add_argument("--check", action="store_true",
-                        help="Verify ACP dependencies and adapter imports, then exit")
+    parser.add_argument("--check", action="store_true", help="Verify ACP dependencies and adapter imports, then exit")
     parser.add_argument("--setup", action="store_true",
                         help="Run interactive Hermes provider/model setup for ACP terminal auth")
     parser.add_argument("--setup-browser", action="store_true",

acp_adapter/events.py

@@ -27,8 +27,7 @@ def _build_plan_update_from_todo_result(result: Any) -> AgentPlanUpdate | None:
     """Translate Hermes' todo tool result into ACP's native plan update.
 
     Zed renders ``sessionUpdate: plan`` as its first-class task panel, so the
-    todo state is exposed natively rather than only as a tool-call transcript.
-    """
+    todo state is exposed natively rather than only as a tool-call transcript."""
     if not isinstance(result, str) or not result.strip():
         return None
     data = _json_loads_maybe(result)
@@ -82,8 +81,7 @@ def make_tool_progress_cb(
     Signature: ``tool_progress_callback(event_type, name, preview, args, **kwargs)``.
     Emits ``ToolCallStart`` for ``tool.started`` and tracks IDs in a FIFO per tool
     name so parallel same-name calls complete against the right ACP tool call.
-    Other event types (``tool.completed``, ``reasoning.available``) are ignored.
-    """
+    Other event types (``tool.completed``, ``reasoning.available``) are ignored."""
 
     def _tool_progress(event_type: str, name: str = None, preview: str = None, args: Any = None, **kwargs) -> None:
         if event_type != "tool.started":

acp_adapter/model_catalog.py

@@ -23,8 +23,7 @@ def _named_custom_provider_catalogs() -> list[tuple[str, str, list[tuple[str, st
     Models = the entry's declared models, refreshed from the live ``/models`` listing when a
     credential exists and ``discover_models`` isn't disabled; declared models survive a failed
     discovery (some endpoints have no ``/models`` route). Slugs use the ``custom:<name>`` shape
-    ``parse_model_input``/``resolve_runtime_provider`` resolve, so choice ids round-trip.
-    """
+    ``parse_model_input``/``resolve_runtime_provider`` resolve, so choice ids round-trip."""
     try:
         from hermes_cli.config import (get_compatible_custom_providers, is_provider_enabled, load_config)
         from hermes_cli.model_switch import (
@@ -165,8 +164,7 @@ class _ModelCatalog:
 
     Dedupes on the encoded choice id AND a semantic ``provider:model`` id (``ollama`` ==
     ``custom:ollama``). A bare/``custom`` current provider whose base_url matches an ollama
-    inventory row is resolved to ``custom:ollama``.
-    """
+    inventory row is resolved to ``custom:ollama``."""
 
     normalize_provider: Callable[[str], str]
     current_model: str
@@ -265,8 +263,7 @@ def build_model_state(model: str, provider: str, base_url: str) -> SessionModelS
     payload = build_models_payload(
         context, explicit_only=True, include_unconfigured=False, picker_hints=False,
         canonical_order=True, pricing=False, capabilities=False, refresh=False,
-        probe_custom_providers=False, probe_current_custom_provider=False,
-        max_models=ACP_MAX_MODELS_PER_PROVIDER,
+        probe_custom_providers=False, probe_current_custom_provider=False, max_models=ACP_MAX_MODELS_PER_PROVIDER,
     )
 
     cat = _ModelCatalog(

acp_adapter/permissions.py

@@ -15,11 +15,7 @@ logger = logging.getLogger(__name__)
 # ACP permission option id -> Hermes approval result. Ids are stable across the
 # ``allow_permanent=True`` and ``False`` paths even though the option list differs.
 _OPTION_ID_TO_HERMES = {
-    "allow_once": "once",
-    "allow_session": "session",
-    "allow_always": "always",
-    "deny": "deny",
-    "deny_always": "deny",
+    "allow_once": "once", "allow_session": "session", "allow_always": "always", "deny": "deny", "deny_always": "deny"
 }
 
 _PERMISSION_REQUEST_IDS = count(1)
@@ -61,11 +57,8 @@ def _build_permission_tool_call(command: str, description: str):
 
     content_text = f"{description}\n$ {command}" if description else f"$ {command}"
     return _acp.update_tool_call(
-        f"perm-check-{next(_PERMISSION_REQUEST_IDS)}",
-        title=f"{description}: {command}" if description else command,
-        kind="execute",
-        status="pending",
-        content=[_acp.tool_content(_acp.text_block(content_text))],
+        f"perm-check-{next(_PERMISSION_REQUEST_IDS)}", title=f"{description}: {command}" if description else command,
+        kind="execute", status="pending", content=[_acp.tool_content(_acp.text_block(content_text))],
         raw_input={"command": command, "description": description},
     )
 
@@ -125,8 +118,6 @@ def make_approval_callback(request_permission_fn: Callable, loop: asyncio.Abstra
             return "timeout"
         if response is None:
             return "deny"
-        return _map_outcome_to_hermes(
-            response.outcome, allowed_option_ids={option.option_id for option in options},
-        )
+        return _map_outcome_to_hermes(response.outcome, allowed_option_ids={option.option_id for option in options})
 
     return _callback

acp_adapter/provenance.py

@@ -30,16 +30,14 @@ def _is_compression_end(row: Any) -> bool:
 
 
 def build_session_provenance(
-    db: Any, acp_session_id: str, current_hermes_session_id: str, *,
-    previous_hermes_session_id: Optional[str] = None,
+    db: Any, acp_session_id: str, current_hermes_session_id: str, *, previous_hermes_session_id: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     """Build ``_meta.hermes.sessionProvenance`` for an ACP session.
 
     ``db`` must expose ``get_session``. ``current_hermes_session_id`` is the live
     internal id (``state.agent.session_id``); ``previous_hermes_session_id`` is
     the id before the most recent turn, supplied by ``prompt()`` to flag a
-    rotation. Returns ``None`` if the session can't be read.
-    """
+    rotation. Returns ``None`` if the session can't be read."""
     row = _get_row(db, current_hermes_session_id)
     if not row:
         return None
@@ -85,8 +83,7 @@ def build_session_provenance(
 
 
 def session_provenance_meta(
-    db: Any, acp_session_id: str, current_hermes_session_id: str, *,
-    previous_hermes_session_id: Optional[str] = None,
+    db: Any, acp_session_id: str, current_hermes_session_id: str, *, previous_hermes_session_id: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     """Return a ready ``_meta`` payload: ``{"hermes": {"sessionProvenance": ...}}``."""
     prov = build_session_provenance(

acp_adapter/server.py

@@ -27,8 +27,7 @@ from acp_adapter.auth import TERMINAL_SETUP_AUTH_METHOD_ID, build_auth_methods,
 from acp_adapter.commands import HERMES_VERSION, SlashCommandsMixin, _estimate_tokens
 from acp_adapter.content import PromptBlock, _content_blocks_to_openai_user_content, _extract_text
 from acp_adapter.events import (
-    _build_plan_update_from_todo_result, make_message_cb, make_step_cb, make_thinking_cb,
-    make_tool_progress_cb,
+    _build_plan_update_from_todo_result, make_message_cb, make_step_cb, make_thinking_cb, make_tool_progress_cb,
 )
 from acp_adapter.model_catalog import (  # noqa: F401  (ACP_MAX_MODELS_PER_PROVIDER re-exported for tests)
     ACP_MAX_MODELS_PER_PROVIDER, build_model_state, encode_model_choice,
@@ -86,8 +85,7 @@ def _history_summary_meta(message: dict[str, Any], text: str) -> dict[str, Any]
     Summaries persist as ordinary messages, standalone (either role) or merged into the first
     preserved tail message. Two keys so clients can't hide real content: ``compactionSummary``
     (whole chunk; safe to collapse) vs ``containsCompactionSummary`` (real content + summary).
-    Uses the in-process flag, falling back to content classification for DB-reloaded sessions.
-    """
+    Uses the in-process flag, falling back to content classification for DB-reloaded sessions."""
     kind = ContextCompressor.classify_summary_content(text)
     if kind is None and message.get(COMPRESSED_SUMMARY_METADATA_KEY):
         # Flagged but unclassified (prefix drift): the flag only marks summaries -> standalone.
@@ -101,8 +99,7 @@ def _history_summary_meta(message: dict[str, Any], text: str) -> dict[str, Any]
 
 # role -> (chunk class, session_update tag) for history replay.
 _HISTORY_CHUNK_TYPES = {
-    "user": (UserMessageChunk, "user_message_chunk"),
-    "assistant": (AgentMessageChunk, "agent_message_chunk"),
+    "user": (UserMessageChunk, "user_message_chunk"), "assistant": (AgentMessageChunk, "agent_message_chunk")
 }
 
 
@@ -341,8 +338,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         endpoint: dict[str, Any] = {}
         if keep_endpoint and not (current_provider and target_provider != current_provider):
             endpoint = {
-                "base_url": getattr(state.agent, "base_url", None),
-                "api_mode": getattr(state.agent, "api_mode", None),
+                "base_url": getattr(state.agent, "base_url", None), "api_mode": getattr(state.agent, "api_mode", None)
             }
         state.agent = self.session_manager._make_agent(
             session_id=state.session_id, cwd=state.cwd, model=new_model,
@@ -472,8 +468,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         Cache safety: only pre-first-turn (nothing cached yet); afterwards the snapshot stays
         frozen and late servers land via the between-turns prologue refresh
         (``agent/turn_context.py``). No-op if discovery finished, join timed out, registry
-        unchanged, or session closed.
-        """
+        unchanged, or session closed."""
         try:
             from hermes_cli.mcp_startup import mcp_discovery_in_flight
         except Exception:
@@ -518,9 +513,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
             except Exception:
                 logger.debug("Session %s: late MCP refresh failed", session_id, exc_info=True)
 
-        threading.Thread(
-            target=_wait_then_refresh, name=f"acp-mcp-late-refresh-{session_id}", daemon=True
-        ).start()
+        threading.Thread(target=_wait_then_refresh, name=f"acp-mcp-late-refresh-{session_id}", daemon=True).start()
 
     # ---- ACP lifecycle ------------------------------------------------------
 
@@ -530,8 +523,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
     ) -> InitializeResponse:
         auth_methods = build_auth_methods()
         logger.info(
-            "Initialize from %s (protocol v%s)",
-            client_info.name if client_info else "unknown",
+            "Initialize from %s (protocol v%s)", client_info.name if client_info else "unknown",
             protocol_version if isinstance(protocol_version, int) else acp.PROTOCOL_VERSION,
         )
 
@@ -583,9 +575,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         except Exception:
             logger.warning(
                 f"ACP history replay raised during session/{verb} for %s — "
-                f"{verb} will still succeed, partial transcript may be missing",
-                state.session_id,
-                exc_info=True,
+                f"{verb} will still succeed, partial transcript may be missing", state.session_id, exc_info=True,
             )
 
     def _session_response_fields(self, state: SessionState) -> dict[str, Any]:
@@ -759,8 +749,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         Approval routing is thread-local, so it MUST be bound here, not on the loop thread.
         Interactive routing is a ``tools.approval`` contextvar, not ``HERMES_INTERACTIVE`` in
         os.environ, so concurrent workers can't race a global flag onto the non-interactive
-        auto-approve path (GHSA-96vc-wcxf-jjff).
-        """
+        auto-approve path (GHSA-96vc-wcxf-jjff)."""
         agent = state.agent
         with contextlib.ExitStack() as stack:
             # HERMES_SESSION_KEY scopes per-session caches (interactive sudo password) to this
@@ -828,9 +817,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         if not has_content:
             return PromptResponse(stop_reason="end_turn")
 
-        user_text, user_content = self._rewrite_prompt_for_interrupt(
-            state, user_text, user_content, text_only_prompt
-        )
+        user_text, user_content = self._rewrite_prompt_for_interrupt(state, user_text, user_content, text_only_prompt)
 
         # Slash commands are text-only; a prompt with media goes to the agent even if it starts with "/".
         if text_only_prompt and isinstance(user_content, str) and user_text.startswith("/"):

acp_adapter/session.py

@@ -156,8 +156,7 @@ class SessionManager:
     """Thread-safe manager for ACP sessions backed by Hermes AIAgent instances.
 
     Sessions are held in-memory for fast access **and** persisted to the shared
-    SessionDB so they survive restarts and are searchable via ``session_search``.
-    """
+    SessionDB so they survive restarts and are searchable via ``session_search``."""
 
     def __init__(self, agent_factory=None, db=None):
         """``agent_factory``: AIAgent-like factory (tests); default builds a real AIAgent from

acp_adapter/tools.py

@@ -144,8 +144,7 @@ def _json_loads_maybe(value: Optional[str]) -> Any:
     """Decode a JSON string; non-strings pass through, undecodable strings yield None.
 
     Some Hermes tools append a human hint after the payload (``{...}\\n\\n[Hint: ...]``),
-    so fall back to decoding the first JSON value to keep the structured rendering path.
-    """
+    so fall back to decoding the first JSON value to keep the structured rendering path."""
     if not isinstance(value, str):
         return value
     try:
@@ -175,8 +174,7 @@ def _tool_result_failed(result: Optional[str], tool_name: str | None = None) ->
     """Return True when a structured Hermes tool result clearly failed.
 
     Deliberately conservative: plain text may legitimately contain "error", so
-    only structured tool-level failures map to ACP failed status.
-    """
+    only structured tool-level failures map to ACP failed status."""
     # The agent's tool executor wraps raised exceptions in a canonical
     # "Error executing tool '<name>': ..." prefix that well-behaved tool output
     # cannot produce; catch it so a tool that blew up is not shown green.
@@ -362,8 +360,7 @@ def _format_execute_code_result(tool_name: str, data: Args, args: Args) -> Optio
             parts.extend(["", "Output truncated."])
     for key, label, value in (
         ("warning", "Warning:", str(data.get("warning") or "").strip()),
-        ("output", "Output:", str(data.get("output") or "")),
-        ("error", "Error:", str(data.get("error") or "")),
+        ("output", "Output:", str(data.get("output") or "")), ("error", "Error:", str(data.get("error") or "")),
     ):
         if value:
             parts.extend(["", label, value])
@@ -921,8 +918,7 @@ def build_tool_start(tool_call_id: str, tool_name: str, arguments: Args, *, edit
     ignoring the schema) must never abort the render — this runs on the live
     tool-progress callback and during history replay — so any failure in the
     title/content/location builders falls back to a minimal valid start event
-    (mirrors ``get_cute_tool_message`` in ``agent/display.py``).
-    """
+    (mirrors ``get_cute_tool_message`` in ``agent/display.py``)."""
     try:
         return _build_tool_start(tool_call_id, tool_name, arguments, edit_diff=edit_diff)
     except Exception as exc:  # noqa: BLE001 — a tool-call render must never abort the turn