From 88a82327e60b885a1ab4954bf2132cada2346d2f Mon Sep 17 00:00:00 2001 From: 0xallam Date: Thu, 19 Mar 2026 17:36:44 -0700 Subject: [PATCH] Fix browser review regressions --- containers/healthcheck.sh | 5 +- strix/interface/main.py | 34 +- .../tool_components/browser_renderer.py | 473 ++++++------------ strix/tools/browser/browser_actions.py | 3 + 4 files changed, 169 insertions(+), 346 deletions(-) diff --git a/containers/healthcheck.sh b/containers/healthcheck.sh index 5bf40d5e..43669bf9 100755 --- a/containers/healthcheck.sh +++ b/containers/healthcheck.sh @@ -14,8 +14,9 @@ if ! curl -sf --max-time 3 -H "Authorization: Bearer ${TOOL_SERVER_TOKEN}" "http exit 1 fi -# 2. Caido proxy must be reachable -if ! curl -sf --max-time 3 -o /dev/null "http://127.0.0.1:${CAIDO_PORT}/graphql/"; then +# 2. Caido proxy must be reachable. A bare /graphql/ probe may legitimately return +# 400, which the entrypoint already treats as ready. +if ! curl -s -o /dev/null -w "%{http_code}" --max-time 3 "http://127.0.0.1:${CAIDO_PORT}/graphql/" | grep -qE "^(200|400)$"; then echo "UNHEALTHY: Caido proxy not responding on port ${CAIDO_PORT}" exit 1 fi diff --git a/strix/interface/main.py b/strix/interface/main.py index a6b759f5..74013e52 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -46,31 +46,17 @@ from strix.telemetry import posthog # noqa: E402 from strix.telemetry.tracer import get_global_tracer # noqa: E402 -_log_dir = Path("strix_runs") -_log_dir.mkdir(exist_ok=True) -_file_handler = logging.FileHandler(_log_dir / "strix.log") -_file_handler.setLevel(logging.INFO) -_file_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")) +def configure_logging() -> None: + log_dir = Path("strix_runs") + log_dir.mkdir(exist_ok=True) -_root = logging.getLogger() -_root.setLevel(logging.INFO) -_root.handlers = [_file_handler] + file_handler = logging.FileHandler(log_dir / "strix.log") + file_handler.setLevel(logging.INFO) + file_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")) -# prevent any library from adding StreamHandlers (console) to the root logger -_orig_add_handler = logging.Logger.addHandler - - -def _guarded_add_handler(self: logging.Logger, handler: logging.Handler) -> None: - if ( - self is _root - and isinstance(handler, logging.StreamHandler) - and not isinstance(handler, logging.FileHandler) - ): - return - _orig_add_handler(self, handler) - - -logging.Logger.addHandler = _guarded_add_handler # type: ignore[assignment] + root = logging.getLogger() + root.setLevel(logging.INFO) + root.handlers = [file_handler] def validate_environment() -> None: # noqa: PLR0912, PLR0915 @@ -565,6 +551,8 @@ def main() -> None: # noqa: PLR0912, PLR0915 if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) + configure_logging() + args = parse_arguments() if args.config: diff --git a/strix/interface/tool_components/browser_renderer.py b/strix/interface/tool_components/browser_renderer.py index 7d2c3838..6dcc385b 100644 --- a/strix/interface/tool_components/browser_renderer.py +++ b/strix/interface/tool_components/browser_renderer.py @@ -1,11 +1,5 @@ -from __future__ import annotations - from functools import cache -from typing import TYPE_CHECKING, Any, ClassVar - - -if TYPE_CHECKING: - from collections.abc import Callable +from typing import Any, ClassVar from pygments.lexers import get_lexer_by_name from pygments.styles import get_style_by_name @@ -27,19 +21,23 @@ class BrowserRenderer(BaseToolRenderer): tool_name: ClassVar[str] = "browser_action" css_classes: ClassVar[list[str]] = ["tool-call", "browser-tool"] - # -- palette (used only for highlights) ---------------------------- - NAV: ClassVar[str] = "#06b6d4" # cyan — links / URLs - INTERACT: ClassVar[str] = "#3b82f6" # blue — targets / values - OBSERVE: ClassVar[str] = "#a78bfa" # purple — data fields - EXEC: ClassVar[str] = "#f59e0b" # amber — task text - LIFE: ClassVar[str] = "#10b981" # teal — lifecycle values - OK: ClassVar[str] = "#22c55e" # green — success - ERR: ClassVar[str] = "#ef4444" # red — error - DIM: ClassVar[str] = "dim" # gray — prose / labels - - # ----------------------------------------------------------------- - # helpers - # ----------------------------------------------------------------- + SIMPLE_ACTIONS: ClassVar[dict[str, str]] = { + "launch": "launching browser", + "go_back": "going back in browser history", + "close_browser": "closing browser", + "close_tab": "closing browser tab", + "screenshot": "taking screenshot of browser tab", + "dropdown_options": "getting dropdown options", + "extract": "extracting content from page", + "wait": "waiting...", + } + PARAM_ACTIONS: ClassVar[dict[str, str]] = { + "click": "clicking", + "find_text": "finding text ", + "send_keys": "pressing key ", + "search_page": "searching page for ", + "find_elements": "finding elements matching ", + } @classmethod def _get_token_color(cls, token_type: Any) -> str | None: @@ -61,326 +59,159 @@ class BrowserRenderer(BaseToolRenderer): text.append(token_value, style=color) return text - @classmethod - def _status_mark(cls, status: str) -> Text: - text = Text() - if status == "completed": - text.append(" ✓", style=f"dim {cls.OK}") - elif status in ("failed", "error"): - text.append(" ✗", style=f"dim {cls.ERR}") - return text - - @classmethod - def _append_fields(cls, text: Text, res: dict[str, Any]) -> None: - """Append a dim summary of returned history fields.""" - fields = res.get("fields") - if not fields or not isinstance(fields, dict): - return - names = sorted(fields) - text.append("\n ") - text.append("fields ", style=cls.OBSERVE) - text.append(" ".join(names), style=cls.DIM) - - # ----------------------------------------------------------------- - # public - # ----------------------------------------------------------------- - @classmethod def render(cls, tool_data: dict[str, Any]) -> Static: args = tool_data.get("args", {}) status = tool_data.get("status", "unknown") - result = tool_data.get("result") action = args.get("action", "") - content = cls._build_content(action, args, status, result) + content = cls._build_content(action, args) css_classes = cls.get_css_classes(status) return Static(content, classes=css_classes) - # ----------------------------------------------------------------- - # content builder - # ----------------------------------------------------------------- - @classmethod - def _build_content( + def _build_url_action( cls, - action: str, - args: dict[str, Any], - status: str, - result: Any, - ) -> Text: - # Dispatch to action-specific builders - builders: dict[str, Callable[[], Text]] = { - "run": lambda: cls._build_run(args, status, result), - "launch": lambda: cls._build_launch(status, result), - "navigate": lambda: cls._build_navigate(args, status), - "click": lambda: cls._build_click(args, status), - "input": lambda: cls._build_input(args, status), - "scroll": lambda: cls._build_scroll(args, status), - "find_text": lambda: cls._build_find_text(args, status), - "send_keys": lambda: cls._build_send_keys(args, status), - "search_page": lambda: cls._build_search_page(args, status), - "find_elements": lambda: cls._build_find_elements(args, status), - "dropdown_options": lambda: cls._build_dropdown_options(args, status), - "select_dropdown": lambda: cls._build_select_dropdown(args, status), - "evaluate": lambda: cls._build_evaluate(args, status), - "wait": lambda: cls._build_wait(args, status), - "switch": lambda: cls._build_switch(args, status), - } - - simple_actions = { - "go_back": "going back", - "close_browser": "closing browser", - "close_tab": "closing tab", - "screenshot": "taking screenshot", - "extract": "extracting content", - } - - if action in builders: - return builders[action]() - if action in simple_actions: - return cls._build_simple(simple_actions[action], status) - return cls._build_fallback(action, status) + text: Text, + label: str, + url: str | None, + suffix: str = "", + ) -> None: + text.append(label, style="#06b6d4") + if url: + text.append(url, style="#06b6d4") + if suffix: + text.append(suffix, style="#06b6d4") @classmethod - def _build_simple(cls, label: str, status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append(label, style=cls.DIM) - text.append_text(cls._status_mark(status)) + def _base_text(cls) -> Text: + text = Text() + text.append("🌐 ") return text @classmethod - def _build_launch(cls, status: str, result: Any) -> Text: - text = Text("◈ ", style=cls.LIFE) - text.append("launching browser", style=f"bold {cls.LIFE}") - text.append_text(cls._status_mark(status)) - res = result if isinstance(result, dict) else {} - warning = res.get("warning") - if warning: - text.append(f"\n ⚠ {warning}", style=f"italic {cls.DIM}") + def _build_simple_action(cls, action: str, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append(cls.SIMPLE_ACTIONS[action], style="#06b6d4") + if action == "wait" and args.get("seconds") is not None: + text.append(f" {args['seconds']}s", style="#06b6d4") return text @classmethod - def _build_navigate(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("navigating to ", style=cls.DIM) - url = args.get("url", "") - if len(url) > 80: - url = url[:77] + "..." - text.append(url, style=f"{cls.NAV} underline") - if args.get("new_tab"): - text.append(" in new tab", style=cls.DIM) - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_click(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("clicking", style=cls.DIM) - index = args.get("index") - if index is not None: - text.append(f" #{index}", style=f"bold {cls.INTERACT}") - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_input(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("inputting", style=cls.DIM) - index = args.get("index") - if index is not None: - text.append(f" #{index}", style=f"bold {cls.INTERACT}") - value = args.get("text") - if value: - preview = value if len(value) <= 60 else value[:57] + "..." - text.append(f' "{preview}"', style=cls.INTERACT) - if args.get("clear"): - text.append(" (clear)", style=cls.DIM) - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_scroll(cls, args: dict[str, Any], status: str) -> Text: - direction = "down" if args.get("down", True) else "up" - text = Text("@ ", style=cls.DIM) - text.append("scrolling ", style=cls.DIM) - text.append(direction, style=cls.INTERACT) - pages = args.get("pages") - if pages is not None: - text.append(f" {pages} page(s)", style=cls.DIM) - index = args.get("index") - if index is not None: - text.append(" on ", style=cls.DIM) - text.append(f"#{index}", style=f"bold {cls.INTERACT}") - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_find_text(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("finding text ", style=cls.DIM) - value = args.get("text", "") - if value: - preview = value if len(value) <= 80 else value[:77] + "..." - text.append(f'"{preview}"', style=cls.OBSERVE) - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_send_keys(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("pressing ", style=cls.DIM) - text.append(args.get("keys", ""), style=f"bold {cls.INTERACT}") - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_search_page(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("searching page for ", style=cls.DIM) - pattern = args.get("pattern", "") - if pattern: - preview = pattern if len(pattern) <= 80 else pattern[:77] + "..." - text.append(f'"{preview}"', style=cls.OBSERVE) - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_find_elements(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("finding elements ", style=cls.DIM) - selector = args.get("selector", "") - if selector: - text.append(selector, style=cls.OBSERVE) - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_dropdown_options(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("reading dropdown options", style=cls.DIM) - index = args.get("index") - if index is not None: - text.append(f" #{index}", style=f"bold {cls.INTERACT}") - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_select_dropdown(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("selecting dropdown value", style=cls.DIM) - index = args.get("index") - if index is not None: - text.append(f" #{index}", style=f"bold {cls.INTERACT}") - value = args.get("text", "") - if value: - text.append(f' "{value}"', style=f"bold {cls.INTERACT}") - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_evaluate(cls, args: dict[str, Any], status: str) -> Text: - js = args.get("code") - text = Text("@ ", style=cls.DIM) - text.append("executing javascript", style=cls.DIM) - text.append_text(cls._status_mark(status)) - if js: - text.append("\n") - text.append_text(cls._highlight_js(js)) - return text - - @classmethod - def _build_wait(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - seconds = args.get("seconds") - if status == "completed": - text.append("waited", style=cls.DIM) - else: - text.append("waiting", style=cls.DIM) - if seconds is not None: - text.append(f" {seconds}s", style=cls.INTERACT) - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_switch(cls, args: dict[str, Any], status: str) -> Text: - text = Text("@ ", style=cls.DIM) - text.append("switching to tab ", style=cls.DIM) - text.append(str(args.get("tab_id", "?")), style=f"bold {cls.NAV}") - text.append_text(cls._status_mark(status)) - return text - - @classmethod - def _build_fallback(cls, action: str, status: str) -> Text: - if not action: - return Text() - text = Text("@ ", style=cls.DIM) - text.append(action, style=cls.DIM) - text.append_text(cls._status_mark(status)) - return text - - # ----------------------------------------------------------------- - # run task - # ----------------------------------------------------------------- - - @classmethod - def _build_run( - cls, - args: dict[str, Any], - status: str, - result: Any, - ) -> Text: - task = args.get("task", "") - - if status == "running": - text = Text("@ ", style=cls.DIM) - text.append("running task", style=f"bold {cls.EXEC}") - if task: - text.append("\n ") - text.append(task, style=cls.EXEC) - rf = args.get("return_fields") - if rf and isinstance(rf, list): - text.append("\n ") - text.append("returning ", style=cls.OBSERVE) - text.append(" ".join(rf), style=cls.DIM) - return text - - # Completed or failed — show result - if status in ("completed", "failed", "error"): - res = result if isinstance(result, dict) else {} - has_error = "error" in res - - if has_error: - text = Text("@ ", style=cls.DIM) - text.append("task failed", style=f"bold {cls.ERR}") - if task: - text.append("\n ") - text.append(task, style="dim strike") - text.append("\n ") - error_msg = str(res["error"]) - if len(error_msg) > 200: - error_msg = error_msg[:197] + "..." - text.append(error_msg, style=cls.ERR) - else: - text = Text("@ ", style=cls.DIM) - text.append("browser task completed", style=f"bold {cls.OK}") - if task: - text.append("\n ") - text.append(task, style=cls.DIM) - output = res.get("result", "") - if output: - text.append("\n ") - output_str = str(output) - if len(output_str) > 300: - output_str = output_str[:297] + "..." - text.append(output_str, style=cls.DIM) - cls._append_fields(text, res) - return text - - # Unknown status - text = Text("@ ", style=cls.DIM) - text.append("running browser task", style=cls.DIM) + def _build_run(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append("running browser task", style="#06b6d4") + task = args.get("task") if task: - text.append("\n ") - text.append(task, style=cls.DIM) + text.append("\n") + text.append(str(task), style="#06b6d4") return text + + @classmethod + def _build_navigate(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + suffix = " in new tab" if args.get("new_tab") else "" + cls._build_url_action(text, "navigating to ", args.get("url"), suffix) + return text + + @classmethod + def _build_param_action(cls, action: str, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append(cls.PARAM_ACTIONS[action], style="#06b6d4") + + value_key = { + "click": "index", + "find_text": "text", + "send_keys": "keys", + "search_page": "pattern", + "find_elements": "selector", + }[action] + value = args.get(value_key) + if value is not None: + if action == "click": + text.append(f" element #{value}", style="#06b6d4") + else: + text.append(str(value), style="#06b6d4") + return text + + @classmethod + def _build_input(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append("typing ", style="#06b6d4") + if args.get("text"): + text.append(str(args["text"]), style="#06b6d4") + index = args.get("index") + if index is not None: + text.append(f" into element #{index}", style="#06b6d4") + if args.get("clear"): + text.append(" with clear enabled", style="#06b6d4") + return text + + @classmethod + def _build_scroll(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + direction = "down" if args.get("down", True) else "up" + text.append(f"scrolling {direction}", style="#06b6d4") + if args.get("pages") is not None: + text.append(f" {args['pages']} page(s)", style="#06b6d4") + if args.get("index") is not None: + text.append(f" on element #{args['index']}", style="#06b6d4") + return text + + @classmethod + def _build_select_dropdown(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append("selecting dropdown option ", style="#06b6d4") + if args.get("text"): + text.append(str(args["text"]), style="#06b6d4") + if args.get("index") is not None: + text.append(f" on element #{args['index']}", style="#06b6d4") + return text + + @classmethod + def _build_evaluate(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append("executing javascript", style="#06b6d4") + js_code = args.get("code") + if js_code: + text.append("\n") + text.append_text(cls._highlight_js(str(js_code))) + return text + + @classmethod + def _build_switch(cls, args: dict[str, Any]) -> Text: + text = cls._base_text() + text.append("switching browser tab", style="#06b6d4") + if args.get("tab_id"): + text.append(" to ", style="#06b6d4") + text.append(str(args["tab_id"]), style="#06b6d4") + return text + + @classmethod + def _build_fallback(cls, action: str) -> Text: + text = cls._base_text() + if action: + text.append(action, style="#06b6d4") + return text + + @classmethod + def _build_content(cls, action: str, args: dict[str, Any]) -> Text: + builders = { + "run": cls._build_run, + "navigate": cls._build_navigate, + "input": cls._build_input, + "scroll": cls._build_scroll, + "select_dropdown": cls._build_select_dropdown, + "evaluate": cls._build_evaluate, + "switch": cls._build_switch, + } + + if action in cls.SIMPLE_ACTIONS: + return cls._build_simple_action(action, args) + if action in cls.PARAM_ACTIONS: + return cls._build_param_action(action, args) + if action in builders: + return builders[action](args) + + return cls._build_fallback(action) diff --git a/strix/tools/browser/browser_actions.py b/strix/tools/browser/browser_actions.py index 4e689e31..28f07a20 100644 --- a/strix/tools/browser/browser_actions.py +++ b/strix/tools/browser/browser_actions.py @@ -107,6 +107,9 @@ async def _run_browser_agent( return_fields: list[str] | None, metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: + if not session.browser.is_cdp_connected: + await session.start() + llm, vision = _build_llm(metadata=metadata) # [monkeypatch] prevent browseruse from killing the cdp connection after execution