diff --git a/openhands-sdk/openhands/sdk/__init__.py b/openhands-sdk/openhands/sdk/__init__.py index 78602c3581..3e6fffb11c 100644 --- a/openhands-sdk/openhands/sdk/__init__.py +++ b/openhands-sdk/openhands/sdk/__init__.py @@ -31,6 +31,7 @@ LLMProfileStore, LLMRegistry, LLMStreamChunk, + MaterializedRef, Message, RedactedThinkingBlock, RegistryEvent, @@ -132,6 +133,7 @@ "Message", "TextContent", "ImageContent", + "MaterializedRef", "ThinkingBlock", "RedactedThinkingBlock", "Tool", diff --git a/openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py b/openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py index 53b660ef36..08aed06ec2 100644 --- a/openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py +++ b/openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py @@ -49,7 +49,14 @@ from openhands.sdk.event.conversation_error import ConversationErrorEvent from openhands.sdk.hooks import HookConfig, HookEventProcessor, create_hook_callback from openhands.sdk.io import FileStore, LocalFileStore -from openhands.sdk.llm import LLM, Message, TextContent, content_to_str +from openhands.sdk.llm import ( + LLM, + BaseContent, + MaterializedRef, + Message, + TextContent, + content_to_str, +) from openhands.sdk.llm.auth.openai import create_subscription_llm_from_config from openhands.sdk.llm.llm import LLMCallContext from openhands.sdk.llm.llm_profile_store import LLMProfileStore @@ -380,7 +387,12 @@ def _default_callback(e): # This runs on first run()/send_message() call and handles both # explicit hooks and plugin hooks in one place self._hook_processor = None - self._on_event = self._tree_stamping(self._rules_injecting(base_callback)) + # In-memory dedup of events whose content has already been materialized, + # so re-emitting the same event (e.g. on replay) does not re-run writes. + self._materialized_event_ids: set[EventID] = set() + self._on_event = self._tree_stamping( + self._rules_injecting(self._materializing(base_callback)) + ) self._on_token = ( BaseConversation.compose_callbacks(token_callbacks) if token_callbacks @@ -547,6 +559,90 @@ def _touched_rule_path(self, event: ObservationEvent) -> str | None: return resolved.relative_to(resolved_root).as_posix() return None # touched a file outside the workspace; rules are repo-scoped + def _materializing( + self, inner: ConversationCallbackType + ) -> ConversationCallbackType: + """Wrap a callback so non-inline content is materialized to the workspace. + + Mirrors :meth:`_rules_injecting`: intercept each event as it enters, + enrich it, and pass it on — under the state lock, before persistence. + Because it dispatches on the polymorphic ``BaseContent.materialize()`` + contract and runs for every event flowing through ``_on_event`` (user + messages *and* tool/MCP observations), it is both source- and + type-agnostic; a new content type is supported by implementing + ``materialize()`` on that type, with no change here. + """ + + def wrapped(event: Event) -> None: + inner(self._maybe_materialize_content(event)) + + return cast(ConversationCallbackType, wrapped) + + def _maybe_materialize_content(self, event: Event) -> Event: + """Return ``event`` with materialized-path pointers injected, or unchanged. + + Walks the content parts an event carries, calls ``materialize()`` on each + (polymorphic; a no-op for text), writes any bytes into the workspace, and + appends a lightweight ``TextContent`` pointer to ``extended_content`` so + the agent is told where each file landed. Materialization is deduped per + event id (in memory) and is content-addressed on disk, so replaying an + event never writes duplicates. + """ + # Only these two event shapes carry externally-sourced content parts; + # both expose ``extended_content`` for the injected pointer. + if not isinstance(event, (MessageEvent, ObservationEvent)): + return event + + content_parts = self._event_content_parts(event) + if not content_parts: + return event + + if event.id in self._materialized_event_ids: + return event + + refs: list[MaterializedRef] = [] + for part in content_parts: + refs.extend(part.materialize(self.workspace)) + + self._materialized_event_ids.add(event.id) + + if not refs: + return event + + pointer = self._format_materialized_pointer(refs) + return event.model_copy( + update={ + "extended_content": list(event.extended_content) + [pointer], + } + ) + + @staticmethod + def _event_content_parts( + event: "MessageEvent | ObservationEvent", + ) -> list[BaseContent]: + """Return the materializable content parts carried by ``event``. + + ``MessageEvent`` carries user/agent message content; ``ObservationEvent`` + carries tool/MCP results. + """ + if isinstance(event, MessageEvent): + return list(event.llm_message.content) + return list(event.observation.to_llm_content) + + @staticmethod + def _format_materialized_pointer(refs: list[MaterializedRef]) -> TextContent: + """Build the pointer text telling the agent where files were written.""" + lines = [ + "The following attached content has been saved to the workspace " + "filesystem so you can operate on it with your tools:" + ] + for ref in refs: + detail = ref.path + if ref.mime_type: + detail += f" ({ref.mime_type})" + lines.append(f"- {detail}") + return TextContent(text="\n".join(lines)) + def _recover_persisted_client_tools( self, persistence_base_dir: str | Path, @@ -1108,7 +1204,9 @@ def _ensure_plugins_loaded(self) -> None: visualizer=self._visualizer, conversation_stats=self._state.stats, ) - self._on_event = self._tree_stamping(self._rules_injecting(raw_on_event)) + self._on_event = self._tree_stamping( + self._rules_injecting(self._materializing(raw_on_event)) + ) self._hook_processor.set_conversation_state(self._state) self._hook_processor.run_session_start() @@ -1181,7 +1279,9 @@ def _merge_runtime_plugin_hooks(self, plugin_hooks: HookConfig) -> None: visualizer=self._visualizer, conversation_stats=self._state.stats, ) - self._on_event = self._tree_stamping(self._rules_injecting(raw_on_event)) + self._on_event = self._tree_stamping( + self._rules_injecting(self._materializing(raw_on_event)) + ) self._hook_processor.set_conversation_state(self._state) self._hook_processor.run_session_start() diff --git a/openhands-sdk/openhands/sdk/llm/__init__.py b/openhands-sdk/openhands/sdk/llm/__init__.py index 9f20c92f33..e6a833ef98 100644 --- a/openhands-sdk/openhands/sdk/llm/__init__.py +++ b/openhands-sdk/openhands/sdk/llm/__init__.py @@ -14,7 +14,9 @@ from openhands.sdk.llm.llm_registry import LLMRegistry, RegistryEvent from openhands.sdk.llm.llm_response import LLMResponse from openhands.sdk.llm.message import ( + BaseContent, ImageContent, + MaterializedRef, Message, MessageToolCall, ReasoningItemModel, @@ -57,8 +59,10 @@ # Messages "Message", "MessageToolCall", + "BaseContent", "TextContent", "ImageContent", + "MaterializedRef", "ThinkingBlock", "RedactedThinkingBlock", "ReasoningItemModel", diff --git a/openhands-sdk/openhands/sdk/llm/message.py b/openhands-sdk/openhands/sdk/llm/message.py index 8b5ba0d8a2..62a2af62a8 100644 --- a/openhands-sdk/openhands/sdk/llm/message.py +++ b/openhands-sdk/openhands/sdk/llm/message.py @@ -1,7 +1,7 @@ import json from abc import abstractmethod from collections.abc import Sequence -from typing import Any, ClassVar, Literal +from typing import TYPE_CHECKING, Any, ClassVar, Literal from litellm import ChatCompletionMessageToolCall, ResponseFunctionToolCall from litellm.types.responses.main import ( @@ -18,6 +18,10 @@ from openhands.sdk.utils.deprecation import handle_deprecated_model_fields +if TYPE_CHECKING: + from openhands.sdk.workspace.base import BaseWorkspace + + logger = get_logger(__name__) @@ -158,6 +162,27 @@ class ReasoningItemModel(BaseModel): status: str | None = Field(default=None) +class MaterializedRef(BaseModel): + """A pointer to content that a :class:`BaseContent` wrote to a workspace. + + Returned by :meth:`BaseContent.materialize`. The ``path`` is where the bytes + now live on the workspace filesystem so the agent's file-oriented tools + (terminal, file editor) can act on it. + """ + + path: str = Field(description="Absolute path of the written file in the workspace") + source_url: str | None = Field( + default=None, + description="The originating URL (data: or http(s)://) if applicable", + ) + mime_type: str | None = Field( + default=None, description="Detected MIME type of the materialized content" + ) + size_bytes: int | None = Field( + default=None, description="Size of the written file in bytes" + ) + + class BaseContent(BaseModel): cache_prompt: bool = False @@ -169,6 +194,28 @@ def to_llm_dict(self) -> list[dict[str, str | dict[str, str]]]: even if they only have a single item. """ + def materialize( + self, + workspace: "BaseWorkspace", # noqa: ARG002 + ) -> list[MaterializedRef]: + """Persist this content to the workspace filesystem, if applicable. + + Mirrors the :meth:`to_llm_dict` contract: each content type knows how to + write *itself* out. The default is a no-op (e.g. :class:`TextContent` + carries nothing to materialize). Content types backed by bytes the agent + cannot reach inline (images, files, audio, blobs) override this to decode + or download their payload, write it into the workspace, and return the + resulting path(s). + + Args: + workspace: The workspace to write into. + + Returns: + A list of :class:`MaterializedRef` for every file written; empty when + there is nothing to materialize. + """ + return [] + class TextContent(BaseContent): type: Literal["text"] = "text" @@ -214,6 +261,48 @@ def to_llm_dict(self) -> list[dict[str, str | dict[str, str]]]: images[-1]["cache_control"] = {"type": "ephemeral"} return images + def materialize(self, workspace: "BaseWorkspace") -> list["MaterializedRef"]: + """Decode/download each image URL and write it into the workspace. + + ``data:`` URLs are decoded from base64; ``http(s)://`` URLs are + downloaded with a size cap. Any URL that cannot be materialized is + logged and skipped rather than aborting the others. + """ + from openhands.sdk.llm.utils.content_materialize import ( + download_url, + parse_data_url, + write_bytes_to_workspace, + ) + + refs: list[MaterializedRef] = [] + for url in self.image_urls: + try: + if url.startswith("data:"): + data, mime_type = parse_data_url(url) + elif url.startswith(("http://", "https://")): + data, mime_type = download_url(url) + else: + logger.warning( + "Skipping image URL with unsupported scheme during " + "materialize: %s", + url[:64], + ) + continue + path, size = write_bytes_to_workspace( + workspace, data, mime_type=mime_type + ) + refs.append( + MaterializedRef( + path=path, + source_url=url if not url.startswith("data:") else None, + mime_type=mime_type, + size_bytes=size, + ) + ) + except Exception as e: + logger.warning("Failed to materialize image content: %s", e) + return refs + class Message(BaseModel): # NOTE: this is not the same as EventSource diff --git a/openhands-sdk/openhands/sdk/llm/utils/content_materialize.py b/openhands-sdk/openhands/sdk/llm/utils/content_materialize.py new file mode 100644 index 0000000000..999ad903f4 --- /dev/null +++ b/openhands-sdk/openhands/sdk/llm/utils/content_materialize.py @@ -0,0 +1,143 @@ +"""Helpers for materializing multipart content to a workspace filesystem. + +These functions decode ``data:`` URLs or download ``http(s)://`` URLs (with a +size cap) and write the bytes into the workspace using a deterministic, +content-addressed path. Content-addressing makes writes idempotent: the same +payload always maps to the same file, so replaying an event never produces a +duplicate write. +""" + +import base64 +import binascii +import hashlib +import mimetypes +import os +import tempfile +from pathlib import PurePosixPath +from typing import TYPE_CHECKING + +import httpx + +from openhands.sdk.logger import get_logger + + +if TYPE_CHECKING: + from openhands.sdk.workspace.base import BaseWorkspace + + +logger = get_logger(__name__) + +# Directory (relative to the workspace working dir) where materialized content +# is written. Kept out of the way of normal project files. +MATERIALIZE_SUBDIR = ".materialized" + +# Default cap for http(s) downloads (bytes). Guards against unbounded fetches. +DEFAULT_MAX_DOWNLOAD_BYTES = 20 * 1024 * 1024 # 20 MiB + + +def parse_data_url(url: str) -> tuple[bytes, str | None]: + """Decode a ``data:;base64,`` URL into ``(bytes, mime_type)``. + + Only base64-encoded data URLs are supported. Raises ``ValueError`` on any + malformed input. + """ + if not url.startswith("data:"): + raise ValueError("Not a data: URL") + header, sep, encoded = url[len("data:") :].partition(",") + if not sep: + raise ValueError("Malformed data URL: missing comma separator") + if ";base64" not in header: + raise ValueError("Only base64-encoded data URLs are supported") + mime_type = header.split(";", 1)[0] or None + try: + raw = base64.b64decode(encoded, validate=True) + except (binascii.Error, ValueError) as e: + raise ValueError(f"Invalid base64 payload in data URL: {e}") from e + return raw, mime_type + + +def download_url( + url: str, *, max_bytes: int = DEFAULT_MAX_DOWNLOAD_BYTES +) -> tuple[bytes, str | None]: + """Download an ``http(s)://`` URL, enforcing a size cap. + + The cap is checked both against the ``Content-Length`` header (when present) + and against the number of bytes actually streamed, so a lying or missing + header cannot bypass it. Raises ``ValueError`` if the cap is exceeded. + """ + with httpx.Client(follow_redirects=True, timeout=30.0) as client: + with client.stream("GET", url) as response: + response.raise_for_status() + declared = response.headers.get("content-length") + if declared is not None and int(declared) > max_bytes: + raise ValueError( + f"Refusing to download {url}: Content-Length {declared} " + f"exceeds cap of {max_bytes} bytes" + ) + chunks: list[bytes] = [] + total = 0 + for chunk in response.iter_bytes(): + total += len(chunk) + if total > max_bytes: + raise ValueError( + f"Refusing to download {url}: stream exceeded cap of " + f"{max_bytes} bytes" + ) + chunks.append(chunk) + mime_type = response.headers.get("content-type") + if mime_type: + mime_type = mime_type.split(";", 1)[0].strip() or None + return b"".join(chunks), mime_type + + +def _extension_for(mime_type: str | None) -> str: + if not mime_type: + return ".bin" + return mimetypes.guess_extension(mime_type) or ".bin" + + +def write_bytes_to_workspace( + workspace: "BaseWorkspace", + data: bytes, + *, + mime_type: str | None, +) -> tuple[str, int]: + """Write ``data`` into the workspace under a content-addressed path. + + Returns ``(absolute_path, size_bytes)``. The filename is derived from the + SHA-256 of the bytes, so repeated calls with identical content resolve to + the same destination and never write duplicates. + """ + digest = hashlib.sha256(data).hexdigest()[:16] + filename = f"{digest}{_extension_for(mime_type)}" + rel_path = str(PurePosixPath(MATERIALIZE_SUBDIR) / filename) + dest_path = str(PurePosixPath(workspace.working_dir) / rel_path) + + # Idempotency: if the content-addressed file already exists, skip the write. + if _workspace_file_exists(workspace, dest_path): + logger.debug("Materialized file already exists, skipping write: %s", dest_path) + return dest_path, len(data) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp.write(data) + tmp_path = tmp.name + try: + result = workspace.file_upload(tmp_path, dest_path) + if not result.success: + raise RuntimeError(f"Failed to write materialized file: {result.error}") + finally: + os.unlink(tmp_path) + return dest_path, len(data) + + +def _workspace_file_exists(workspace: "BaseWorkspace", path: str) -> bool: + """Best-effort check whether ``path`` already exists in the workspace.""" + try: + result = workspace.execute_command(f"test -f {_shquote(path)}") + return result.exit_code == 0 + except Exception: + return False + + +def _shquote(s: str) -> str: + return "'" + s.replace("'", "'\"'\"'") + "'" diff --git a/tests/cross/test_remote_conversation_live_server.py b/tests/cross/test_remote_conversation_live_server.py index c4d4d73ff0..34c776ad84 100644 --- a/tests/cross/test_remote_conversation_live_server.py +++ b/tests/cross/test_remote_conversation_live_server.py @@ -2294,3 +2294,83 @@ async def slow_acompletion(self, messages, tools=None, **kwargs): # type: ignor assert events_resp.status_code == 200 items = events_resp.json()["items"] assert len(items) >= 1, f"Expected at least one InterruptEvent, got: {items}" + + +def test_image_content_materialized_over_real_server(server_env, patched_llm): + """End-to-end: an image in a user message is materialized to the workspace. + + Exercises the full ``_on_event`` seam (not the helper directly) against a + real agent server: sending a ``Message`` carrying an ``ImageContent`` data + URL must (1) write the decoded bytes into the server-side workspace under + ``.materialized/`` via the real ``file_upload`` path, and (2) inject a + path pointer into the persisted user ``MessageEvent``'s ``extended_content``. + + Pointing the ``RemoteWorkspace`` at ``server_env['workspace_path']`` means + the server's ``LocalWorkspace`` writes to a directory the test can inspect + directly on disk. + """ + import base64 + + from openhands.sdk import Message, TextContent + from openhands.sdk.llm import ImageContent + from openhands.sdk.llm.utils.content_materialize import MATERIALIZE_SUBDIR + + png_bytes = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9" + "awAAAABJRU5ErkJggg==" + ) + data_url = f"data:image/png;base64,{base64.b64encode(png_bytes).decode()}" + + workspace_dir = server_env["workspace_path"] + + llm = LLM(model="gpt-4o-mini", api_key=SecretStr("test")) + agent = Agent(llm=llm, tools=[]) + workspace = RemoteWorkspace(host=server_env["host"], working_dir=str(workspace_dir)) + conv: RemoteConversation = Conversation(agent=agent, workspace=workspace) + + try: + conv.send_message( + Message( + role="user", + content=[ + TextContent(text="here is an attachment"), + ImageContent(image_urls=[data_url]), + ], + ) + ) + conv.run() + + # 1. The decoded bytes landed in the server-side workspace, content-addressed. + materialized_dir = workspace_dir / MATERIALIZE_SUBDIR + written: list[Path] = [] + for _ in range(50): # up to ~5s for the server to flush the write + if materialized_dir.exists(): + written = list(materialized_dir.iterdir()) + if written: + break + time.sleep(0.1) + assert len(written) == 1, ( + f"Expected exactly one materialized file, found: {written}" + ) + assert written[0].read_bytes() == png_bytes + + # 2. The persisted user MessageEvent gained a path pointer. + found_pointer = False + for _ in range(50): + for e in conv.state.events: + if ( + isinstance(e, MessageEvent) + and e.source == "user" + and any(MATERIALIZE_SUBDIR in c.text for c in e.extended_content) + ): + found_pointer = True + break + if found_pointer: + break + time.sleep(0.1) + assert found_pointer, ( + "Expected the user MessageEvent's extended_content to carry a " + "materialized-path pointer" + ) + finally: + conv.close() diff --git a/tests/sdk/conversation/test_content_materialization.py b/tests/sdk/conversation/test_content_materialization.py new file mode 100644 index 0000000000..f3e5219a21 --- /dev/null +++ b/tests/sdk/conversation/test_content_materialization.py @@ -0,0 +1,158 @@ +"""Tests for the content-materialization seam in LocalConversation. + +As each event enters ``_on_event``, content parts that carry non-inline bytes +(images today) are written to the workspace and a path pointer is appended to +the event's ``extended_content`` — mirroring the path-rule injection seam. +The interceptor is source-agnostic (user messages *and* tool observations) and +type-agnostic (dispatches on ``BaseContent.materialize()``). +""" + +import base64 +from pathlib import Path + +from openhands.sdk.agent import Agent +from openhands.sdk.conversation.impl.local_conversation import LocalConversation +from openhands.sdk.event import MessageEvent, ObservationEvent +from openhands.sdk.llm import ImageContent, Message, TextContent +from openhands.sdk.llm.utils.content_materialize import MATERIALIZE_SUBDIR +from openhands.sdk.testing import TestLLM +from openhands.sdk.tool.builtins.finish import FinishObservation + + +PNG_BYTES = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9" + "awAAAABJRU5ErkJggg==" +) + + +def _data_url(data: bytes, mime: str = "image/png") -> str: + return f"data:{mime};base64,{base64.b64encode(data).decode()}" + + +def _conversation(tmp_path: Path) -> LocalConversation: + agent = Agent( + llm=TestLLM.from_messages( + [Message(role="assistant", content=[TextContent(text="ok")])], + model="test-model", + ), + tools=[], + include_default_tools=[], + ) + return LocalConversation( + agent=agent, + workspace=tmp_path, + persistence_dir=tmp_path / "conversation", + delete_on_close=True, + ) + + +def _materialize(conv: LocalConversation, event): + result = conv._maybe_materialize_content(event) + return result + + +def test_user_message_image_materialized_and_pointer_injected(tmp_path: Path) -> None: + conv = _conversation(tmp_path) + try: + event = MessageEvent( + source="user", + llm_message=Message( + role="user", + content=[ + TextContent(text="attach this"), + ImageContent(image_urls=[_data_url(PNG_BYTES)]), + ], + ), + ) + result = _materialize(conv, event) + assert isinstance(result, MessageEvent) + + # A pointer was appended to extended_content. + assert len(result.extended_content) == 1 + pointer = result.extended_content[0].text + assert MATERIALIZE_SUBDIR in pointer + assert "image/png" in pointer + + # The file actually landed in the workspace with correct bytes. + subdir = tmp_path / MATERIALIZE_SUBDIR + files = list(subdir.iterdir()) + assert len(files) == 1 + assert files[0].read_bytes() == PNG_BYTES + finally: + conv.close() + + +def test_observation_image_materialized(tmp_path: Path) -> None: + conv = _conversation(tmp_path) + try: + obs = ObservationEvent( + observation=FinishObservation( + content=[ + TextContent(text="here is a tool image"), + ImageContent(image_urls=[_data_url(PNG_BYTES)]), + ] + ), + action_id="a1", + tool_name="some_tool", + tool_call_id="tc1", + ) + result = _materialize(conv, obs) + assert isinstance(result, ObservationEvent) + assert len(result.extended_content) == 1 + assert MATERIALIZE_SUBDIR in result.extended_content[0].text + assert len(list((tmp_path / MATERIALIZE_SUBDIR).iterdir())) == 1 + finally: + conv.close() + + +def test_text_only_message_is_unchanged(tmp_path: Path) -> None: + conv = _conversation(tmp_path) + try: + event = MessageEvent( + source="user", + llm_message=Message( + role="user", content=[TextContent(text="no attachments here")] + ), + ) + result = _materialize(conv, event) + assert result.extended_content == [] + assert not (tmp_path / MATERIALIZE_SUBDIR).exists() + finally: + conv.close() + + +def test_event_without_content_parts_is_unchanged(tmp_path: Path) -> None: + conv = _conversation(tmp_path) + try: + # ObservationEvent whose observation has only text -> no materializable parts. + obs = ObservationEvent( + observation=FinishObservation(content=[TextContent(text="just text")]), + action_id="a1", + tool_name="some_tool", + tool_call_id="tc1", + ) + result = _materialize(conv, obs) + assert result.extended_content == [] + finally: + conv.close() + + +def test_replay_does_not_write_duplicates(tmp_path: Path) -> None: + conv = _conversation(tmp_path) + try: + event = MessageEvent( + source="user", + llm_message=Message( + role="user", + content=[ImageContent(image_urls=[_data_url(PNG_BYTES)])], + ), + ) + first = _materialize(conv, event) + assert len(first.extended_content) == 1 + + # Re-emitting the same event id must not re-materialize or re-inject. + second = _materialize(conv, event) + assert second.extended_content == [] + assert len(list((tmp_path / MATERIALIZE_SUBDIR).iterdir())) == 1 + finally: + conv.close() diff --git a/tests/sdk/llm/test_content_materialize.py b/tests/sdk/llm/test_content_materialize.py new file mode 100644 index 0000000000..57cf8b904f --- /dev/null +++ b/tests/sdk/llm/test_content_materialize.py @@ -0,0 +1,166 @@ +"""Tests for the ``BaseContent.materialize()`` contract. + +Covers: data-URL decode -> file, http-URL download -> file (with size cap), +a ``TextContent`` no-op, and content-addressed idempotency (no duplicate +writes across repeated materialization). +""" + +import base64 +from pathlib import Path + +import httpx +import pytest + +from openhands.sdk.llm import ImageContent, MaterializedRef, TextContent +from openhands.sdk.llm.utils import content_materialize +from openhands.sdk.llm.utils.content_materialize import ( + download_url, + parse_data_url, +) +from openhands.sdk.workspace import LocalWorkspace + + +PNG_BYTES = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9" + "awAAAABJRU5ErkJggg==" +) + + +def _data_url(data: bytes, mime: str = "image/png") -> str: + return f"data:{mime};base64,{base64.b64encode(data).decode()}" + + +def test_parse_data_url_decodes_bytes_and_mime() -> None: + data, mime = parse_data_url(_data_url(PNG_BYTES)) + assert data == PNG_BYTES + assert mime == "image/png" + + +@pytest.mark.parametrize( + "url", + [ + "http://example.com/x.png", # not a data URL + "data:image/png,notbase64", # missing ;base64 + "data:image/png;base64", # missing comma + "data:image/png;base64,!!!notbase64!!!", # invalid payload + ], +) +def test_parse_data_url_rejects_malformed(url: str) -> None: + with pytest.raises(ValueError): + parse_data_url(url) + + +def test_text_content_materialize_is_noop(tmp_path: Path) -> None: + ws = LocalWorkspace(working_dir=str(tmp_path)) + assert TextContent(text="hello").materialize(ws) == [] + + +def test_image_content_materialize_data_url_writes_file(tmp_path: Path) -> None: + ws = LocalWorkspace(working_dir=str(tmp_path)) + content = ImageContent(image_urls=[_data_url(PNG_BYTES)]) + + refs = content.materialize(ws) + + assert len(refs) == 1 + ref = refs[0] + assert isinstance(ref, MaterializedRef) + written = Path(ref.path) + assert written.is_file() + assert written.read_bytes() == PNG_BYTES + assert ref.mime_type == "image/png" + assert ref.size_bytes == len(PNG_BYTES) + # data: URLs are not echoed back as a source URL. + assert ref.source_url is None + # Written under the workspace working dir. + assert str(tmp_path) in ref.path + + +def test_image_content_materialize_http_url_downloads( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + url = "https://example.com/pic.png" + + def handler(request: httpx.Request) -> httpx.Response: + assert str(request.url) == url + return httpx.Response( + 200, content=PNG_BYTES, headers={"content-type": "image/png"} + ) + + transport = httpx.MockTransport(handler) + real_client = httpx.Client + + def fake_client(*args, **kwargs): + kwargs["transport"] = transport + return real_client(*args, **kwargs) + + monkeypatch.setattr(content_materialize.httpx, "Client", fake_client) + + ws = LocalWorkspace(working_dir=str(tmp_path)) + refs = ImageContent(image_urls=[url]).materialize(ws) + + assert len(refs) == 1 + assert Path(refs[0].path).read_bytes() == PNG_BYTES + assert refs[0].source_url == url + + +def test_download_url_enforces_size_cap_via_content_length( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, content=b"x" * 100, headers={"content-length": "100"} + ) + + transport = httpx.MockTransport(handler) + real_client = httpx.Client + + def fake_client(*args, **kwargs): + kwargs["transport"] = transport + return real_client(*args, **kwargs) + + monkeypatch.setattr(content_materialize.httpx, "Client", fake_client) + + with pytest.raises(ValueError, match="exceeds cap"): + download_url("https://example.com/big.bin", max_bytes=10) + + +def test_download_url_enforces_size_cap_while_streaming( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # An understated content-length must not let the real payload bypass the + # cap: the byte counter enforces it while streaming. + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"x" * 100, headers={"content-length": "5"}) + + transport = httpx.MockTransport(handler) + real_client = httpx.Client + + def fake_client(*args, **kwargs): + kwargs["transport"] = transport + return real_client(*args, **kwargs) + + monkeypatch.setattr(content_materialize.httpx, "Client", fake_client) + + with pytest.raises(ValueError, match="exceeded cap"): + download_url("https://example.com/big.bin", max_bytes=10) + + +def test_materialize_is_idempotent_no_duplicate_writes(tmp_path: Path) -> None: + ws = LocalWorkspace(working_dir=str(tmp_path)) + content = ImageContent(image_urls=[_data_url(PNG_BYTES)]) + + first = content.materialize(ws) + second = content.materialize(ws) + + # Content-addressed: both resolve to the same path. + assert first[0].path == second[0].path + # Exactly one file exists in the materialized subdir. + subdir = tmp_path / content_materialize.MATERIALIZE_SUBDIR + files = list(subdir.iterdir()) + assert len(files) == 1 + + +def test_image_content_materialize_skips_unsupported_scheme(tmp_path: Path) -> None: + ws = LocalWorkspace(working_dir=str(tmp_path)) + refs = ImageContent(image_urls=["ftp://example.com/x.png"]).materialize(ws) + assert refs == [] diff --git a/tests/sdk/workspace/remote/test_materialize_remote_workspace.py b/tests/sdk/workspace/remote/test_materialize_remote_workspace.py new file mode 100644 index 0000000000..365ee5a214 --- /dev/null +++ b/tests/sdk/workspace/remote/test_materialize_remote_workspace.py @@ -0,0 +1,132 @@ +"""Materialization over a remote workspace transport. + +The local-workspace tests in ``tests/sdk/llm/test_content_materialize.py`` cover +the decode/download/write logic against a ``LocalWorkspace``. This module fills +the remaining gap: ``write_bytes_to_workspace`` reaches the filesystem through +``workspace.file_upload`` and gates duplicate writes with +``workspace.execute_command("test -f ...")``. Those two calls behave differently +for a ``RemoteWorkspace`` than for a local one, so we exercise them here through +an in-memory fake remote transport rather than real HTTP. + +The fake records every ``file_upload`` and honours the ``test -f`` existence +check, so we can assert the content-addressed destination path, that the bytes +are handed to the remote upload, and that content-addressing makes repeated +materialization idempotent (no duplicate remote writes). +""" + +import base64 +from pathlib import Path + +import pytest +from pydantic import PrivateAttr + +from openhands.sdk.llm import ImageContent, MaterializedRef +from openhands.sdk.llm.utils.content_materialize import MATERIALIZE_SUBDIR +from openhands.sdk.workspace import RemoteWorkspace +from openhands.sdk.workspace.models import CommandResult, FileOperationResult + + +PNG_BYTES = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9" + "awAAAABJRU5ErkJggg==" +) + + +def _data_url(data: bytes, mime: str = "image/png") -> str: + return f"data:{mime};base64,{base64.b64encode(data).decode()}" + + +class FakeRemoteWorkspace(RemoteWorkspace): + """A ``RemoteWorkspace`` whose transport is an in-memory remote filesystem. + + Overriding only ``file_upload`` and ``execute_command`` keeps the real + ``working_dir`` / path-building behaviour of ``RemoteWorkspace`` while + letting the test observe exactly what the materialization layer sends over + the wire. + """ + + _remote_files: dict[str, bytes] = PrivateAttr(default_factory=dict) + _uploads: list[str] = PrivateAttr(default_factory=list) + + def file_upload(self, source_path, destination_path) -> FileOperationResult: # type: ignore[override] + data = Path(source_path).read_bytes() + dest = str(destination_path) + self._remote_files[dest] = data + self._uploads.append(dest) + return FileOperationResult( + success=True, + source_path=str(source_path), + destination_path=dest, + file_size=len(data), + ) + + def execute_command(self, command, cwd=None, timeout=30.0) -> CommandResult: # type: ignore[override] + # Only the content-addressed existence probe is exercised here. + exists = False + if command.startswith("test -f "): + quoted = command[len("test -f ") :].strip() + path = quoted.strip("'") + exists = path in self._remote_files + return CommandResult( + command=command, + exit_code=0 if exists else 1, + stdout="", + stderr="", + timeout_occurred=False, + ) + + +@pytest.fixture +def remote_ws() -> FakeRemoteWorkspace: + return FakeRemoteWorkspace( + host="http://remote.invalid:8000", working_dir="/remote/project" + ) + + +def test_image_materialize_uploads_to_remote_workspace( + remote_ws: FakeRemoteWorkspace, +) -> None: + refs = ImageContent(image_urls=[_data_url(PNG_BYTES)]).materialize(remote_ws) + + assert len(refs) == 1 + ref = refs[0] + assert isinstance(ref, MaterializedRef) + + # Written under the remote working dir's materialized subdir, content-addressed. + assert ref.path.startswith(f"/remote/project/{MATERIALIZE_SUBDIR}/") + assert ref.path.endswith(".png") + assert ref.mime_type == "image/png" + assert ref.size_bytes == len(PNG_BYTES) + + # Exactly one upload, carrying the real bytes to the content-addressed path. + assert remote_ws._uploads == [ref.path] + assert remote_ws._remote_files[ref.path] == PNG_BYTES + + +def test_image_materialize_is_idempotent_over_remote( + remote_ws: FakeRemoteWorkspace, +) -> None: + content = ImageContent(image_urls=[_data_url(PNG_BYTES)]) + + first = content.materialize(remote_ws) + second = content.materialize(remote_ws) + + # Content-addressed: both resolve to the same remote path. + assert first[0].path == second[0].path + # The ``test -f`` probe short-circuits the second write: only one upload. + assert remote_ws._uploads == [first[0].path] + assert len(remote_ws._remote_files) == 1 + + +def test_distinct_content_yields_distinct_remote_paths( + remote_ws: FakeRemoteWorkspace, +) -> None: + other = PNG_BYTES + b"\x00extra" + + first = ImageContent(image_urls=[_data_url(PNG_BYTES)]).materialize(remote_ws) + second = ImageContent(image_urls=[_data_url(other)]).materialize(remote_ws) + + assert first[0].path != second[0].path + assert len(remote_ws._remote_files) == 2 + assert remote_ws._remote_files[first[0].path] == PNG_BYTES + assert remote_ws._remote_files[second[0].path] == other