diff --git a/packages/atif-converter/src/atif_converter/domain/blobs.py b/packages/atif-converter/src/atif_converter/domain/blobs.py index cec269a..c9bbf21 100644 --- a/packages/atif-converter/src/atif_converter/domain/blobs.py +++ b/packages/atif-converter/src/atif_converter/domain/blobs.py @@ -36,8 +36,15 @@ In a Codex rollout the shape is ``{"type": "input_image", "image_url": "data:;base64,"}`` inside a ``response_item`` message's content -or a tool output list. A base64 value anywhere else (a tool CALL's own -arguments, for instance) is the model's input and is left alone. +or a tool output list. A tool output list can also carry an MCP image block, +``{"type": "image", "data": "", "mimeType": "image/png"}``. The +converter's harbor port names every inline image in a tool output it can save +by a placeholder (:func:`codex_tool_image_ref` is its rule: a supported image +type, a payload that decodes once whitespace is removed, data URL parameters +allowed), so a tool output's MCP image, and an ``input_image`` the strict data +URL match above misses, is lifted by that same rule and the placeholder names +bytes the collector holds. A base64 value anywhere else (a tool CALL's own +arguments, an ``image`` block in message content) is left alone. Replacement is by block: an attachment block becomes a TEXT block holding the placeholder (``input_text`` for Codex), which is the one block type every @@ -84,6 +91,15 @@ #: A Codex data URL: ``data:;base64,``. _DATA_URL_RE = re.compile(r"^data:([^;,]{1,128});base64,", re.ASCII) +#: The image types harbor 0.24.0 saves from a Codex tool output +#: (``Codex._image_media_type``); any other type is ``[image omitted]`` there. +CODEX_TOOL_IMAGE_MEDIA_TYPES = frozenset({"image/png", "image/jpeg", "image/gif", "image/webp"}) + +#: harbor's data URL for a Codex tool output image: parameters allowed before ``;base64``. +_TOOL_IMAGE_DATA_URL = re.compile( + r"data:(?P[^;,]+)(?:;[^;,]*)*;base64,(?P.*)", re.DOTALL +) + @dataclass(frozen=True, slots=True) class BlobRef: @@ -320,9 +336,54 @@ def extract_claude_code_blobs(records: Iterable[Any], collector: BlobCollector) # --------------------------------------------------------------------------- -def _codex_image_ref(item: Any, collector: BlobCollector) -> BlobRef | None: - if not isinstance(item, dict) or item.get("type") != "input_image": +def codex_tool_image_media_type(value: str) -> str | None: + """The supported image MIME type for ``value``, as harbor's ``Codex._image_media_type`` maps it.""" + media_type = value.strip().lower() + if media_type == "image/jpg": + media_type = "image/jpeg" + return media_type if media_type in CODEX_TOOL_IMAGE_MEDIA_TYPES else None + + +def codex_tool_image_ref(data: str, media_type: Any, collector: BlobCollector) -> BlobRef | None: + """Register one inline Codex tool output image the way harbor 0.24.0 would save it. + + ``data`` is base64 or a ``data:`` URL (whose media type then wins over + ``media_type``); harbor saves it only when the media type is one of + :data:`CODEX_TOOL_IMAGE_MEDIA_TYPES` and the payload, with whitespace + removed, decodes. ``None`` means harbor would write ``[image omitted]``: + nothing is registered. The converter's port and the pre-pass both call + this, so the placeholder text and the stored blob agree byte for byte. + """ + data_url = _TOOL_IMAGE_DATA_URL.fullmatch(data) + base64_text = data + if data_url is not None: + media_type, base64_text = data_url.group("media_type"), str(data_url.group("data")) + if not isinstance(media_type, str): return None + valid_media_type = codex_tool_image_media_type(media_type) + if valid_media_type is None: + return None + return collector.add_base64("".join(base64_text.split()), valid_media_type) + + +def _mcp_image_ref(item: dict[str, Any], collector: BlobCollector) -> BlobRef | None: + """An MCP ``image`` block's ref, read the way the converter's port reads it.""" + data = item.get("data") + if not isinstance(data, str) or ("://" in data and not data.startswith("data:")): + return None + return codex_tool_image_ref(data, item.get("mimeType") or item.get("mime_type"), collector) + + +def _tool_input_image_ref(item: dict[str, Any], collector: BlobCollector) -> BlobRef | None: + """A tool output ``input_image``'s ref by the port's rule (see :func:`codex_tool_image_ref`).""" + url = item.get("image_url") + if not isinstance(url, str) or ("://" in url and not url.startswith("data:")): + return None + return codex_tool_image_ref(url, None, collector) + + +def _input_image_ref(item: dict[str, Any], collector: BlobCollector) -> BlobRef | None: + """An ``input_image`` item's ref when its ``image_url`` is a base64 data URL.""" url = item.get("image_url") if not isinstance(url, str): return None @@ -332,10 +393,28 @@ def _codex_image_ref(item: Any, collector: BlobCollector) -> BlobRef | None: return collector.add_base64(url[match.end() :], match.group(1)) -def _rewrite_codex_items(items: list[Any], collector: BlobCollector) -> list[BlobRef]: +def _codex_image_ref(item: Any, collector: BlobCollector, *, tool_output: bool) -> BlobRef | None: + if not isinstance(item, dict): + return None + item_type = item.get("type") + if item_type == "input_image": + ref = _input_image_ref(item, collector) + if ref is None and tool_output: + # The port's rule is wider (data URL parameters, whitespace in + # the payload); whatever it would name, the collector must hold. + ref = _tool_input_image_ref(item, collector) + return ref + if item_type == "image" and tool_output: + return _mcp_image_ref(item, collector) + return None + + +def _rewrite_codex_items( + items: list[Any], collector: BlobCollector, *, tool_output: bool = False +) -> list[BlobRef]: refs: list[BlobRef] = [] for position, item in enumerate(items): - ref = _codex_image_ref(item, collector) + ref = _codex_image_ref(item, collector, tool_output=tool_output) if ref is None: continue items[position] = {"type": "input_text", "text": ref.placeholder()} @@ -348,7 +427,11 @@ def extract_codex_blobs( collector: BlobCollector, record_keys: Sequence[str] | None = None, ) -> BlobIndex: - """Rewrite every inline ``input_image`` data URL in a rollout's response items. + """Rewrite every inline image in a rollout's response items. + + An ``input_image`` data URL is lifted from message content and tool + outputs; an MCP ``image`` block with inline base64 only from a tool output + list, by :func:`codex_tool_image_ref`'s rule. ``record_keys`` (parallel to ``records``) names each record the way the enrichment pass names it in ``source_uuids``; without it, attachments in @@ -371,16 +454,20 @@ def extract_codex_blobs( output = payload.get("output") if isinstance(output, list): index.add_tool_result( - payload.get("call_id"), _rewrite_codex_items(output, collector) + payload.get("call_id"), + _rewrite_codex_items(output, collector, tool_output=True), ) return index __all__ = [ + "CODEX_TOOL_IMAGE_MEDIA_TYPES", "Blob", "BlobCollector", "BlobIndex", "BlobRef", + "codex_tool_image_media_type", + "codex_tool_image_ref", "extract_claude_code_blobs", "extract_codex_blobs", ] diff --git a/packages/atif-converter/src/atif_converter/domain/codex_conversion.py b/packages/atif-converter/src/atif_converter/domain/codex_conversion.py index 401cf03..cc57765 100644 --- a/packages/atif-converter/src/atif_converter/domain/codex_conversion.py +++ b/packages/atif-converter/src/atif_converter/domain/codex_conversion.py @@ -65,7 +65,6 @@ from __future__ import annotations import json -import re from typing import Any, Literal from loguru import logger @@ -82,7 +81,7 @@ ToolCall, Trajectory, ) -from atif_converter.domain.blobs import BlobCollector +from atif_converter.domain.blobs import BlobCollector, codex_tool_image_ref from atif_converter.domain.codex_enrichment import message_text #: The schema version harbor 0.22.0 stamps on a Codex trajectory. @@ -101,11 +100,6 @@ | _TOOL_OUTPUT_IMAGE_TYPES | frozenset({"audio", "input_audio", "resource", "resource_link", "encrypted_content"}) ) -#: The image types harbor stores; anything else is ``[image omitted]``. -_IMAGE_MEDIA_TYPES = frozenset({"image/png", "image/jpeg", "image/gif", "image/webp"}) -_BASE64_DATA_URL = re.compile( - r"data:(?P[^;,]+)(?:;[^;,]*)*;base64,(?P.*)", re.DOTALL -) #: harbor's text for an image it cannot reference. _IMAGE_OMITTED = "[image omitted]" @@ -224,32 +218,21 @@ def _tool_output_image_text(block: dict[str, Any]) -> str: def _inline_image_text(data: str, media_type: Any) -> str: - """An inline image's placeholder, or harbor's text when harbor would not save it.""" - data_url = _BASE64_DATA_URL.fullmatch(data) - base64_text = data - if data_url is not None: - media_type, base64_text = data_url.group("media_type"), str(data_url.group("data")) - if not isinstance(media_type, str): - return _IMAGE_OMITTED - valid_media_type = _image_media_type(media_type) - if valid_media_type is None: - logger.debug("Unsupported Codex tool image type: {}", media_type) - return _IMAGE_OMITTED - ref = BlobCollector().add_base64("".join(base64_text.split()), valid_media_type) + """An inline image's placeholder, or harbor's text when harbor would not save it. + + The rule is :func:`~atif_converter.domain.blobs.codex_tool_image_ref`, the + one the pre-pass lifts the same image by, so in production the pre-pass has + already replaced the image with this placeholder and stored its bytes; the + throwaway collector here only names the bytes when the records were not + pre-passed. + """ + ref = codex_tool_image_ref(data, media_type, BlobCollector()) if ref is None: - logger.debug("Failed to decode Codex tool image") + logger.debug("Unsupported or undecodable Codex tool image: {}", media_type) return _IMAGE_OMITTED return ref.placeholder() -def _image_media_type(value: str) -> str | None: - """The supported image MIME type for ``value``, as harbor's ``Codex._image_media_type`` maps it.""" - media_type = value.strip().lower() - if media_type == "image/jpg": - media_type = "image/jpeg" - return media_type if media_type in _IMAGE_MEDIA_TYPES else None - - def _group_events_by_api_call_id( normalized_events: list[dict[str, Any]], ) -> list[dict[str, Any]]: diff --git a/packages/atif-converter/src/atif_converter/domain/schema_version.py b/packages/atif-converter/src/atif_converter/domain/schema_version.py index a5b01a7..de8ed0b 100644 --- a/packages/atif-converter/src/atif_converter/domain/schema_version.py +++ b/packages/atif-converter/src/atif_converter/domain/schema_version.py @@ -65,12 +65,19 @@ keeps its raw text (harbor #3467). A Codex ``web_search_call`` carries its item id, or ``_web_search_``, where its id was empty (harbor #2972). +* 8 — every Codex tool output image the port writes a blob placeholder for is + lifted by the pre-pass too: an MCP ``image`` block with inline base64 (by + ``mimeType`` or ``mime_type``), and an ``input_image`` whose data URL + carries parameters or whose base64 is wrapped. Its bytes reach the blob + store, its result carries ``extra.images``, and a tool output list the port + dumps as JSON (one element of an unknown type) holds the placeholder item + where it held the base64. Placeholder text is unchanged. """ from __future__ import annotations from typing import Final -CONVERTER_SCHEMA_VERSION: Final = 7 +CONVERTER_SCHEMA_VERSION: Final = 8 __all__ = ["CONVERTER_SCHEMA_VERSION"] diff --git a/packages/atif-converter/tests/test_codex_blobs.py b/packages/atif-converter/tests/test_codex_blobs.py new file mode 100644 index 0000000..bdbcf70 --- /dev/null +++ b/packages/atif-converter/tests/test_codex_blobs.py @@ -0,0 +1,286 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Every Codex tool output image the converter names by placeholder is a stored blob. + +The converter's harbor 0.24.0 port turns an inline image in a tool output list +into ``[image sha256: bytes]``, and the pre-pass +(:func:`~atif_converter.domain.blobs.extract_codex_blobs`) is what hands those +bytes to the corpus blob store. Both read the image by +:func:`~atif_converter.domain.blobs.codex_tool_image_ref`, so these tests hold +the two to one rule: whatever placeholder the port would write, the collector +holds the bytes it names, the index maps them to the output's ``call_id``, and +whatever the port would call ``[image omitted]`` stays in the record untouched. +""" + +from __future__ import annotations + +import base64 +import copy +import hashlib +import json +import re +from pathlib import Path +from typing import Any + +import pytest +from codex_fixtures import codex_rollout_records, write_codex_rollout +from subagent_fixtures import PASTED_PNG, READ_PNG + +from atif_converter.application.convert_codex import convert_codex_and_audit +from atif_converter.domain.blobs import BlobCollector, BlobIndex, extract_codex_blobs +from atif_converter.domain.codex_conversion import convert_codex_records + +_TS = "2026-09-11T17:27:02.{:03d}Z" +_PLACEHOLDER = re.compile(r"\[image sha256:(?P[0-9a-f]{64}) (?P\S+) (?P\d+) bytes\]") + +_READ_B64 = base64.b64encode(READ_PNG).decode() +_PASTED_B64 = base64.b64encode(PASTED_PNG).decode() + + +def _sha(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _placeholder(data: bytes, media_type: str = "image/png") -> str: + return f"[image sha256:{_sha(data)} {media_type} {len(data)} bytes]" + + +def _item(payload: dict[str, Any], n: int) -> dict[str, Any]: + return {"timestamp": _TS.format(n), "type": "response_item", "payload": payload} + + +def _output(kind: str, call_id: str, output: Any) -> dict[str, Any]: + return _item({"type": kind, "call_id": call_id, "output": output}, 2) + + +def _mcp(data: str, media_type: str | None = "image/png", key: str = "mimeType") -> dict[str, Any]: + block: dict[str, Any] = {"type": "image", "data": data} + if media_type is not None: + block[key] = media_type + return block + + +def _extract(records: list[dict[str, Any]]) -> tuple[BlobCollector, BlobIndex]: + collector = BlobCollector() + return collector, extract_codex_blobs(records, collector) + + +class TestMcpImageInAToolOutput: + @pytest.mark.parametrize("kind", ["function_call_output", "custom_tool_call_output"]) + def test_is_lifted_stored_and_indexed_by_call_id(self, kind: str) -> None: + records = [_output(kind, "c1", [{"type": "text", "text": "shot:"}, _mcp(_READ_B64)])] + collector, index = _extract(records) + output = records[0]["payload"]["output"] + assert output == [ + {"type": "text", "text": "shot:"}, + {"type": "input_text", "text": _placeholder(READ_PNG)}, + ] + assert [(blob.ref.sha256, blob.data) for blob in collector.blobs] == [ + (_sha(READ_PNG), READ_PNG) + ] + ((ref,),) = index.by_tool_call_id.values() + assert list(index.by_tool_call_id) == ["c1"] + assert (ref.sha256, ref.media_type, ref.width, ref.height) == ( + _sha(READ_PNG), + "image/png", + 3, + 2, + ) + assert index.by_record_key == {} + + @pytest.mark.parametrize( + ("block", "media_type"), + [ + (_mcp(_READ_B64, "image/png", key="mime_type"), "image/png"), + (_mcp(_READ_B64, " IMAGE/JPG "), "image/jpeg"), + (_mcp(f"data:image/webp;base64,{_READ_B64}", None), "image/webp"), + (_mcp(f"data:image/gif;name=x.gif;base64,{_READ_B64}", "image/bmp"), "image/gif"), + (_mcp(_READ_B64[:20] + "\n" + _READ_B64[20:]), "image/png"), + ], + ids=["mime_type-key", "jpg-alias", "data-url", "data-url-params-win", "wrapped-base64"], + ) + def test_reads_the_media_type_and_payload_the_way_the_port_does( + self, block: dict[str, Any], media_type: str + ) -> None: + records = [_output("function_call_output", "c1", [block])] + collector, index = _extract(records) + assert records[0]["payload"]["output"] == [ + {"type": "input_text", "text": _placeholder(READ_PNG, media_type)} + ] + assert [blob.ref.media_type for blob in collector.blobs] == [media_type] + assert [ref.sha256 for ref in index.by_tool_call_id["c1"]] == [_sha(READ_PNG)] + + @pytest.mark.parametrize( + "block", + [ + _mcp(_READ_B64, "image/bmp"), + _mcp(_READ_B64, "text/plain"), + _mcp("not base64!"), + _mcp(""), + _mcp(_READ_B64, None), + _mcp("https://example.com/shot.png"), + {"type": "image", "data": 7, "mimeType": "image/png"}, + {"type": "image", "mimeType": "image/png"}, + ], + ids=[ + "unsupported-bmp", + "not-an-image", + "bad-base64", + "empty", + "no-media-type", + "remote-url", + "data-not-a-string", + "no-data", + ], + ) + def test_an_image_the_port_omits_stays_unlifted(self, block: dict[str, Any]) -> None: + records = [_output("function_call_output", "c1", [{"type": "text", "text": "a"}, block])] + before = copy.deepcopy(records) + collector, index = _extract(records) + assert records == before + assert collector.blobs == () + assert collector.references == 0 + assert index.by_tool_call_id == {} + + def test_an_image_block_in_message_content_stays(self) -> None: + """Message content is not a tool output: the port reads only its ``text`` fields.""" + records = [ + _item( + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "q"}, _mcp(_READ_B64)], + }, + 1, + ) + ] + before = copy.deepcopy(records) + collector, index = _extract(records) + assert records == before + assert collector.blobs == () + assert index.by_record_key == {} + + +class TestInputImageInAToolOutput: + @pytest.mark.parametrize( + "url", + [ + f"data:image/png;charset=binary;base64,{_READ_B64}", + f"data:image/png;base64,{_READ_B64[:20]}\n{_READ_B64[20:]}", + ], + ids=["data-url-params", "wrapped-base64"], + ) + def test_the_ports_wider_data_urls_are_lifted_too(self, url: str) -> None: + records = [ + _output("function_call_output", "c1", [{"type": "input_image", "image_url": url}]) + ] + collector, index = _extract(records) + assert records[0]["payload"]["output"] == [ + {"type": "input_text", "text": _placeholder(READ_PNG)} + ] + assert [blob.data for blob in collector.blobs] == [READ_PNG] + assert [ref.sha256 for ref in index.by_tool_call_id["c1"]] == [_sha(READ_PNG)] + + def test_in_message_content_only_the_strict_data_url_is_lifted(self) -> None: + url = f"data:image/png;charset=binary;base64,{_READ_B64}" + records = [ + _item( + { + "type": "message", + "role": "user", + "content": [{"type": "input_image", "image_url": url}], + }, + 1, + ) + ] + before = copy.deepcopy(records) + collector, _ = _extract(records) + assert records == before + assert collector.blobs == () + + +def _tool_call_records(output: list[dict[str, Any]]) -> list[dict[str, Any]]: + records = codex_rollout_records() + extra = [ + _item( + { + "type": "function_call", + "call_id": "call_mcp", + "name": "mcp__browser__screenshot", + "arguments": "{}", + }, + 310, + ), + _item({"type": "function_call_output", "call_id": "call_mcp", "output": output}, 320), + ] + return records[:10] + extra + records[10:] + + +def _result(trajectory: dict[str, Any], call_id: str) -> dict[str, Any]: + for step in trajectory["steps"]: + for result in (step.get("observation") or {}).get("results") or []: + if result.get("source_call_id") == call_id: + return result + raise AssertionError(call_id) + + +_MIXED_OUTPUT: list[dict[str, Any]] = [ + {"type": "text", "text": "two shots:"}, + _mcp(_READ_B64), + _mcp(_PASTED_B64, "image/jpg"), + _mcp(_READ_B64, "image/bmp"), +] + + +class TestEndToEnd: + def test_every_placeholder_names_a_stored_blob(self, tmp_path: Path) -> None: + """The production use case: each placeholder's sha256 is a blob the result carries.""" + rollout = write_codex_rollout(tmp_path / "sessions", _tool_call_records(_MIXED_OUTPUT)) + result, _ = convert_codex_and_audit(rollout) + content = _result(result.trajectory, "call_mcp")["content"] + assert content == "\n".join( + [ + "two shots:", + _placeholder(READ_PNG), + _placeholder(PASTED_PNG, "image/jpeg"), + "[image omitted]", + ] + ) + stored = {blob.ref.sha256: blob for blob in result.blobs} + named = [match.group("sha") for match in _PLACEHOLDER.finditer(content)] + assert named == [_sha(READ_PNG), _sha(PASTED_PNG)] + for sha in named: + assert hashlib.sha256(stored[sha].data).hexdigest() == sha + assert stored[_sha(PASTED_PNG)].data == PASTED_PNG + images = _result(result.trajectory, "call_mcp")["extra"]["images"] + assert [(i["sha256"], i["media_type"], i["width"], i["height"]) for i in images] == [ + (_sha(READ_PNG), "image/png", 3, 2), + (_sha(PASTED_PNG), "image/jpeg", 5, 4), + ] + text = json.dumps(result.trajectory) + assert _PASTED_B64 not in text + assert result.validation_errors == () + + def test_the_pre_pass_changes_no_converted_byte(self) -> None: + """Raw and pre-passed records convert to the same trajectory: placeholders are identical.""" + records = _tool_call_records(copy.deepcopy(_MIXED_OUTPUT)) + raw = convert_codex_records(copy.deepcopy(records), fallback_session_id="s") + lifted = copy.deepcopy(records) + _, index = _extract(lifted) + assert list(index.by_tool_call_id) == ["call_mcp"] + pre_passed = convert_codex_records(lifted, fallback_session_id="s") + assert raw is not None + assert pre_passed is not None + assert raw.to_json_dict() == pre_passed.to_json_dict() + + def test_a_list_the_port_dumps_as_json_keeps_no_base64(self, tmp_path: Path) -> None: + """One unknown block makes the port dump the list: the dump holds the placeholder item.""" + output = [_mcp(_PASTED_B64), {"type": "mystery"}] + rollout = write_codex_rollout(tmp_path / "sessions", _tool_call_records(output)) + result, _ = convert_codex_and_audit(rollout) + content = _result(result.trajectory, "call_mcp")["content"] + assert json.loads(content) == [ + {"type": "input_text", "text": _placeholder(PASTED_PNG)}, + {"type": "mystery"}, + ] + assert [blob.data for blob in result.blobs] == [PASTED_PNG] diff --git a/packages/atif-converter/tests/test_converter_schema_version.py b/packages/atif-converter/tests/test_converter_schema_version.py index 9051683..07cb3a5 100644 --- a/packages/atif-converter/tests/test_converter_schema_version.py +++ b/packages/atif-converter/tests/test_converter_schema_version.py @@ -28,8 +28,8 @@ #: The reviewed pair. Update BOTH in the commit that changes converter code: #: the version only when output can change, the digest always. -PINNED_VERSION = 7 -PINNED_SOURCE_DIGEST = "ec3d4dd26231542d1083122e863ecf1f2286b8d5e7ab3e5b2477970b47f14338" +PINNED_VERSION = 8 +PINNED_SOURCE_DIGEST = "bc8ed5cc906d36d3c739931a4eaf8603f134e9914f46ba192b92a7b09216c81d" _SRC = Path(__file__).resolve().parents[1] / "src" / "atif_converter"