fix(gateway): recover explicit computer-use media paths
This commit is contained in:
Gille
+154
@@ -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<path>(?:[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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user