fix(artifacts): bind ranged reads to validated generation (#7978)

* fix(artifacts): bind ranged reads to validated generation

* fix(artifacts): drain pinned reads without Windows hard links

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -3367,8 +3367,9 @@ external image URLs keep their existing behavior.
 
 If the sample cuts a CRLF line ending in half, the preview keeps the earlier complete rows and omits the incomplete final record, including when its last field is quoted.
 
-Text artifacts are streamed with HTTP byte-range support. The Web UI initially
-loads at most 1 MiB, shows the preview size when a file is larger, and waits for
+Text artifacts are streamed with HTTP byte-range support. A resumed download
+validates its ETag against the file version that supplies its response bytes.
+The Web UI initially loads at most 1 MiB, shows the preview size for larger files, and waits for
 an explicit **Load full file** action before fetching the remainder or mounting
 the full code editor. Editing becomes available once the complete file has loaded;
 unsaved-change detection uses the full content, including when reverting edits

backend/app/gateway/routers/AGENTS.md

@@ -1,5 +1,11 @@
 # Router contracts
 
+Artifact range validators and bytes come from one opened file. POSIX streams
+the descriptor. Windows copies only the selected response bytes into a
+private temporary file and closes the public handle before sending headers.
+Preserve the ZIP single-link checks and drain preparation before cancellation
+can release an artifact response's resources.
+
 Trash listing resolves the live retention configuration off-loop before its
 best-effort sweep. Configuration failures skip automatic cleanup and preserve
 the listing; never substitute a default policy for an unreadable one. Resolve

backend/app/gateway/routers/artifacts.py

@@ -4,17 +4,21 @@ import hashlib
 import logging
 import mimetypes
 import os
+import secrets
 import stat
 import tempfile
+import threading
 import zipfile
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from pathlib import Path
+from typing import BinaryIO
 from urllib.parse import quote
 
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import FileResponse, Response
 from pydantic import BaseModel, Field
+from starlette.datastructures import MutableHeaders
 
 from app.gateway.authz import SandboxRequestLease, require_permission, try_acquire_sandbox_for_request
 from app.gateway.deps import get_run_manager
@@ -41,6 +45,8 @@ _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-"
+_ARTIFACT_COPY_BEFORE_STREAM = os.name == "nt"
+_artifact_digest_source = threading.local()
 
 
 class ArtifactUpdateRequest(BaseModel):
@@ -334,7 +340,7 @@ def _read_artifact_payload(actual_path: Path, path: str, download: bool) -> tupl
     return ("inline_file", mime_type)
 
 
-def _sha256_of_file(path: Path) -> str:
+def _sha256_of_file(path: Path, *, pinned: tuple[BinaryIO, os.stat_result] | None = None) -> str:
     """Return the hex SHA-256 digest of *path* without loading it whole.
 
     Computing the digest on the Gateway lets the browser skip its own
@@ -346,45 +352,217 @@ def _sha256_of_file(path: Path) -> str:
     requests a browser issues while scrubbing/paginating a preview do not each
     re-hash a potentially huge artifact from scratch (raised in PR review).
     """
-    stat = path.stat()
+    file_stat = path.stat() if pinned is None else pinned[1]
     # Sandbox syncs and other writers can atomically replace a same-size file
     # while preserving its mtime. Identity separates those generations; ctime
     # also invalidates in-place writes where the filesystem exposes change time.
-    return _sha256_of_file_cached(
+    key = (
         str(path),
-        stat.st_dev,
-        stat.st_ino,
-        stat.st_ctime_ns,
-        stat.st_mtime_ns,
-        stat.st_size,
+        file_stat.st_dev,
+        file_stat.st_ino,
+        file_stat.st_ctime_ns,
+        file_stat.st_mtime_ns,
+        file_stat.st_size,
     )
+    if pinned is None:
+        return _sha256_of_file_cached(*key)
+
+    previous = getattr(_artifact_digest_source, "pinned", None)
+    _artifact_digest_source.pinned = (key, pinned[0])
+    try:
+        return _sha256_of_file_cached(*key)
+    finally:
+        _artifact_digest_source.pinned = previous
 
 
 @functools.lru_cache(maxsize=256)
 def _sha256_of_file_cached(path: str, device: int, inode: int, ctime_ns: int, mtime_ns: int, size: int) -> str:
     """Cached SHA-256 of *path*; metadata arguments separate file generations."""
-    digest = hashlib.sha256()
-    with open(path, "rb") as handle:
-        for chunk in iter(lambda: handle.read(1 << 20), b""):
-            digest.update(chunk)
-    return digest.hexdigest()
+    pinned = getattr(_artifact_digest_source, "pinned", None)
+    # A nested lookup for a different file must hash its own bytes.
+    if pinned is None or pinned[0] != (path, device, inode, ctime_ns, mtime_ns, size):
+        with open(path, "rb") as handle:
+            return hashlib.file_digest(handle, "sha256").hexdigest()
+
+    handle = pinned[1]
+    previous_offset = handle.tell()
+    try:
+        handle.seek(0)
+        return hashlib.file_digest(handle, "sha256").hexdigest()
+    finally:
+        handle.seek(previous_offset)
 
 
 def _artifact_etag(path: Path) -> str:
-    """Keep content revisions for editable files and metadata-only tags for large files."""
+    """Keep content revisions for editable files and metadata tags for large files."""
     file_stat = path.stat()
     if file_stat.st_size <= MAX_EDITABLE_ARTIFACT_BYTES:
-        return f'"{_sha256_of_file(path)}"'
+        # Populate the shared cache only with metadata from the hashed handle.
+        with path.open("rb") as handle:
+            return _artifact_etag_from_handle(path, handle, os.fstat(handle.fileno()))
+    return _artifact_stat_etag(file_stat)
 
+
+def _artifact_stat_etag(file_stat: os.stat_result) -> str:
     # Do not read oversized artifacts just to validate a small byte range.
     # The prefix keeps this opaque validator distinct from an editable SHA-256.
     identity = f"{file_stat.st_dev}:{file_stat.st_ino}:{file_stat.st_ctime_ns}:{file_stat.st_mtime_ns}:{file_stat.st_size}"
     return f'"stat-{hashlib.sha256(identity.encode("ascii")).hexdigest()}"'
 
 
+def _artifact_etag_from_handle(path: Path, handle: BinaryIO, file_stat: os.stat_result) -> str:
+    """Build the validator from the same file generation the response will read."""
+    if file_stat.st_size <= MAX_EDITABLE_ARTIFACT_BYTES:
+        digest = _sha256_of_file(path, pinned=(handle, file_stat))
+        return f'"{digest}"'
+    return _artifact_stat_etag(file_stat)
+
+
+def _prepare_artifact_stream(path: Path) -> tuple[BinaryIO, os.stat_result, str]:
+    """Pin one file generation before response headers authorize a byte range."""
+    handle = path.open("rb")
+    try:
+        file_stat = os.fstat(handle.fileno())
+        if not stat.S_ISREG(file_stat.st_mode):
+            raise RuntimeError(f"File at path {path} is not a file.")
+        return handle, file_stat, _artifact_etag_from_handle(path, handle, file_stat)
+    except BaseException:
+        _close_artifact_handle(handle)
+        raise
+
+
+def _copy_artifact_ranges(handle: BinaryIO, ranges: list[tuple[int, int]]) -> BinaryIO:
+    """Pack the selected ranges into a private temporary file in send order."""
+    snapshot = tempfile.TemporaryFile(mode="w+b")
+    try:
+        for start, end in ranges:
+            handle.seek(start)
+            remaining = end - start
+            while remaining:
+                chunk = handle.read(min(1 << 20, remaining))
+                if not chunk:
+                    raise OSError("Artifact was truncated while preparing the response")
+                snapshot.write(chunk)
+                remaining -= len(chunk)
+        snapshot.seek(0)
+        return snapshot
+    except BaseException:
+        _close_artifact_handle(snapshot)
+        raise
+
+
+def _close_artifact_handle(handle: BinaryIO) -> None:
+    try:
+        handle.close()
+    except OSError:
+        logger.warning("Could not close artifact response handle", exc_info=True)
+
+
 class _ArtifactFileResponse(FileResponse):
     """Require an ETag to authorize conditional ranges of replaceable artifacts."""
 
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._artifact_handle: BinaryIO | None = None
+
+    async def __call__(self, scope, receive, send) -> None:
+        async def prepare() -> None:
+            self._artifact_handle, self.stat_result, etag = await asyncio.to_thread(_prepare_artifact_stream, Path(self.path))
+            self.headers["etag"] = etag
+            self.set_stat_headers(self.stat_result)
+
+        try:
+            await await_drained(prepare())
+            await super().__call__(scope, receive, send)
+        finally:
+            if self._artifact_handle is not None:
+                handle, self._artifact_handle = self._artifact_handle, None
+                await await_drained(asyncio.to_thread(_close_artifact_handle, handle))
+
+    async def _snapshot_windows_body(self, ranges: list[tuple[int, int]], send_header_only: bool) -> bool:
+        if not _ARTIFACT_COPY_BEFORE_STREAM:
+            return False
+
+        source = self._artifact_handle
+        assert source is not None
+
+        async def snapshot() -> None:
+            # Windows replacement needs the public handle closed. A private
+            # copy also leaves the artifact's link count unchanged for ZIP.
+            self._artifact_handle = None if send_header_only else await asyncio.to_thread(_copy_artifact_ranges, source, ranges)
+            await asyncio.to_thread(_close_artifact_handle, source)
+
+        await await_drained(snapshot())
+        return True
+
+    async def _handle_simple(self, send, send_header_only: bool, send_pathsend: bool) -> None:
+        assert self.stat_result is not None
+        await self._snapshot_windows_body([(0, self.stat_result.st_size)], send_header_only)
+
+        await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
+        if send_header_only:
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+            return
+
+        handle = self._artifact_handle
+        assert handle is not None
+        await await_drained(asyncio.to_thread(handle.seek, 0))
+        more_body = True
+        while more_body:
+            chunk = await await_drained(asyncio.to_thread(handle.read, self.chunk_size))
+            more_body = len(chunk) == self.chunk_size
+            await send({"type": "http.response.body", "body": chunk, "more_body": more_body})
+
+    async def _handle_single_range(self, send, start: int, end: int, file_size: int, send_header_only: bool) -> None:
+        snapshot = await self._snapshot_windows_body([(start, end)], send_header_only)
+
+        headers = MutableHeaders(raw=list(self.raw_headers))
+        headers["content-range"] = f"bytes {start}-{end - 1}/{file_size}"
+        headers["content-length"] = str(end - start)
+        await send({"type": "http.response.start", "status": 206, "headers": headers.raw})
+        if send_header_only:
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+            return
+
+        handle = self._artifact_handle
+        assert handle is not None
+        if not snapshot:
+            await await_drained(asyncio.to_thread(handle.seek, start))
+        more_body = True
+        while more_body:
+            chunk = await await_drained(asyncio.to_thread(handle.read, min(self.chunk_size, end - start)))
+            start += len(chunk)
+            more_body = len(chunk) == self.chunk_size and start < end
+            await send({"type": "http.response.body", "body": chunk, "more_body": more_body})
+
+    async def _handle_multiple_ranges(self, send, ranges: list[tuple[int, int]], file_size: int, send_header_only: bool) -> None:
+        snapshot = await self._snapshot_windows_body(ranges, send_header_only)
+
+        boundary = secrets.token_hex(13)
+        content_length, header_generator = self.generate_multipart(ranges, boundary, file_size, self.headers["content-type"])
+        headers = MutableHeaders(raw=list(self.raw_headers))
+        headers["content-type"] = f"multipart/byteranges; boundary={boundary}"
+        headers["content-length"] = str(content_length)
+        await send({"type": "http.response.start", "status": 206, "headers": headers.raw})
+        if send_header_only:
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+            return
+
+        handle = self._artifact_handle
+        assert handle is not None
+        for start, end in ranges:
+            await send({"type": "http.response.body", "body": header_generator(start, end), "more_body": True})
+            if not snapshot:
+                await await_drained(asyncio.to_thread(handle.seek, start))
+            while start < end:
+                chunk = await await_drained(asyncio.to_thread(handle.read, min(self.chunk_size, end - start)))
+                if not chunk:
+                    raise OSError("Artifact was truncated while streaming a byte range")
+                start += len(chunk)
+                await send({"type": "http.response.body", "body": chunk, "more_body": True})
+            await send({"type": "http.response.body", "body": b"\r\n", "more_body": True})
+        await send({"type": "http.response.body", "body": f"--{boundary}--".encode("latin-1"), "more_body": False})
+
     def _should_use_range(self, http_if_range: str) -> bool:
         # Sandbox syncs can preserve mtimes across replacements, so Last-Modified
         # is not a strong validator. Date-form If-Range must return the full file.

backend/tests/test_artifacts_router.py

@@ -18,6 +18,7 @@ from starlette.requests import Request
 from starlette.responses import FileResponse
 
 import app.gateway.routers.artifacts as artifacts_router
+from app.gateway.artifact_archive import build_artifact_archive
 from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE
 from deerflow.config.paths import Paths, make_safe_user_id
 from deerflow.sandbox.lease import get_sandbox_lease_manager
@@ -1041,6 +1042,134 @@ def test_replaced_artifact_rejects_previous_if_range_validator(tmp_path, monkeyp
     assert response.headers["etag"] == f'"{hashlib.sha256(b"new report").hexdigest()}"'
 
 
+@pytest.mark.parametrize("ranges", ["bytes=4-9", "bytes=0-1,4-5"], ids=["single", "multipart"])
+@pytest.mark.parametrize(
+    "copy_before_stream",
+    [pytest.param(False, marks=pytest.mark.skipif(os.name == "nt", reason="Windows requires a private response copy")), True],
+    ids=["descriptor", "private-copy"],
+)
+def test_if_range_stream_keeps_the_validated_artifact_generation(tmp_path, monkeypatch, ranges, copy_before_stream) -> None:
+    outputs = tmp_path / "outputs"
+    outputs.mkdir()
+    artifact_path = outputs / "note.txt"
+    original = b"old lesson"
+    artifact_path.write_bytes(original)
+    artifacts_router._sha256_of_file_cached.cache_clear()
+    monkeypatch.setattr(artifacts_router, "_ARTIFACT_COPY_BEFORE_STREAM", copy_before_stream)
+    monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda _thread_id, _path, user_id=None: artifact_path)
+    app = make_authed_test_app()
+    app.include_router(artifacts_router.router)
+    url = "/api/threads/thread-1/artifacts/mnt/user-data/outputs/note.txt"
+
+    replace_after_headers = False
+    replacement_confirmed = False
+
+    async def replace_between_headers_and_body(scope, receive, send):
+        async def send_and_replace(message):
+            nonlocal replacement_confirmed
+            await send(message)
+            if replace_after_headers and message["type"] == "http.response.start":
+                archive = await asyncio.to_thread(build_artifact_archive, outputs, ["mnt/user-data/outputs/note.txt"], user_data_dir=tmp_path)
+                with archive.file, zipfile.ZipFile(archive.file) as zipped:
+                    assert zipped.read("note.txt") == original
+                _replace_preserving_artifact_mtime(artifact_path, b"new slides")
+                replacement_confirmed = True
+
+        await app(scope, receive, send_and_replace)
+
+    with TestClient(replace_between_headers_and_body) as client:
+        first = client.get(url)
+        replace_after_headers = True
+        resumed = client.get(url, headers={"Range": ranges, "If-Range": first.headers["etag"]})
+
+    assert replacement_confirmed
+    assert artifact_path.read_bytes() == b"new slides"
+    assert resumed.status_code == 206
+    assert resumed.headers["etag"] == first.headers["etag"]
+    assert int(resumed.headers["content-length"]) == len(resumed.content)
+    if ranges == "bytes=4-9":
+        assert resumed.content == b"lesson"
+        assert resumed.headers["content-range"] == "bytes 4-9/10"
+    else:
+        media_type, _, boundary = resumed.headers["content-type"].partition("; boundary=")
+        assert media_type == "multipart/byteranges" and boundary
+        parts = []
+        for start, end in [(0, 2), (4, 6)]:
+            part_header = f"--{boundary}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Range: bytes {start}-{end - 1}/10\r\n\r\n"
+            parts.append(part_header.encode("ascii") + original[start:end] + b"\r\n")
+        assert resumed.content == b"".join(parts) + f"--{boundary}--".encode("ascii")
+    assert list(outputs.iterdir()) == [artifact_path]
+
+
+@pytest.mark.parametrize("stage", ["_prepare_artifact_stream", "_copy_artifact_ranges"])
+def test_cancelled_artifact_response_drains_file_ownership(tmp_path, monkeypatch, stage) -> None:
+    artifact_path = tmp_path / "note.txt"
+    artifact_path.write_bytes(b"original")
+    monkeypatch.setattr(artifacts_router, "_ARTIFACT_COPY_BEFORE_STREAM", True)
+    worker = getattr(artifacts_router, stage)
+    release_worker = threading.Event()
+    handles = []
+
+    async def exercise():
+        loop = asyncio.get_running_loop()
+        started = asyncio.Event()
+        finished = asyncio.Event()
+
+        def gated_worker(*args):
+            result = worker(*args)
+            handles.append(result[0] if stage == "_prepare_artifact_stream" else result)
+            loop.call_soon_threadsafe(started.set)
+            try:
+                assert release_worker.wait(5)
+                return result
+            finally:
+                loop.call_soon_threadsafe(finished.set)
+
+        async def send(_message):
+            pytest.fail("Cancellation before preparation completes must send no headers or body")
+
+        monkeypatch.setattr(artifacts_router, stage, gated_worker)
+        response = artifacts_router._ArtifactFileResponse(artifact_path)
+        scope = {"type": "http", "method": "GET", "headers": [(b"range", b"bytes=1-2")]}
+        task = asyncio.create_task(response(scope, AsyncMock(), send))
+        try:
+            await asyncio.wait_for(started.wait(), timeout=5)
+            task.cancel()
+            await asyncio.sleep(0)
+            assert not task.done()
+        finally:
+            release_worker.set()
+            if started.is_set():
+                await asyncio.wait_for(finished.wait(), timeout=5)
+            await asyncio.gather(task, return_exceptions=True)
+        assert task.cancelled()
+        assert len(handles) == 1 and handles[0].closed
+        assert artifact_path.stat().st_nlink == 1
+
+    asyncio.run(exercise())
+
+
+def test_pinned_digest_cannot_populate_another_artifacts_cache_key(tmp_path, monkeypatch) -> None:
+    first = tmp_path / "first.txt"
+    other = tmp_path / "other.txt"
+    first.write_bytes(b"first file")
+    other.write_bytes(b"other file")
+    cached = artifacts_router._sha256_of_file_cached
+    cached.cache_clear()
+
+    def nested_lookup(path, *metadata):
+        if path == str(first):
+            assert artifacts_router._sha256_of_file(other) == hashlib.sha256(b"other file").hexdigest()
+        return cached(path, *metadata)
+
+    monkeypatch.setattr(artifacts_router, "_sha256_of_file_cached", nested_lookup)
+    with first.open("rb") as handle:
+        handle.seek(2)
+        assert artifacts_router._artifact_etag_from_handle(first, handle, os.fstat(handle.fileno())) == f'"{hashlib.sha256(b"first file").hexdigest()}"'
+        assert handle.tell() == 2
+    assert artifacts_router._sha256_of_file(other) == hashlib.sha256(b"other file").hexdigest()
+
+
 @pytest.mark.parametrize("changed_field", ["st_dev", "st_ino", "st_ctime_ns"])
 def test_artifact_digest_invalidates_when_file_identity_changes(tmp_path, monkeypatch, changed_field) -> None:
     artifact_path = tmp_path / "note.txt"