diff --git a/gateway/run.py b/gateway/run.py index a7f7caae3c4d2..115cdd8d37570 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1506,6 +1506,19 @@ def _collect_history_media_paths(agent_history: List[Dict[str, Any]]) -> set: """ paths: set = set() tool_name_by_call_id: Dict[str, str] = {} + + def _add_text_media_paths(content: str) -> None: + for match in _TOOL_MEDIA_RE.finditer(content): + path = match.group(1).strip().rstrip('",}') + if path: + paths.add(path) + # The regex alone misses quoted and spaced paths that the delivery + # pipeline's extract_media grammar accepts — collect through the same + # extractor so the dedup set sees every path that could actually have + # been delivered. + media_files, _ = BasePlatformAdapter.extract_media(content) + paths.update(path for path, _is_voice in media_files) + for msg in agent_history: if msg.get("role") == "assistant": for call in msg.get("tool_calls") or []: @@ -1519,19 +1532,13 @@ def _collect_history_media_paths(agent_history: List[Dict[str, Any]]) -> set: if role == "assistant": content = str(msg.get("content", "") or "") if "MEDIA:" in content: - for match in _TOOL_MEDIA_RE.finditer(content): - p = match.group(1).strip().rstrip('",}') - if p: - paths.add(p) + _add_text_media_paths(content) continue if role not in {"tool", "function"}: continue content = str(msg.get("content", "") or "") if "MEDIA:" in content: - for match in _TOOL_MEDIA_RE.finditer(content): - p = match.group(1).strip().rstrip('",}') - if p: - paths.add(p) + _add_text_media_paths(content) continue cid = str(msg.get("tool_call_id") or msg.get("call_id") or "") if tool_name_by_call_id.get(cid) == "image_generate": diff --git a/tests/gateway/test_media_spaced_paths_and_history_dedupe.py b/tests/gateway/test_media_spaced_paths_and_history_dedupe.py index 3478211c863cd..bc214dcdd8962 100644 --- a/tests/gateway/test_media_spaced_paths_and_history_dedupe.py +++ b/tests/gateway/test_media_spaced_paths_and_history_dedupe.py @@ -81,4 +81,22 @@ class TestHistoryMediaDedupe: paths = _collect_history_media_paths(history) assert "/tmp/chart.png" in paths + def test_quoted_spaced_home_path_is_collected_in_delivery_form( + self, + tmp_path, + monkeypatch, + ): + monkeypatch.setenv("HOME", str(tmp_path)) + history = [ + { + "role": "assistant", + "content": 'MEDIA:"~/audio cache/old.ogg"', + }, + ] + paths = _collect_history_media_paths(history) + + assert str(tmp_path / "audio cache" / "old.ogg") in paths + + def test_empty_history_empty_set(self): + assert _collect_history_media_paths([]) == set()