From 1fac44086519112fc7f346ffa1c8b41de0556ee1 Mon Sep 17 00:00:00 2001 From: Gille <4317663+helix4u@users.noreply.github.com> Date: Mon, 24 Aug 2026 22:06:51 -0600 Subject: [PATCH] fix(gateway): recover explicit computer-use media paths --- gateway/run.py | 154 +++++++++++++++++++++++++ tests/gateway/test_media_extraction.py | 109 +++++++++++++++++ 2 files changed, 263 insertions(+) diff --git a/gateway/run.py b/gateway/run.py index 71aef94ef7..c9bb0f7536 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1823,6 +1823,146 @@ _TOOL_MEDIA_RE = re.compile( ) +_COMPUTER_USE_CAPTURE_BASENAME_RE = re.compile( + r"^computer_use_[0-9a-f]{32}\.(?:png|jpe?g)$", + re.IGNORECASE, +) +_COMPUTER_USE_CAPTURE_SUMMARY_RE = re.compile( + r"\(shareable screenshot saved to " + r"(?P(?:[A-Za-z]:[/\\]|/|\\\\)[^\r\n]*?" + r"computer_use_[0-9a-f]{32}\.(?:png|jpe?g))\)", + re.IGNORECASE, +) + + +def _computer_use_capture_basename(path: Any) -> str: + """Return a canonical capture basename for either path separator style.""" + value = str(path or "").strip().strip("`\"'") + basename = re.split(r"[/\\]", value)[-1] + if _COMPUTER_USE_CAPTURE_BASENAME_RE.fullmatch(basename): + return basename.lower() + return "" + + +def _iter_computer_use_capture_paths(content: Any): + """Yield persisted screenshot paths from computer_use result content. + + The tool can return JSON, a multimodal content list, or a text fallback. + The latter two retain the canonical path in the human-readable summary + even though the multimodal envelope's ``meta`` dictionary is not stored in + the tool message. + """ + if isinstance(content, str): + for match in _COMPUTER_USE_CAPTURE_SUMMARY_RE.finditer(content): + yield match.group("path").strip() + stripped = content.strip() + if stripped.startswith(("{", "[")): + try: + payload = json.loads(stripped) + except Exception: + payload = None + if isinstance(payload, (dict, list)): + yield from _iter_computer_use_capture_paths(payload) + return + + if isinstance(content, list): + for part in content: + yield from _iter_computer_use_capture_paths(part) + return + + if not isinstance(content, dict): + return + + screenshot_path = content.get("screenshot_path") + if isinstance(screenshot_path, str): + yield screenshot_path + meta = content.get("meta") + if isinstance(meta, dict) and isinstance(meta.get("screenshot_path"), str): + yield meta["screenshot_path"] + for field in ("content", "text", "text_summary", "summary"): + nested = content.get(field) + if isinstance(nested, (str, dict, list)): + yield from _iter_computer_use_capture_paths(nested) + + +def _repair_explicit_computer_use_media_paths( + response: str, + messages: List[Dict[str, Any]], + history_offset: int = 0, +) -> str: + """Recover model-mangled paths for explicitly requested screenshots. + + ``computer_use`` persists a bounded screenshot and gives its absolute path + to the model. Some models rewrite a Windows path into a POSIX-looking path + before emitting ``MEDIA:``, which makes the gateway reject the nonexistent + path. Repair only an already-explicit directive whose unique generated + basename matches a canonical screenshot path from this turn. This does not + auto-attach ordinary computer-use captures, and normal media path + validation still runs after the repair. + """ + if "MEDIA:" not in response: + return response + + if history_offset and len(messages) >= history_offset: + turn_messages = messages[history_offset:] + elif history_offset: + # Compression can invalidate the original slice boundary. Recover the + # current turn from its last user message; fail closed if none remains. + last_user = next( + ( + index + for index in range(len(messages) - 1, -1, -1) + if messages[index].get("role") == "user" + ), + None, + ) + turn_messages = messages[last_user:] if last_user is not None else [] + else: + turn_messages = messages + + tool_name_by_call_id: Dict[str, str] = {} + for msg in turn_messages: + if msg.get("role") != "assistant": + continue + for call in msg.get("tool_calls") or []: + call_id = call.get("id") or call.get("call_id") + fn = call.get("function") or {} + name = str(fn.get("name") or call.get("name") or "") + if call_id and name: + tool_name_by_call_id[str(call_id)] = name + + canonical_by_basename: Dict[str, str] = {} + for msg in turn_messages: + if msg.get("role") not in {"tool", "function"}: + continue + call_id = str(msg.get("tool_call_id") or msg.get("call_id") or "") + tool_name = str( + msg.get("name") + or msg.get("tool_name") + or tool_name_by_call_id.get(call_id) + or "" + ) + if tool_name != "computer_use": + continue + for path in _iter_computer_use_capture_paths(msg.get("content")): + basename = _computer_use_capture_basename(path) + if basename and re.match(r"^(?:[A-Za-z]:[/\\]|/|\\\\)", path): + canonical_by_basename[basename] = path + + if not canonical_by_basename: + return response + + media_files, _ = BasePlatformAdapter.extract_media(response) + repaired = response + for emitted_path, _is_voice in media_files: + canonical = canonical_by_basename.get( + _computer_use_capture_basename(emitted_path) + ) + if canonical and emitted_path != canonical: + repaired = repaired.replace(emitted_path, canonical) + return repaired + + def _collect_auto_append_media_tags( messages: List[Dict[str, Any]], history_offset: int = 0, @@ -6429,6 +6569,20 @@ class TurnRunner: except Exception: pass reset_current_session_key(_approval_session_token) + # Canonicalize an explicitly emitted computer-use screenshot path at + # the common result boundary. The streaming finalizer below and the + # normal non-streaming delivery path must see the same response; + # repairing only during later media scanning leaves streaming with the + # model-mangled path and a rejected attachment. + if isinstance(result, dict): + _result_final = result.get("final_response") + if isinstance(_result_final, str): + result["final_response"] = _repair_explicit_computer_use_media_paths( + _result_final, + result.get("messages", []), + history_offset=len(agent_history), + ) + ctx.result_holder[0] = result # Signal the stream consumer that the agent is done. Pass the diff --git a/tests/gateway/test_media_extraction.py b/tests/gateway/test_media_extraction.py index 65f64e6650..b651d4ebe6 100644 --- a/tests/gateway/test_media_extraction.py +++ b/tests/gateway/test_media_extraction.py @@ -11,6 +11,7 @@ make_image tool several turns earlier must not leak onto a later text-only reply, even when the path-based dedup set fails to capture it. """ +import json import re from unittest.mock import MagicMock @@ -105,6 +106,114 @@ def extract_media_tags_broken(result_messages): class TestMediaExtraction: """Tests for MEDIA tag extraction from tool results.""" + def test_repairs_explicit_computer_use_media_path_from_json_result(self): + from gateway.run import _repair_explicit_computer_use_media_paths + + capture_name = "computer_use_0123456789abcdef0123456789abcdef.png" + canonical = rf"C:\Users\Alice\AppData\Local\hermes\cache\images\{capture_name}" + response = ( + "Here is the screenshot.\n" + f"MEDIA:/Users/Alice/AppData/Local/hermes/cache/images/{capture_name}" + ) + messages = [ + { + "role": "assistant", + "tool_calls": [ + {"id": "capture", "function": {"name": "computer_use"}} + ], + }, + { + "role": "tool", + "tool_call_id": "capture", + "content": json.dumps({"screenshot_path": canonical}), + }, + ] + + repaired = _repair_explicit_computer_use_media_paths(response, messages) + + assert repaired == f"Here is the screenshot.\nMEDIA:{canonical}" + + def test_repairs_explicit_path_from_multimodal_text_summary(self): + from gateway.run import _repair_explicit_computer_use_media_paths + + capture_name = "computer_use_fedcba9876543210fedcba9876543210.jpg" + canonical = rf"D:\Hermes Data\cache\images\{capture_name}" + response = f'MEDIA:"/Users/Alice/Hermes Data/cache/images/{capture_name}"' + messages = [ + { + "role": "tool", + "name": "computer_use", + "tool_call_id": "capture", + "content": [ + { + "type": "text", + "text": ( + "capture mode=screen 1920x1080\n" + f" (shareable screenshot saved to {canonical})" + ), + }, + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,AAAA"}, + }, + ], + } + ] + + repaired = _repair_explicit_computer_use_media_paths(response, messages) + + assert repaired == f'MEDIA:"{canonical}"' + + def test_does_not_auto_attach_computer_use_capture(self): + from gateway.run import _repair_explicit_computer_use_media_paths + + capture_name = "computer_use_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.png" + canonical = rf"C:\Users\Alice\AppData\Local\hermes\cache\images\{capture_name}" + messages = [ + { + "role": "tool", + "name": "computer_use", + "content": json.dumps({"screenshot_path": canonical}), + } + ] + + assert ( + _repair_explicit_computer_use_media_paths("Done.", messages) + == "Done." + ) + + def test_does_not_rewrite_unmatched_or_previous_turn_capture(self): + from gateway.run import _repair_explicit_computer_use_media_paths + + old_name = "computer_use_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.png" + current_name = "computer_use_cccccccccccccccccccccccccccccccc.png" + old_canonical = rf"C:\cache\images\{old_name}" + current_canonical = rf"C:\cache\images\{current_name}" + history = [ + { + "role": "tool", + "name": "computer_use", + "content": json.dumps({"screenshot_path": old_canonical}), + }, + ] + current_turn = [ + {"role": "user", "content": "Send the current screenshot."}, + { + "role": "tool", + "name": "computer_use", + "content": json.dumps({"screenshot_path": current_canonical}), + }, + ] + response = f"MEDIA:/Users/Alice/cache/images/{old_name}" + + repaired = _repair_explicit_computer_use_media_paths( + response, + history + current_turn, + history_offset=len(history), + ) + + assert repaired == response + def test_gateway_auto_append_ignores_media_examples_in_skill_docs(self): """Skill/documentation examples must not be appended as real attachments.""" from gateway.run import _collect_auto_append_media_tags