fix(artifacts): resolve viewed images from shared blobs (#6329)

Co-authored-by: mikamikasuki <211269698+mikamikasuki@users.noreply.github.com>

backend/app/gateway/routers/artifacts.py

@@ -39,6 +39,7 @@ router = APIRouter(prefix="/api", tags=["artifacts"])
 MAX_SKILL_ARCHIVE_MEMBER_BYTES = 16 * 1024 * 1024
 _SKILL_ARCHIVE_READ_CHUNK_SIZE = 64 * 1024
 MAX_EDITABLE_ARTIFACT_BYTES = 2 * 1024 * 1024
+MAX_VIEWED_IMAGE_BLOB_BYTES = 20 * 1024 * 1024
 _ARTIFACT_EDIT_TEMP_PREFIX = ".artifact-edit-"
 
 
@@ -495,7 +496,15 @@ async def get_artifact(thread_id: ThreadId, path: str, request: Request, downloa
     # Offload path stat + MIME sniff (blocking filesystem IO). Every regular
     # artifact response is streamed by FileResponse; the worker only reports
     # disposition and media type.
-    kind, mime_type = await asyncio.to_thread(_read_artifact_payload, actual_path, path, download)
+    try:
+        kind, mime_type = await asyncio.to_thread(_read_artifact_payload, actual_path, path, download)
+    except HTTPException as exc:
+        if exc.status_code != 404:
+            raise
+        blob_response = await _get_viewed_image_blob_response(thread_id, path, request, download=download)
+        if blob_response is not None:
+            return blob_response
+        raise
 
     if kind == "file":
         # Always force download for active content types to prevent script
@@ -522,6 +531,86 @@ async def get_artifact(thread_id: ThreadId, path: str, request: Request, downloa
     raise AssertionError(f"Unhandled artifact response kind: {kind!r}")
 
 
+async def _get_viewed_image_blob_response(
+    thread_id: str,
+    path: str,
+    request: Request,
+    *,
+    download: bool,
+) -> Response | None:
+    """Resolve a checkpointed viewed-image blob when this Gateway lacks the file.
+
+    The checkpoint binds the content-addressed reference to the authorized
+    thread and virtual path. Metadata and bytes are revalidated before serving;
+    an absent or unusable reference leaves the normal artifact 404 in place.
+    """
+    from deerflow.storage import BlobRef, get_blob_store_if_enabled
+
+    virtual_path = f"/{path.lstrip('/')}"
+    try:
+        from app.gateway.services import build_thread_checkpoint_state_accessor
+
+        accessor, config = await build_thread_checkpoint_state_accessor(request, thread_id=thread_id)
+        snapshot = await accessor.aget(config)
+        values = getattr(snapshot, "values", None)
+        viewed_images = values.get("viewed_images") if isinstance(values, dict) else None
+        metadata = viewed_images.get(virtual_path) if isinstance(viewed_images, dict) else None
+        if metadata is None and isinstance(viewed_images, dict):
+            metadata = viewed_images.get(path)
+        if not isinstance(metadata, dict):
+            return None
+
+        expected_size = metadata.get("size")
+        expected_sha256 = metadata.get("sha256")
+        mime_type = metadata.get("mime_type")
+        if (
+            isinstance(expected_size, bool)
+            or not isinstance(expected_size, int)
+            or not 0 <= expected_size <= MAX_VIEWED_IMAGE_BLOB_BYTES
+            or not isinstance(expected_sha256, str)
+            or not isinstance(mime_type, str)
+            or not mime_type.startswith("image/")
+        ):
+            return None
+        path_mime_type, _ = mimetypes.guess_type(virtual_path)
+        if path_mime_type != mime_type:
+            return None
+        blob_ref_data = metadata.get("blob_ref")
+        if not isinstance(blob_ref_data, dict):
+            return None
+        blob_ref = BlobRef.model_validate(blob_ref_data)
+        if blob_ref.kind != "viewed-image" or blob_ref.sha256 != expected_sha256 or blob_ref.size != expected_size or blob_ref.content_type != mime_type:
+            return None
+        store = await asyncio.to_thread(get_blob_store_if_enabled)
+        if store is None:
+            return None
+        content = await asyncio.to_thread(store.get_bytes, blob_ref)
+        if len(content) != expected_size or hashlib.sha256(content).hexdigest() != expected_sha256:
+            return None
+    except Exception:
+        logger.debug("Could not resolve viewed-image blob for thread artifact", exc_info=True)
+        return None
+
+    etag = f'"{expected_sha256}"'
+    headers = {
+        "ETag": etag,
+        "X-Content-Type-Options": "nosniff",
+        "Accept-Ranges": "bytes",
+    }
+    if download or _is_active_content_mime_type(mime_type):
+        headers.update(_build_attachment_headers(Path(virtual_path).name, headers))
+    else:
+        headers["Content-Disposition"] = _build_content_disposition("inline", Path(virtual_path).name)
+
+    range_header = request.headers.get("range") if request is not None else None
+    if_range = request.headers.get("if-range") if request is not None else None
+    if if_range and if_range != etag:
+        range_header = None
+    ranged_content, status_code, range_headers = _slice_byte_range(content, range_header)
+    headers.update(range_headers)
+    return Response(content=ranged_content, status_code=status_code, media_type=mime_type, headers=headers)
+
+
 @router.put(
     "/threads/{thread_id}/artifacts/{path:path}",
     response_model=ArtifactUpdateResponse,

backend/docs/blob-storage.md

@@ -4,7 +4,7 @@ Resolves the multi-instance half of [#4189](https://github.com/bytedance/deer-fl
 
 | Producer | Write | Reads |
 |---|---|---|
-| Viewed images | `view_image_tool` → `ViewedImageData.blob_ref` plus compatibility `actual_path` | `ViewImageMiddleware._read_image_as_data_url`, gateway artifact routes, IM channels, `present_file_tool` |
+| Viewed images | `view_image_tool` → `ViewedImageData.blob_ref` plus compatibility `actual_path` | `ViewImageMiddleware._read_image_as_data_url`; thread artifact GET resolves a matching checkpoint-bound ref when the local file is absent; IM channels and `present_file_tool` retain local-path reads |
 | Externalized tool results | `ToolOutputBudgetMiddleware` → server-owned `tool-output` ref beside the virtual path | pre-model restoration into the current thread-data mount, then model `read_file` |
 
 On a single gateway both are correct. Behind a load balancer, the instance handling the read is frequently not the instance that wrote the file. The blob store replaces **"where on this machine"** with **"which content"**, so every instance that can reach the backing store resolves the same bytes.
@@ -62,10 +62,12 @@ The producer migrations are independently revertible:
 1. **Viewed images — migrated.** `ViewedImageData` carries an optional
    `blob_ref` while retaining `actual_path`; `view_image_tool` writes validated
    bytes when the store is enabled, and `ViewImageMiddleware` resolves the
-   digest-bound blob before the sandbox/local compatibility paths. The gateway
-   artifact routes and IM channels keep their local-path reads until they can be
-   exercised against a multi-instance deployment. Store initialization or write
-   failures return a generic tool error; read failures fall through to the
+   digest-bound blob before the sandbox/local compatibility paths. Thread
+   artifact GET also serves a checkpoint-bound image blob when the reader
+   Gateway lacks the local file; the reference is checked against the thread's
+   virtual path, digest, size, kind and MIME type. IM channels and
+   `present_file_tool` still use local-path reads. Store initialization or write
+   failures return a generic tool error; model-read failures fall through to the
    validated sandbox/host compatibility paths.
 2. **Externalized tool results — migrated.** Host externalization stores the
    exact UTF-8 bytes as `kind="tool-output"` and checkpoints a versioned,

backend/packages/harness/deerflow/storage/AGENTS.md

@@ -27,7 +27,11 @@ bytes as `kind="viewed-image"`, checkpoints the JSON-safe ref beside
 tool call rather than claiming cross-instance durability. `ViewImageMiddleware` accepts a
 ref only when its kind, digest, size and content type agree with the separately
 recorded image metadata; reads are blob-first and retain the existing
-sandbox/host compatibility path as fallback after any store failure.
+sandbox/host compatibility path as fallback after any store failure. The
+thread artifact GET route also resolves the checkpoint-bound image ref when the
+reader Gateway cannot find the local file; it confines this fallback to the
+authorized thread and exact virtual path and verifies the bytes before serving.
+Other artifact and IM-channel reads still use their existing local-path flow.
 
 `ToolOutputBudgetMiddleware` is a migrated producer for host-externalized
 results. It persists exact UTF-8 bytes as `kind="tool-output"`, checkpoints a

backend/tests/test_artifacts_router.py

@@ -63,6 +63,142 @@ def test_get_artifact_reads_utf8_text_file_on_windows_locale(tmp_path, monkeypat
     assert response.headers["accept-ranges"] == "bytes"
 
 
+def test_get_artifact_reads_shared_viewed_image_blob_when_local_path_is_missing(tmp_path, monkeypatch) -> None:
+    """A second Gateway can serve an image from its shared blob ref without the writer's disk."""
+    from deerflow.storage import BlobRef
+
+    image_bytes = b"\x89PNG\r\n\x1a\nimage"
+    virtual_path = "/mnt/user-data/outputs/chart.png"
+    digest = hashlib.sha256(image_bytes).hexdigest()
+    blob_ref = BlobRef(sha256=digest, size=len(image_bytes), kind="viewed-image", content_type="image/png")
+    absent_on_reader = tmp_path / "writer-only" / "chart.png"
+    monkeypatch.setattr(
+        artifacts_router,
+        "resolve_thread_virtual_path",
+        lambda _thread_id, _path, user_id=None: absent_on_reader,
+    )
+
+    from app.gateway import services
+
+    class Accessor:
+        async def aget(self, _config):
+            return SimpleNamespace(
+                values={
+                    "viewed_images": {
+                        virtual_path: {
+                            "actual_path": str(absent_on_reader),
+                            "mime_type": "image/png",
+                            "size": len(image_bytes),
+                            "sha256": digest,
+                            "blob_ref": blob_ref.model_dump(mode="json"),
+                        }
+                    }
+                }
+            )
+
+    async def build_accessor(_request, *, thread_id):
+        assert thread_id == "thread-1"
+        return Accessor(), {"configurable": {"thread_id": thread_id}}
+
+    monkeypatch.setattr(services, "build_thread_checkpoint_state_accessor", build_accessor)
+
+    class SharedStore:
+        def get_bytes(self, requested_ref):
+            assert requested_ref == blob_ref
+            return image_bytes
+
+    monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", lambda: SharedStore())
+
+    response = asyncio.run(
+        call_unwrapped(
+            artifacts_router.get_artifact,
+            "thread-1",
+            virtual_path.lstrip("/"),
+            _make_request(),
+        )
+    )
+
+    assert response.status_code == 200
+    assert response.body == image_bytes
+    assert response.media_type == "image/png"
+    assert response.headers["ETag"] == f'"{digest}"'
+
+    range_request = Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "path": "/",
+            "headers": [(b"range", b"bytes=1-3"), (b"if-range", f'"{digest}"'.encode())],
+            "query_string": b"",
+        }
+    )
+    ranged = asyncio.run(
+        call_unwrapped(
+            artifacts_router.get_artifact,
+            "thread-1",
+            virtual_path.lstrip("/"),
+            range_request,
+        )
+    )
+    assert ranged.status_code == 206
+    assert ranged.body == image_bytes[1:4]
+    assert ranged.headers["Content-Range"] == f"bytes 1-3/{len(image_bytes)}"
+
+
+def test_get_artifact_rejects_checkpoint_image_blob_metadata_mismatch(tmp_path, monkeypatch) -> None:
+    from deerflow.storage import BlobRef
+
+    image_bytes = b"\x89PNG\r\n\x1a\nimage"
+    virtual_path = "/mnt/user-data/outputs/chart.png"
+    actual_digest = hashlib.sha256(image_bytes).hexdigest()
+    wrong_digest = hashlib.sha256(b"different bytes").hexdigest()
+    blob_ref = BlobRef(sha256=actual_digest, size=len(image_bytes), kind="viewed-image", content_type="image/png")
+    absent_on_reader = tmp_path / "writer-only" / "chart.png"
+    monkeypatch.setattr(
+        artifacts_router,
+        "resolve_thread_virtual_path",
+        lambda _thread_id, _path, user_id=None: absent_on_reader,
+    )
+
+    from app.gateway import services
+
+    class Accessor:
+        async def aget(self, _config):
+            return SimpleNamespace(
+                values={
+                    "viewed_images": {
+                        virtual_path: {
+                            "mime_type": "image/png",
+                            "size": len(image_bytes),
+                            "sha256": wrong_digest,
+                            "blob_ref": blob_ref.model_dump(mode="json"),
+                        }
+                    }
+                }
+            )
+
+    async def build_accessor(_request, *, thread_id):
+        return Accessor(), {"configurable": {"thread_id": thread_id}}
+
+    monkeypatch.setattr(services, "build_thread_checkpoint_state_accessor", build_accessor)
+
+    class MustNotReadStore:
+        def get_bytes(self, _requested_ref):
+            pytest.fail("checkpoint metadata must be validated before reading a blob")
+
+    monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", lambda: MustNotReadStore())
+
+    with pytest.raises(HTTPException, match="Artifact not found"):
+        asyncio.run(
+            call_unwrapped(
+                artifacts_router.get_artifact,
+                "thread-1",
+                virtual_path.lstrip("/"),
+                _make_request(),
+            )
+        )
+
+
 @asynccontextmanager
 async def _allow_artifact_write(*_args, **_kwargs):
     yield