chore: ruff auto-fix UP045 — non-pep604-annotation-optional

acp_adapter/auth.py

@@ -8,7 +8,7 @@
 TERMINAL_SETUP_AUTH_METHOD_ID = "hermes-setup"
 
 
-def detect_provider() -> Optional[str]:
+def detect_provider() -> str | None:
     """Resolve the active Hermes runtime provider, or None if unavailable.
 
     Treats a ``Callable`` ``api_key`` (Azure Foundry Entra ID bearer

acp_adapter/server.py

@@ -516,7 +516,7 @@ class HermesACPAgent(acp.Agent):
     def __init__(self, session_manager: SessionManager | None = None):
         super().__init__()
         self.session_manager = session_manager or SessionManager()
-        self._conn: Optional[acp.Client] = None
+        self._conn: acp.Client | None = None
 
     # ---- Connection lifecycle -----------------------------------------------
 

acp_adapter/session.py

@@ -228,7 +228,7 @@ def create_session(self, cwd: str = ".") -> SessionState:
         logger.info("Created ACP session %s (cwd=%s)", session_id, cwd)
         return state
 
-    def get_session(self, session_id: str) -> Optional[SessionState]:
+    def get_session(self, session_id: str) -> SessionState | None:
         """Return the session for *session_id*, or ``None``.
 
         If the session is not in memory but exists in the database (e.g. after
@@ -250,7 +250,7 @@ def remove_session(self, session_id: str) -> bool:
             _clear_task_cwd(session_id)
         return existed or db_existed
 
-    def fork_session(self, session_id: str, cwd: str = ".") -> Optional[SessionState]:
+    def fork_session(self, session_id: str, cwd: str = ".") -> SessionState | None:
         """Deep-copy a session's history into a new session."""
         import threading
 
@@ -354,7 +354,7 @@ def list_sessions(self, cwd: str | None = None) -> List[Dict[str, Any]]:
         results.sort(key=lambda item: _updated_at_sort_key(item.get("updated_at")), reverse=True)
         return results
 
-    def update_cwd(self, session_id: str, cwd: str) -> Optional[SessionState]:
+    def update_cwd(self, session_id: str, cwd: str) -> SessionState | None:
         """Update the working directory for a session and its tool overrides."""
         cwd = _translate_acp_cwd(cwd)
         state = self.get_session(session_id)  # checks DB too
@@ -473,7 +473,7 @@ def _persist(self, state: SessionState) -> None:
         except Exception:
             logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True)
 
-    def _restore(self, session_id: str) -> Optional[SessionState]:
+    def _restore(self, session_id: str) -> SessionState | None:
         """Load a session from the database into memory, recreating the AIAgent."""
         import threading
 

acp_adapter/tools.py

@@ -184,7 +184,7 @@ def _text(content: str) -> Any:
     return acp.tool_content(acp.text_block(content))
 
 
-def _json_loads_maybe(value: Optional[str]) -> Any:
+def _json_loads_maybe(value: str | None) -> Any:
     if not isinstance(value, str):
         return value
     try:
@@ -202,7 +202,7 @@ def _json_loads_maybe(value: Optional[str]) -> Any:
         return None
 
 
