From e2e8ba7aecaed6a485380d715df7551de5008476 Mon Sep 17 00:00:00 2001 From: mika Date: Mon, 26 Oct 2026 23:21:42 -0700 Subject: [PATCH] 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 | 91 +++++++++++- backend/docs/blob-storage.md | 12 +- .../harness/deerflow/storage/AGENTS.md | 6 +- backend/tests/test_artifacts_router.py | 136 ++++++++++++++++++ 4 files changed, 238 insertions(+), 7 deletions(-) diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py index d19afc02e4..e3ca9b2bd7 100644 --- a/backend/app/gateway/routers/artifacts.py +++ b/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, diff --git a/backend/docs/blob-storage.md b/backend/docs/blob-storage.md index 83aed43acf..0d265e3c99 100644 --- a/backend/docs/blob-storage.md +++ b/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, diff --git a/backend/packages/harness/deerflow/storage/AGENTS.md b/backend/packages/harness/deerflow/storage/AGENTS.md index 0f92a3070f..d5fe2d2943 100644 --- a/backend/packages/harness/deerflow/storage/AGENTS.md +++ b/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 diff --git a/backend/tests/test_artifacts_router.py b/backend/tests/test_artifacts_router.py index e207512559..dcb902a3f9 100644 --- a/backend/tests/test_artifacts_router.py +++ b/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