From 8d8ca506bafba4b91a5057238605b941f116c83e Mon Sep 17 00:00:00 2001 From: Ryker_Feng <90562015+18062706139fcz@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:01:21 +0800 Subject: [PATCH] feat(artifacts): download run files as zip (#5117) * feat(artifacts): download run files as zip * fix(artifacts): address archive review feedback * fix(artifacts): gate unavailable archive downloads * fix(artifacts): verify archive availability * fix(artifacts): harden archive consistency * fix(artifacts): reject archive path aliases --- README.md | 2 +- backend/app/gateway/AGENTS.md | 4 +- backend/app/gateway/artifact_archive.py | 270 ++++++++ backend/app/gateway/auth/pat.py | 1 + backend/app/gateway/routers/thread_runs.py | 161 ++++- .../harness/deerflow/runtime/journal.py | 15 +- .../harness/deerflow/runtime/runs/schemas.py | 1 + .../blocking_io/test_artifact_archive.py | 34 + backend/tests/test_artifact_archive.py | 614 ++++++++++++++++++ backend/tests/test_run_journal.py | 28 + .../[agent_name]/chats/[thread_id]/page.tsx | 3 + .../artifacts/artifact-file-list.tsx | 183 ++++-- .../components/workspace/chats/chat-page.tsx | 3 + .../workspace/messages/message-list.tsx | 23 +- frontend/src/core/artifacts/api.ts | 55 +- frontend/src/core/artifacts/utils.ts | 10 + frontend/src/core/i18n/locales/en-US.ts | 8 + frontend/src/core/i18n/locales/types.ts | 6 + frontend/src/core/i18n/locales/zh-CN.ts | 7 + .../src/core/messages/artifact-archive.ts | 32 + .../src/core/threads/thread-branch-tree.ts | 8 +- .../artifacts/artifact-file-list.dom.test.tsx | 299 +++++++++ .../tests/unit/core/artifacts/api.test.ts | 52 +- .../core/messages/artifact-archive.test.ts | 65 ++ .../core/threads/thread-branch-tree.test.ts | 23 +- 25 files changed, 1840 insertions(+), 67 deletions(-) create mode 100644 backend/app/gateway/artifact_archive.py create mode 100644 backend/tests/blocking_io/test_artifact_archive.py create mode 100644 backend/tests/test_artifact_archive.py create mode 100644 frontend/src/core/messages/artifact-archive.ts create mode 100644 frontend/tests/unit/components/workspace/artifacts/artifact-file-list.dom.test.tsx create mode 100644 frontend/tests/unit/core/messages/artifact-archive.test.ts diff --git a/README.md b/README.md index 943cbef3c..89e2123d4 100644 --- a/README.md +++ b/README.md @@ -1274,7 +1274,7 @@ Image bytes loaded for a vision-model call are transient: DeerFlow removes the h After each run, DeerFlow records a workspace change summary for the run-owned `workspace` and `outputs` directories. The Web UI shows a compact "files changed" badge on the assistant turn; opening it reveals created, modified, and deleted files with text diffs when safe to display. Uploads are excluded because they are user inputs, not agent-generated changes, and stdio MCP temporary/debug files under the DeerFlow-owned `.mcp/` namespace are excluded because they are process-internal state (like `.git/` and `node_modules/`, any directory named `.mcp` is excluded at any depth). Large, binary, or sensitive-looking files are shown as metadata only. -Files presented through `present_files` remain part of the thread's artifact state, and the Web UI restores the artifact panel and selected document after a page refresh. The currently selected formal artifact is refreshed once when the run finishes so edits become visible without a manual reload. Existing UTF-8 text artifacts under `/mnt/user-data/outputs` can also be edited and explicitly saved from the panel on Unix and Windows while the thread is idle; saves use content revisions to prevent overwriting agent changes. +Files presented through `present_files` remain part of the thread's artifact state, and the Web UI restores the artifact panel and selected document after a page refresh. When a completed response successfully presents between 2 and 50 files, its final file card also offers one ZIP download. Archive membership comes from the terminal delivery receipt rather than browser-supplied paths, and the ZIP contains the current file versions, which may have changed since the response. The currently selected formal artifact is refreshed once when the run finishes so edits become visible without a manual reload. Existing UTF-8 text artifacts under `/mnt/user-data/outputs` can also be edited and explicitly saved from the panel on Unix and Windows while the thread is idle; saves use content revisions to prevent overwriting agent changes. 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 diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index e0f001ce7..252e8f3f2 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -57,7 +57,7 @@ owner-scoped assistant version selection remains enabled. | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update explicitly. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | -| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty event feed from an existing checkpoint so legacy checkpoint-only history keeps earlier thread-global ordering and stays visible; skip without a checkpoint or when the feed is populated. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. | +| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. | | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | | **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback | | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | @@ -152,7 +152,7 @@ JSONL event stores when `GATEWAY_WORKERS > 1`. - Store-only hydrated runs are readable history. In multi-worker mode with heartbeat enabled, cancel on a store-only run records `runs.cancel_action` / `cancel_requested_at` while the owner's lease is live; the first action wins even if a retry later lands on the owner. `RunStore.request_cancel()` and owner completion through `finalize_if_not_cancelled()` are competing active-row CAS operations, so an accepted cancel cannot be overwritten by a later success. `RunStore.renew_lease()` renews and observes the request atomically in the SQL implementation. The owner then executes the normal process-local interrupt/rollback and terminal stream path without transferring the lease. An expired owner is still taken over and marked `error`. `wait=true` and cancel-then-stream use the shared bridge to observe owner finalization; a non-standard process-local bridge returns accepted 202 instead of subscribing to an unreachable stream. In single-worker mode (heartbeat off), store-only runs still return 409. - A local worker's `RunRecord.lease_expires_at` is the last durably confirmed ownership deadline. `_renew_leases()` bounds each renewal attempt by that deadline: transient store exceptions remain retryable while it is valid, but an exception or blocked call that reaches expiry sets the process-local `ownership_lost` fence, raises `abort_event`, and cancels the run task. Successful renewals collect durable cancellation actions; after all local renewals have been attempted, heartbeat only signals the corresponding process-local tasks, leaving status writes and rollback cleanup to the worker finalization path. Fenced workers do not perform subsequent journal/delivery-receipt, progress/completion/status, checkpoint/thread-metadata, or `on_run_completed` writes; the peer recovery path owns the terminal receipt. `RunStore.update_run_completion()` also refuses to replace a different terminal status, closing the peer-takeover/late-finalization race. `grace_seconds` delays peer reclamation for clock skew but is not extra execution time for an owner that can no longer confirm its lease. Already-committed remote tool side effects remain outside this local cancellation boundary. - Startup/orphan reconciliation must claim stale active rows with `RunStore.claim_for_takeover()`, not a plain `update_status()`. The final claim re-checks `status` and lease expiry atomically, so a heartbeat renewal between the candidate scan and the recovery write keeps the run active. -- Run admission and independent writes are first-class thread operations. `runs.operation_kind` distinguishes `run` from `checkpoint_write`, `artifact_write`, `branch`, and `delete`; every active kind shares the durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds. +- Run admission and independent writes are first-class thread operations. `runs.operation_kind` distinguishes `run` from `checkpoint_write`, `artifact_write`, `artifact_archive`, `branch`, and `delete`; every active kind shares the durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds. - Gateway checkpoint mutations outside run execution must use `services.reserve_checkpoint_write()`, which composes the process-local thread lock with the durable `checkpoint_write` reservation. Manual compaction, `POST /threads/{id}/state`, and both goal mutation routes (`PUT` / `DELETE /threads/{id}/goal`, including creation of a missing goal checkpoint) use this boundary, so an existing run blocks the write and the reservation blocks new reject/interrupt/rollback runs across workers. - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy. diff --git a/backend/app/gateway/artifact_archive.py b/backend/app/gateway/artifact_archive.py new file mode 100644 index 000000000..e3fc88250 --- /dev/null +++ b/backend/app/gateway/artifact_archive.py @@ -0,0 +1,270 @@ +"""Fail-closed ZIP construction for files presented by one run.""" + +from __future__ import annotations + +import os +import stat +import tempfile +import time +import unicodedata +import zipfile +from collections.abc import Iterable +from dataclasses import dataclass +from hashlib import sha256 +from pathlib import Path +from typing import BinaryIO + +from deerflow.constants import BROWSER_FRAMES_DIRNAME, TOOL_RESULTS_DIRNAME + +_VIRTUAL_PREFIX = "mnt/user-data/outputs/" +_EDIT_TEMP_PREFIX = ".artifact-edit-" +_ALLOWED_FORMAT_CHARS = frozenset({"\u200c", "\u200d"}) +_WINDOWS_INVALID_CHARS = frozenset('<>:"|?*') +_WINDOWS_DEVICE_NAMES = frozenset({"con", "prn", "aux", "nul"} | {f"com{number}" for number in range(1, 10)} | {f"lpt{number}" for number in range(1, 10)}) +MAX_FILES = 50 +MAX_FILE_BYTES = 50 * 1024 * 1024 +MAX_TOTAL_BYTES = 100 * 1024 * 1024 +MAX_ENTRY_BYTES = 1024 +BUILD_TIMEOUT_SECONDS = 60.0 +_CHUNK_BYTES = 1024 * 1024 + + +class ArtifactArchiveError(ValueError): + def __init__(self, detail: str, status_code: int = 409) -> None: + super().__init__(detail) + self.detail = detail + self.status_code = status_code + + +@dataclass(frozen=True) +class ArtifactArchiveResult: + file: BinaryIO + size: int + member_count: int + input_bytes: int + + +@dataclass(frozen=True) +class _ArchiveMember: + path: Path + entry: str + initial: os.stat_result + components: tuple[tuple[Path, int, int], ...] + + +def _reject() -> ArtifactArchiveError: + return ArtifactArchiveError("The files listed by this response are not available for archive download") + + +def _too_large(detail: str) -> ArtifactArchiveError: + return ArtifactArchiveError(detail, 413) + + +def _check_deadline(deadline: float) -> None: + if time.monotonic() > deadline: + raise ArtifactArchiveError("Artifact archive creation timed out", 503) + + +def _is_link_like(path: Path, metadata: os.stat_result) -> bool: + return stat.S_ISLNK(metadata.st_mode) or path.is_junction() + + +def _member( + root: Path, + virtual_path: str, + reserved: frozenset[str], + deadline: float, + root_components: tuple[tuple[Path, int, int], ...], +) -> _ArchiveMember: + _check_deadline(deadline) + if not virtual_path or virtual_path.startswith("//") or "\\" in virtual_path or "\x00" in virtual_path: + raise _reject() + stripped = virtual_path.removeprefix("/") + if not stripped.startswith(_VIRTUAL_PREFIX): + raise _reject() + + parts = stripped.removeprefix(_VIRTUAL_PREFIX).split("/") + if any(part in {"", ".", ".."} for part in parts): + raise _reject() + if any(any(char in _WINDOWS_INVALID_CHARS for char in part) or part.endswith((" ", ".")) or part.split(".", 1)[0].rstrip().casefold() in _WINDOWS_DEVICE_NAMES for part in parts): + raise _reject() + if any(any(unicodedata.category(char).startswith("C") and char not in _ALLOWED_FORMAT_CHARS for char in part) for part in parts): + raise _reject() + if any(part.casefold() in reserved or part.casefold().startswith(_EDIT_TEMP_PREFIX) for part in parts): + raise _reject() + if any(part.casefold().endswith(".skill") for part in parts[:-1]): + raise _reject() + + entry = "/".join(parts) + if len(entry.encode()) > MAX_ENTRY_BYTES: + raise _too_large("An artifact path is too long to include in an archive") + + candidate = root.joinpath(*parts) + current = root + components = list(root_components) + try: + for part in parts: + _check_deadline(deadline) + current /= part + metadata = os.lstat(current) + if _is_link_like(current, metadata): + raise _reject() + components.append((current, metadata.st_dev, metadata.st_ino)) + initial = os.lstat(candidate) + if not stat.S_ISREG(initial.st_mode) or initial.st_nlink != 1: + raise _reject() + resolved = candidate.resolve(strict=True) + resolved.relative_to(root) + except ArtifactArchiveError: + raise + except (OSError, ValueError) as exc: + raise _reject() from exc + _check_deadline(deadline) + return _ArchiveMember(resolved, entry, initial, tuple(components)) + + +def _hash_descriptor(descriptor: int, size: int, deadline: float) -> bytes: + digest = sha256() + remaining = size + try: + os.lseek(descriptor, 0, os.SEEK_SET) + while remaining: + _check_deadline(deadline) + chunk = os.read(descriptor, min(_CHUNK_BYTES, remaining)) + if not chunk: + raise _reject() + digest.update(chunk) + remaining -= len(chunk) + if os.read(descriptor, 1): + raise _reject() + except OSError as exc: + raise _reject() from exc + return digest.digest() + + +def _copy_member( + archive: zipfile.ZipFile, + member: _ArchiveMember, + deadline: float, + remaining_total_bytes: int, +) -> int: + _check_deadline(deadline) + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(member.path, flags) + except OSError as exc: + raise _reject() from exc + + try: + _check_deadline(deadline) + before = os.fstat(descriptor) + identity = (before.st_dev, before.st_ino) + if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or identity != (member.initial.st_dev, member.initial.st_ino): + raise _reject() + if before.st_size > MAX_FILE_BYTES: + raise _too_large(f"Each archived artifact must be at most {MAX_FILE_BYTES} bytes") + if before.st_size > remaining_total_bytes: + raise _too_large(f"Archived artifacts must total at most {MAX_TOTAL_BYTES} bytes") + + info = zipfile.ZipInfo(member.entry) + info.create_system = 0 + info.compress_type = zipfile.ZIP_STORED + remaining = before.st_size + copied_digest = sha256() + with archive.open(info, "w", force_zip64=False) as destination: + while remaining: + _check_deadline(deadline) + chunk = os.read(descriptor, min(_CHUNK_BYTES, remaining)) + if not chunk: + raise _reject() + destination.write(chunk) + copied_digest.update(chunk) + remaining -= len(chunk) + if os.read(descriptor, 1): + raise _reject() + + _check_deadline(deadline) + after = os.fstat(descriptor) + if after.st_nlink != 1 or (after.st_dev, after.st_ino) != identity or (after.st_size, after.st_mtime_ns) != (before.st_size, before.st_mtime_ns): + raise _reject() + if _hash_descriptor(descriptor, before.st_size, deadline) != copied_digest.digest(): + raise _reject() + after_verification = os.fstat(descriptor) + if after_verification.st_nlink != 1 or (after_verification.st_dev, after_verification.st_ino) != identity or (after_verification.st_size, after_verification.st_mtime_ns) != (before.st_size, before.st_mtime_ns): + raise _reject() + try: + for component, device, inode in member.components: + current = os.lstat(component) + if _is_link_like(component, current) or (current.st_dev, current.st_ino) != (device, inode): + raise _reject() + except OSError as exc: + raise _reject() from exc + return before.st_size + finally: + os.close(descriptor) + + +def build_artifact_archive( + outputs_dir: Path, + virtual_paths: Iterable[str], + *, + user_data_dir: Path, + extra_reserved_dir_names: Iterable[str] = (), +) -> ArtifactArchiveResult: + deadline = time.monotonic() + BUILD_TIMEOUT_SECONDS + try: + if outputs_dir.parent != user_data_dir: + raise _reject() + user_data_metadata = os.lstat(user_data_dir) + outputs_metadata = os.lstat(outputs_dir) + if _is_link_like(user_data_dir, user_data_metadata) or not stat.S_ISDIR(user_data_metadata.st_mode) or _is_link_like(outputs_dir, outputs_metadata) or not stat.S_ISDIR(outputs_metadata.st_mode): + raise _reject() + user_data_root = user_data_dir.resolve(strict=True) + root = outputs_dir.resolve(strict=True) + if root.parent != user_data_root: + raise _reject() + except ArtifactArchiveError: + raise + except OSError as exc: + raise _reject() from exc + _check_deadline(deadline) + + root_components = ( + (user_data_dir, user_data_metadata.st_dev, user_data_metadata.st_ino), + (outputs_dir, outputs_metadata.st_dev, outputs_metadata.st_ino), + ) + + paths = list(dict.fromkeys(virtual_paths)) + if not paths: + raise _reject() + if len(paths) > MAX_FILES: + raise _too_large(f"An artifact archive can contain at most {MAX_FILES} files") + + reserved = frozenset(name.casefold() for name in {BROWSER_FRAMES_DIRNAME, TOOL_RESULTS_DIRNAME, *extra_reserved_dir_names}) + members = [_member(root, path, reserved, deadline, root_components) for path in paths] + collision_keys = [unicodedata.normalize("NFC", member.entry).casefold() for member in members] + if len(collision_keys) != len(set(collision_keys)): + raise _reject() + + sizes = [member.initial.st_size for member in members] + if any(size > MAX_FILE_BYTES for size in sizes): + raise _too_large(f"Each archived artifact must be at most {MAX_FILE_BYTES} bytes") + if sum(sizes) > MAX_TOTAL_BYTES: + raise _too_large(f"Archived artifacts must total at most {MAX_TOTAL_BYTES} bytes") + + _check_deadline(deadline) + output = tempfile.TemporaryFile("w+b") + try: + if hasattr(os, "fchmod"): + os.fchmod(output.fileno(), 0o600) + input_bytes = 0 + with zipfile.ZipFile(output, "w", zipfile.ZIP_STORED, allowZip64=False) as archive: + for member in members: + input_bytes += _copy_member(archive, member, deadline, MAX_TOTAL_BYTES - input_bytes) + _check_deadline(deadline) + size = output.tell() + output.seek(0) + return ArtifactArchiveResult(output, size, len(members), input_bytes) + except Exception: + output.close() + raise diff --git a/backend/app/gateway/auth/pat.py b/backend/app/gateway/auth/pat.py index 6e7fe4fec..34c97e9d2 100644 --- a/backend/app/gateway/auth/pat.py +++ b/backend/app/gateway/auth/pat.py @@ -74,6 +74,7 @@ _PAT_ROUTE_RULES: tuple[tuple[frozenset[str], re.Pattern[str]], ...] = ( frozenset({"GET"}), re.compile(r"^/api/threads/[^/]+/runs/[^/]+/(join|messages|events|workspace-changes)$"), ), + (frozenset({"GET", "POST"}), re.compile(r"^/api/threads/[^/]+/runs/[^/]+/artifacts/archive$")), (frozenset({"GET", "POST"}), re.compile(r"^/api/threads/[^/]+/runs/[^/]+/stream$")), (frozenset({"POST"}), re.compile(r"^/api/runs/(stream|wait)$")), (frozenset({"GET"}), re.compile(r"^/api/runs/[^/]+/(messages|feedback)$")), diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 66b5b3d3a..a274bc5ed 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -13,6 +13,7 @@ from __future__ import annotations import asyncio import logging +import re import uuid from copy import deepcopy from datetime import UTC, datetime @@ -22,7 +23,9 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import Response, StreamingResponse from langchain_core.messages import BaseMessage from pydantic import BaseModel, Field +from starlette.background import BackgroundTask +from app.gateway.artifact_archive import ArtifactArchiveError, ArtifactArchiveResult, build_artifact_archive from app.gateway.authz import require_cancel_permission_if, require_permission from app.gateway.checkpoint_lineage import ( CheckpointLineageError, @@ -35,13 +38,17 @@ from app.gateway.checkpoint_lineage import ( ) from app.gateway.context_usage import build_context_usage from app.gateway.deps import get_current_user, get_feedback_repo, get_run_event_store, get_run_manager, get_run_store, get_stream_bridge +from app.gateway.internal_auth import get_trusted_internal_owner_user_id from app.gateway.pagination import trim_run_message_page from app.gateway.run_models import RunCreateRequest from app.gateway.services import build_checkpoint_state_accessor, build_thread_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion from app.gateway.utils import sanitize_log_param from deerflow.agents.middlewares.dynamic_context_middleware import strip_injected_user_message_id_suffix -from deerflow.runtime import CancelOutcome, RunRecord, RunStatus, serialize_channel_values_for_api +from deerflow.authz.sandbox_authz import safe_app_config_async +from deerflow.config.paths import get_paths, make_safe_user_id +from deerflow.runtime import CancelOutcome, ConflictError, RunRecord, RunStatus, ThreadOperationKind, serialize_channel_values_for_api from deerflow.runtime.secret_context import redact_config_secrets, redact_metadata_secrets +from deerflow.runtime.user_context import get_effective_user_id from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text from deerflow.utils.thread_id import ThreadId from deerflow.workspace_changes import get_workspace_changes_response @@ -53,6 +60,7 @@ REGENERATE_HISTORY_SCAN_LIMIT = 200 # (one per successful run in steady state) consume roughly half of history. REGENERATE_HISTORY_RAW_SCAN_LIMIT = REGENERATE_HISTORY_SCAN_LIMIT * 2 THREAD_MESSAGE_PAGE_SCAN_BATCH = 201 +_artifact_archive_slots = asyncio.Semaphore(4) _MISSING_REGENERATE_BASE_DETAIL = "Could not find an addressable checkpoint before the target user message" _UNSAFE_REGENERATE_LINEAGE_DETAIL = "Could not safely resolve the checkpoint before the target user message" THREAD_MESSAGE_LEGACY_SCAN_BATCH = 201 @@ -168,6 +176,10 @@ class RunResponse(BaseModel): stop_reason: str | None = None +class ArtifactArchiveManifestResponse(BaseModel): + file_count: int + + class ThreadTokenUsageModelBreakdown(BaseModel): tokens: int = 0 runs: int = Field( @@ -1462,6 +1474,153 @@ async def list_run_messages( return {"data": data, "has_more": has_more} +def _archive_response_chunks(result: ArtifactArchiveResult): + try: + while chunk := result.file.read(1024 * 1024): + yield chunk + finally: + result.file.close() + + +async def _build_archive_without_abandoning_worker( + outputs_dir, + user_data_dir, + presented_paths: list[str], + *, + extra_reserved_dir_names: set[str], +) -> ArtifactArchiveResult: + if _artifact_archive_slots.locked(): + raise ArtifactArchiveError("Too many artifact archives are being created; try again shortly", 429) + await _artifact_archive_slots.acquire() + build_task = asyncio.create_task( + asyncio.to_thread( + build_artifact_archive, + outputs_dir, + presented_paths, + user_data_dir=user_data_dir, + extra_reserved_dir_names=extra_reserved_dir_names, + ) + ) + try: + return await asyncio.shield(build_task) + except asyncio.CancelledError: + while not build_task.done(): + try: + await asyncio.shield(build_task) + except asyncio.CancelledError: + continue + except Exception: + break + if not build_task.cancelled(): + try: + build_task.result().file.close() + except Exception: + pass + raise + finally: + _artifact_archive_slots.release() + + +def _presented_files_from_delivery(events: list[dict]) -> list[str]: + if len(events) != 1: + raise HTTPException(status_code=409, detail="This response has no verified artifact delivery") + content = events[0].get("content") + by_tool = content.get("by_tool") if isinstance(content, dict) else None + presented = by_tool.get("present_files") if isinstance(by_tool, dict) else None + if not isinstance(presented, list) or not presented or any(not isinstance(path, str) for path in presented): + raise HTTPException(status_code=409, detail="This response has no verified artifact delivery") + return presented + + +async def _archive_presented_paths(thread_id: ThreadId, run_id: str, request: Request) -> list[str]: + run = await get_run_store(request).get(run_id) + if run is None or run.get("thread_id") != thread_id or run.get("operation_kind", "run") != "run": + raise HTTPException(status_code=404, detail=f"Run {run_id} not found") + if run.get("status") in {RunStatus.pending.value, RunStatus.running.value}: + raise HTTPException(status_code=409, detail="This run has not finished") + + events = await get_run_event_store(request).list_events( + thread_id, + run_id, + event_types=["run.delivery"], + limit=2, + ) + return _presented_files_from_delivery(events) + + +@router.get( + "/{thread_id}/runs/{run_id}/artifacts/archive", + response_model=ArtifactArchiveManifestResponse, +) +@require_permission("runs", "read", owner_check=True, require_existing=True) +async def get_run_artifact_archive_manifest( + thread_id: ThreadId, + run_id: str, + request: Request, +) -> ArtifactArchiveManifestResponse: + """Return the verified terminal delivery count used by the archive.""" + presented_paths = await _archive_presented_paths(thread_id, run_id, request) + return ArtifactArchiveManifestResponse(file_count=len(dict.fromkeys(presented_paths))) + + +@router.post("/{thread_id}/runs/{run_id}/artifacts/archive") +@require_permission("runs", "read", owner_check=True, require_existing=True) +async def create_run_artifact_archive( + thread_id: ThreadId, + run_id: str, + request: Request, +) -> StreamingResponse: + """Download the current contents of the files presented by one terminal run.""" + presented_paths = await _archive_presented_paths(thread_id, run_id, request) + + raw_owner_user_id = get_trusted_internal_owner_user_id(request) + effective_user_id = make_safe_user_id(raw_owner_user_id) if raw_owner_user_id else get_effective_user_id() + app_config = await safe_app_config_async() + custom_tool_output_dir = getattr(getattr(app_config, "tool_output", None), "storage_subdir", None) + extra_reserved_dir_names = {custom_tool_output_dir} if isinstance(custom_tool_output_dir, str) else set() + paths = get_paths() + user_data_dir = paths.sandbox_user_data_dir(thread_id, user_id=effective_user_id) + outputs_dir = paths.sandbox_outputs_dir(thread_id, user_id=effective_user_id) + + try: + async with get_run_manager(request).reserve_thread_operation( + thread_id, + kind=ThreadOperationKind.artifact_archive, + user_id=effective_user_id, + ): + result = await _build_archive_without_abandoning_worker( + outputs_dir, + user_data_dir, + presented_paths, + extra_reserved_dir_names=extra_reserved_dir_names, + ) + except ConflictError as exc: + raise HTTPException(status_code=409, detail="Artifacts are currently being modified; try again shortly") from exc + except ArtifactArchiveError as exc: + raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc + + safe_run_id = re.sub(r"[^A-Za-z0-9_-]", "", run_id)[:32] or "run" + logger.info( + "Created artifact archive thread_id=%s run_id=%s members=%d input_bytes=%d output_bytes=%d", + sanitize_log_param(thread_id), + sanitize_log_param(run_id), + result.member_count, + result.input_bytes, + result.size, + ) + return StreamingResponse( + _archive_response_chunks(result), + media_type="application/zip", + headers={ + "Content-Disposition": f'attachment; filename="artifacts-{safe_run_id}.zip"', + "Content-Length": str(result.size), + "Cache-Control": "private, no-store", + "X-Content-Type-Options": "nosniff", + }, + background=BackgroundTask(result.file.close), + ) + + @router.get("/{thread_id}/runs/{run_id}/events") @require_permission("runs", "read", owner_check=True) async def list_run_events( diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 4e24adcca..9bf7eae46 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -283,6 +283,7 @@ class RunJournal(BaseCallbackHandler): self._llm_call_index = 0 self._seen_llm_starts: set[str] = set() # langchain run_ids that fired on_chat_model_start self._current_run_tool_call_names: dict[str, str] = {} + self._active_tool_names: dict[str, str] = {} self._persisted_tool_message_identities: set[str] = set() # Artifact-production tracking for the terminal run.delivery event @@ -524,12 +525,16 @@ class RunJournal(BaseCallbackHandler): ) def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, tags=None, metadata=None, inputs=None, **kwargs): - """Handle tool start event, cache tool call ID for later correlation""" - tool_call_id = str(run_id) - logger.debug("Tool start for node %s, tool_call_id=%s, tags=%s", run_id, tool_call_id, tags) + """Cache the executing tool name for artifact attribution.""" + tool_run_id = str(run_id) + tool_name = serialized.get("name") if isinstance(serialized, Mapping) else None + if isinstance(tool_name, str) and tool_name: + self._active_tool_names[tool_run_id] = tool_name + logger.debug("Tool start for node %s, tool_run_id=%s, tags=%s", run_id, tool_run_id, tags) def on_tool_end(self, output, *, run_id, parent_run_id=None, **kwargs): """Handle tool end event, append message and clear node data""" + active_tool_name = self._active_tool_names.pop(str(run_id), None) try: if isinstance(output, ToolMessage): msg = cast(ToolMessage, output) @@ -555,7 +560,9 @@ class RunJournal(BaseCallbackHandler): else: logger.warning(f"on_tool_end {run_id}: command update message is not BaseMessage: {type(message)}") if artifacts: - artifact_tool_name = next(iter(artifact_tool_names)) if len(artifact_tool_names) == 1 else None + artifact_tool_name = active_tool_name + if artifact_tool_name is None and len(artifact_tool_names) == 1: + artifact_tool_name = next(iter(artifact_tool_names)) self._record_produced_artifacts(artifacts, artifact_tool_name) else: logger.warning(f"on_tool_end {run_id}: output is not ToolMessage: {type(output)}") diff --git a/backend/packages/harness/deerflow/runtime/runs/schemas.py b/backend/packages/harness/deerflow/runtime/runs/schemas.py index 15ace3ae9..b5bb31d54 100644 --- a/backend/packages/harness/deerflow/runtime/runs/schemas.py +++ b/backend/packages/harness/deerflow/runtime/runs/schemas.py @@ -9,6 +9,7 @@ class ThreadOperationKind(StrEnum): run = "run" checkpoint_write = "checkpoint_write" artifact_write = "artifact_write" + artifact_archive = "artifact_archive" branch = "branch" delete = "delete" diff --git a/backend/tests/blocking_io/test_artifact_archive.py b/backend/tests/blocking_io/test_artifact_archive.py new file mode 100644 index 000000000..2aa998e26 --- /dev/null +++ b/backend/tests/blocking_io/test_artifact_archive.py @@ -0,0 +1,34 @@ +"""Artifact archive construction must stay off the Gateway event loop.""" + +from __future__ import annotations + +import asyncio +import io +import zipfile +from pathlib import Path + +import pytest + +from app.gateway.routers.thread_runs import _build_archive_without_abandoning_worker + +pytestmark = pytest.mark.asyncio + + +async def test_artifact_archive_build_does_not_block_event_loop(tmp_path: Path) -> None: + outputs = tmp_path / "outputs" + await asyncio.to_thread(outputs.mkdir) + await asyncio.to_thread((outputs / "report.txt").write_text, "report", encoding="utf-8") + + result = await _build_archive_without_abandoning_worker( + outputs, + outputs.parent, + ["/mnt/user-data/outputs/report.txt"], + extra_reserved_dir_names=set(), + ) + try: + payload = await asyncio.to_thread(result.file.read) + finally: + result.file.close() + + with zipfile.ZipFile(io.BytesIO(payload)) as archive: + assert archive.read("report.txt") == b"report" diff --git a/backend/tests/test_artifact_archive.py b/backend/tests/test_artifact_archive.py new file mode 100644 index 000000000..83de6abaf --- /dev/null +++ b/backend/tests/test_artifact_archive.py @@ -0,0 +1,614 @@ +from __future__ import annotations + +import asyncio +import io +import os +import threading +import zipfile +from pathlib import Path +from uuid import UUID + +import pytest +from _router_auth_helpers import make_authed_test_app +from fastapi.testclient import TestClient + +from app.gateway import artifact_archive +from app.gateway.auth.models import User +from app.gateway.routers import thread_runs +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.runs.manager import RunManager +from deerflow.runtime.runs.store.memory import MemoryRunStore + +THREAD_ID = "thread-archive" +RUN_ID = "run-archive" +USER_ID = UUID("00000000-0000-0000-0000-000000000123") +ARCHIVE_URL = f"/api/threads/{THREAD_ID}/runs/{RUN_ID}/artifacts/archive" + + +class _FakePaths: + def __init__(self, outputs_dir: Path) -> None: + self._outputs_dir = outputs_dir + + def sandbox_outputs_dir(self, _thread_id: str, *, user_id: str | None = None) -> Path: + return self._outputs_dir + + def sandbox_user_data_dir(self, _thread_id: str, *, user_id: str | None = None) -> Path: + return self._outputs_dir.parent + + +def _user() -> User: + return User( + id=USER_ID, + email="archive-test@example.com", + password_hash="x", + system_role="user", + ) + + +def _archive_app( + monkeypatch: pytest.MonkeyPatch, + outputs_dir: Path, + *, + paths: list[str] | None = None, + run_thread_id: str = THREAD_ID, + run_status: str = "success", + with_receipt: bool = True, +) -> tuple[TestClient, MemoryRunStore, MemoryRunEventStore]: + run_store = MemoryRunStore() + event_store = MemoryRunEventStore() + run_manager = RunManager(store=run_store) + + asyncio.run( + run_store.put( + RUN_ID, + thread_id=run_thread_id, + user_id=None, + status=run_status, + ) + ) + if with_receipt: + presented = paths or [] + asyncio.run( + event_store.put( + thread_id=THREAD_ID, + run_id=RUN_ID, + event_type="run.delivery", + category="outputs", + content={ + "presented": len(presented), + "paths": presented, + "by_tool": {"present_files": presented}, + }, + ) + ) + + monkeypatch.setattr( + thread_runs, + "get_paths", + lambda: _FakePaths(outputs_dir), + raising=False, + ) + + app = make_authed_test_app(user_factory=_user) + app.state.run_store = run_store + app.state.run_event_store = event_store + app.state.run_manager = run_manager + app.include_router(thread_runs.router) + return TestClient(app), run_store, event_store + + +def test_archive_download_contains_only_presented_files(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + (outputs / "reports").mkdir(parents=True) + (outputs / "reports" / "summary.txt").write_text("summary", encoding="utf-8") + (outputs / "data.csv").write_text("a,b\n1,2\n", encoding="utf-8") + (outputs / "not-presented.txt").write_text("secret", encoding="utf-8") + paths = [ + "/mnt/user-data/outputs/reports/summary.txt", + "/mnt/user-data/outputs/data.csv", + "/mnt/user-data/outputs/data.csv", + ] + client, _, _ = _archive_app(monkeypatch, outputs, paths=paths) + + with client: + response = client.post( + ARCHIVE_URL, + json={"paths": ["/mnt/user-data/outputs/not-presented.txt"]}, + ) + + assert response.status_code == 200 + assert response.headers["content-type"] == "application/zip" + assert response.headers["cache-control"] == "private, no-store" + assert response.headers["x-content-type-options"] == "nosniff" + assert "attachment" in response.headers["content-disposition"] + with zipfile.ZipFile(io.BytesIO(response.content)) as archive: + assert archive.namelist() == ["reports/summary.txt", "data.csv"] + assert archive.read("reports/summary.txt") == b"summary" + assert archive.read("data.csv") == b"a,b\n1,2\n" + assert "not-presented.txt" not in archive.namelist() + + +def test_archive_manifest_counts_only_verified_delivery_paths(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=[ + "/mnt/user-data/outputs/report.txt", + "/mnt/user-data/outputs/data.csv", + "/mnt/user-data/outputs/data.csv", + ], + ) + + with client: + response = client.get(ARCHIVE_URL) + + assert response.status_code == 200 + assert response.json() == {"file_count": 2} + + +@pytest.mark.parametrize( + "presented_path", + [ + "/mnt/user-data/uploads/private.txt", + "/mnt/user-data/outputs/../uploads/private.txt", + "/mnt/user-data/outputs/.tool-results/raw.txt", + "/mnt/user-data/outputs/.browser-frames/frame.png", + ], +) +def test_archive_rejects_paths_outside_public_outputs( + tmp_path, + monkeypatch, + presented_path: str, +) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + client, _, _ = _archive_app(monkeypatch, outputs, paths=[presented_path]) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + assert response.json()["detail"] == "The files listed by this response are not available for archive download" + + +def test_archive_rejects_a_directory_without_recursing(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + folder = outputs / "site" + folder.mkdir(parents=True) + (folder / "index.html").write_text("hidden descendant", encoding="utf-8") + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=["/mnt/user-data/outputs/site"], + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +def test_archive_rejects_a_symlink(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + outside = tmp_path / "outside.txt" + outside.write_text("outside", encoding="utf-8") + link = outputs / "linked.txt" + try: + link.symlink_to(outside) + except OSError: + pytest.skip("symlinks are unavailable on this platform") + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=["/mnt/user-data/outputs/linked.txt"], + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +def test_archive_rejects_a_symlinked_outputs_root(tmp_path, monkeypatch) -> None: + user_data = tmp_path / "user-data" + user_data.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (outside / "secret.txt").write_text("outside", encoding="utf-8") + outputs = user_data / "outputs" + try: + outputs.symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("symlinks are unavailable on this platform") + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=["/mnt/user-data/outputs/secret.txt"], + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +def test_archive_rejects_a_hard_linked_member(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + internal = outputs / ".tool-results" + internal.mkdir(parents=True) + secret = internal / "raw-secret.txt" + secret.write_text("internal", encoding="utf-8") + public_alias = outputs / "report.txt" + try: + public_alias.hardlink_to(secret) + except OSError: + pytest.skip("hard links are unavailable on this platform") + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=["/mnt/user-data/outputs/report.txt"], + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +@pytest.mark.parametrize("junction_relative", [Path(), Path("site")]) +def test_archive_rejects_junction_like_roots_and_components( + tmp_path, + monkeypatch, + junction_relative: Path, +) -> None: + outputs = tmp_path / "outputs" + target = outputs / "site" / "report.txt" + target.parent.mkdir(parents=True) + target.write_text("report", encoding="utf-8") + junction = outputs / junction_relative + monkeypatch.setattr(Path, "is_junction", lambda self: self == junction) + + with pytest.raises(artifact_archive.ArtifactArchiveError): + artifact_archive.build_artifact_archive( + outputs, + ["/mnt/user-data/outputs/site/report.txt"], + user_data_dir=outputs.parent, + ) + + +def test_archive_rejects_missing_or_nonterminal_delivery(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + client, _, _ = _archive_app( + monkeypatch, + outputs, + run_status="running", + with_receipt=False, + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +@pytest.mark.parametrize("method", ["get", "post"]) +def test_archive_hides_a_run_from_another_thread(tmp_path, monkeypatch, method: str) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + client, _, _ = _archive_app( + monkeypatch, + outputs, + run_thread_id="another-thread", + paths=["/mnt/user-data/outputs/report.txt"], + ) + + with client: + response = getattr(client, method)(ARCHIVE_URL) + + assert response.status_code == 404 + + +def test_archive_conflicts_with_an_active_run(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + (outputs / "report.txt").write_text("report", encoding="utf-8") + client, run_store, _ = _archive_app( + monkeypatch, + outputs, + paths=["/mnt/user-data/outputs/report.txt"], + ) + asyncio.run( + run_store.put( + "active-run", + thread_id=THREAD_ID, + user_id=None, + status="running", + ) + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +def test_archive_rejects_when_the_worker_is_at_capacity(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + (outputs / "report.txt").write_text("report", encoding="utf-8") + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=["/mnt/user-data/outputs/report.txt"], + ) + slots = asyncio.Semaphore(1) + asyncio.run(slots.acquire()) + monkeypatch.setattr(thread_runs, "_artifact_archive_slots", slots, raising=False) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 429 + + +def test_archive_rejects_casefolded_entry_collisions(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + upper = outputs / "Report.txt" + lower = outputs / "report.txt" + upper.write_text("upper", encoding="utf-8") + lower.write_text("lower", encoding="utf-8") + if upper.samefile(lower): + pytest.skip("case-sensitive filenames are unavailable on this filesystem") + client, _, _ = _archive_app( + monkeypatch, + outputs, + paths=[ + "/mnt/user-data/outputs/Report.txt", + "/mnt/user-data/outputs/report.txt", + ], + ) + + with client: + response = client.post(ARCHIVE_URL) + + assert response.status_code == 409 + + +def test_archive_rechecks_size_after_open(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + artifact = outputs / "report.txt" + artifact.write_bytes(b"x") + original_member = artifact_archive._member + + def grow_after_validation(*args, **kwargs): + member = original_member(*args, **kwargs) + artifact.write_bytes(b"12345") + return member + + monkeypatch.setattr(artifact_archive, "MAX_FILE_BYTES", 4) + monkeypatch.setattr(artifact_archive, "MAX_TOTAL_BYTES", 8) + monkeypatch.setattr(artifact_archive, "_member", grow_after_validation) + + with pytest.raises(artifact_archive.ArtifactArchiveError) as exc_info: + artifact_archive.build_artifact_archive( + outputs, + ["/mnt/user-data/outputs/report.txt"], + user_data_dir=outputs.parent, + ) + + assert exc_info.value.status_code == 413 + + +def test_archive_enforces_file_count_and_total_size_limits(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + paths = [] + for name in ("one.txt", "two.txt"): + (outputs / name).write_bytes(b"x") + paths.append(f"/mnt/user-data/outputs/{name}") + + monkeypatch.setattr(artifact_archive, "MAX_FILES", 1) + with pytest.raises(artifact_archive.ArtifactArchiveError) as count_error: + artifact_archive.build_artifact_archive(outputs, paths, user_data_dir=outputs.parent) + assert count_error.value.status_code == 413 + + monkeypatch.setattr(artifact_archive, "MAX_FILES", 2) + monkeypatch.setattr(artifact_archive, "MAX_TOTAL_BYTES", 1) + with pytest.raises(artifact_archive.ArtifactArchiveError) as size_error: + artifact_archive.build_artifact_archive(outputs, paths, user_data_dir=outputs.parent) + assert size_error.value.status_code == 413 + + +@pytest.mark.parametrize( + ("relative_path", "reserved"), + [ + (".ARTIFACT-EDIT-draft", set()), + ("private-cache/result.txt", {"private-cache"}), + ], +) +def test_archive_rejects_internal_output_names( + tmp_path, + relative_path: str, + reserved: set[str], +) -> None: + outputs = tmp_path / "outputs" + target = outputs / relative_path + target.parent.mkdir(parents=True) + target.write_text("internal", encoding="utf-8") + + with pytest.raises(artifact_archive.ArtifactArchiveError): + artifact_archive.build_artifact_archive( + outputs, + [f"/mnt/user-data/outputs/{relative_path}"], + user_data_dir=outputs.parent, + extra_reserved_dir_names=reserved, + ) + + +def test_archive_rejects_a_path_replaced_during_read(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + artifact = outputs / "report.txt" + artifact.write_bytes(b"old-content") + replacement = outputs / "replacement.txt" + replacement.write_bytes(b"new-content") + original_read = artifact_archive.os.read + replaced = False + + def replace_after_read(descriptor: int, count: int) -> bytes: + nonlocal replaced + data = original_read(descriptor, count) + if not replaced: + replaced = True + replacement.replace(artifact) + return data + + monkeypatch.setattr(artifact_archive.os, "read", replace_after_read) + + with pytest.raises(artifact_archive.ArtifactArchiveError): + artifact_archive.build_artifact_archive( + outputs, + ["/mnt/user-data/outputs/report.txt"], + user_data_dir=outputs.parent, + ) + + +def test_archive_rejects_same_size_content_change_with_restored_mtime(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + artifact = outputs / "report.bin" + chunk_size = artifact_archive._CHUNK_BYTES + artifact.write_bytes(b"A" * (chunk_size * 2)) + original_mtime_ns = artifact.stat().st_mtime_ns + original_read = artifact_archive.os.read + reads = 0 + + def change_middle_chunk_then_restore(descriptor: int, count: int) -> bytes: + nonlocal reads + data = original_read(descriptor, count) + reads += 1 + if reads == 1: + with artifact.open("r+b") as stream: + stream.seek(chunk_size) + stream.write(b"B" * chunk_size) + os.utime(artifact, ns=(artifact.stat().st_atime_ns, original_mtime_ns)) + elif reads == 2: + with artifact.open("r+b") as stream: + stream.seek(chunk_size) + stream.write(b"A" * chunk_size) + os.utime(artifact, ns=(artifact.stat().st_atime_ns, original_mtime_ns)) + return data + + monkeypatch.setattr(artifact_archive.os, "read", change_middle_chunk_then_restore) + + with pytest.raises(artifact_archive.ArtifactArchiveError): + artifact_archive.build_artifact_archive( + outputs, + ["/mnt/user-data/outputs/report.bin"], + user_data_dir=outputs.parent, + ) + + +def test_archive_deadline_applies_to_empty_files(tmp_path, monkeypatch) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + (outputs / "empty.txt").touch() + monkeypatch.setattr(artifact_archive, "BUILD_TIMEOUT_SECONDS", -1) + + with pytest.raises(artifact_archive.ArtifactArchiveError) as exc_info: + artifact_archive.build_artifact_archive( + outputs, + ["/mnt/user-data/outputs/empty.txt"], + user_data_dir=outputs.parent, + ) + + assert exc_info.value.status_code == 503 + + +@pytest.mark.parametrize( + "filename", + [ + "report.txt.", + "C:report.txt", + "CON.txt", + "report.txt", + 'report"draft.txt', + "report|draft.txt", + "report?.txt", + "report*.txt", + ], +) +def test_archive_rejects_nonportable_zip_names(tmp_path, filename: str) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + target = outputs / filename + try: + target.write_text("report", encoding="utf-8") + except OSError: + pytest.skip("the platform cannot create this nonportable filename") + + with pytest.raises(artifact_archive.ArtifactArchiveError): + artifact_archive.build_artifact_archive( + outputs, + [f"/mnt/user-data/outputs/{filename}"], + user_data_dir=outputs.parent, + ) + + +def test_archive_allows_emoji_joiner_sequences(tmp_path) -> None: + outputs = tmp_path / "outputs" + outputs.mkdir() + filename = "report-🧑‍💻.txt" + (outputs / filename).write_text("report", encoding="utf-8") + + result = artifact_archive.build_artifact_archive( + outputs, + [f"/mnt/user-data/outputs/{filename}"], + user_data_dir=outputs.parent, + ) + + with result.file, zipfile.ZipFile(result.file) as archive: + assert archive.namelist() == [filename] + + +@pytest.mark.asyncio +async def test_repeated_cancellation_keeps_the_archive_slot_until_worker_exit(monkeypatch) -> None: + started = threading.Event() + release = threading.Event() + + def blocking_build(*_args, **_kwargs): + started.set() + release.wait(timeout=5) + return artifact_archive.ArtifactArchiveResult(io.BytesIO(), 0, 0, 0) + + slots = asyncio.Semaphore(1) + monkeypatch.setattr(thread_runs, "_artifact_archive_slots", slots) + monkeypatch.setattr(thread_runs, "build_artifact_archive", blocking_build) + task = asyncio.create_task( + thread_runs._build_archive_without_abandoning_worker( + Path("unused"), + Path("unused-parent"), + [], + extra_reserved_dir_names=set(), + ) + ) + assert await asyncio.to_thread(started.wait, 1) + + task.cancel() + await asyncio.sleep(0) + task.cancel() + done, _ = await asyncio.wait({task}, timeout=0.1) + + try: + assert not done + assert slots.locked() + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index b09f41177..9adee7690 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -1518,6 +1518,34 @@ class TestDeliveryTracking: assert content["by_tool"] == {"present_files": ["/mnt/user-data/outputs/report.md"]} assert delivery[0]["category"] == "outputs" + @pytest.mark.anyio + async def test_tool_callback_name_preserves_attribution_when_message_lookup_misses(self, journal_setup): + from langchain_core.messages import ToolMessage + from langgraph.types import Command + + j, store = journal_setup + tool_run_id = uuid4() + j.on_tool_start( + {"name": "present_files"}, + "", + run_id=tool_run_id, + ) + j.on_tool_end( + Command( + update={ + "artifacts": ["/mnt/user-data/outputs/report.md"], + "messages": [ToolMessage("Successfully presented files", tool_call_id="call_missing")], + } + ), + run_id=tool_run_id, + ) + j.record_delivery() + await j.flush() + + events = await store.list_events("t1", "r1") + content = next(e for e in events if e["event_type"] == "run.delivery")["content"] + assert content["by_tool"] == {"present_files": ["/mnt/user-data/outputs/report.md"]} + @pytest.mark.anyio async def test_command_with_multiple_messages_records_artifacts_once(self, journal_setup): from langchain_core.messages import ToolMessage diff --git a/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx b/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx index 8cf663a32..4f90f8520 100644 --- a/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx +++ b/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx @@ -327,6 +327,9 @@ export default function AgentChatPage() {
(null); + const staticWebsiteOnly = isStaticWebsiteOnly(); + const { data: archiveManifest } = useQuery({ + queryKey: ["artifact-archive-manifest", threadId, runId], + queryFn: () => getArtifactArchiveManifest({ threadId, runId: runId! }), + enabled: + archiveDownloadsEnabled && runId !== undefined && !staticWebsiteOnly, + retry: false, + staleTime: Infinity, + }); + const archiveCount = archiveManifest?.fileCount; const handleClick = useCallback( (filepath: string) => { @@ -78,58 +101,116 @@ export function ArtifactFileList({ [threadId, installingFile, t], ); + const handleDownloadArchive = useCallback(async () => { + if (!runId || downloadingArchive) return; + + setDownloadingArchive(true); + let objectUrl: string | undefined; + try { + const { blob, filename } = await downloadArtifactArchive({ + threadId, + runId, + }); + objectUrl = URL.createObjectURL(blob); + const link = document.createElement("a"); + link.href = objectUrl; + link.download = filename; + document.body.append(link); + link.click(); + link.remove(); + } catch (error) { + console.error("Failed to download artifact archive:", error); + toast.error( + error instanceof ArtifactRequestError + ? error.message + : t.artifactArchive.downloadFailed, + ); + } finally { + if (objectUrl) URL.revokeObjectURL(objectUrl); + setDownloadingArchive(false); + } + }, [downloadingArchive, runId, t, threadId]); + + const canDownloadArchive = + archiveDownloadsEnabled && + archiveCount !== undefined && + archiveCount > 1 && + archiveCount <= MAX_ARTIFACT_ARCHIVE_FILES && + !staticWebsiteOnly; + return ( -
    - {files.map((file) => ( - handleClick(file)} - > - - -
    {getFileName(file)}
    -
    - {getFileIcon(file, "size-6")} -
    -     - - {getFileExtensionDisplayName(file)} file - - - {file.endsWith(".skill") && isAdmin && ( - +

    + {t.artifactArchive.currentVersionNotice} +

    +
+ )} + + + + + ))} + + ); } diff --git a/frontend/src/components/workspace/chats/chat-page.tsx b/frontend/src/components/workspace/chats/chat-page.tsx index 5c00fea25..a6c650dde 100644 --- a/frontend/src/components/workspace/chats/chat-page.tsx +++ b/frontend/src/components/workspace/chats/chat-page.tsx @@ -329,6 +329,9 @@ export default function ChatPage() {
getRunDurationDisplaysByGroupIndex(groupedMessages), [groupedMessages], ); + const artifactArchiveCandidatesByGroupIndex = useMemo( + () => getArtifactArchiveCandidatesByGroupIndex(groupedMessages), + [groupedMessages], + ); const workspaceChangeAnchorGroupIndices = useMemo( () => getWorkspaceChangeAnchorGroupIndices(groupedMessages), [groupedMessages], @@ -1269,14 +1276,17 @@ export function MessageList({ } return withRunDuration(group, groupIndex, null); } else if (group.type === "assistant:present-files") { - const files: string[] = []; + const files = new Set(); for (const message of group.messages) { if (hasPresentFiles(message)) { const presentFiles = extractPresentFilesFromMessage(message); - files.push(...presentFiles); + for (const file of presentFiles) files.add(file); } } + const presentedFiles = [...files]; + const archiveCandidate = + artifactArchiveCandidatesByGroupIndex[groupIndex]; return withRunDuration( group, groupIndex, @@ -1288,7 +1298,14 @@ export function MessageList({ className="mb-4" /> )} - + {renderTokenUsage({ messages: group.messages, turnUsageMessages, diff --git a/frontend/src/core/artifacts/api.ts b/frontend/src/core/artifacts/api.ts index fc67abd35..f93276853 100644 --- a/frontend/src/core/artifacts/api.ts +++ b/frontend/src/core/artifacts/api.ts @@ -1,6 +1,6 @@ import { fetch } from "@/core/api/fetcher"; -import { urlOfArtifact } from "./utils"; +import { urlOfArtifact, urlOfArtifactArchive } from "./utils"; export interface ArtifactUpdateResponse { path: string; @@ -8,6 +8,17 @@ export interface ArtifactUpdateResponse { size: number; } +export interface ArtifactArchiveDownload { + blob: Blob; + filename: string; +} + +export interface ArtifactArchiveManifest { + fileCount: number; +} + +export const MAX_ARTIFACT_ARCHIVE_FILES = 50; + export class ArtifactRequestError extends Error { readonly status: number; @@ -52,3 +63,45 @@ export async function updateArtifactContent({ } return response.json() as Promise; } + +export async function downloadArtifactArchive({ + threadId, + runId, +}: { + threadId: string; + runId: string; +}): Promise { + const response = await fetch(urlOfArtifactArchive({ threadId, runId }), { + method: "POST", + }); + if (!response.ok) { + throw new ArtifactRequestError( + response.status, + await readErrorDetail(response), + ); + } + const disposition = response.headers.get("Content-Disposition") ?? ""; + const filename = /filename="([^"]+)"/.exec(disposition)?.[1]; + return { + blob: await response.blob(), + filename: filename ?? `artifacts-${runId}.zip`, + }; +} + +export async function getArtifactArchiveManifest({ + threadId, + runId, +}: { + threadId: string; + runId: string; +}): Promise { + const response = await fetch(urlOfArtifactArchive({ threadId, runId })); + if (!response.ok) { + throw new ArtifactRequestError( + response.status, + await readErrorDetail(response), + ); + } + const data = (await response.json()) as { file_count: number }; + return { fileCount: data.file_count }; +} diff --git a/frontend/src/core/artifacts/utils.ts b/frontend/src/core/artifacts/utils.ts index eb31d3256..f948c1ace 100644 --- a/frontend/src/core/artifacts/utils.ts +++ b/frontend/src/core/artifacts/utils.ts @@ -73,6 +73,16 @@ export function urlOfArtifact({ return `${getBackendBaseURL()}/api/threads/${encodedThreadId}/artifacts${encodedFilepath}${download ? "?download=true" : ""}`; } +export function urlOfArtifactArchive({ + threadId, + runId, +}: { + threadId: string; + runId: string; +}) { + return `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/runs/${encodeURIComponent(runId)}/artifacts/archive`; +} + export function extractArtifactsFromThread(thread: { values: Pick; }) { diff --git a/frontend/src/core/i18n/locales/en-US.ts b/frontend/src/core/i18n/locales/en-US.ts index c3183abc1..1b6f05308 100644 --- a/frontend/src/core/i18n/locales/en-US.ts +++ b/frontend/src/core/i18n/locales/en-US.ts @@ -128,6 +128,14 @@ export const enUS: Translations = { missingTarget: "This link does not say which artifact to display.", }, + artifactArchive: { + downloadCurrent: (count) => + `Download current versions (${count} ${count === 1 ? "file" : "files"})`, + currentVersionNotice: + "The file list comes from this response. Contents are the current versions and may have changed.", + downloadFailed: "Failed to download artifact archive.", + }, + // Citations citations: { sourcesSummary: (count) => diff --git a/frontend/src/core/i18n/locales/types.ts b/frontend/src/core/i18n/locales/types.ts index 9a7b943ca..c8349e534 100644 --- a/frontend/src/core/i18n/locales/types.ts +++ b/frontend/src/core/i18n/locales/types.ts @@ -106,6 +106,12 @@ export interface Translations { missingTarget: string; }; + artifactArchive: { + downloadCurrent: (count: number) => string; + currentVersionNotice: string; + downloadFailed: string; + }; + // Citations citations: { sourcesSummary: (count: number) => string; diff --git a/frontend/src/core/i18n/locales/zh-CN.ts b/frontend/src/core/i18n/locales/zh-CN.ts index 146381712..b816d0dbf 100644 --- a/frontend/src/core/i18n/locales/zh-CN.ts +++ b/frontend/src/core/i18n/locales/zh-CN.ts @@ -125,6 +125,13 @@ export const zhCN: Translations = { missingTarget: "该链接没有指明要展示哪个文件。", }, + artifactArchive: { + downloadCurrent: (count) => `下载当前版本（${count} 个文件）`, + currentVersionNotice: + "文件列表来自此回复；内容为当前版本，可能已发生变化。", + downloadFailed: "文件压缩包下载失败。", + }, + // Citations citations: { sourcesSummary: (count) => `使用了 ${count} 个来源`, diff --git a/frontend/src/core/messages/artifact-archive.ts b/frontend/src/core/messages/artifact-archive.ts new file mode 100644 index 000000000..0e5c72cad --- /dev/null +++ b/frontend/src/core/messages/artifact-archive.ts @@ -0,0 +1,32 @@ +import { getMessageRunId } from "./run-duration"; +import { hasPresentFiles, type MessageGroup } from "./utils"; + +export interface ArtifactArchiveCandidate { + runId: string; +} + +export function getArtifactArchiveCandidatesByGroupIndex( + groups: MessageGroup[], +): Array { + const candidates = Array( + groups.length, + ).fill(undefined); + const lastGroupIndexByRunId = new Map(); + + groups.forEach((group, groupIndex) => { + if (group.type !== "assistant:present-files") return; + + for (const message of group.messages) { + if (!hasPresentFiles(message)) continue; + const runId = getMessageRunId(message); + if (!runId) continue; + lastGroupIndexByRunId.set(runId, groupIndex); + } + }); + + for (const [runId, groupIndex] of lastGroupIndexByRunId) { + candidates[groupIndex] = { runId }; + } + + return candidates; +} diff --git a/frontend/src/core/threads/thread-branch-tree.ts b/frontend/src/core/threads/thread-branch-tree.ts index 5dcb3aefb..0d42034b0 100644 --- a/frontend/src/core/threads/thread-branch-tree.ts +++ b/frontend/src/core/threads/thread-branch-tree.ts @@ -16,8 +16,8 @@ function recencyOfThread(thread: AgentThread) { return Number.isFinite(timestamp) ? timestamp : 0; } -function branchParentId(thread: AgentThread) { - if (thread.metadata?.[THREAD_BRANCH_METADATA_KEY] !== true) { +function branchParentId(thread: AgentThread | null | undefined) { + if (thread?.metadata?.[THREAD_BRANCH_METADATA_KEY] !== true) { return null; } const parentId = thread.metadata?.[THREAD_BRANCH_PARENT_METADATA_KEY]; @@ -27,6 +27,10 @@ function branchParentId(thread: AgentThread) { return parentId.trim() || null; } +export function isBranchThread(thread: AgentThread | null | undefined) { + return branchParentId(thread) !== null; +} + /** * Project the loaded flat thread page into a safe visual lineage. * diff --git a/frontend/tests/unit/components/workspace/artifacts/artifact-file-list.dom.test.tsx b/frontend/tests/unit/components/workspace/artifacts/artifact-file-list.dom.test.tsx new file mode 100644 index 000000000..e86bf62ae --- /dev/null +++ b/frontend/tests/unit/components/workspace/artifacts/artifact-file-list.dom.test.tsx @@ -0,0 +1,299 @@ +import { afterEach, describe, expect, it, rs } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { + cleanup, + fireEvent, + render, + screen, + waitFor, +} from "@testing-library/react"; + +const artifactState = rs.hoisted(() => ({ + select: rs.fn(), + setOpen: rs.fn(), +})); +const archiveState = rs.hoisted(() => { + class RequestError extends Error { + readonly status: number; + + constructor(status: number, message: string) { + super(message); + this.status = status; + } + } + + return { + download: rs.fn(), + manifest: rs.fn(), + RequestError, + toastError: rs.fn(), + }; +}); + +rs.mock("@/core/auth/AuthProvider", () => ({ + useAuth: () => ({ user: null }), +})); +rs.mock("@/components/workspace/artifacts/context", () => ({ + useArtifacts: () => artifactState, +})); +rs.mock("@/core/artifacts/api", () => ({ + ArtifactRequestError: archiveState.RequestError, + downloadArtifactArchive: archiveState.download, + getArtifactArchiveManifest: archiveState.manifest, + MAX_ARTIFACT_ARCHIVE_FILES: 50, +})); +rs.mock("sonner", () => ({ + toast: { error: archiveState.toastError, success: rs.fn() }, +})); + +import { ArtifactFileList } from "@/components/workspace/artifacts/artifact-file-list"; +import { ArtifactRequestError } from "@/core/artifacts/api"; +import { I18nContext } from "@/core/i18n/context"; +import { enUS } from "@/core/i18n/locales/en-US"; + +const files = [ + "/mnt/user-data/outputs/report.md", + "/mnt/user-data/outputs/data.csv", +]; + +function renderList( + props: Partial> = {}, +) { + const componentProps = { files, threadId: "thread-1", ...props }; + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const component = ( + nextProps: React.ComponentProps, + ) => ( + + undefined, t: enUS }} + > + + + + ); + const result = render(component(componentProps)); + return { + ...result, + rerenderList: ( + nextProps: Partial>, + ) => result.rerender(component({ ...componentProps, ...nextProps })), + }; +} + +afterEach(cleanup); +afterEach(() => { + rs.restoreAllMocks(); +}); + +describe("ArtifactFileList archive download", () => { + it("offers a branch-local run accepted by the manifest ownership check", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 2 }); + renderList({ runId: "branch-run", threadId: "branch-thread" }); + + expect( + await screen.findByRole("button", { + name: "Download current versions (2 files)", + }), + ).toBeTruthy(); + expect(archiveState.manifest).toHaveBeenCalledWith({ + runId: "branch-run", + threadId: "branch-thread", + }); + expect( + screen.getByText( + "The file list comes from this response. Contents are the current versions and may have changed.", + ), + ).toBeTruthy(); + }); + + it("uses the verified receipt count instead of attempted tool arguments", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 3 }); + renderList({ runId: "run-1" }); + + expect( + await screen.findByRole("button", { + name: "Download current versions (3 files)", + }), + ).toBeTruthy(); + }); + + it("does not offer an archive when only one attempted file was delivered", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 1 }); + renderList({ runId: "run-1" }); + + await waitFor(() => { + expect(archiveState.manifest).toHaveBeenCalledWith({ + runId: "run-1", + threadId: "thread-1", + }); + }); + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("does not offer an archive outside a run-scoped delivery", () => { + renderList(); + + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("does not offer an archive for a single verified file", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 1 }); + renderList({ files: files.slice(0, 1), runId: "run-1" }); + + await waitFor(() => { + expect(archiveState.manifest).toHaveBeenCalled(); + }); + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("does not offer an archive above the server file-count limit", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 51 }); + renderList({ runId: "run-1" }); + + await waitFor(() => { + expect(archiveState.manifest).toHaveBeenCalled(); + }); + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("does not offer an archive when the current thread cannot download it", () => { + renderList({ archiveDownloadsEnabled: false, runId: "run-1" }); + + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("hides a cached archive while downloads are disabled", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 2 }); + const { rerenderList } = renderList({ runId: "run-1" }); + await screen.findByRole("button", { + name: "Download current versions (2 files)", + }); + + rerenderList({ archiveDownloadsEnabled: false }); + + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("hides an inherited run when the thread ownership check rejects it", async () => { + archiveState.manifest.mockRejectedValue( + new ArtifactRequestError(404, "Run parent-run not found"), + ); + renderList({ runId: "parent-run" }); + + await waitFor(() => { + expect(archiveState.manifest).toHaveBeenCalledWith({ + runId: "parent-run", + threadId: "thread-1", + }); + }); + expect( + screen.queryByRole("button", { + name: /Download current versions/, + }), + ).toBeNull(); + }); + + it("downloads the archive and releases its object URL", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 2 }); + const blob = new Blob(["zip"]); + archiveState.download.mockResolvedValue({ + blob, + filename: "artifacts-run-1.zip", + }); + const createObjectURL = rs + .spyOn(URL, "createObjectURL") + .mockReturnValue("blob:archive"); + const revokeObjectURL = rs.spyOn(URL, "revokeObjectURL"); + let downloadedFilename: string | undefined; + rs.spyOn(HTMLAnchorElement.prototype, "click").mockImplementation(function ( + this: HTMLAnchorElement, + ) { + downloadedFilename = this.download; + }); + renderList({ runId: "run-1" }); + + fireEvent.click( + await screen.findByRole("button", { + name: "Download current versions (2 files)", + }), + ); + + await waitFor(() => { + expect(archiveState.download).toHaveBeenCalledWith({ + runId: "run-1", + threadId: "thread-1", + }); + expect(downloadedFilename).toBe("artifacts-run-1.zip"); + expect(createObjectURL).toHaveBeenCalledWith(blob); + expect(revokeObjectURL).toHaveBeenCalledWith("blob:archive"); + }); + }); + + it("reports archive download failures", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 2 }); + archiveState.download.mockRejectedValue(new Error("network down")); + renderList({ runId: "run-1" }); + + fireEvent.click( + await screen.findByRole("button", { + name: "Download current versions (2 files)", + }), + ); + + await waitFor(() => { + expect(archiveState.toastError).toHaveBeenCalledWith( + "Failed to download artifact archive.", + ); + }); + }); + + it("shows actionable archive errors returned by the server", async () => { + archiveState.manifest.mockResolvedValue({ fileCount: 2 }); + archiveState.download.mockRejectedValue( + new ArtifactRequestError( + 413, + "An artifact archive can contain at most 50 files", + ), + ); + renderList({ runId: "run-1" }); + + fireEvent.click( + await screen.findByRole("button", { + name: "Download current versions (2 files)", + }), + ); + + await waitFor(() => { + expect(archiveState.toastError).toHaveBeenCalledWith( + "An artifact archive can contain at most 50 files", + ); + }); + }); +}); diff --git a/frontend/tests/unit/core/artifacts/api.test.ts b/frontend/tests/unit/core/artifacts/api.test.ts index 0a054fe26..d3fea8de5 100644 --- a/frontend/tests/unit/core/artifacts/api.test.ts +++ b/frontend/tests/unit/core/artifacts/api.test.ts @@ -1,6 +1,10 @@ import { afterEach, describe, expect, it, rs } from "@rstest/core"; -import { updateArtifactContent } from "@/core/artifacts/api"; +import { + downloadArtifactArchive, + getArtifactArchiveManifest, + updateArtifactContent, +} from "@/core/artifacts/api"; afterEach(() => { rs.restoreAllMocks(); @@ -58,3 +62,49 @@ describe("updateArtifactContent", () => { ).rejects.toMatchObject({ status: 412 }); }); }); + +describe("downloadArtifactArchive", () => { + it("posts to the encoded run endpoint and preserves the server filename", async () => { + const blob = new Blob(["zip"]); + const fetchMock = rs.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(blob, { + headers: { + "Content-Disposition": 'attachment; filename="artifacts-run-1.zip"', + }, + }), + ); + + const result = await downloadArtifactArchive({ + threadId: "thread #1", + runId: "run/1", + }); + + expect(fetchMock).toHaveBeenCalledWith( + "/api/threads/thread%20%231/runs/run%2F1/artifacts/archive", + expect.objectContaining({ method: "POST", credentials: "include" }), + ); + expect(result.filename).toBe("artifacts-run-1.zip"); + expect(await result.blob.text()).toBe("zip"); + }); +}); + +describe("getArtifactArchiveManifest", () => { + it("reads the verified delivery count from the encoded run endpoint", async () => { + const fetchMock = rs.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(JSON.stringify({ file_count: 3 }), { + headers: { "Content-Type": "application/json" }, + }), + ); + + const result = await getArtifactArchiveManifest({ + threadId: "thread #1", + runId: "run/1", + }); + + expect(fetchMock).toHaveBeenCalledWith( + "/api/threads/thread%20%231/runs/run%2F1/artifacts/archive", + expect.objectContaining({ credentials: "include" }), + ); + expect(result).toEqual({ fileCount: 3 }); + }); +}); diff --git a/frontend/tests/unit/core/messages/artifact-archive.test.ts b/frontend/tests/unit/core/messages/artifact-archive.test.ts new file mode 100644 index 000000000..081cbc3d3 --- /dev/null +++ b/frontend/tests/unit/core/messages/artifact-archive.test.ts @@ -0,0 +1,65 @@ +import type { Message } from "@langchain/langgraph-sdk"; +import { describe, expect, test } from "@rstest/core"; + +import { getArtifactArchiveCandidatesByGroupIndex } from "@/core/messages/artifact-archive"; +import { getMessageGroups } from "@/core/messages/utils"; + +function presentFiles(id: string, runId: string, filepaths: string[]): Message { + return { + id, + type: "ai", + content: "", + run_id: runId, + tool_calls: [ + { + id: `call-${id}`, + name: "present_files", + args: { filepaths }, + }, + ], + } as Message; +} + +describe("artifact archive display placement", () => { + test("anchors one archive action after a run's final file presentation", () => { + const groups = getMessageGroups([ + presentFiles("first", "run-1", ["/mnt/user-data/outputs/a.txt"]), + presentFiles("second", "run-1", [ + "/mnt/user-data/outputs/a.txt", + "/mnt/user-data/outputs/b.txt", + ]), + ]); + + expect(getArtifactArchiveCandidatesByGroupIndex(groups)).toEqual([ + undefined, + { runId: "run-1" }, + ]); + }); + + test("verifies even a single attempted path against the terminal receipt", () => { + const groups = getMessageGroups([ + presentFiles("only", "run-1", ["/mnt/user-data/outputs/a.txt"]), + ]); + + expect(getArtifactArchiveCandidatesByGroupIndex(groups)).toEqual([ + { runId: "run-1" }, + ]); + }); + + test("keeps archive actions independent across runs", () => { + const groups = getMessageGroups([ + presentFiles("run-1-first", "run-1", ["/mnt/user-data/outputs/a.txt"]), + presentFiles("run-2", "run-2", [ + "/mnt/user-data/outputs/c.txt", + "/mnt/user-data/outputs/d.txt", + ]), + presentFiles("run-1-last", "run-1", ["/mnt/user-data/outputs/b.txt"]), + ]); + + expect(getArtifactArchiveCandidatesByGroupIndex(groups)).toEqual([ + undefined, + { runId: "run-2" }, + { runId: "run-1" }, + ]); + }); +}); diff --git a/frontend/tests/unit/core/threads/thread-branch-tree.test.ts b/frontend/tests/unit/core/threads/thread-branch-tree.test.ts index 51c75c999..0a6b60160 100644 --- a/frontend/tests/unit/core/threads/thread-branch-tree.test.ts +++ b/frontend/tests/unit/core/threads/thread-branch-tree.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from "@rstest/core"; -import { flattenThreadBranches } from "@/core/threads/thread-branch-tree"; +import { + flattenThreadBranches, + isBranchThread, +} from "@/core/threads/thread-branch-tree"; import type { AgentThread } from "@/core/threads/types"; function thread( @@ -40,6 +43,24 @@ function summarize(entries: ReturnType) { })); } +describe("isBranchThread", () => { + it("recognizes a thread with valid branch provenance", () => { + expect( + isBranchThread(branch("child", "parent", "2026-01-01T00:00:00Z")), + ).toBe(true); + }); + + it("rejects missing or malformed branch provenance", () => { + expect(isBranchThread(thread("regular", "2026-01-01T00:00:00Z"))).toBe( + false, + ); + expect( + isBranchThread(branch("malformed", 42, "2026-01-01T00:00:00Z")), + ).toBe(false); + expect(isBranchThread(null)).toBe(false); + }); +}); + describe("flattenThreadBranches", () => { it("nests loaded siblings and lifts the group by its freshest descendant", () => { const parent = thread("parent", "2026-01-01T00:00:00Z");