-def _tool_result_failed(result: Optional[str], tool_name: str | None = None) -> bool:
+def _tool_result_failed(result: str | None, tool_name: str | None = None) -> bool:
     """Return True when a structured Hermes tool result clearly failed.
 
     Keep this deliberately conservative. Plain text can contain words like
@@ -253,7 +253,7 @@ def _fenced_text(text: str, language: str = "") -> str:
     return f"{fence}{language}\n{text}\n{fence}"
 
 
-def _format_todo_result(result: Optional[str]) -> Optional[str]:
+def _format_todo_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict) or not isinstance(data.get("todos"), list):
         return None
@@ -285,7 +285,7 @@ def _format_todo_result(result: Optional[str]) -> Optional[str]:
     return "\n".join(lines)
 
 
-def _format_read_file_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+def _format_read_file_result(result: str | None, args: Dict[str, Any] | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -312,7 +312,7 @@ def _format_read_file_result(result: Optional[str], args: Optional[Dict[str, Any
     return _truncate_text(f"{header}\n\n{_fenced_text(content)}")
 
 
-def _format_search_files_result(result: Optional[str]) -> Optional[str]:
+def _format_search_files_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -371,7 +371,7 @@ def _format_search_files_result(result: Optional[str]) -> Optional[str]:
     return _truncate_text("\n".join(lines), limit=7000)
 
 
-def _format_execute_code_result(result: Optional[str]) -> Optional[str]:
+def _format_execute_code_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return result if isinstance(result, str) and result.strip() else None
@@ -399,7 +399,7 @@ def _extract_markdown_headings(content: str, limit: int = 8) -> list[str]:
     return headings
 
 
-def _format_skill_view_result(result: Optional[str]) -> Optional[str]:
+def _format_skill_view_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -432,7 +432,7 @@ def _format_skill_view_result(result: Optional[str]) -> Optional[str]:
     return "\n".join(lines)
 
 
-def _format_skill_manage_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+def _format_skill_manage_result(result: str | None, args: Dict[str, Any] | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -462,7 +462,7 @@ def _format_skill_manage_result(result: Optional[str], args: Optional[Dict[str,
     return "\n".join(lines)
 
 
-def _format_web_search_result(result: Optional[str]) -> Optional[str]:
+def _format_web_search_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -482,7 +482,7 @@ def _format_web_search_result(result: Optional[str]) -> Optional[str]:
     return _truncate_text("\n".join(lines))
 
 
-def _format_web_extract_result(result: Optional[str]) -> Optional[str]:
+def _format_web_extract_result(result: str | None) -> str | None:
     """Return only web_extract errors for ACP; success stays compact via title."""
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
@@ -513,7 +513,7 @@ def _format_web_extract_result(result: Optional[str]) -> Optional[str]:
     return "\n".join(lines)
 
 
-def _format_process_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+def _format_process_result(result: str | None, args: Dict[str, Any] | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return result if isinstance(result, str) and result.strip() else None
@@ -560,7 +560,7 @@ def _format_process_result(result: Optional[str], args: Optional[Dict[str, Any]]
     return _truncate_text("\n".join(lines), limit=7000)
 
 
-def _format_delegate_result(result: Optional[str]) -> Optional[str]:
+def _format_delegate_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -606,7 +606,7 @@ def _format_delegate_result(result: Optional[str]) -> Optional[str]:
     return _truncate_text("\n".join(lines), limit=8000)
 
 
-def _format_session_search_result(result: Optional[str]) -> Optional[str]:
+def _format_session_search_result(result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -637,7 +637,7 @@ def _format_session_search_result(result: Optional[str]) -> Optional[str]:
     return _truncate_text("\n".join(lines), limit=7000)
 
 
-def _format_memory_result(result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+def _format_memory_result(result: str | None, args: Dict[str, Any] | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return None
@@ -664,7 +664,7 @@ def _format_memory_result(result: Optional[str], args: Optional[Dict[str, Any]])
     return "\n".join(lines)
 
 
-def _format_edit_result(tool_name: str, result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+def _format_edit_result(tool_name: str, result: str | None, args: Dict[str, Any] | None) -> str | None:
     data = _json_loads_maybe(result)
     path = str((args or {}).get("path") or "file").strip()
     if isinstance(data, dict):
@@ -687,7 +687,7 @@ def _format_edit_result(tool_name: str, result: Optional[str], args: Optional[Di
     return f"✅ {tool_name} completed" + (f" for `{path}`" if path else "")
 
 
-def _format_browser_result(tool_name: str, result: Optional[str], args: Optional[Dict[str, Any]]) -> Optional[str]:
+def _format_browser_result(tool_name: str, result: str | None, args: Dict[str, Any] | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return result if isinstance(result, str) and result.strip() else None
@@ -713,7 +713,7 @@ def _format_browser_result(tool_name: str, result: Optional[str], args: Optional
     return _truncate_text("\n".join(lines), limit=7000)
 
 
-def _format_media_or_cron_result(tool_name: str, result: Optional[str]) -> Optional[str]:
+def _format_media_or_cron_result(tool_name: str, result: str | None) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, dict):
         return result if isinstance(result, str) and result.strip() else None
@@ -818,10 +818,10 @@ def _format_structured_value(
 
 def _format_generic_structured_result(
     tool_name: str,
-    result: Optional[str],
+    result: str | None,
     *,
     fallback_to_text: bool = True,
-) -> Optional[str]:
+) -> str | None:
     data = _json_loads_maybe(result)
     if not isinstance(data, (dict, list)):
         return result if fallback_to_text and isinstance(result, str) and result.strip() else None
@@ -870,9 +870,9 @@ def _format_generic_structured_result(
 
 def _build_polished_completion_content(
     tool_name: str,
-    result: Optional[str],
-    function_args: Optional[Dict[str, Any]],
-) -> Optional[List[Any]]:
+    result: str | None,
+    function_args: Dict[str, Any] | None,
+) -> List[Any] | None:
     formatter = {
         "todo": lambda: _format_todo_result(result),
         "read_file": lambda: _format_read_file_result(result, function_args),
@@ -986,8 +986,8 @@ def _parse_unified_diff_content(diff_text: str) -> List[Any]:
         return []
 
     content: List[Any] = []
-    current_old_path: Optional[str] = None
-    current_new_path: Optional[str] = None
+    current_old_path: str | None = None
+    current_new_path: str | None = None
     old_lines: list[str] = []
     new_lines: list[str] = []
 
@@ -1041,9 +1041,9 @@ def _flush() -> None:
 
 def _build_tool_complete_content(
     tool_name: str,
-    result: Optional[str],
+    result: str | None,
     *,
-    function_args: Optional[Dict[str, Any]] = None,
+    function_args: Dict[str, Any] | None = None,
     snapshot: Any = None,
 ) -> List[Any]:
     """Build structured ACP completion content, falling back to plain text."""
@@ -1308,15 +1308,15 @@ def build_tool_start(
     )
 
 
-def _is_structured_json_result(result: Optional[str]) -> bool:
+def _is_structured_json_result(result: str | None) -> bool:
     return isinstance(_json_loads_maybe(result), (dict, list))
 
 
 def build_tool_complete(
     tool_call_id: str,
     tool_name: str,
-    result: Optional[str] = None,
-    function_args: Optional[Dict[str, Any]] = None,
+    result: str | None = None,
+    function_args: Dict[str, Any] | None = None,
     snapshot: Any = None,
 ) -> ToolCallProgress:
     """Create a ToolCallUpdate (progress) event for a completed tool call."""

agent/account_usage.py

@@ -18,9 +18,9 @@ def _utc_now() -> datetime:
 @dataclass(frozen=True)
 class AccountUsageWindow:
     label: str
-    used_percent: Optional[float] = None
-    reset_at: Optional[datetime] = None
-    detail: Optional[str] = None
+    used_percent: float | None = None
+    reset_at: datetime | None = None
+    detail: str | None = None
 
 
 @dataclass(frozen=True)
@@ -29,24 +29,24 @@ class AccountUsageSnapshot:
     source: str
     fetched_at: datetime
     title: str = "Account limits"
-    plan: Optional[str] = None
+    plan: str | None = None
     windows: tuple[AccountUsageWindow, ...] = ()
     details: tuple[str, ...] = ()
-    unavailable_reason: Optional[str] = None
+    unavailable_reason: str | None = None
 
     @property
     def available(self) -> bool:
         return bool(self.windows or self.details) and not self.unavailable_reason
 
 
-def _title_case_slug(value: Optional[str]) -> Optional[str]:
+def _title_case_slug(value: str | None) -> str | None:
     cleaned = str(value or "").strip()
     if not cleaned:
         return None
     return cleaned.replace("_", " ").replace("-", " ").title()
 
 
-def _parse_dt(value: Any) -> Optional[datetime]:
+def _parse_dt(value: Any) -> datetime | None:
     if value in {None, ""}:
         return None
     if isinstance(value, (int, float)):
@@ -65,7 +65,7 @@ def _parse_dt(value: Any) -> Optional[datetime]:
     return None
 
 
-def _format_reset(dt: Optional[datetime]) -> str:
+def _format_reset(dt: datetime | None) -> str:
     if not dt:
         return "unknown"
     local_dt = dt.astimezone()
@@ -85,7 +85,7 @@ def _format_reset(dt: Optional[datetime]) -> str:
     return f"{rel} ({local_dt.strftime('%Y-%m-%d %H:%M %Z')})"
 
 
-def render_account_usage_lines(snapshot: Optional[AccountUsageSnapshot], *, markdown: bool = False) -> list[str]:
+def render_account_usage_lines(snapshot: AccountUsageSnapshot | None, *, markdown: bool = False) -> list[str]:
     if not snapshot:
         return []
     header = f"📈 {'**' if markdown else ''}{snapshot.title}{'**' if markdown else ''}"
@@ -124,7 +124,7 @@ def _resolve_codex_usage_url(base_url: str) -> str:
     return normalized + "/api/codex/usage"
 
 
-def _fetch_codex_account_usage() -> Optional[AccountUsageSnapshot]:
+def _fetch_codex_account_usage() -> AccountUsageSnapshot | None:
     creds = resolve_codex_runtime_credentials(refresh_if_expiring=True)
     token_data = _read_codex_tokens()
     tokens = token_data.get("tokens") or {}
@@ -172,7 +172,7 @@ def _fetch_codex_account_usage() -> Optional[AccountUsageSnapshot]:
     )
 
 
-def _fetch_anthropic_account_usage() -> Optional[AccountUsageSnapshot]:
+def _fetch_anthropic_account_usage() -> AccountUsageSnapshot | None:
     token = (resolve_anthropic_token() or "").strip()
     if not token:
         return None
@@ -233,7 +233,7 @@ def _fetch_anthropic_account_usage() -> Optional[AccountUsageSnapshot]:
     )
 
 
-def _fetch_openrouter_account_usage(base_url: Optional[str], api_key: Optional[str]) -> Optional[AccountUsageSnapshot]:
+def _fetch_openrouter_account_usage(base_url: str | None, api_key: str | None) -> AccountUsageSnapshot | None:
     runtime = resolve_runtime_provider(
         requested="openrouter",
         explicit_base_url=base_url,
@@ -306,11 +306,11 @@ def _fetch_openrouter_account_usage(base_url: Optional[str], api_key: Optional[s
 
 
 def fetch_account_usage(
-    provider: Optional[str],
+    provider: str | None,
     *,
-    base_url: Optional[str] = None,
-    api_key: Optional[str] = None,
-) -> Optional[AccountUsageSnapshot]:
+    base_url: str | None = None,
+    api_key: str | None = None,
+) -> AccountUsageSnapshot | None:
     normalized = str(provider or "").strip().lower()
     if normalized in {"", "auto", "custom"}:
         return None