fix(gateway): collect quoted/spaced/home-relative MEDIA paths into the history dedup set

Salvaged from PR #73982 by Alexander Russell (@AlexxRussell) — the
collector half only. _collect_history_media_paths used only
_TOOL_MEDIA_RE, which misses quoted and spaced paths that the delivery
pipeline's extract_media grammar accepts; run text content through the
same extractor so the surviving dedup consumers (auto-append lane and
bare-path filter) see every path that could actually have been
delivered.

The PR's other halves (post-stream dedup snapshot plumbing, canonical
path comparison in _deliver_media_from_response, queued-followup
snapshot union) are moot after #74495 removed the post-stream history
filter entirely.
This commit is contained in:
Alexander Russell 2026-07-29 18:49:36 -07:00 committed by Teknium
parent 8f4122efd2
commit cea4c3362d
2 changed files with 33 additions and 8 deletions

View File

@ -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":

View File

@ -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()