feat(conversation): materialize non-inline multipart content to the workspace

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",

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

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",

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