refactor: type Claude native bridge boundaries (#3885)

* refactor: type Claude native bridge boundaries

Signed-off-by: Pat Sukprasert <pat.sukprasert@databricks.com>

* test: validate OpenCode MCP config strings

Signed-off-by: Pat Sukprasert <pat.sukprasert@databricks.com>

---------

Signed-off-by: Pat Sukprasert <pat.sukprasert@databricks.com>

omnigent/claude_native_bridge.py

@@ -48,14 +48,17 @@ from datetime import datetime
 from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, cast
 from urllib import error, request
 
 from omnigent._platform import stable_user_id
 from omnigent.claude_native_message_display_hook import MESSAGE_DELTAS_FILE
+from omnigent.json_types import JsonObject as _JsonObject
 from omnigent.kiro_native_bridge import bridge_root as kiro_bridge_root
 
 if TYPE_CHECKING:
+    import httpx
+
     from omnigent.llms.context_window import ModelPricing
 
 from omnigent.inner.bundle_skills import claude_native_skill_args
@@ -172,7 +175,7 @@ _DRAFT_NEEDLE_MAX_CHARS = 24
 _TERMINAL_FAILURE_TAIL_LINES = 12
 _TERMINAL_FAILURE_TAIL_CHARS = 800
 
-ToolExecutor = Callable[[str, dict[str, Any]], Awaitable[Any]]
+ToolExecutor = Callable[[str, _JsonObject], Awaitable[object]]
 
 
 def _absolute_syntactic_path(path: Path) -> Path:
@@ -325,7 +328,7 @@ class ClaudeTranscriptItem:
 
     source_id: str
     item_type: str
-    data: dict[str, Any]
+    data: _JsonObject
     response_id: str
     is_compact_summary: bool = False
 
@@ -436,7 +439,7 @@ class ClaudeHookRecord:
     clear_rotated_to: str | None = None
     fork_detected: bool = False
     fork_rotated_to: str | None = None
-    todos: list[dict[str, Any]] | None = None
+    todos: list[_JsonObject] | None = None
     task_id: str | None = None
     task_subject: str | None = None
     task_status: str | None = None
@@ -944,7 +947,7 @@ def ensure_claude_workspace_trusted(workspace: Path) -> None:
     _atomic_write_user_json(config_path, data)
 
 
-def _atomic_write_user_json(path: Path, payload: dict[str, Any]) -> None:
+def _atomic_write_user_json(path: Path, payload: _JsonObject) -> None:
     """
     Atomically rewrite a user-owned JSON config file in place.
 
@@ -1055,7 +1058,7 @@ def write_active_session_id(bridge_dir: Path, session_id: str) -> None:
     _write_json_file(bridge_dir / _CONFIG_FILE, config)
 
 
-def read_permission_hook_config(bridge_dir: Path) -> dict[str, Any]:
+def read_permission_hook_config(bridge_dir: Path) -> _JsonObject:
     """
     Read Omnigent routing details for the permission command hook.
 
@@ -1089,7 +1092,7 @@ def update_permission_hook_auth_headers(
     return True
 
 
-def build_mcp_config(bridge_dir: Path, *, python_executable: str | None = None) -> dict[str, Any]:
+def build_mcp_config(bridge_dir: Path, *, python_executable: str | None = None) -> _JsonObject:
     """
     Build the Claude Code MCP config for the Omnigent bridge server.
 
@@ -1130,7 +1133,7 @@ def build_hook_settings(
     launch_model: str | None = None,
     launch_permission_mode: str | None = None,
     launch_effort: str | None = None,
-) -> dict[str, Any]:
+) -> _JsonObject:
     """
     Build invocation-local Claude Code hook settings.
 
@@ -1198,7 +1201,7 @@ def build_hook_settings(
         "type": "command",
         "command": shlex.join(message_display_command_parts),
     }
-    hooks: dict[str, Any] = {
+    hooks: dict[str, list[_JsonObject]] = {
         "SessionStart": [{"hooks": [session_start_hook]}],
         "Stop": [{"hooks": [hook]}],
         "StopFailure": [{"hooks": [hook]}],
@@ -1270,7 +1273,7 @@ def build_hook_settings(
             "--bridge-dir",
             str(bridge_dir),
         ]
-        permission_hook: dict[str, Any] = {
+        permission_hook: _JsonObject = {
             "type": "command",
             "command": shlex.join(permission_command_parts),
             # Wait up to a day for the verdict. Claude Code's default
@@ -1294,7 +1297,7 @@ def build_hook_settings(
             "--bridge-dir",
             str(bridge_dir),
         ]
-        evaluate_policy_hook: dict[str, Any] = {
+        evaluate_policy_hook: _JsonObject = {
             "type": "command",
             "command": shlex.join(evaluate_policy_command_parts),
         }
@@ -1310,7 +1313,7 @@ def build_hook_settings(
             "--bridge-dir",
             str(bridge_dir),
         ]
-        ask_uq_hook: dict[str, Any] = {
+        ask_uq_hook: _JsonObject = {
             "type": "command",
             "command": shlex.join(ask_uq_command_parts),
             # Short timeout: if the web-UI elicitation isn't answered
@@ -1342,7 +1345,7 @@ def build_hook_settings(
         # server-side. Covers both web-UI-injected and direct-terminal
         # prompts, since both fire UserPromptSubmit.
         hooks["UserPromptSubmit"].append({"hooks": [evaluate_policy_hook]})
-    settings: dict[str, Any] = {"hooks": hooks}
+    settings: _JsonObject = {"hooks": hooks}
     if launch_model:
         settings["model"] = launch_model
     if launch_permission_mode:
@@ -1552,7 +1555,7 @@ def _merge_disallowed_tools(args: list[str], extra: tuple[str, ...]) -> list[str
     return args
 
 
-def record_hook_event(bridge_dir: Path, payload: dict[str, Any]) -> None:
+def record_hook_event(bridge_dir: Path, payload: _JsonObject) -> None:
     """
     Record one Claude Code hook payload in the bridge directory.
 
@@ -2348,7 +2351,7 @@ def _hook_record_from_jsonl_record(record: _JsonlRecord) -> ClaudeHookRecord:
     # Extract todos from PostToolUse/TodoWrite hook payloads. Claude Code
     # fires this hook after every TodoWrite call with ``tool_input.todos``
     # containing the updated list. Other PostToolUse events have no todos.
-    todos: list[dict[str, Any]] | None = None
+    todos: list[_JsonObject] | None = None
     task_id: str | None = None
     task_subject: str | None = None
     task_status: str | None = None
@@ -2542,7 +2545,7 @@ def write_tmux_target(
     :returns: None.
     """
     _ensure_secure_dir(bridge_dir)
-    payload: dict[str, Any] = {
+    payload: _JsonObject = {
         "socket_path": str(socket_path),
         "tmux_target": tmux_target,
         "updated_at": time.time(),
@@ -3296,10 +3299,10 @@ def _wait_for_tmux_info(bridge_dir: Path, *, timeout_s: float) -> dict[str, str]
 def start_tool_relay(
     *,
     bridge_dir: Path,
-    tools: list[dict[str, Any]],
+    tools: list[_JsonObject],
     tool_executor: ToolExecutor,
     loop: asyncio.AbstractEventLoop,
-    policy_client: Any | None = None,
+    policy_client: httpx.AsyncClient | None = None,
     session_id: str | None = None,
 ) -> ClaudeNativeToolRelay:
     """
@@ -3334,7 +3337,7 @@ def start_tool_relay(
     )
     httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler_cls)
     host, port = _http_server_host_port(httpd)
-    relay_info: dict[str, Any] = {
+    relay_info: _JsonObject = {
         "url": f"http://{host}:{port}",
         "token": token,
         "tools": _normalize_relay_tool_specs(tools),
@@ -3398,7 +3401,7 @@ def _serve_mcp(bridge_dir: Path) -> None:
     if not isinstance(token, str) or not token:
         raise SystemExit("bridge config missing token")
 
-    notification_queue: queue.Queue[dict[str, Any] | None] = queue.Queue()
+    notification_queue: queue.Queue[_JsonObject | None] = queue.Queue()
     stdout_lock = threading.Lock()
     httpd = _start_http_ingress(bridge_dir, token, notification_queue)
     tools, close_tools = _build_tools(config)
@@ -3421,7 +3424,7 @@ def _serve_mcp(bridge_dir: Path) -> None:
 def _start_http_ingress(
     bridge_dir: Path,
     token: str,
-    notification_queue: queue.Queue[dict[str, Any] | None],
+    notification_queue: queue.Queue[_JsonObject | None],
 ) -> ThreadingHTTPServer:
     """
     Start the localhost control HTTP server.
@@ -3457,7 +3460,7 @@ def _start_http_ingress(
 
 def _handler_factory(
     token: str,
-    notification_queue: queue.Queue[dict[str, Any] | None],
+    notification_queue: queue.Queue[_JsonObject | None],
 ) -> type[BaseHTTPRequestHandler]:
     """
     Create an HTTP handler class bound to the MCP notification queue.
@@ -3471,7 +3474,7 @@ def _handler_factory(
     class _ControlHandler(BaseHTTPRequestHandler):
         """HTTP handler for the local MCP control endpoint."""
 
-        def log_message(self, format: str, *args: Any) -> None:
+        def log_message(self, format: str, *args: object) -> None:
             """
             Suppress default HTTP server logging.
 
@@ -3514,7 +3517,7 @@ def _handler_factory(
             )
             self._send_json({"ok": True})
 
-        def _send_json(self, payload: dict[str, Any]) -> None:
+        def _send_json(self, payload: _JsonObject) -> None:
             """
             Send a JSON response body.
 
@@ -3536,7 +3539,7 @@ def _tool_relay_handler_factory(
     tool_executor: ToolExecutor,
     loop: asyncio.AbstractEventLoop,
     *,
-    policy_client: Any | None = None,
+    policy_client: httpx.AsyncClient | None = None,
     session_id: str | None = None,
 ) -> type[BaseHTTPRequestHandler]:
     """
@@ -3555,7 +3558,7 @@ def _tool_relay_handler_factory(
     class _ToolRelayHandler(BaseHTTPRequestHandler):
         """HTTP handler for active Omnigent tool relay calls."""
 
-        def log_message(self, format: str, *args: Any) -> None:
+        def log_message(self, format: str, *args: object) -> None:
             """
             Suppress default HTTP server logging.
 
@@ -3594,7 +3597,7 @@ def _tool_relay_handler_factory(
                 arguments = {}
             self._send_json(_run_relay_tool(tool_executor, loop, name, arguments))
 
-        def _handle_policy_evaluate(self, payload: dict[str, Any]) -> None:
+        def _handle_policy_evaluate(self, payload: _JsonObject) -> None:
             if policy_client is None or session_id is None:
                 self.send_error(HTTPStatus.SERVICE_UNAVAILABLE)
                 return
@@ -3619,7 +3622,7 @@ def _tool_relay_handler_factory(
             self.end_headers()
             self.wfile.write(raw)
 
-        def _read_json_body(self) -> dict[str, Any] | None:
+        def _read_json_body(self) -> _JsonObject | None:
             """
             Read and decode a JSON request body.
 
@@ -3637,7 +3640,7 @@ def _tool_relay_handler_factory(
                 return None
             return payload if isinstance(payload, dict) else None
 
-        def _send_json(self, payload: dict[str, Any]) -> None:
+        def _send_json(self, payload: _JsonObject) -> None:
             """
             Send a JSON response body.
 
@@ -3658,8 +3661,8 @@ def _run_relay_tool(
     tool_executor: ToolExecutor,
     loop: asyncio.AbstractEventLoop,
     name: str,
-    arguments: dict[str, Any],
-) -> dict[str, Any]:
+    arguments: _JsonObject,
+) -> _JsonObject:
     """
     Execute one relay tool call on the harness event loop.
 
@@ -3670,7 +3673,9 @@ def _run_relay_tool(
     :param arguments: Decoded tool arguments.
     :returns: MCP tool-call response.
     """
-    future = asyncio.run_coroutine_threadsafe(tool_executor(name, arguments), loop)
+    future = asyncio.run_coroutine_threadsafe(
+        _await_tool_result(tool_executor(name, arguments)), loop
+    )
     try:
         result = future.result(timeout=_TOOL_CALL_TIMEOUT_S)
     except Exception as exc:  # noqa: BLE001 - relay converts callback failures to MCP errors.
@@ -3678,7 +3683,11 @@ def _run_relay_tool(
     return _mcp_response_from_tool_result(result)
 
 
-def _mcp_response_from_tool_result(result: Any) -> dict[str, Any]:
+async def _await_tool_result(result: Awaitable[object]) -> object:
+    return await result
+
+
+def _mcp_response_from_tool_result(result: object) -> _JsonObject:
     """
     Convert a harness tool result into MCP response shape.
 
@@ -3687,7 +3696,7 @@ def _mcp_response_from_tool_result(result: Any) -> dict[str, Any]:
     :returns: MCP tool-call response.
     """
     payload = result if isinstance(result, dict) else {"result": result}
-    response: dict[str, Any] = {
+    response: _JsonObject = {
         "content": [{"type": "text", "text": json.dumps(payload)}],
     }
     if payload.get("blocked") is True or ("error" in payload and payload.get("error")):
@@ -3696,7 +3705,7 @@ def _mcp_response_from_tool_result(result: Any) -> dict[str, Any]:
 
 
 def _notification_writer(
-    notification_queue: queue.Queue[dict[str, Any] | None],
+    notification_queue: queue.Queue[_JsonObject | None],
     stdout_lock: threading.Lock,
 ) -> None:
     """
@@ -3773,7 +3782,7 @@ def _stdio_jsonrpc_loop(
         # tool, or an OSError that slipped a narrower except).
         try:
             result = _handle_mcp_request(method, message.get("params"), tools, bridge_dir)
-            response: dict[str, Any] = {
+            response: _JsonObject = {
                 "jsonrpc": "2.0",
                 "id": request_id,
                 "result": result,
@@ -3790,10 +3799,10 @@ def _stdio_jsonrpc_loop(
 
 def _handle_mcp_request(
     method: str,
-    params: Any,
+    params: object,
     tools: dict[str, Tool],
     bridge_dir: Path,
-) -> dict[str, Any]:
+) -> _JsonObject:
     """
     Handle one MCP request.
 
@@ -3830,7 +3839,7 @@ def _handle_mcp_request(
     return {}
 
 
-def _mcp_tool_schema(tool: Tool) -> dict[str, Any]:
+def _mcp_tool_schema(tool: Tool) -> _JsonObject:
     """
     Convert an Omnigent tool schema into MCP tool-list shape.
 
@@ -3848,7 +3857,7 @@ def _mcp_tool_schema(tool: Tool) -> dict[str, Any]:
 def _combined_mcp_tool_schemas(
     local_tools: dict[str, Tool],
     bridge_dir: Path,
-) -> list[dict[str, Any]]:
+) -> list[_JsonObject]:
     """
     Return local and active-turn relay tools in MCP list shape.
 
@@ -3869,7 +3878,7 @@ def _combined_mcp_tool_schemas(
     return list(schemas.values())
 
 
-def _mcp_tool_schema_from_spec(tool_spec: dict[str, Any]) -> dict[str, Any]:
+def _mcp_tool_schema_from_spec(tool_spec: _JsonObject) -> _JsonObject:
     """
     Convert an Omnigent tool schema dict into MCP tool-list shape.
 
@@ -3888,10 +3897,10 @@ def _mcp_tool_schema_from_spec(tool_spec: dict[str, Any]) -> dict[str, Any]:
 
 
 def _call_mcp_tool(
-    params: Any,
+    params: object,
     tools: dict[str, Tool],
     bridge_dir: Path,
-) -> dict[str, Any]:
+) -> _JsonObject:
     """
     Execute one MCP tool call.
 
@@ -3943,7 +3952,7 @@ def _read_relay_tool_names(bridge_dir: Path) -> set[str]:
     }
 
 
-def _read_relay_tool_specs(bridge_dir: Path) -> list[dict[str, Any]]:
+def _read_relay_tool_specs(bridge_dir: Path) -> list[_JsonObject]:
     """
     Return active relay tool schemas.
 
@@ -3962,8 +3971,8 @@ def _read_relay_tool_specs(bridge_dir: Path) -> list[dict[str, Any]]:
 def _call_relay_tool(
     bridge_dir: Path,
     name: str,
-    arguments: dict[str, Any],
-) -> dict[str, Any]:
+    arguments: _JsonObject,
+) -> _JsonObject:
     """
     Call the active harness turn's tool relay.
 
@@ -4011,7 +4020,7 @@ def _call_relay_tool(
     return decoded
 
 
-def _mcp_error(message: str) -> dict[str, Any]:
+def _mcp_error(message: str) -> _JsonObject:
     """
     Build an MCP error-content tool result.
 
@@ -4021,7 +4030,7 @@ def _mcp_error(message: str) -> dict[str, Any]:
     return {"content": [{"type": "text", "text": json.dumps({"error": message})}], "isError": True}
 
 
-def _normalize_relay_tool_specs(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
+def _normalize_relay_tool_specs(tools: list[_JsonObject]) -> list[_JsonObject]:
     """
     Normalize active-turn tool schemas before advertising them.
 
@@ -4029,7 +4038,7 @@ def _normalize_relay_tool_specs(tools: list[dict[str, Any]]) -> list[dict[str, A
         ``[{"name": "sys_os_read", "parameters": {...}}]``.
     :returns: Schemas containing only fields the MCP bridge needs.
     """
-    normalized: list[dict[str, Any]] = []
+    normalized: list[_JsonObject] = []
     for tool in tools:
         name = tool.get("name")
         if not isinstance(name, str) or not name:
@@ -4048,7 +4057,7 @@ def _normalize_relay_tool_specs(tools: list[dict[str, Any]]) -> list[dict[str, A
     return normalized
 
 
-def _empty_object_schema() -> dict[str, Any]:
+def _empty_object_schema() -> _JsonObject:
     """
     Return a minimal JSON object schema.
 
@@ -4057,7 +4066,7 @@ def _empty_object_schema() -> dict[str, Any]:
     return {"type": "object", "properties": {}}
 
 
-def _build_tools(config: dict[str, Any]) -> tuple[dict[str, Tool], Callable[[], None]]:
+def _build_tools(config: _JsonObject) -> tuple[dict[str, Tool], Callable[[], None]]:
     """
     Build Omnigent MCP tools served by the bridge.
 
@@ -4087,7 +4096,7 @@ def _build_tools(config: dict[str, Any]) -> tuple[dict[str, Tool], Callable[[],
 
 
 def _write_jsonrpc(
-    payload: dict[str, Any],
+    payload: _JsonObject,
     stdout_lock: threading.Lock,
     *,
     framed: bool = False,
@@ -4111,7 +4120,7 @@ def _write_jsonrpc(
             print(raw, flush=True)
 
 
-def _model_from_transcript_entry(entry: dict[str, Any]) -> str | None:
+def _model_from_transcript_entry(entry: _JsonObject) -> str | None:
     """
     Return ``message.model`` from an assistant transcript record.
 
@@ -4133,7 +4142,7 @@ def _model_from_transcript_entry(entry: dict[str, Any]) -> str | None:
     return None
 
 
-def read_claude_context_state(bridge_dir: Path) -> dict[str, Any] | None:
+def read_claude_context_state(bridge_dir: Path) -> _JsonObject | None:
     """
     Read the most recent statusLine snapshot from ``context.json``.
 
@@ -4252,7 +4261,7 @@ def read_user_effort_level() -> str | None:
     return None
 
 
-def _usage_from_transcript_entry(entry: dict[str, Any]) -> dict[str, int] | None:
+def _usage_from_transcript_entry(entry: _JsonObject) -> dict[str, int] | None:
     """
     Extract token-usage from one Claude assistant transcript entry.
 
@@ -4325,7 +4334,7 @@ def _assistant_text_from_transcript_line(line: str) -> str | None:
 
 
 def _transcript_items_from_entry(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     *,
     line_number: int,
     record_offset: int | None = None,
@@ -4396,7 +4405,7 @@ def _transcript_items_from_entry(
 
 
 def _attachment_transcript_items_from_entry(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     *,
     line_number: int,
     record_offset: int | None,
@@ -4650,7 +4659,7 @@ def _is_task_notification_text(text: str) -> bool:
 
 
 def _local_command_transcript_items_from_entry(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     *,
     line_number: int,
     record_offset: int | None,
@@ -4754,7 +4763,7 @@ def _terminal_command_items_from_content(
 
 
 def _user_transcript_items_from_entry(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     *,
     line_number: int,
     record_offset: int | None,
@@ -4830,7 +4839,7 @@ def _user_transcript_items_from_entry(
             if payload is None or payload.name in _CLAUDE_CLI_DROPPED_COMMANDS:
                 return current_response_id, []
             kind = "command" if payload.name in _CLAUDE_CLI_SURFACED_COMMANDS else "skill"
-            data: dict[str, Any] = {
+            data: _JsonObject = {
                 "agent": agent_name,
                 "kind": kind,
                 "name": payload.name,
@@ -4902,7 +4911,7 @@ def _user_transcript_items_from_entry(
     if not isinstance(content, list):
         return current_response_id, []
 
-    user_blocks: list[dict[str, Any]] = []
+    user_blocks: list[_JsonObject] = []
     saw_user_text = False
     item_index = 0
     for block in content:
@@ -4982,7 +4991,7 @@ def _user_transcript_items_from_entry(
 
 
 def _assistant_transcript_items_from_entry(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     *,
     line_number: int,
     record_offset: int | None,
@@ -5112,7 +5121,7 @@ def _assistant_message_item(
     )
 
 
-def _stripped_image_placeholder(source: dict[str, Any]) -> str:
+def _stripped_image_placeholder(source: _JsonObject) -> str:
     """
     Build the placeholder text for a stripped inline image block.
 
@@ -5132,7 +5141,7 @@ def _stripped_image_placeholder(source: dict[str, Any]) -> str:
     )
 
 
-def _strip_inline_image_data(value: Any) -> Any:
+def _strip_inline_image_data(value: object) -> object:
     """
     Replace base64 image payloads with a lightweight placeholder.
 
@@ -5159,7 +5168,7 @@ def _strip_inline_image_data(value: Any) -> Any:
     return value
 
 
-def _tool_result_output(entry: dict[str, Any], block: dict[str, Any]) -> str:
+def _tool_result_output(entry: _JsonObject, block: _JsonObject) -> str:
     """
     Return the UI-facing output string for a Claude tool result.
 
@@ -5186,7 +5195,7 @@ def _tool_result_output(entry: dict[str, Any], block: dict[str, Any]) -> str:
 
 
 def _transcript_source_key(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     line_number: int,
     record_offset: int | None = None,
 ) -> str:
@@ -5210,7 +5219,7 @@ def _transcript_source_key(
 
 
 def _parent_or_record_source_key(
-    entry: dict[str, Any],
+    entry: _JsonObject,
     line_number: int,
     record_offset: int | None = None,
 ) -> str:
@@ -5252,7 +5261,7 @@ def _source_id(source_key: str, item_index: int, item_type: str) -> str:
     return f"{source_key}:{item_index}:{item_type}"
 
 
-def _summary_text_from_blocks(content: Any) -> str:
+def _summary_text_from_blocks(content: object) -> str:
     """
     Join the text of a compact-summary record's content blocks.
 
@@ -5280,7 +5289,7 @@ def _summary_text_from_blocks(content: Any) -> str:
     return "\n".join(parts)
 
 
-def _wait_for_server_info(bridge_dir: Path, *, timeout_s: float) -> dict[str, Any]:
+def _wait_for_server_info(bridge_dir: Path, *, timeout_s: float) -> _JsonObject:
     """
     Wait for the bridge control HTTP endpoint file.
 
@@ -5302,7 +5311,7 @@ def _wait_for_server_info(bridge_dir: Path, *, timeout_s: float) -> dict[str, An
     )
 
 
-def _read_json_file(path: Path) -> dict[str, Any]:
+def _read_json_file(path: Path) -> _JsonObject:
     """
     Read a JSON object file.
 
@@ -5320,7 +5329,7 @@ def _read_json_file(path: Path) -> dict[str, Any]:
     return payload if isinstance(payload, dict) else {}
 
 
-def _write_json_file(path: Path, payload: dict[str, Any]) -> None:
+def _write_json_file(path: Path, payload: _JsonObject) -> None:
     """
     Atomically write a JSON object file with owner-only permissions.
 

omnigent/claude_native_forwarder.py

@@ -3560,9 +3560,12 @@ async def _forward_available_items(
     # numerator fallback only when the statusLine hasn't fired yet
     # (e.g. cold-resume before the first render tick).
     status_state = await asyncio.to_thread(read_claude_context_state, bridge_dir)
-    resolved_context_window = (
+    context_window_value = (
         status_state.get("context_window_size") if status_state is not None else None
     )
+    resolved_context_window = (
+        context_window_value if isinstance(context_window_value, int) else None
+    )
     usage_from_status = (
         _usage_from_status_state(status_state) if status_state is not None else None
     )

omnigent/opencode_native_provider.py

@@ -213,13 +213,34 @@ def build_opencode_omnigent_mcp_server(
     claude_cfg = build_mcp_config(bridge_dir, python_executable=python_executable)
     # build_mcp_config returns {"mcpServers": {"<name>": {command, args, env}}};
     # opencode wants a flat command list + ``environment``.
-    name, server = next(iter(claude_cfg["mcpServers"].items()))
+    servers = claude_cfg.get("mcpServers")
+    if not isinstance(servers, dict) or not servers:
+        raise ValueError("Claude MCP config is missing mcpServers")
+    name, server = next(iter(servers.items()))
+    if not isinstance(server, dict):
+        raise ValueError("Claude MCP server config is malformed")
+    command = server.get("command")
+    args = server.get("args", [])
+    if (
+        not isinstance(command, str)
+        or not isinstance(args, list)
+        or not all(isinstance(arg, str) for arg in args)
+    ):
+        raise ValueError("Claude MCP server command is malformed")
     entry: dict[str, object] = {
         "type": "local",
-        "command": [server["command"], *server.get("args", [])],
+        "command": [command, *args],
         "enabled": True,
     }
-    env = dict(server.get("env", {}) or {})
+    env_value = server.get("env")
+    if env_value is None:
+        env: dict[str, str] = {}
+    elif isinstance(env_value, dict) and all(
+        isinstance(key, str) and isinstance(value, str) for key, value in env_value.items()
+    ):
+        env = dict(env_value)
+    else:
+        raise ValueError("Claude MCP server environment is malformed")
     if env:
         entry["environment"] = env
     return {str(name): entry}

tests/test_opencode_native_provider.py

@@ -52,6 +52,26 @@ def test_build_omnigent_mcp_server_honors_python_executable() -> None:
     assert block["omnigent"]["command"][0] == "/custom/python"
 
 
+@pytest.mark.parametrize(
+    "server",
+    [
+        {"command": "python", "args": [1], "env": {}},
+        {"command": "python", "args": [], "env": {"TOKEN": 1}},
+    ],
+)
+def test_build_omnigent_mcp_server_rejects_non_string_values(
+    monkeypatch: pytest.MonkeyPatch,
+    server: dict[str, object],
+) -> None:
+    monkeypatch.setattr(
+        "omnigent.claude_native_bridge.build_mcp_config",
+        lambda bridge_dir, *, python_executable=None: {"mcpServers": {"omnigent": server}},
+    )
+
+    with pytest.raises(ValueError, match="Claude MCP server"):
+        build_opencode_omnigent_mcp_server(Path("/tmp/b"))
+
+
 def test_build_model_default_config_pins_model_without_provider_block() -> None:
     cfg = build_opencode_model_default_config("anthropic/claude-sonnet-4-5")
     assert cfg == {