From a0cc8cdb253a881369efa1ecbfe8b0d84254e401 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:26:51 -0700 Subject: [PATCH] refactor(agent/transports): drop _Pending/_take_id and __enter__/__exit__ shims, lift file-change summary + scope-id helpers, tighten projector/bedrock/types/vertex --- agent/transports/bedrock.py | 30 ++-- agent/transports/codex_app_server.py | 27 +--- agent/transports/codex_app_server_session.py | 161 ++++++++----------- agent/transports/codex_event_projector.py | 44 +++-- agent/transports/hermes_tools_mcp_server.py | 38 +---- agent/transports/types.py | 9 +- agent/vertex_adapter.py | 20 +-- 7 files changed, 113 insertions(+), 216 deletions(-) diff --git a/agent/transports/bedrock.py b/agent/transports/bedrock.py index 9a6f35ea3b..1e905f592e 100644 --- a/agent/transports/bedrock.py +++ b/agent/transports/bedrock.py @@ -15,12 +15,8 @@ class BedrockTransport(ProviderTransport): # The adapter already maps inside normalize_converse_response; this serves raw-response access. _STOP_REASON_MAP = { - "end_turn": "stop", - "tool_use": "tool_calls", - "max_tokens": "length", - "stop_sequence": "stop", - "guardrail_intervened": "content_filter", - "content_filtered": "content_filter", + "end_turn": "stop", "tool_use": "tool_calls", "max_tokens": "length", "stop_sequence": "stop", + "guardrail_intervened": "content_filter", "content_filtered": "content_filter", } @property @@ -61,24 +57,18 @@ class BedrockTransport(ProviderTransport): choice = ns.choices[0] msg = choice.message - tool_calls = None - if msg.tool_calls: - tool_calls = [ToolCall(id=tc.id, name=tc.function.name, arguments=tc.function.arguments) for tc in msg.tool_calls] - usage = None - if hasattr(ns, "usage") and ns.usage: - usage = Usage.from_openai(ns.usage) - - provider_data = {} - if getattr(msg, "reasoning_details", None): - provider_data["reasoning_details"] = msg.reasoning_details - if getattr(msg, "bedrock_content_blocks", None): - provider_data["bedrock_content_blocks"] = msg.bedrock_content_blocks - + tool_calls = ( + [ToolCall(id=tc.id, name=tc.function.name, arguments=tc.function.arguments) for tc in msg.tool_calls] + if msg.tool_calls else None + ) + provider_data = { + key: getattr(msg, key) for key in ("reasoning_details", "bedrock_content_blocks") if getattr(msg, key, None) + } return NormalizedResponse( content=msg.content, tool_calls=tool_calls, finish_reason=choice.finish_reason or "stop", reasoning=getattr(msg, "reasoning", None) or getattr(msg, "reasoning_content", None), - usage=usage, provider_data=provider_data or None, + usage=Usage.from_openai(ns.usage) if getattr(ns, "usage", None) else None, provider_data=provider_data or None, ) def validate_response(self, response: Any) -> bool: diff --git a/agent/transports/codex_app_server.py b/agent/transports/codex_app_server.py index 89e36d106f..330448ed83 100644 --- a/agent/transports/codex_app_server.py +++ b/agent/transports/codex_app_server.py @@ -15,8 +15,7 @@ import queue import re import subprocess import threading -import time -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Any, Optional from tools.environments.local import hermes_subprocess_env @@ -36,13 +35,6 @@ class CodexAppServerError(RuntimeError): return f"codex app-server error {self.code}: {self.message}" -@dataclass -class _Pending: - queue: queue.Queue - method: str - sent_at: float = field(default_factory=time.time) - - class CodexAppServerClient: """Minimal synchronous JSON-RPC 2.0 client for ``codex app-server`` over stdio. @@ -90,7 +82,7 @@ class CodexAppServerClient: bufsize=0, env=spawn_env, creationflags=windows_hide_flags(), ) self._next_id = 1 - self._pending: dict[int, _Pending] = {} + self._pending: dict[int, queue.Queue] = {} # request id -> single-slot reply queue self._pending_lock = threading.Lock() self._notifications: queue.Queue = queue.Queue() self._server_requests: queue.Queue = queue.Queue() @@ -148,10 +140,10 @@ class CodexAppServerClient: def request(self, method: str, params: Optional[dict] = None, timeout: float = 30.0) -> dict: """Send a request and block for ``result``; raise CodexAppServerError on ``error``.""" - rid = self._take_id() + rid, self._next_id = self._next_id, self._next_id + 1 q: queue.Queue = queue.Queue(maxsize=1) with self._pending_lock: - self._pending[rid] = _Pending(queue=q, method=method) + self._pending[rid] = q self._send({"id": rid, "method": method, "params": params or {}}) try: msg = q.get(timeout=timeout) @@ -182,9 +174,7 @@ class CodexAppServerClient: @staticmethod def _take(q: queue.Queue, timeout: float) -> Optional[dict]: try: - if timeout <= 0: - return q.get_nowait() - return q.get(timeout=timeout) + return q.get_nowait() if timeout <= 0 else q.get(timeout=timeout) except queue.Empty: return None @@ -204,11 +194,6 @@ class CodexAppServerClient: def is_alive(self) -> bool: return self._proc.poll() is None - def _take_id(self) -> int: - rid = self._next_id - self._next_id += 1 - return rid - def _send(self, obj: dict) -> None: if self._closed: raise RuntimeError("codex app-server client is closed") @@ -250,7 +235,7 @@ class CodexAppServerClient: pending = self._pending.pop(msg["id"], None) if pending is not None: try: - pending.queue.put_nowait(msg) + pending.put_nowait(msg) except queue.Full: # pragma: no cover - defensive pass elif "method" in msg: # server-initiated request (has id) or notification diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index 446f52ae1d..d14d2fee9a 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -19,9 +19,7 @@ from typing import Any, Callable, Optional from agent.codex_responses_adapter import _format_responses_error from agent.redact import redact_sensitive_text -from agent.transports.codex_app_server import ( - CodexAppServerClient, CodexAppServerError -) +from agent.transports.codex_app_server import CodexAppServerClient, CodexAppServerError from agent.transports.codex_event_projector import CodexEventProjector, ProjectionResult logger = logging.getLogger(__name__) @@ -63,25 +61,25 @@ class TurnResult: _TURN_ABORTED_MARKERS = ("", "") +def _first_scope_id(*lookups: tuple[Any, str, str]) -> Any: + """``src.get(a) or src.get(b)`` over successive dict sources until one is not None.""" + for src, primary, fallback in lookups: + if isinstance(src, dict): + observed = src.get(primary) or src.get(fallback) + if observed is not None: + return observed + return None + + def _notification_scope_ids(note: dict) -> tuple[Optional[str], Optional[str]]: """Extract the thread/turn identity carried by a notification (top-level, then turn/item).""" params = (note.get("params") or {}) if isinstance(note, dict) else None if not isinstance(params, dict): return None, None turn, item = params.get("turn") or {}, params.get("item") or {} - - def first(*lookups: tuple[Any, str, str]) -> Any: - """``src.get(a) or src.get(b)`` over successive dict sources until one is not None.""" - for src, primary, fallback in lookups: - if isinstance(src, dict): - observed = src.get(primary) or src.get(fallback) - if observed is not None: - return observed - return None - return ( - first((params, "threadId", "thread_id"), (turn, "threadId", "thread_id"), (item, "threadId", "thread_id")), - first((params, "turnId", "turn_id"), (turn, "id", "turnId"), (item, "turnId", "turn_id")), + _first_scope_id((params, "threadId", "thread_id"), (turn, "threadId", "thread_id"), (item, "threadId", "thread_id")), + _first_scope_id((params, "turnId", "turn_id"), (turn, "id", "turnId"), (item, "turnId", "turn_id")), ) @@ -95,9 +93,11 @@ def _notification_belongs_to_turn(note: dict, *, thread_id: Optional[str], turn_ if not isinstance(note, dict): return False observed_thread_id, observed_turn_id = _notification_scope_ids(note) - if thread_id is not None and observed_thread_id is not None and str(observed_thread_id) != str(thread_id): - return False - return not (turn_id is not None and observed_turn_id is not None and str(observed_turn_id) != str(turn_id)) + + def foreign(expected: Optional[str], observed: Optional[str]) -> bool: + return expected is not None and observed is not None and str(observed) != str(expected) + + return not (foreign(thread_id, observed_thread_id) or foreign(turn_id, observed_turn_id)) def _coerce_turn_input_text(user_input: Any) -> str: @@ -113,8 +113,7 @@ def _coerce_turn_input_text(user_input: Any) -> str: parts: list[str] = [] for item in user_input: if not isinstance(item, dict): - keep = item.strip() if isinstance(item, str) else item is not None - if keep: + if item.strip() if isinstance(item, str) else item is not None: parts.append(str(item)) continue item_type = item.get("type") @@ -149,9 +148,7 @@ _OAUTH_REAUTH_HINT = ( def _classify_oauth_failure(*parts: str) -> Optional[str]: """Re-auth hint if any part looks like a codex OAuth/token-refresh failure, else None.""" haystack = " ".join(p for p in parts if p).lower() - if haystack and any(needle in haystack for needle in _OAUTH_REFRESH_FAILURE_HINTS): - return _OAUTH_REAUTH_HINT - return None + return _OAUTH_REAUTH_HINT if any(needle in haystack for needle in _OAUTH_REFRESH_FAILURE_HINTS) else None @dataclass @@ -216,9 +213,7 @@ class CodexAppServerSession: result = self._client.request("thread/start", {"cwd": self._cwd}, timeout=15) # Different codex versions serialize the id under thread.id / sessionId / threadId. thread_obj = result.get("thread") or {} - thread_id = ( - thread_obj.get("id") or thread_obj.get("sessionId") or result.get("sessionId") or result.get("threadId") - ) + thread_id = thread_obj.get("id") or thread_obj.get("sessionId") or result.get("sessionId") or result.get("threadId") if not thread_id: raise CodexAppServerError( code=-32603, @@ -244,12 +239,6 @@ class CodexAppServerSession: self._client = None self._thread_id = None - def __enter__(self) -> "CodexAppServerSession": - return self - - def __exit__(self, *exc: Any) -> None: - self.close() - def request_interrupt(self) -> None: """Idempotent: signal the active turn loop to issue turn/interrupt and unwind.""" self._interrupt_event.set() @@ -285,7 +274,7 @@ class CodexAppServerSession: becomes diagnosable. Specific classifications (OAuth, wedge watchdog) produce their own clean hint instead. """ - exc_str = str(exc) if exc != "" and exc is not None else "" + exc_str = "" if exc is None else str(exc) base = f"{prefix}: {exc_str}" if exc_str else prefix try: tail = self._client.stderr_tail(tail_lines) if self._client is not None else [] @@ -344,10 +333,7 @@ class CodexAppServerSession: if self._client.is_alive(): return False hint = _classify_oauth_failure(self._stderr_blob(60)) - self._retire( - result, - hint or self._format_error_with_stderr("codex app-server subprocess exited unexpectedly", tail_lines=20), - ) + self._retire(result, hint or self._format_error_with_stderr("codex app-server subprocess exited unexpectedly", tail_lines=20)) return True def _absorb_notification( @@ -373,8 +359,8 @@ class CodexAppServerSession: if projection.final_text is not None: # Multiple agentMessage items per turn: the last one is canonical. result.final_text = projection.final_text - if _has_turn_aborted_marker(projection.final_text): - aborted = True + aborted = _has_turn_aborted_marker(projection.final_text) + if aborted: result.interrupted = True result.error = result.error or "codex reported turn_aborted" return projection, aborted @@ -418,19 +404,15 @@ class CodexAppServerSession: last_tool_completion_at: Optional[float] = None def watchdog_tripped() -> bool: - if ( - last_tool_completion_at is not None - and (time.monotonic() - last_tool_completion_at) > post_tool_quiet_timeout - ): - self._issue_interrupt(result.turn_id) - result.interrupted = True - self._retire( - result, - f"codex went silent for {post_tool_quiet_timeout:.0f}s after a tool result; " - f"retiring app-server session.", - ) - return True - return False + if last_tool_completion_at is None or (time.monotonic() - last_tool_completion_at) <= post_tool_quiet_timeout: + return False + self._issue_interrupt(result.turn_id) + result.interrupted = True + self._retire( + result, + f"codex went silent for {post_tool_quiet_timeout:.0f}s after a tool result; retiring app-server session.", + ) + return True def on_server_request(sreq: dict) -> bool: nonlocal last_tool_completion_at @@ -442,12 +424,9 @@ class CodexAppServerSession: pending = self._client.take_notification(timeout=0) if pending is None: break - if not _notification_belongs_to_turn( - pending, thread_id=self._thread_id, turn_id=result.turn_id - ): + if not _notification_belongs_to_turn(pending, thread_id=self._thread_id, turn_id=result.turn_id): logger.debug( - "ignoring foreign codex notification while draining " - "server request: method=%s", + "ignoring foreign codex notification while draining server request: method=%s", pending.get("method"), ) continue @@ -472,13 +451,9 @@ class CodexAppServerSession: return aborted turn_obj = (note.get("params") or {}).get("turn") or {} turn_status = turn_obj.get("status") - if turn_status and turn_status not in {"completed", "interrupted"}: - err_obj = turn_obj.get("error") - if err_obj: - err_msg = _format_responses_error(err_obj, str(turn_status)) - self._set_classified_error( - result, f"turn ended status={turn_status}", err_msg, err_msg - ) + if turn_status and turn_status not in {"completed", "interrupted"} and turn_obj.get("error"): + err_msg = _format_responses_error(turn_obj["error"], str(turn_status)) + self._set_classified_error(result, f"turn ended status={turn_status}", err_msg, err_msg) return True self._drive_turn( @@ -529,25 +504,16 @@ class CodexAppServerSession: method = note.get("method", "") if pre_scope_filter is not None and not pre_scope_filter(note, method): continue - if not _notification_belongs_to_turn( - note, thread_id=self._thread_id, turn_id=result.turn_id - ): + if not _notification_belongs_to_turn(note, thread_id=self._thread_id, turn_id=result.turn_id): logger.debug("ignoring foreign codex notification: method=%s", method) continue if on_note(note, method): turn_complete = True - if ( - accept_final_text_at_deadline - and not turn_complete - and not result.interrupted - and result.final_text - and result.error is None - ): + if accept_final_text_at_deadline and not turn_complete and not result.interrupted and result.final_text and result.error is None: logger.warning( - "codex app-server turn reached deadline after a completed " - "assistant message but before turn/completed; accepting " - "the assistant text as the terminal response" + "codex app-server turn reached deadline after a completed assistant message but before " + "turn/completed; accepting the assistant text as the terminal response" ) turn_complete = True @@ -573,9 +539,7 @@ class CodexAppServerSession: self._interrupt_event.clear() projector = CodexEventProjector() - if self._request_for( - result, "thread/compact/start", {"threadId": self._thread_id}, "thread/compact/start" - ) is None: + if self._request_for(result, "thread/compact/start", {"threadId": self._thread_id}, "thread/compact/start") is None: return result def pre_scope_filter(note: dict, method: str) -> bool: @@ -611,9 +575,7 @@ class CodexAppServerSession: result.error = result.error or "compact turn interrupted" elif turn_status and turn_status != "completed": err_msg = _format_responses_error(turn_obj.get("error"), str(turn_status)) - self._set_classified_error( - result, f"compact turn ended status={turn_status}", err_msg, err_msg - ) + self._set_classified_error(result, f"compact turn ended status={turn_status}", err_msg, err_msg) return True def on_server_request(sreq: dict) -> bool: @@ -730,21 +692,24 @@ class CodexAppServerSession: if method == "item/completed": self._pending_file_changes.pop(item_id, None) elif method == "item/started": - raw_changes = item.get("changes") or [] - if not raw_changes: - self._pending_file_changes[item_id] = "1 change pending" - return - changes = [ch for ch in raw_changes if isinstance(ch, dict)] - kinds: dict[str, int] = {} - for ch in changes: - kind = (ch.get("kind") or {}).get("type") or "update" - kinds[kind] = kinds.get(kind, 0) + 1 - paths: list[str] = [ch["path"] for ch in changes if ch.get("path")] - counts = ", ".join(f"{n} {k}" for k, n in sorted(kinds.items())) - preview = ", ".join(paths[:3]) - if len(paths) > 3: - preview += f", +{len(paths) - 3} more" - self._pending_file_changes[item_id] = f"{counts}: {preview}" if preview else counts + self._pending_file_changes[item_id] = _summarize_file_changes(item.get("changes") or []) + + +def _summarize_file_changes(raw_changes: list) -> str: + """One-line ``" add, update: a.py, b.py, +k more"`` summary of a fileChange item's changes.""" + if not raw_changes: + return "1 change pending" + changes = [ch for ch in raw_changes if isinstance(ch, dict)] + kinds: dict[str, int] = {} + for ch in changes: + kind = (ch.get("kind") or {}).get("type") or "update" + kinds[kind] = kinds.get(kind, 0) + 1 + paths: list[str] = [ch["path"] for ch in changes if ch.get("path")] + counts = ", ".join(f"{n} {k}" for k, n in sorted(kinds.items())) + preview = ", ".join(paths[:3]) + if len(paths) > 3: + preview += f", +{len(paths) - 3} more" + return f"{counts}: {preview}" if preview else counts def _apply_token_usage_notification(result: TurnResult, note: dict) -> None: diff --git a/agent/transports/codex_event_projector.py b/agent/transports/codex_event_projector.py index 649e44507f..af96848acd 100644 --- a/agent/transports/codex_event_projector.py +++ b/agent/transports/codex_event_projector.py @@ -64,11 +64,9 @@ class CodexEventProjector: Streaming deltas are display-only, mirroring how Hermes writes the assistant message only after the streaming completion event. """ - method = notification.get("method", "") - params = notification.get("params", {}) or {} - if method != "item/completed": + if notification.get("method", "") != "item/completed": return ProjectionResult() - item = params.get("item") or {} + item = (notification.get("params", {}) or {}).get("item") or {} item_type = item.get("type") or "" item_id = item.get("id") or "" if item_type == "agentMessage": @@ -98,16 +96,15 @@ class CodexEventProjector: text = item.get("text") or "" return ProjectionResult(messages=[self._assistant_message(text)], final_text=text) - def _project_user_message(self, item: dict) -> ProjectionResult: + @staticmethod + def _project_user_message(item: dict) -> ProjectionResult: # userMessage content is a list of UserInput variants; flatten text # fragments and drop non-text parts (Hermes' messages store text only). - text_parts: list[str] = [] - for fragment in item.get("content") or []: - if isinstance(fragment, dict): - if fragment.get("type") == "text": - text_parts.append(fragment.get("text") or "") - elif "text" in fragment: - text_parts.append(str(fragment["text"])) + text_parts = [ + (fragment.get("text") or "") if fragment.get("type") == "text" else str(fragment["text"]) + for fragment in item.get("content") or [] + if isinstance(fragment, dict) and (fragment.get("type") == "text" or "text" in fragment) + ] return ProjectionResult(messages=[{"role": "user", "content": "\n".join(text_parts)}]) def _project_tool_item( @@ -139,10 +136,7 @@ class CodexEventProjector: def _file_change_spec(item: dict) -> tuple[str, str, dict, str]: # Per-file change kinds only — full file contents can be huge. changes_summary = [ - { - "kind": (change.get("kind") or {}).get("type") or "update", - "path": change.get("path") or "", - } + {"kind": (change.get("kind") or {}).get("type") or "update", "path": change.get("path") or ""} for change in item.get("changes") or [] ] status = item.get("status") or "unknown" @@ -153,14 +147,11 @@ class CodexEventProjector: def _mcp_tool_call_spec(item: dict) -> tuple[str, str, dict, str]: server = item.get("server") or "mcp" tool = item.get("tool") or "unknown" - result = item.get("result") - error = item.get("error") + result, error = item.get("result"), item.get("error") if error: content = f"[error] {json.dumps(error, ensure_ascii=False)[:1000]}" - elif result is not None: - content = json.dumps(result, ensure_ascii=False)[:4000] else: - content = "" + content = json.dumps(result, ensure_ascii=False)[:4000] if result is not None else "" # Mirror the native MCP name convention (mcp__server__tool) in the call id # so it stays consistent with registration names. return f"mcp__{server}__{tool}", f"mcp.{server}.{tool}", _dict_args(item.get("arguments")), content @@ -169,10 +160,10 @@ class CodexEventProjector: def _dynamic_tool_call_spec(item: dict) -> tuple[str, str, dict, str]: tool = item.get("tool") or "unknown" content_items = item.get("contentItems") or [] - if isinstance(content_items, list) and content_items: - content = json.dumps(content_items, ensure_ascii=False)[:4000] - else: - content = f"success={item.get('success')}" + content = ( + json.dumps(content_items, ensure_ascii=False)[:4000] if isinstance(content_items, list) and content_items + else f"success={item.get('success')}" + ) return f"dyn_{tool}", tool, _dict_args(item.get("arguments")), content _TOOL_PROJECTIONS: dict[str, Callable[[dict], tuple[str, str, dict, str]]] = { @@ -182,7 +173,8 @@ class CodexEventProjector: "dynamicToolCall": _dynamic_tool_call_spec, } - def _project_opaque(self, item: dict, item_type: str) -> ProjectionResult: + @staticmethod + def _project_opaque(item: dict, item_type: str) -> ProjectionResult: try: payload = json.dumps(item, ensure_ascii=False)[:1500] except (TypeError, ValueError): diff --git a/agent/transports/hermes_tools_mcp_server.py b/agent/transports/hermes_tools_mcp_server.py index a3801325e4..05c619a932 100644 --- a/agent/transports/hermes_tools_mcp_server.py +++ b/agent/transports/hermes_tools_mcp_server.py @@ -45,37 +45,16 @@ def _signature_from_schema(schema: dict | None) -> tuple[inspect.Signature, dict # approval UI; delegate_task/memory/session_search/todo — ``_AGENT_LOOP_TOOLS`` # need the running AIAgent context, which a stateless MCP callback lacks. EXPOSED_TOOLS: tuple[str, ...] = ( - "web_search", - "web_extract", - "browser_navigate", - "browser_click", - "browser_type", - "browser_press", - "browser_snapshot", - "browser_scroll", - "browser_back", - "browser_get_images", - "browser_console", - "browser_vision", - "vision_analyze", - "image_generate", - "skill_view", - "skills_list", - "text_to_speech", + "web_search", "web_extract", + "browser_navigate", "browser_click", "browser_type", "browser_press", "browser_snapshot", "browser_scroll", + "browser_back", "browser_get_images", "browser_console", "browser_vision", + "vision_analyze", "image_generate", "skill_view", "skills_list", "text_to_speech", # Kanban handoff tools: stateless (read HERMES_KANBAN_TASK, write kanban.db). # Without them a codex-runtime worker can't report completion and hangs. - "kanban_complete", - "kanban_block", - "kanban_request_review", - "kanban_request_changes", - "kanban_comment", - "kanban_heartbeat", - "kanban_show", - "kanban_list", + "kanban_complete", "kanban_block", "kanban_request_review", "kanban_request_changes", "kanban_comment", + "kanban_heartbeat", "kanban_show", "kanban_list", # Orchestrator-only (the kanban tool gates them on HERMES_KANBAN_TASK unset). - "kanban_create", - "kanban_unblock", - "kanban_link", + "kanban_create", "kanban_unblock", "kanban_link", ) @@ -135,9 +114,8 @@ def _build_server() -> Any: continue description = spec.get("description") or f"Hermes {name} tool" params_schema = spec.get("parameters") or {"type": "object", "properties": {}} - handler = _make_handler(name, params_schema, description) try: - mcp.add_tool(handler, name=name, description=description) + mcp.add_tool(_make_handler(name, params_schema, description), name=name, description=description) except TypeError: # Older mcp SDK: decorator-style registration; __signature__ still drives schema. mcp.tool(name=name, description=description)(_make_handler(name, params_schema, description)) diff --git a/agent/transports/types.py b/agent/transports/types.py index e572fdab5d..82c5f5c40c 100644 --- a/agent/transports/types.py +++ b/agent/transports/types.py @@ -30,13 +30,8 @@ class ToolCall: # Back-compat: run_agent reads tc.function.name / tc.function.arguments (45+ # sites) and getattr()s the provider fields, so expose them as properties. - @property - def type(self) -> str: - return "function" - - @property - def function(self) -> ToolCall: - return self + type = property(lambda self: "function") + function = property(lambda self: self) def _pd(self, key: str) -> Any: return (self.provider_data or {}).get(key) diff --git a/agent/vertex_adapter.py b/agent/vertex_adapter.py index bb6c18238a..f7eecfd634 100644 --- a/agent/vertex_adapter.py +++ b/agent/vertex_adapter.py @@ -58,10 +58,7 @@ def _vertex_config() -> dict: def _env_or_config(env_var: str, config_key: str) -> str: """Setting precedence: env/secret > config.yaml; "" when neither is set.""" - env_value = (_get_secret(env_var) or "").strip() - if env_value: - return env_value - return str(_vertex_config().get(config_key) or "").strip() + return (_get_secret(env_var) or "").strip() or str(_vertex_config().get(config_key) or "").strip() def _resolve_region(explicit: Optional[str] = None) -> str: @@ -95,12 +92,10 @@ def _sa_snapshot(resolved_path: Optional[str]) -> Tuple[Optional[bytes], Tuple[A - Readable file: (bytes, (path, sha256)). - Unreadable file: (None, (path,)) — the caller falls back to the SDK's own file read. - The key fingerprints file CONTENT, not stat metadata: a metadata-preserving - atomic replacement (equal-length JSON, restored mtime) yields a different - private key under an identical stat signature, and this cache guards an - identity. Returning the bytes lets the caller build credentials from the SAME - snapshot the key was computed from (no stat->read TOCTOU); one read + sha256 - per probe is noise next to the OAuth token mint the cache avoids. + The key fingerprints file CONTENT, not stat metadata (a metadata-preserving + atomic replacement can swap the private key under an identical stat signature, + and this cache guards an identity). Returning the bytes lets the caller build + credentials from the SAME snapshot the key was computed from (no stat->read TOCTOU). """ if not resolved_path: return None, ("__adc__",) @@ -140,10 +135,7 @@ def _needs_refresh(creds) -> bool: return ( not getattr(creds, "token", None) or getattr(creds, "expired", False) - or ( - getattr(creds, "expiry", None) is not None - and (creds.expiry.timestamp() - time.time()) < 300 - ) + or (getattr(creds, "expiry", None) is not None and (creds.expiry.timestamp() - time.time()) < 300) )