feat(sharing): add revocable read-only conversation snapshots (#8211)

* feat(sharing): add revocable read-only conversation snapshots

* fix(sharing): preserve fork idempotency and allow snapshot cleanup
This commit is contained in:
Wenchao An 2026-11-14 12:22:22 +08:00 committed by GitHub
parent 9cb2bde557
commit 4f3a8b790d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
45 changed files with 3050 additions and 56 deletions

View File

@ -138,6 +138,7 @@ DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI-
- [Install.md](Install.md) - Setup instructions written for coding agents
- [Configuration Reference](backend/docs/CONFIGURATION.md) - Every `config.yaml` option
- [Backend Architecture](backend/README.md) - Backend architecture and API reference
- [Conversation Sharing](docs/guide/sharing.md) - Expiring read-only snapshots, artifact previews and workspace copies
- [Architecture Overview](docs/ARCHITECTURE.md) - How the services fit together
- [Contributing Guide](CONTRIBUTING.md) - Development environment, workflow and tests
- [Changelog](CHANGELOG.md) - Release notes

View File

@ -22,6 +22,7 @@ replay, archive), [run event stream](../../docs/RUN_EVENT_STREAM.md),
## Trust boundary
- **Share tokens read only frozen snapshots.** See [sharing](../../../docs/guide/sharing.md).
- **Server-owned run context.** A client can send run context in `body.context`
and in `body.config`. Admit server-owned keys only from internally authenticated
callers and strip them from both places for everyone else. When you add a

View File

@ -50,6 +50,7 @@ from app.gateway.routers import (
runs,
scheduled_tasks,
search,
shares,
skill_mutations,
skills,
subagent_batches,
@ -1382,6 +1383,7 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for
# Thread cleanup API is mounted at /api/threads/{thread_id}
app.include_router(search.router)
app.include_router(threads.router)
app.include_router(shares.router)
# Per-user activity feed (own prefix, so /api/threads/{thread_id} never captures it)
app.include_router(thread_activity.router)

View File

@ -10,6 +10,7 @@ Fine-grained permission checks remain in authz.py decorators.
"""
import logging
import re
from collections.abc import Callable
from fastapi import HTTPException, Request, Response
@ -60,7 +61,11 @@ _PUBLIC_EXACT_PATHS: frozenset[str] = frozenset(
)
def _is_public(path: str) -> bool:
def _is_public(path: str, method: str = "GET") -> bool:
# A share token grants only these reads. Forks and any future write routes
# under /shares still pass through session authentication and permissions.
if method in {"GET", "HEAD"} and re.fullmatch(r"/api/shares/[A-Za-z0-9_-]{43}(?:/artifacts/.+)?", path):
return True
stripped = path.rstrip("/")
if stripped in _PUBLIC_EXACT_PATHS:
return True
@ -91,7 +96,7 @@ class AuthMiddleware(BaseHTTPMiddleware):
super().__init__(app)
async def dispatch(self, request: Request, call_next: Callable) -> Response:
if _is_public(get_request_route_path(request)):
if _is_public(get_request_route_path(request), request.method):
return await call_next(request)
internal_user = None

View File

@ -178,7 +178,7 @@ async def import_snapshot(request: Request, user_id: str, thread_id: str, namesp
return result
async def read_snapshot(request: Request, user_id: str, thread_id: str, path: str) -> bytes:
async def read_snapshot(request: Request, user_id: str, thread_id: str, path: str, *, max_bytes: int = MAX_BYTES) -> bytes:
"""Capture bounded output bytes through host storage or an authorized lease.
Non-output paths raise HTTP 400; unsafe files raise 422 and host mutation
@ -191,8 +191,8 @@ async def read_snapshot(request: Request, user_id: str, thread_id: str, path: st
path = normalize_outputs_virtual_path(path)
async with snapshot_sandbox(request, user_id, thread_id) as sandbox:
if sandbox is not None:
return await run_file_io(read_remote, sandbox, path.removeprefix("/mnt/user-data/outputs/"))
return await run_file_io(read_output_snapshot, user_id, thread_id, path)
return await run_file_io(read_remote, sandbox, path.removeprefix("/mnt/user-data/outputs/"), max_bytes=max_bytes)
return await run_file_io(read_output_snapshot, user_id, thread_id, path, max_bytes=max_bytes)
async def verify_imports(request: Request, user_id: str, thread_id: str, receipts: list[dict[str, Any]], *, lifetime: AsyncExitStack | None = None) -> None:
@ -220,7 +220,7 @@ async def verify_imports(request: Request, user_id: str, thread_id: str, receipt
raise AgentRunError(422, "Attached artifact copy is unavailable or changed; inspect the team artifacts and send a new request") from exc
def read_output_snapshot(user_id: str, thread_id: str, path: str) -> bytes:
def read_output_snapshot(user_id: str, thread_id: str, path: str, *, max_bytes: int = MAX_BYTES) -> bytes:
"""Read bounded host output bytes without following links.
Non-output paths raise HTTP 400; unsafe files raise 422 and mutation during
@ -229,10 +229,10 @@ def read_output_snapshot(user_id: str, thread_id: str, path: str) -> bytes:
if os.name != "posix" or not isinstance(path, str):
raise AgentRunError(422, "Output capture requires POSIX and an output path")
path = normalize_outputs_virtual_path(path)
return _read_host_snapshot(user_id, thread_id, path.removeprefix("/mnt/user-data/outputs/"), "outputs")
return _read_host_snapshot(user_id, thread_id, path.removeprefix("/mnt/user-data/outputs/"), "outputs", max_bytes=max_bytes)
def _read_host_snapshot(user_id: str, thread_id: str, path: str, area: str) -> bytes:
def _read_host_snapshot(user_id: str, thread_id: str, path: str, area: str, *, max_bytes: int = MAX_BYTES) -> bytes:
paths = get_paths()
root = paths.base_dir.resolve()
base = paths.sandbox_outputs_dir(thread_id, user_id=user_id) if area == "outputs" else paths.sandbox_uploads_dir(thread_id, user_id=user_id)
@ -247,11 +247,11 @@ def _read_host_snapshot(user_id: str, thread_id: str, path: str, area: str) -> b
fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
with os.fdopen(fd, "rb") as source:
before = os.fstat(source.fileno())
if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_BYTES or (area == "uploads" and before.st_nlink != 1):
raise AgentRunError(422, "Artifact must be a regular file of at most 2 MiB")
content = source.read(MAX_BYTES + 1)
if not stat.S_ISREG(before.st_mode) or before.st_size > max_bytes or before.st_nlink != 1:
raise AgentRunError(422, "Artifact must be a regular, unlinked file within the capture limit")
content = source.read(max_bytes + 1)
after = os.fstat(source.fileno())
if len(content) > MAX_BYTES or (before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns):
if len(content) > max_bytes or (before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns):
raise AgentRunError(409, "Artifact changed during capture")
return content
except OSError as exc:

View File

@ -0,0 +1,130 @@
"""Owner-managed snapshots, public read capabilities, and authenticated imports."""
import logging
from datetime import UTC, datetime, timedelta
from typing import Annotated
from uuid import UUID
from deerflow_extension_api import AgentRunError
from fastapi import APIRouter, HTTPException, Path, Request
from fastapi.responses import JSONResponse, Response
from pydantic import BaseModel, Field
from app.gateway.authz import require_permission
from app.gateway.routers.artifacts import _build_content_disposition, _slice_byte_range
from app.gateway.services import reserve_checkpoint_write
from app.gateway.share_fork import fork_share
from app.gateway.share_snapshot import capture_share
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.thread_shares.sql import ThreadShareRepository
from deerflow.runtime import ConflictError
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.file_io import await_drained
from deerflow.utils.thread_id import ThreadId
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api", tags=["shares"])
ShareToken = Annotated[str, Path(pattern=r"^[A-Za-z0-9_-]{43}$")]
PRIVATE_HEADERS = {"Cache-Control": "no-store", "Referrer-Policy": "no-referrer", "X-Robots-Tag": "noindex, nofollow", "X-Content-Type-Options": "nosniff"}
class ShareCreateRequest(BaseModel):
checkpoint_id: str | None = Field(default=None, max_length=128)
expires_in_seconds: int = Field(default=7 * 86400, ge=3600, le=30 * 86400)
class ShareForkRequest(BaseModel):
request_id: UUID
def share_repository() -> ThreadShareRepository:
factory = get_session_factory()
if factory is None:
raise HTTPException(503, "Sharing requires SQL persistence")
return ThreadShareRepository(factory)
@router.post("/threads/{thread_id}/shares", status_code=201)
@require_permission("threads", "write", owner_check=True, require_existing=True)
@require_permission("runs", "read")
async def create_share(thread_id: ThreadId, body: ShareCreateRequest, request: Request):
async def create():
repo = share_repository()
async with reserve_checkpoint_write(request, thread_id, user_id=get_effective_user_id()):
if await repo.count_active(thread_id, get_effective_user_id()) >= 10:
raise HTTPException(409, "Revoke an existing link before creating another")
captured = await capture_share(request, thread_id, body.checkpoint_id)
return await repo.create(**captured, expires_at=datetime.now(UTC) + timedelta(seconds=body.expires_in_seconds))
try:
result = await await_drained(create())
return JSONResponse(result, status_code=201, headers=PRIVATE_HEADERS)
except ConflictError:
raise HTTPException(409, "Thread has work in flight; share it after the work finishes") from None
except AgentRunError as exc:
raise HTTPException(exc.status_code, str(exc)) from None
except HTTPException:
raise
except Exception:
logger.exception("Failed to create conversation share")
raise HTTPException(500, "Unable to create share") from None
@router.get("/threads/{thread_id}/shares")
@require_permission("threads", "read", owner_check=True, require_existing=True)
async def list_shares(thread_id: ThreadId, request: Request):
return JSONResponse(await share_repository().list_owned(thread_id, get_effective_user_id()), headers=PRIVATE_HEADERS)
@router.delete("/threads/{thread_id}/shares/{share_id}", status_code=204)
@require_permission("threads", "write", owner_check=True, require_existing=True)
async def revoke_share(thread_id: ThreadId, share_id: str, request: Request):
if not await await_drained(share_repository().revoke(thread_id, share_id, get_effective_user_id())):
raise HTTPException(404, "Share not found")
return Response(status_code=204, headers=PRIVATE_HEADERS)
@router.get("/shares/{share_token}")
async def read_share(share_token: ShareToken):
share = await share_repository().get_public(share_token)
if share is None:
raise HTTPException(404, "Share unavailable", headers=PRIVATE_HEADERS)
return JSONResponse({**share["snapshot"], "checkpoint_id": share["checkpoint_id"], "seq": share["seq"], "created_at": share["created_at"], "expires_at": share["expires_at"], "artifacts": share["artifacts"]}, headers=PRIVATE_HEADERS)
@router.get("/shares/{share_token}/artifacts/{path:path}")
@router.head("/shares/{share_token}/artifacts/{path:path}")
async def read_share_artifact(share_token: ShareToken, path: str, request: Request, download: bool = False):
artifact = await share_repository().get_artifact(share_token, path)
if artifact is None:
raise HTTPException(404, "Shared artifact unavailable", headers=PRIVATE_HEADERS)
content, status, range_headers = _slice_byte_range(artifact["content"], request.headers.get("range"))
# Even a direct tab navigation cannot run active content with our origin.
# HTML previews render in an opaque sandbox and cannot contact the network.
headers = {
**PRIVATE_HEADERS,
**range_headers,
"Content-Disposition": _build_content_disposition("attachment" if download else "inline", path.rsplit("/", 1)[-1]),
"Content-Length": str(len(content)),
}
# Chromium's built-in PDF viewer cannot load in a CSP sandbox. The PDF
# media type and nosniff keep it outside the active HTML document surface.
if artifact["mime_type"] != "application/pdf":
headers["Content-Security-Policy"] = "sandbox; default-src 'none'; style-src 'unsafe-inline'; img-src data:; font-src data:; base-uri 'none'; form-action 'none'"
return Response(content if request.method != "HEAD" else b"", status_code=status, media_type=artifact["mime_type"], headers=headers)
@router.post("/shares/{share_token}/fork")
@require_permission("threads", "write")
async def import_share(share_token: ShareToken, body: ShareForkRequest, request: Request):
try:
return JSONResponse(await await_drained(fork_share(request, share_repository(), share_token, str(body.request_id))), headers=PRIVATE_HEADERS)
except ConflictError:
raise HTTPException(409, "Import is already in progress; retry shortly") from None
except AgentRunError as exc:
raise HTTPException(exc.status_code, str(exc)) from None
except HTTPException:
raise
except Exception:
logger.exception("Failed to import conversation share")
raise HTTPException(500, "Unable to import share") from None

View File

@ -0,0 +1,135 @@
"""Import a share into a fresh owner-scoped default-agent conversation."""
import os
import uuid
from contextlib import asynccontextmanager
from fastapi import HTTPException, Request
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.types import Overwrite
from app.gateway.deps import get_checkpointer, get_run_event_store, get_run_manager, get_thread_store
from app.gateway.extension_artifacts import snapshot_sandbox
from app.gateway.history_seed import reserve_history_seed, write_history_seed
from app.gateway.persistent_writes import run_drained_write
from app.gateway.services import build_checkpoint_state_mutation_accessor
from app.gateway.share_snapshot import MAX_SHARE_FILE_BYTES, OUTPUT_PREFIX, artifact_path
from deerflow.config.paths import get_paths
from deerflow.persistence.thread_shares.sql import ThreadShareRepository
from deerflow.runtime import ThreadOperationKind
from deerflow.runtime.events.store.db import DbRunEventStore
from deerflow.runtime.goal import goal_thread_lock
from deerflow.runtime.journal import build_branch_history_seed_events
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.sandbox.artifact_snapshot import publish_snapshot
from deerflow.utils.file_io import run_file_io
def _write_fork_files(owner: str, thread_id: str, artifacts: list[dict]) -> None:
"""A fresh UUID directory, descriptor-confined descendants and exclusive files."""
root = get_paths().sandbox_outputs_dir(thread_id, user_id=owner)
root.mkdir(mode=0o700, parents=True, exist_ok=False)
directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
root_fd = os.open(root, directory_flags)
try:
for artifact in artifacts:
parts = artifact_path(artifact["path"]).split("/")
directory = os.dup(root_fd)
try:
for part in parts[:-1]:
try:
os.mkdir(part, 0o700, dir_fd=directory)
except FileExistsError:
pass
following = os.open(part, directory_flags, dir_fd=directory)
os.close(directory)
directory = following
os.fchmod(directory, 0o777)
fd = os.open(parts[-1], os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
with os.fdopen(fd, "wb") as target:
target.write(artifact["content"])
target.flush()
os.fchmod(target.fileno(), 0o666)
os.fsync(target.fileno())
finally:
os.close(directory)
# The bind mount is writable by the sandbox UID even when Gateway runs
# under a different UID or a restrictive umask.
os.fchmod(root_fd, 0o777)
finally:
os.close(root_fd)
@asynccontextmanager
async def _reserve_import(request: Request, thread_id: str, owner: str):
if isinstance(get_run_event_store(request), DbRunEventStore):
async with reserve_history_seed(request, thread_id) as reservation:
yield reservation
else:
async with get_run_manager(request).reserve_thread_operation(thread_id, kind=ThreadOperationKind.checkpoint_write, user_id=owner):
yield None
async def fork_share(request: Request, repo: ThreadShareRepository, token: str, request_id: str) -> dict:
share = await repo.get_public(token)
if share is None:
raise HTTPException(404, "Share unavailable")
owner = get_effective_user_id()
thread_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"deerflow-share:{owner}:{share['share_id']}:{request_id}"))
threads = get_thread_store(request)
async with goal_thread_lock(thread_id):
existing = await threads.get(thread_id, user_id=owner)
if existing is not None:
if existing.get("metadata", {}).get("share_import_complete") is True:
return {"thread_id": thread_id}
raise HTTPException(409, "The previous import did not finish; create a new copy")
artifacts = []
for descriptor in share["artifacts"]:
artifact = await repo.get_artifact(token, descriptor["path"])
if artifact is None:
raise HTTPException(404, "Share unavailable")
artifacts.append(artifact)
# Tool results become ordinary historical assistant text. The new thread
# inherits no tool execution, system prompt, Agent, project or auth grants.
messages = []
for item in share["snapshot"]["messages"]:
content = item["content"]
if item["type"] == "tool":
content = f"{item['name']}:\n{content}"
cls = HumanMessage if item["type"] == "human" else AIMessage
messages.append(cls(content=content, id=str(uuid.uuid4())))
title = str(share["snapshot"]["title"])[:256]
# Recheck after loading all bytes, before creating any new workspace state.
if await repo.get_public(token) is None:
raise HTTPException(404, "Share unavailable")
async with _reserve_import(request, thread_id, owner) as reservation:
existing = await threads.get(thread_id, user_id=owner)
if existing is not None:
if existing.get("metadata", {}).get("share_import_complete") is True:
return {"thread_id": thread_id}
raise HTTPException(409, "The previous import did not finish; create a new copy")
created = False
try:
record = await threads.create(thread_id, user_id=owner, display_name=title)
created = True
if artifacts:
await run_drained_write("Import shared artifacts", _write_fork_files, (), owner, thread_id, artifacts)
async with snapshot_sandbox(request, owner, thread_id) as sandbox:
if sandbox is not None:
for artifact in artifacts:
await run_file_io(publish_snapshot, sandbox, artifact["path"], artifact["content"], area="outputs", max_bytes=MAX_SHARE_FILE_BYTES)
accessor, config = build_checkpoint_state_mutation_accessor(request, thread_id=thread_id, as_node="branch")
await accessor.aupdate(config, {"messages": Overwrite(messages), "title": title, "artifacts": Overwrite([OUTPUT_PREFIX + a["path"] for a in artifacts])}, as_node="branch")
events = build_branch_history_seed_events(messages, thread_id=thread_id, run_id_prefix=f"branch-seed-{thread_id}", parent_thread_id=thread_id)
await write_history_seed(request, events, reservation=reservation, seed_kind="branch", thread=record)
await threads.update_metadata(thread_id, {"share_import_complete": True}, user_id=owner)
except Exception:
if created:
# Keep admission until rollback finishes, so a failed import
# cannot be continued while its files/checkpoints are removed.
await get_checkpointer(request).adelete_thread(thread_id)
await get_run_event_store(request).delete_by_thread(thread_id, user_id=owner)
await run_drained_write("Remove failed share import", get_paths().delete_thread_dir, (), thread_id, user_id=owner)
await threads.delete(thread_id, user_id=owner)
raise
return {"thread_id": thread_id}

View File

@ -0,0 +1,119 @@
"""Build a portable transcript under the caller's exclusive thread reservation."""
import json
import mimetypes
from pathlib import PurePosixPath
from fastapi import HTTPException, Request
from app.gateway.conversation_access import _visible_text
from app.gateway.conversation_reader import read_visible_message_page
from app.gateway.deps import get_run_event_store, get_run_manager, get_thread_store
from app.gateway.extension_artifacts import read_snapshot
from app.gateway.services import build_thread_checkpoint_state_accessor
from deerflow.runtime import serialize_channel_values_for_api
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.file_io import run_file_io
MAX_SHARE_MESSAGES = 10_000
MAX_SHARE_JSON_BYTES = 10 * 1024 * 1024
MAX_SHARE_ARTIFACTS = 64
MAX_SHARE_FILE_BYTES = 20 * 1024 * 1024
MAX_SHARE_TOTAL_BYTES = 100 * 1024 * 1024
OUTPUT_PREFIX = "/mnt/user-data/outputs/"
def artifact_path(value: str) -> str:
"""Only canonical relative output paths are ever persisted or imported."""
path = value.removeprefix(OUTPUT_PREFIX)
if not path or len(path) > 1024 or path.startswith("/") or "\\" in path or any(ord(c) < 32 for c in path) or any(p in {"", ".", ".."} for p in path.split("/")) or str(PurePosixPath(path)) != path:
raise HTTPException(422, "Invalid shared artifact path")
return path
def project_message(message: dict, index: int) -> dict | None:
"""Share visible text, not execution metadata, prompts or private attachments."""
extra = message.get("additional_kwargs") or {}
if not isinstance(extra, dict) or extra.get("hide_from_ui") or message.get("name") == "summary":
return None
visible = _visible_text({"content": message})
identity = str(message.get("id") or f"message-{index}")
if visible:
role, content = visible
return {"id": identity, "type": "human" if role == "user" else "ai", "content": content}
if message.get("type") == "tool":
content = message.get("content")
if not isinstance(content, str):
content = json.dumps(content, ensure_ascii=False)
return {"id": identity, "type": "tool", "name": str(message.get("name") or "tool"), "content": content}
return None
def project_transcript(messages: list[dict]) -> list[dict]:
projected: list[dict] = []
positions: dict[str, int] = {}
for index, message in enumerate(messages):
if not isinstance(message, dict):
continue
item = project_message(message, index)
if item is None:
continue
identity = item["id"].removesuffix("__user") if item["type"] == "human" else item["id"]
if identity in positions:
projected[positions[identity]] = item
else:
positions[identity] = len(projected)
projected.append(item)
return projected
async def capture_share(request: Request, thread_id: str, expected_checkpoint: str | None) -> dict:
owner = get_effective_user_id()
record = await get_thread_store(request).get(thread_id, user_id=owner)
if record is None:
raise HTTPException(404, "Thread not found")
accessor, config = await build_thread_checkpoint_state_accessor(request, thread_id=thread_id, fail_closed=True)
state = await accessor.aget(config)
checkpoint_id = (state.config or {}).get("configurable", {}).get("checkpoint_id")
if not checkpoint_id or state.next:
raise HTTPException(409, "Finish the current turn before sharing")
if expected_checkpoint is not None and checkpoint_id != expected_checkpoint:
raise HTTPException(409, "Thread changed; refresh before sharing")
rows: list[dict] = []
cursor = None
while True:
page, more = await read_visible_message_page(event_store=get_run_event_store(request), run_manager=get_run_manager(request), thread_id=thread_id, user_id=owner, limit=200, before_seq=cursor)
rows = page + rows
if len(rows) > MAX_SHARE_MESSAGES:
raise HTTPException(413, "Conversation exceeds the share limit")
if not more:
break
if not page or (cursor is not None and page[0]["seq"] >= cursor):
raise HTTPException(500, "Unable to capture complete conversation history")
cursor = page[0]["seq"]
values = serialize_channel_values_for_api(state.values)
source_messages = [row["content"] for row in rows] if rows else values.get("messages", [])
if len(source_messages) > MAX_SHARE_MESSAGES:
raise HTTPException(413, "Conversation exceeds the share limit")
messages = await run_file_io(project_transcript, source_messages)
if not messages:
raise HTTPException(409, "There are no visible messages to share")
snapshot = {"title": record.get("display_name") or values.get("title") or "Shared conversation", "messages": messages}
if len(json.dumps(snapshot, ensure_ascii=False).encode()) > MAX_SHARE_JSON_BYTES:
raise HTTPException(413, "Conversation exceeds the share limit")
paths = list(dict.fromkeys(values.get("artifacts") or []))
if len(paths) > MAX_SHARE_ARTIFACTS:
raise HTTPException(413, "Too many artifacts to share")
artifacts: list[dict] = []
total = 0
for path in paths:
if not isinstance(path, str) or not path.startswith(OUTPUT_PREFIX):
raise HTTPException(422, "Only output artifacts can be shared")
relative = artifact_path(path)
content = await read_snapshot(request, owner, thread_id, path, max_bytes=MAX_SHARE_FILE_BYTES)
total += len(content)
if total > MAX_SHARE_TOTAL_BYTES:
raise HTTPException(413, "Artifacts exceed the total share limit")
mime = await run_file_io(lambda: mimetypes.guess_type(relative)[0] or "application/octet-stream")
artifacts.append({"path": relative, "content": content, "mime_type": mime})
return {"thread_id": thread_id, "owner_id": owner, "incarnation": record.get("incarnation"), "checkpoint_id": checkpoint_id, "seq": max((row["seq"] for row in rows), default=0), "snapshot": snapshot, "artifacts": artifacts}

View File

@ -0,0 +1,45 @@
"""Add immutable conversation shares and copied artifact bytes."""
import sqlalchemy as sa
from alembic import op
revision = "0055_thread_shares"
down_revision = "0054_project_memory"
branch_labels = None
depends_on = None
def upgrade():
if not sa.inspect(op.get_bind()).has_table("thread_shares"):
op.create_table(
"thread_shares",
sa.Column("share_id", sa.String(32), primary_key=True),
sa.Column("token_hash", sa.String(64), nullable=False, unique=True),
sa.Column("thread_id", sa.String(64), sa.ForeignKey("threads_meta.thread_id", ondelete="CASCADE"), nullable=False),
sa.Column("owner_id", sa.String(64), nullable=False),
sa.Column("incarnation", sa.String(32)),
sa.Column("checkpoint_id", sa.String(128), nullable=False),
sa.Column("seq", sa.BigInteger(), nullable=False),
sa.Column("snapshot", sa.JSON(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("revoked_at", sa.DateTime(timezone=True)),
)
op.create_index("ix_thread_shares_owner_thread", "thread_shares", ["owner_id", "thread_id"])
if not sa.inspect(op.get_bind()).has_table("share_artifacts"):
op.create_table(
"share_artifacts",
sa.Column("share_id", sa.String(32), sa.ForeignKey("thread_shares.share_id", ondelete="CASCADE"), primary_key=True),
sa.Column("path", sa.String(1024), primary_key=True),
sa.Column("mime_type", sa.String(128), nullable=False),
sa.Column("sha256", sa.String(64), nullable=False),
sa.Column("size", sa.BigInteger(), nullable=False),
sa.Column("content", sa.LargeBinary(), nullable=False),
)
def downgrade():
if sa.inspect(op.get_bind()).has_table("share_artifacts"):
op.drop_table("share_artifacts")
if sa.inspect(op.get_bind()).has_table("thread_shares"):
op.drop_table("thread_shares")

View File

@ -42,12 +42,15 @@ from deerflow.persistence.skill_mutations.model import SkillAssetRow, SkillOpera
from deerflow.persistence.subagent_batches.model import SubagentBatchItemRow, SubagentBatchRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.thread_reads.model import ThreadReadMarkerRow, ThreadReadVersionRow
from deerflow.persistence.thread_shares.model import ShareArtifactRow, ThreadShareRow
from deerflow.persistence.tool_approvals.model import ToolApprovalRow
from deerflow.persistence.user.im_notifications import ImNotificationPreferenceRow
from deerflow.persistence.user.model import UserPreferenceRow, UserRow
from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow
__all__ = [
"ShareArtifactRow",
"ThreadShareRow",
"ToolApprovalRow",
"ExtensionExecutionLeaseRow",
"AgentRow",

View File

@ -0,0 +1 @@
"""Immutable, revocable conversation share snapshots."""

View File

@ -0,0 +1,37 @@
"""Share payloads and their copied artifacts live in the application database."""
from datetime import datetime
from sqlalchemy import JSON, BigInteger, DateTime, ForeignKey, Index, LargeBinary, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
class ThreadShareRow(Base):
__tablename__ = "thread_shares"
share_id: Mapped[str] = mapped_column(String(32), primary_key=True)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
thread_id: Mapped[str] = mapped_column(String(64), ForeignKey("threads_meta.thread_id", ondelete="CASCADE"), nullable=False)
owner_id: Mapped[str] = mapped_column(String(64), nullable=False)
incarnation: Mapped[str | None] = mapped_column(String(32))
checkpoint_id: Mapped[str] = mapped_column(String(128), nullable=False)
seq: Mapped[int] = mapped_column(BigInteger, nullable=False)
snapshot: Mapped[dict] = mapped_column(JSON, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
__table_args__ = (Index("ix_thread_shares_owner_thread", "owner_id", "thread_id"),)
class ShareArtifactRow(Base):
__tablename__ = "share_artifacts"
share_id: Mapped[str] = mapped_column(String(32), ForeignKey("thread_shares.share_id", ondelete="CASCADE"), primary_key=True)
path: Mapped[str] = mapped_column(String(1024), primary_key=True)
mime_type: Mapped[str] = mapped_column(String(128), nullable=False)
sha256: Mapped[str] = mapped_column(String(64), nullable=False)
size: Mapped[int] = mapped_column(BigInteger, nullable=False)
content: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)

View File

@ -0,0 +1,120 @@
"""Only token digests are stored; each public read rechecks expiry and revocation."""
import hashlib
import secrets
import uuid
from datetime import UTC, datetime
from sqlalchemy import case, delete, func, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.thread_shares.model import ShareArtifactRow, ThreadShareRow
from deerflow.utils.time import coerce_iso
class ThreadShareRepository:
def __init__(self, session_factory: async_sessionmaker[AsyncSession]):
self._sf = session_factory
@staticmethod
def _summary(row: ThreadShareRow) -> dict:
return {
key: coerce_iso(value) if isinstance(value, datetime) else value
for key, value in {
"share_id": row.share_id,
"checkpoint_id": row.checkpoint_id,
"seq": row.seq,
"created_at": row.created_at,
"expires_at": row.expires_at,
"revoked_at": row.revoked_at,
}.items()
}
@staticmethod
def _live(token: str):
return (
select(ThreadShareRow)
.join(ThreadMetaRow, ThreadMetaRow.thread_id == ThreadShareRow.thread_id)
.where(
ThreadShareRow.token_hash == hashlib.sha256(token.encode()).hexdigest(),
ThreadShareRow.revoked_at.is_(None),
ThreadShareRow.expires_at > datetime.now(UTC),
ThreadMetaRow.user_id == ThreadShareRow.owner_id,
or_(ThreadMetaRow.incarnation == ThreadShareRow.incarnation, ThreadMetaRow.incarnation.is_(None) & ThreadShareRow.incarnation.is_(None)),
)
)
async def create(self, *, thread_id: str, owner_id: str, incarnation: str | None, checkpoint_id: str, seq: int, snapshot: dict, artifacts: list[dict], expires_at: datetime) -> dict:
token = secrets.token_urlsafe(32)
row = ThreadShareRow(
share_id=uuid.uuid4().hex,
token_hash=hashlib.sha256(token.encode()).hexdigest(),
thread_id=thread_id,
owner_id=owner_id,
incarnation=incarnation,
checkpoint_id=checkpoint_id,
seq=seq,
snapshot=snapshot,
created_at=datetime.now(UTC),
expires_at=expires_at,
)
async with self._sf() as session:
session.add(row)
await session.flush()
for artifact in artifacts:
content = artifact["content"]
session.add(ShareArtifactRow(share_id=row.share_id, path=artifact["path"], content=content, mime_type=artifact["mime_type"], size=len(content), sha256=hashlib.sha256(content).hexdigest()))
await session.commit()
return {**self._summary(row), "token": token}
async def count_active(self, thread_id: str, owner_id: str) -> int:
async with self._sf() as session:
return (
await session.scalar(
select(func.count()).select_from(ThreadShareRow).where(ThreadShareRow.thread_id == thread_id, ThreadShareRow.owner_id == owner_id, ThreadShareRow.revoked_at.is_(None), ThreadShareRow.expires_at > datetime.now(UTC))
)
or 0
)
async def list_owned(self, thread_id: str, owner_id: str) -> list[dict]:
async with self._sf() as session:
active = ThreadShareRow.revoked_at.is_(None) & (ThreadShareRow.expires_at > datetime.now(UTC))
unrevoked = ThreadShareRow.revoked_at.is_(None)
rows = await session.execute(
select(ThreadShareRow.share_id, ThreadShareRow.checkpoint_id, ThreadShareRow.seq, ThreadShareRow.created_at, ThreadShareRow.expires_at, ThreadShareRow.revoked_at)
.where(ThreadShareRow.thread_id == thread_id, ThreadShareRow.owner_id == owner_id)
.order_by(case((active, 0), (unrevoked, 1), else_=2), ThreadShareRow.created_at.desc())
.limit(100)
)
return [self._summary(row) for row in rows]
async def get_public(self, token: str) -> dict | None:
async with self._sf() as session:
row = await session.scalar(self._live(token))
if row is None:
return None
artifacts = await session.execute(select(ShareArtifactRow.path, ShareArtifactRow.mime_type, ShareArtifactRow.sha256, ShareArtifactRow.size).where(ShareArtifactRow.share_id == row.share_id).order_by(ShareArtifactRow.path))
return {**self._summary(row), "snapshot": row.snapshot, "artifacts": [dict(a._mapping) for a in artifacts]}
async def get_artifact(self, token: str, path: str) -> dict | None:
# One statement checks the grant AND retrieves the bytes. No cached grant
# and no path resolution against the live owner's filesystem.
live_ids = self._live(token).with_only_columns(ThreadShareRow.share_id)
async with self._sf() as session:
row = await session.scalar(select(ShareArtifactRow).where(ShareArtifactRow.share_id.in_(live_ids), ShareArtifactRow.path == path))
return row.to_dict() if row else None
async def revoke(self, thread_id: str, share_id: str, owner_id: str) -> bool:
async with self._sf() as session:
result = await session.execute(
update(ThreadShareRow)
.where(ThreadShareRow.share_id == share_id, ThreadShareRow.thread_id == thread_id, ThreadShareRow.owner_id == owner_id)
.values(revoked_at=datetime.now(UTC), snapshot={})
.returning(ThreadShareRow.share_id)
)
if result.scalar_one_or_none() is None:
return False
await session.execute(delete(ShareArtifactRow).where(ShareArtifactRow.share_id == share_id))
await session.commit()
return True

View File

@ -22,21 +22,22 @@ logger = logging.getLogger(__name__)
_SCRIPT = r"""
import base64, hashlib, json, os, stat, sys
p = json.loads(sys.argv[1])
limit = p.get('max_bytes', 2097152)
directory = None
def stamp(s):
return [s.st_dev, s.st_ino, s.st_size, s.st_mtime_ns, s.st_ctime_ns]
def opened(name):
fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
s = os.fstat(fd)
if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size > 2097152:
if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size > limit:
os.close(fd)
raise ValueError('Unsafe or oversized file')
return os.fdopen(fd, 'rb'), s
def digest(name):
source, before = opened(name)
with source:
data = source.read(2097153)
if len(data) > 2097152 or stamp(before) != stamp(os.fstat(source.fileno())):
data = source.read(limit + 1)
if len(data) > limit or stamp(before) != stamp(os.fstat(source.fileno())):
raise ValueError('File changed during snapshot')
return hashlib.sha256(data).hexdigest(), len(data)
try:
@ -65,7 +66,7 @@ try:
fd = os.open(p['temporary'], flags, 0o600, dir_fd=directory)
with os.fdopen(fd, 'ab') as target:
s = os.fstat(target.fileno())
if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size != p['offset'] or s.st_size + len(data) > 2097152:
if not stat.S_ISREG(s.st_mode) or s.st_nlink != 1 or s.st_size != p['offset'] or s.st_size + len(data) > limit:
raise ValueError('Staging file changed')
target.write(data)
target.flush()
@ -132,23 +133,23 @@ def validate_snapshot_support(sandbox: Sandbox) -> None:
_call(sandbox, "probe")
def publish_snapshot(sandbox: Sandbox, filename: str, content: bytes) -> None:
"""Publish at most 2 MiB into uploads, reusing only identical regular files.
def publish_snapshot(sandbox: Sandbox, filename: str, content: bytes, *, area: str = "uploads", max_bytes: int = MAX_BYTES) -> None:
"""Publish bounded bytes, reusing only identical regular files.
Unsafe paths, changed existing files or rejected remote operations raise
AgentRunError(422). Existing edits are retained; staging cleanup is attempted
even on failure. Run this synchronous command sequence off the event loop.
"""
if not isinstance(content, bytes) or len(content) > MAX_BYTES:
raise AgentRunError(422, "Artifact exceeds 2 MiB")
if area not in {"uploads", "outputs"} or not isinstance(content, bytes) or len(content) > max_bytes:
raise AgentRunError(422, "Artifact exceeds the snapshot size limit")
temporary = ".artifact-" + uuid.uuid4().hex + ".part"
try:
for offset in range(0, max(1, len(content)), CHUNK_BYTES):
_call(sandbox, "append", area="uploads", name=filename, temporary=temporary, offset=offset, data=base64.b64encode(content[offset : offset + CHUNK_BYTES]).decode("ascii"))
_call(sandbox, "publish", area="uploads", name=filename, temporary=temporary, sha256=hashlib.sha256(content).hexdigest(), size=len(content))
_call(sandbox, "append", area=area, max_bytes=max_bytes, name=filename, temporary=temporary, offset=offset, data=base64.b64encode(content[offset : offset + CHUNK_BYTES]).decode("ascii"))
_call(sandbox, "publish", area=area, max_bytes=max_bytes, name=filename, temporary=temporary, sha256=hashlib.sha256(content).hexdigest(), size=len(content))
finally:
try:
_call(sandbox, "cleanup", area="uploads", name=filename, temporary=temporary)
_call(sandbox, "cleanup", area=area, name=filename, temporary=temporary)
except Exception:
logger.warning("Failed to clean remote artifact staging file", exc_info=True)
@ -158,20 +159,20 @@ def verify_snapshot(sandbox: Sandbox, filename: str, sha256: str, size: int) ->
_call(sandbox, "verify", area="uploads", name=filename, sha256=sha256, size=size)
def read_snapshot(sandbox: Sandbox, relative_path: str) -> bytes:
"""Read at most 2 MiB from outputs with file identity checked across chunks.
def read_snapshot(sandbox: Sandbox, relative_path: str, *, max_bytes: int = MAX_BYTES) -> bytes:
"""Read bounded output bytes with file identity checked across chunks.
Raise AgentRunError(422) for unsafe paths, changed files or invalid remote
replies. Run this synchronous command sequence off the event loop.
"""
content, stamp = bytearray(), None
while True:
result = _call(sandbox, "read", area="outputs", name=relative_path, offset=len(content), stamp=stamp)
result = _call(sandbox, "read", area="outputs", name=relative_path, offset=len(content), stamp=stamp, max_bytes=max_bytes)
try:
data = base64.b64decode(result["data"], validate=True)
size = result["size"]
current_stamp = result["stamp"]
valid = type(size) is int and 0 <= size <= MAX_BYTES and len(data) <= CHUNK_BYTES and len(content) + len(data) <= size
valid = type(size) is int and 0 <= size <= max_bytes and len(data) <= CHUNK_BYTES and len(content) + len(data) <= size
if not valid or (stamp is not None and stamp != current_stamp) or (not data and len(content) != size):
raise ValueError("Invalid remote snapshot")
except (ValueError, KeyError, TypeError) as exc:

View File

@ -0,0 +1,27 @@
"""Snapshot capture and Fork copy real files without blocking the event loop."""
import asyncio
import os
from types import SimpleNamespace
import pytest
from app.gateway import extension_artifacts, share_fork
from app.gateway.persistent_writes import run_drained_write
from deerflow.config.paths import Paths
@pytest.mark.asyncio
@pytest.mark.skipif(os.name != "posix", reason="Descriptor-confined snapshots require POSIX")
async def test_share_capture_and_fork_file_publication(tmp_path, monkeypatch):
paths = await asyncio.to_thread(Paths, base_dir=tmp_path)
monkeypatch.setattr(extension_artifacts, "get_paths", lambda: paths)
monkeypatch.setattr(share_fork, "get_paths", lambda: paths)
monkeypatch.setattr(extension_artifacts, "get_sandbox_provider", lambda: SimpleNamespace(uses_thread_data_mounts=True))
content = b"share data\x00\xff" * 1024
artifact = {"path": "nested/report.txt", "content": content}
await run_drained_write("Write source outputs", share_fork._write_fork_files, (), "alice", "source", [artifact])
captured = await extension_artifacts.read_snapshot(None, "alice", "source", "/mnt/user-data/outputs/nested/report.txt", max_bytes=20 * 1024 * 1024)
await run_drained_write("Import shared outputs", share_fork._write_fork_files, (), "bob", "copy", [{**artifact, "content": captured}])
actual = await asyncio.to_thread((paths.sandbox_outputs_dir("copy", user_id="bob") / artifact["path"]).read_bytes)
assert actual == content

View File

@ -44,9 +44,10 @@ pytestmark = pytest.mark.asyncio
async def test_0047_remains_in_the_single_migration_chain():
assert bootstrap._get_head_revision() == "0054_project_memory"
assert bootstrap._get_head_revision() == "0055_thread_shares"
script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
assert script.get_heads() == ["0054_project_memory"]
assert script.get_heads() == ["0055_thread_shares"]
assert script.get_revision("0055_thread_shares").down_revision == "0054_project_memory"
assert script.get_revision("0054_project_memory").down_revision == "0053_tool_approvals"
assert script.get_revision("0053_tool_approvals").down_revision == "0052_batch_continuation"
assert script.get_revision("0052_batch_continuation").down_revision == "0051_merge_run_inputs_fencing"

View File

@ -52,7 +52,7 @@ async def test_both_migration_branches_upgrade_to_one_head(tmp_path, backend, pr
)
await asyncio.to_thread(command.upgrade, cfg, "head")
async with engine.connect() as conn:
assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == ["0054_project_memory"]
assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == [bootstrap._get_head_revision()]
columns = await conn.run_sync(lambda sync: {c["name"] for c in inspect(sync).get_columns("runs")})
assert {"run_inputs_json", "run_incarnation", "producer_epoch", "producer_state"} <= columns
assert (await conn.execute(text("SELECT status,run_inputs_json,run_incarnation,producer_epoch,producer_state FROM runs WHERE run_id='old'"))).one() == ("success", None, None, None, None)

View File

@ -45,7 +45,7 @@ async def test_bootstrap_upgrades_tool_approvals_database_without_losing_data(tm
await conn.execute(text("INSERT INTO tool_approvals (request_id,user_id,thread_id,incarnation,run_id,fingerprint,kind,status,payload,created_at) VALUES ('r1','u','t','i','run','fingerprint','audit','consumed','{}',1)"))
await bootstrap.bootstrap_schema(engine, backend="sqlite")
async with engine.connect() as conn:
assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == ["0054_project_memory"]
assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars().all() == [bootstrap._get_head_revision()]
assert (await conn.execute(text("SELECT instructions, memory FROM projects WHERE id='p1'"))).one() == ("Keep this instruction", None)
assert (await conn.execute(text("SELECT status FROM tool_approvals WHERE request_id='r1'"))).scalar_one() == "consumed"
finally:

View File

@ -0,0 +1,59 @@
"""Migration and ORM shapes agree; upgrades preserve existing conversations."""
import asyncio
import os
import uuid
from datetime import UTC, datetime
import pytest
from alembic import command
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from support.postgres import asyncpg_test_url
from deerflow.persistence import bootstrap
from deerflow.persistence.base import Base
from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
from deerflow.persistence.thread_meta.model import ThreadMetaRow
@pytest.mark.asyncio
@pytest.mark.parametrize("backend", ["sqlite", "postgres"])
async def test_share_upgrade_downgrade_and_schema(tmp_path, backend):
schema = None
if backend == "sqlite":
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'migration.db'}")
else:
uri = os.environ.get("TEST_POSTGRES_URI")
if not uri:
pytest.skip("requires TEST_POSTGRES_URI")
schema = "share_migration_" + uuid.uuid4().hex
engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema))
config = bootstrap._get_alembic_config(engine, postgres_schema=schema or "")
try:
if schema:
async with engine.begin() as conn:
await conn.execute(text(f'CREATE SCHEMA "{schema}"'))
await asyncio.to_thread(command.upgrade, config, "0054_project_memory")
async with engine.begin() as conn:
await conn.execute(
ThreadMetaRow.__table__.insert().values(thread_id="existing", user_id="alice", display_name="Existing conversation", status="idle", metadata_json={}, created_at=datetime.now(UTC), updated_at=datetime.now(UTC))
)
await asyncio.to_thread(command.upgrade, config, "0055_thread_shares")
async with engine.connect() as conn:
assert (await conn.execute(text("SELECT display_name FROM threads_meta WHERE thread_id='existing'"))).scalar_one() == "Existing conversation"
def schema_diff(sync):
context = MigrationContext.configure(sync, opts={"compare_type": True, "include_object": lambda obj, name, type_, reflected, compare_to: type_ != "table" or name in {"thread_shares", "share_artifacts"}})
return compare_metadata(context, Base.metadata)
assert await conn.run_sync(schema_diff) == []
await asyncio.to_thread(command.downgrade, config, "0054_project_memory")
await asyncio.to_thread(command.upgrade, config, "0055_thread_shares")
finally:
if schema:
async with engine.begin() as conn:
await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
await engine.dispose()

View File

@ -109,3 +109,13 @@ def test_remote_capture_rejects_replacement_between_chunks(remote, tmp_path):
remote.execute_command = command
with pytest.raises(AgentRunError):
read_snapshot(remote, "report.txt")
def test_shared_outputs_keep_larger_bytes_and_nested_paths(remote):
from deerflow.sandbox.artifact_snapshot import publish_snapshot, read_snapshot
content = b"x" * (MAX_BYTES + 1)
publish_snapshot(remote, "charts/report.bin", content, area="outputs", max_bytes=3 * 1024 * 1024)
assert read_snapshot(remote, "charts/report.bin", max_bytes=3 * 1024 * 1024) == content
with pytest.raises(AgentRunError):
read_snapshot(remote, "charts/report.bin")

View File

@ -0,0 +1,141 @@
"""Share capabilities persist copies, never grant access to the source thread."""
import os
import uuid
from datetime import UTC, datetime, timedelta
import pytest
import pytest_asyncio
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from support.postgres import asyncpg_test_url
from deerflow.persistence.base import Base
from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.thread_shares.model import ShareArtifactRow, ThreadShareRow
from deerflow.persistence.thread_shares.sql import ThreadShareRepository
@pytest_asyncio.fixture(params=["sqlite", "postgres"])
async def repository(tmp_path, request):
schema = None
if request.param == "sqlite":
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'shares.db'}")
else:
uri = os.environ.get("TEST_POSTGRES_URI")
if not uri:
pytest.skip("requires TEST_POSTGRES_URI")
schema = "share_repository_" + uuid.uuid4().hex
engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema))
async with engine.begin() as connection:
if schema:
await connection.execute(text(f'CREATE SCHEMA "{schema}"'))
await connection.run_sync(Base.metadata.create_all, tables=[ThreadMetaRow.__table__, ThreadShareRow.__table__, ShareArtifactRow.__table__])
factory = async_sessionmaker(engine, expire_on_commit=False)
async with factory() as session:
session.add(ThreadMetaRow(thread_id="source", user_id="alice", incarnation="original"))
await session.commit()
try:
yield ThreadShareRepository(factory), factory
finally:
if schema:
async with engine.begin() as connection:
await connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
await engine.dispose()
async def create_share(repo, *, expires_at=None):
return await repo.create(
thread_id="source",
owner_id="alice",
incarnation="original",
checkpoint_id="checkpoint-1",
seq=42,
snapshot={"title": "Report", "messages": [{"type": "human", "content": "Question"}]},
artifacts=[{"path": "report.html", "content": b"<h1>Original</h1>", "mime_type": "text/html"}],
expires_at=expires_at or datetime.now(UTC) + timedelta(days=1),
)
@pytest.mark.asyncio
async def test_snapshot_is_durable_and_token_is_not_stored(repository):
repo, factory = repository
created = await create_share(repo)
restarted = ThreadShareRepository(factory)
snapshot = await restarted.get_public(created["token"])
assert snapshot["snapshot"]["messages"][0]["content"] == "Question"
assert snapshot["checkpoint_id"] == "checkpoint-1"
assert snapshot["seq"] == 42
assert (await restarted.get_artifact(created["token"], "report.html"))["content"] == b"<h1>Original</h1>"
async with factory() as session:
row = await session.get(ThreadShareRow, created["share_id"])
assert created["token"] not in str(row.to_dict())
assert "token" not in (await restarted.list_owned("source", "alice"))[0]
assert await restarted.list_owned("source", "bob") == []
assert await restarted.get_public("invalid") is None
assert await restarted.get_artifact(created["token"], "../secret") is None
@pytest.mark.asyncio
async def test_revocation_checks_owner_and_closes_every_surface(repository):
repo, _ = repository
share = await create_share(repo)
assert not await repo.revoke("source", share["share_id"], "bob")
assert await repo.get_public(share["token"]) is not None
assert await repo.revoke("source", share["share_id"], "alice")
assert await repo.get_public(share["token"]) is None
assert await repo.get_artifact(share["token"], "report.html") is None
@pytest.mark.asyncio
async def test_expired_and_recreated_sources_cannot_authorize_old_links(repository):
repo, factory = repository
expired = await create_share(repo, expires_at=datetime.now(UTC) - timedelta(seconds=1))
assert await repo.get_public(expired["token"]) is None
assert await repo.get_artifact(expired["token"], "report.html") is None
current = await create_share(repo)
async with factory() as session:
source = await session.get(ThreadMetaRow, "source")
source.incarnation = "replacement"
await session.commit()
assert await repo.get_public(current["token"]) is None
@pytest.mark.asyncio
async def test_active_link_remains_revocable_after_many_inactive_links(repository):
repo, factory = repository
active = await create_share(repo)
now = datetime.now(UTC)
async with factory() as session:
session.add_all(
[
ThreadShareRow(
share_id=f"revoked-{i}", token_hash=f"digest-{i}", thread_id="source", owner_id="alice", incarnation="original", checkpoint_id="cp", seq=0, snapshot={}, created_at=now, expires_at=now + timedelta(days=1), revoked_at=now
)
for i in range(101)
]
)
await session.commit()
assert await repo.count_active("source", "alice") == 1
assert await repo.count_active("source", "bob") == 0
assert (await repo.list_owned("source", "alice"))[0]["share_id"] == active["share_id"]
await repo.revoke("source", active["share_id"], "alice")
assert await repo.count_active("source", "alice") == 0
@pytest.mark.asyncio
async def test_expired_snapshot_remains_listed_and_revocation_purges_its_copy(repository):
repo, factory = repository
expired = await create_share(repo, expires_at=datetime.now(UTC) - timedelta(seconds=1))
for _ in range(101):
current = await create_share(repo)
assert await repo.revoke("source", current["share_id"], "alice")
assert (await repo.list_owned("source", "alice"))[0]["share_id"] == expired["share_id"]
assert await repo.revoke("source", expired["share_id"], "alice")
async with factory() as session:
row = await session.get(ThreadShareRow, expired["share_id"])
assert row.revoked_at is not None
assert row.snapshot == {}
assert await session.scalar(select(ShareArtifactRow).where(ShareArtifactRow.share_id == expired["share_id"])) is None
assert (await session.get(ThreadMetaRow, "source")).incarnation == "original"

View File

@ -0,0 +1,535 @@
"""Real HTTP/auth, SQL shares, checkpoints, history and host file copies."""
import asyncio
import json
import os
import stat
from contextlib import asynccontextmanager
from types import SimpleNamespace
from uuid import UUID, uuid4
import httpx
import pytest
import pytest_asyncio
from fastapi import FastAPI, HTTPException, Request
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langgraph.checkpoint.memory import InMemorySaver
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from support.postgres import asyncpg_test_url
import deerflow.persistence.models # noqa: F401
from app.gateway import extension_artifacts, history_seed, share_fork, share_snapshot
from app.gateway.auth.models import User
from app.gateway.auth_middleware import AuthMiddleware, _is_public
from app.gateway.csrf_middleware import CSRFMiddleware
from app.gateway.routers import shares, threads
from app.gateway.services import build_checkpoint_state_mutation_accessor
from deerflow.config.authorization_config import AuthorizationConfig
from deerflow.config.paths import Paths
from deerflow.persistence.base import Base
from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
from deerflow.persistence.run.sql import RunRepository
from deerflow.persistence.thread_meta import ThreadMetaRepository
from deerflow.runtime import ConflictError, RunManager, ThreadOperationKind
from deerflow.runtime.events.store.db import DbRunEventStore
from deerflow.runtime.events.store.memory import MemoryRunEventStore
from deerflow.runtime.journal import build_branch_history_seed_events
from deerflow.runtime.user_context import reset_current_user, set_current_user
ALICE = User(id=UUID(int=1), email="alice@example.com", password_hash="unused", system_role="admin")
BOB = User(id=UUID(int=2), email="bob@example.com", password_hash="unused", system_role="user")
pytestmark = [pytest.mark.asyncio, pytest.mark.skipif(os.name != "posix", reason="descriptor-confined artifact capture requires POSIX")]
@pytest_asyncio.fixture(params=[("full", "memory", "sqlite"), ("delta", "memory", "sqlite"), ("full", "db", "sqlite"), ("delta", "db", "sqlite"), ("full", "db", "postgres")])
async def share_app(tmp_path, monkeypatch, request):
schema = None
if request.param[2] == "postgres":
uri = os.environ.get("TEST_POSTGRES_URI")
if not uri:
pytest.skip("requires TEST_POSTGRES_URI")
schema = "share_http_" + uuid4().hex
engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema))
else:
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'http.db'}")
async with engine.begin() as connection:
if schema:
await connection.execute(text(f'CREATE SCHEMA "{schema}"'))
await connection.run_sync(Base.metadata.create_all)
factory = async_sessionmaker(engine, expire_on_commit=False)
monkeypatch.setattr(shares, "get_session_factory", lambda: factory)
monkeypatch.setattr("app.gateway.auth_middleware.is_auth_disabled", lambda: False)
monkeypatch.setattr("app.gateway.csrf_middleware.is_auth_disabled", lambda: False)
monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: AuthorizationConfig())
async def session_user(req):
user = {"alice": ALICE, "bob": BOB}.get(req.cookies.get("access_token"))
if not user:
raise HTTPException(401, "Invalid session")
return user
monkeypatch.setattr("app.gateway.deps.get_current_user_from_request", session_user)
paths = Paths(base_dir=tmp_path / "home")
monkeypatch.setattr(extension_artifacts, "get_paths", lambda: paths)
monkeypatch.setattr(share_fork, "get_paths", lambda: paths)
monkeypatch.setattr(extension_artifacts, "get_sandbox_provider", lambda: SimpleNamespace(uses_thread_data_mounts=True))
app = FastAPI()
app.add_middleware(AuthMiddleware)
app.add_middleware(CSRFMiddleware)
app.state.thread_store = ThreadMetaRepository(factory)
app.state.checkpointer = InMemorySaver()
app.state.run_event_store = DbRunEventStore(factory) if request.param[1] == "db" else MemoryRunEventStore()
app.state.run_manager = RunManager(store=RunRepository(factory), event_store=app.state.run_event_store)
app.state.checkpoint_channel_mode = request.param[0]
app.state.store = None
app.include_router(shares.router)
app.include_router(threads.router)
req = Request({"type": "http", "app": app})
async def accessor(req, *, thread_id, **kwargs):
# Replace agent/model assembly only. Materialization and writes use
# the actual full/delta accessor over the real checkpoint store.
return build_checkpoint_state_mutation_accessor(req, thread_id=thread_id, as_node="branch")
monkeypatch.setattr(share_snapshot, "build_thread_checkpoint_state_accessor", accessor)
await app.state.thread_store.create("source", user_id=str(ALICE.id), display_name="Shared report")
paths.sandbox_outputs_dir("source", user_id=str(ALICE.id)).mkdir(parents=True)
output = paths.sandbox_outputs_dir("source", user_id=str(ALICE.id)) / "report.html"
output.write_text("<h1>Original report</h1><script>parent.pwned=true</script>", encoding="utf-8")
messages = [HumanMessage(content="Visible question", id="h1", additional_kwargs={"files": [{"path": "private-upload"}]})]
messages += [AIMessage(content=f"Step {i}", id=f"ai-{i}") for i in range(205)]
messages += [ToolMessage(content="tool result", name="search", tool_call_id="c1", id="tool"), AIMessage(content="Final report", id="answer")]
hidden = AIMessage(content="private control", id="hidden", additional_kwargs={"hide_from_ui": True})
events = build_branch_history_seed_events([*messages, hidden, SystemMessage(content="secret prompt")], thread_id="source", run_id_prefix="source-run", parent_thread_id="source")
user_token = set_current_user(ALICE)
try:
await app.state.run_event_store.put_batch(events)
finally:
reset_current_user(user_token)
if request.param[1] == "db":
app.state.run_event_store._producer_fencing = "strict"
state, config = await accessor(req, thread_id="source")
config = await state.aupdate(config, {"messages": messages[-1:], "artifacts": ["/mnt/user-data/outputs/report.html"]}, as_node="branch")
try:
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
yield SimpleNamespace(app=app, client=client, paths=paths, output=output, state=state, config=config, request=req, factory=factory)
finally:
if schema:
async with engine.begin() as connection:
await connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
await engine.dispose()
def headers(user="alice", csrf=True):
return {"Cookie": f"access_token={user}; csrf_token=test-csrf", **({"X-CSRF-Token": "test-csrf"} if csrf else {})}
async def make_share(ctx):
response = await ctx.client.post("/api/threads/source/shares", json={}, headers=headers())
assert response.status_code == 201, response.text
return response.json()
async def test_anonymous_snapshot_and_files_are_immutable_and_scoped(share_app):
ctx = share_app
share = await make_share(ctx)
url = f"/api/shares/{share['token']}"
response = await ctx.client.get(url)
assert response.status_code == 200
body = response.json()
assert len(body["messages"]) == 208 # whole paginated history, not the compacted head
assert body["messages"][0]["content"] == "Visible question"
assert body["messages"][-1]["content"] == "Final report"
assert "private-upload" not in response.text and "secret prompt" not in response.text and "private control" not in response.text
assert response.headers["cache-control"] == "no-store"
ctx.output.write_text("changed file", encoding="utf-8")
await ctx.state.aupdate(ctx.config, {"messages": [AIMessage(content="Later private answer", id="later")]}, as_node="branch")
assert (await ctx.client.get(url)).json() == body
artifact = await ctx.client.get(url + "/artifacts/report.html")
assert "Original report" in artifact.text
assert "sandbox" in artifact.headers["content-security-policy"]
assert "allow-scripts" not in artifact.headers["content-security-policy"]
assert artifact.headers["referrer-policy"] == "no-referrer"
ranged = await ctx.client.get(url + "/artifacts/report.html", headers={"Range": "bytes=0-3"})
assert ranged.status_code == 206 and ranged.content == b"<h1>"
assert (await ctx.client.get(url + "/artifacts/private-upload")).status_code == 404
assert (await ctx.client.get("/api/threads/source/state")).status_code == 401
assert (await ctx.client.get("/api/threads/source/state", headers=headers("bob"))).status_code == 404
async def test_share_mutations_require_owner_and_csrf(share_app):
ctx = share_app
assert (await ctx.client.post("/api/threads/source/shares", json={}, headers=headers("bob"))).status_code == 404
assert (await ctx.client.post("/api/threads/source/shares", json={}, headers=headers(csrf=False))).status_code == 403
assert (await ctx.client.post("/api/threads/source/shares", json={"expires_in_seconds": 0}, headers=headers())).status_code == 422
assert (await ctx.client.post("/api/threads/source/shares", json={"checkpoint_id": "old"}, headers=headers())).status_code == 409
async with ctx.app.state.run_manager.reserve_thread_operation("source", kind=ThreadOperationKind.artifact_write, user_id=str(ALICE.id)):
assert (await ctx.client.post("/api/threads/source/shares", json={}, headers=headers())).status_code == 409
share = await make_share(ctx)
path = f"/api/threads/source/shares/{share['share_id']}"
assert (await ctx.client.delete(path, headers=headers("bob"))).status_code == 404
assert (await ctx.client.delete(path, headers=headers())).status_code == 204
public = f"/api/shares/{share['token']}"
assert (await ctx.client.get(public)).status_code == 404
assert (await ctx.client.get(public + "/artifacts/report.html")).status_code == 404
assert (await ctx.client.post(public + "/fork", json={"request_id": str(uuid4())}, headers=headers("bob"))).status_code == 404
async def test_pdf_preview_uses_browser_viewer_without_sandbox(share_app):
ctx = share_app
pdf = b"%PDF-1.4\n% Frozen PDF bytes\n%%EOF"
ctx.output.with_name("report.pdf").write_bytes(pdf)
await ctx.state.aupdate(ctx.config, {"artifacts": ["/mnt/user-data/outputs/report.pdf"]}, as_node="branch")
share = await make_share(ctx)
response = await ctx.client.get(f"/api/shares/{share['token']}/artifacts/report.pdf")
assert response.content == pdf
assert response.headers["content-type"] == "application/pdf"
assert response.headers["x-content-type-options"] == "nosniff"
assert response.headers["cache-control"] == "no-store"
assert "sandbox" not in response.headers.get("content-security-policy", "")
async def test_fork_is_owned_by_visitor_and_survives_revocation(share_app):
ctx = share_app
share = await make_share(ctx)
url = f"/api/shares/{share['token']}/fork"
body = {"request_id": str(uuid4())}
assert (await ctx.client.post(url, json=body)).status_code in {401, 403}
response = await ctx.client.post(url, json=body, headers=headers("bob"))
assert response.status_code == 200, response.text
new_id = response.json()["thread_id"]
assert (await ctx.client.post(url, json=body, headers=headers("bob"))).json() == {"thread_id": new_id}
row = await ctx.app.state.thread_store.get(new_id, user_id=str(BOB.id))
assert row and row["assistant_id"] is None
assert await ctx.app.state.thread_store.get(new_id, user_id=str(ALICE.id)) is None
accessor, config = build_checkpoint_state_mutation_accessor(ctx.request, thread_id=new_id, as_node="branch")
state = await accessor.aget(config)
assert len(state.values["messages"]) == 208
assert all(m.type in {"human", "ai"} for m in state.values["messages"])
assert "sandbox" not in state.values
assert len(await ctx.app.state.run_event_store.list_messages(new_id, limit=1000, user_id=str(BOB.id))) == 208
copied = ctx.paths.sandbox_outputs_dir(new_id, user_id=str(BOB.id)) / "report.html"
assert copied.read_bytes() == ctx.output.read_bytes()
await ctx.client.delete(f"/api/threads/source/shares/{share['share_id']}", headers=headers())
assert copied.exists()
# Continue with a real checkpoint write in the new owner's conversation.
await accessor.aupdate(config, {"messages": [HumanMessage(content="Follow up", id="followup")]}, as_node="branch")
assert (await accessor.aget(config)).values["messages"][-1].content == "Follow up"
async def test_fork_rechecks_completion_after_durable_admission(share_app, monkeypatch):
ctx = share_app
if not isinstance(ctx.app.state.run_event_store, DbRunEventStore):
pytest.skip("requires database-backed history and admission")
share = await make_share(ctx)
url = f"/api/shares/{share['token']}/fork"
body = {"request_id": str(uuid4())}
delayed = asyncio.Event()
resume = asyncio.Event()
reserves = 0
original_reserve = share_fork._reserve_import
first_manager = ctx.app.state.run_manager
second_manager = RunManager(store=RunRepository(ctx.factory), event_store=ctx.app.state.run_event_store)
@asynccontextmanager
async def separate_worker_lock(_thread_id):
yield
@asynccontextmanager
async def park_before_reservation(*args):
nonlocal reserves
reserves += 1
if reserves == 1:
delayed.set()
await resume.wait()
async with original_reserve(*args) as reservation:
yield reservation
monkeypatch.setattr(share_fork, "goal_thread_lock", separate_worker_lock)
monkeypatch.setattr(share_fork, "_reserve_import", park_before_reservation)
monkeypatch.setattr(history_seed, "get_run_manager", lambda request: second_manager if request.headers.get("X-Test-Worker") == "second" else first_manager)
first = asyncio.create_task(ctx.client.post(url, json=body, headers=headers("bob")))
async with asyncio.timeout(10):
try:
await delayed.wait()
second = await ctx.client.post(url, json=body, headers={**headers("bob"), "X-Test-Worker": "second"})
assert second.status_code == 200, second.text
thread_id = second.json()["thread_id"]
copied = ctx.paths.sandbox_outputs_dir(thread_id, user_id=str(BOB.id)) / "report.html"
copied.write_bytes(b"Edited by the recipient")
finally:
resume.set()
duplicate = await first
assert duplicate.status_code == 200, duplicate.text
assert duplicate.json() == {"thread_id": thread_id}
assert [row["thread_id"] for row in await ctx.app.state.thread_store.search(user_id=str(BOB.id))] == [thread_id]
assert copied.read_bytes() == b"Edited by the recipient"
assert len(await ctx.app.state.run_event_store.list_messages(thread_id, limit=1000, user_id=str(BOB.id))) == 208
async def test_fork_continues_under_recipient_run_authority(share_app, monkeypatch, tmp_path):
from _agent_e2e_helpers import FakeToolCallingModel
from app.gateway.routers import thread_runs
from deerflow.agents.lead_agent import agent
from deerflow.config import app_config
from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.mcp import cache, tools
from deerflow.persistence.projects import ProjectRepository
from deerflow.persistence.thread_meta.base import THREAD_PROJECT_METADATA_KEY
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
ctx = share_app
if not isinstance(ctx.app.state.run_event_store, DbRunEventStore) or ctx.app.state.checkpoint_channel_mode != "full":
pytest.skip("exercises the gateway runtime with durable full checkpoints")
publisher, recipient = str(ALICE.id), str(BOB.id)
monkeypatch.setenv("DEER_FLOW_HOME", str(ctx.paths.base_dir))
monkeypatch.setattr("deerflow.config.paths._paths", ctx.paths)
skill_dir = ctx.paths.user_custom_skills_dir(publisher) / "publisher-skill"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text("---\nname: publisher-skill\ndescription: Publisher-only skill\n---\nPublisher-only skill body.", encoding="utf-8")
custom_agent = ctx.paths.user_agent_dir(publisher, "publisher-agent")
custom_agent.mkdir(parents=True)
(custom_agent / "config.yaml").write_text("name: publisher-agent\nskills: [publisher-skill]\n", encoding="utf-8")
(custom_agent / "SOUL.md").write_text("Publisher-only agent instructions.", encoding="utf-8")
from deerflow.mcp.user_config import load_user_mcp_config, user_mcp_config_path
personal_config = user_mcp_config_path(publisher)
personal_config.parent.mkdir(parents=True, exist_ok=True)
personal_config.write_text(json.dumps({"mcpServers": {"publisher": {"type": "http", "url": "https://fixture.example/personal", "headers": {"Authorization": "Bearer personal-publisher"}}}}), encoding="utf-8")
assert load_user_mcp_config(publisher).mcp_servers
assert not load_user_mcp_config(recipient).mcp_servers
async with ctx.factory() as session:
source = await session.get(ThreadMetaRow, "source")
source.assistant_id = "publisher-agent"
await session.commit()
project = await ProjectRepository(ctx.factory).create(name="Publisher project", instructions="Publisher-only instructions", user_id=publisher)
await ctx.app.state.thread_store.set_project("source", project["id"], user_id=publisher)
await ctx.state.aupdate(ctx.config, {"messages": [HumanMessage(content="Publish this", additional_kwargs={"slash_skill_activation": {"skill_name": "publisher-skill"}})]}, as_node="branch")
share = await make_share(ctx)
fork = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": str(uuid4())}, headers=headers("bob"))
assert fork.status_code == 200, fork.text
destination = fork.json()["thread_id"]
copied = await ctx.app.state.thread_store.get(destination, user_id=recipient)
assert THREAD_PROJECT_METADATA_KEY not in copied["metadata"] and copied["assistant_id"] is None
accessor, checkpoint = build_checkpoint_state_mutation_accessor(ctx.request, thread_id=destination, as_node="branch")
copied_messages = (await accessor.aget(checkpoint)).values["messages"]
assert all("slash_skill_activation" not in message.additional_kwargs for message in copied_messages)
config = app_config.AppConfig.model_validate(
{
"models": [{"name": "share-test", "display_name": "Share test", "use": "langchain_openai:ChatOpenAI", "model": "unused"}],
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"skills": {"path": str(tmp_path / "skills")},
"tool_search": {"enabled": False},
"memory": {"enabled": False},
"title": {"enabled": False},
"authorization": {
"enabled": True,
"provider": {
"use": "deerflow.authz.rbac:RbacAuthorizationProvider",
"config": {"roles": {"admin": {}, "user": {"tools": {"allow": "*", "deny": ["denied_tool"]}, "skills": {"allow": "*", "deny": ["restricted-skill"]}}}},
},
},
}
)
restricted = tmp_path / "skills" / "public" / "restricted-skill"
restricted.mkdir(parents=True)
(restricted / "SKILL.md").write_text("---\nname: restricted-skill\ndescription: Role-restricted skill\n---\nRole-restricted skill body.", encoding="utf-8")
monkeypatch.setattr(app_config, "_app_config", config)
monkeypatch.setattr(app_config, "_app_config_is_custom", True)
monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config.authorization)
ctx.app.state.project_repo = ProjectRepository(ctx.factory)
ctx.app.state.stream_bridge = MemoryStreamBridge()
ctx.app.include_router(thread_runs.router)
requests = []
visited_paths = []
def mcp_http(request):
visited_paths.append(request.url.path)
if request.method != "POST":
return httpx.Response(405)
payload = json.loads(request.content)
method = payload["method"]
if "id" not in payload:
return httpx.Response(202)
if method == "initialize":
result = {"protocolVersion": payload["params"]["protocolVersion"], "capabilities": {"tools": {}}, "serverInfo": {"name": "fixture", "version": "1"}}
elif method == "tools/list":
names = {"scoped": ["scoped_probe"], "deployment": ["deployment_probe", "denied_tool"], "personal": ["publisher_only"]}[request.url.path.lstrip("/")]
result = {"tools": [{"name": name, "inputSchema": {"type": "object"}, "description": name} for name in names]}
else:
assert method == "tools/call"
requests.append((payload["params"]["name"], request.headers.get("authorization")))
result = {"content": [{"type": "text", "text": "fixture success"}]}
return httpx.Response(200, json={"jsonrpc": "2.0", "id": payload["id"], "result": result})
def client_factory(headers=None, timeout=None, auth=None):
return httpx.AsyncClient(headers=headers, timeout=timeout, auth=auth, transport=httpx.MockTransport(mcp_http))
original_build = tools.build_servers_config
def build_servers(extensions):
return {name: {**params, "httpx_client_factory": client_factory} for name, params in original_build(extensions).items()}
monkeypatch.setattr(tools, "build_servers_config", build_servers)
calls = []
model_tool_names = []
class RecordingModel(FakeToolCallingModel):
def bind_tools(self, available, **kwargs):
model_tool_names.append({tool.name for tool in available})
return self
def _generate(self, messages, *args, **kwargs):
prompt = "\n".join(str(message.content) for message in messages)
assert "Publisher-only instructions" not in prompt
assert "Publisher-only agent instructions" not in prompt
assert "Publisher-only skill" not in prompt
assert "Role-restricted skill" not in prompt
return super()._generate(messages, *args, **kwargs)
def create_model(**kwargs):
return RecordingModel(
responses=[
AIMessage(
content="",
tool_calls=[{"name": name, "args": {}, "id": name} for name in ("scoped_probe", "deployment_probe", "denied_tool")],
),
AIMessage(content="Recipient continuation finished."),
]
)
async def run_tool_observer(request, handler):
context = request.runtime.context
calls.append((context["user_id"], context["user_role"], context.get("agent_name"), context.get("project_id")))
return await handler(request)
from deerflow.mcp import interceptors
original_interceptors = tools.build_mcp_tool_interceptors
def build_interceptors(*args, **kwargs):
return [run_tool_observer, *original_interceptors(*args, **kwargs)]
monkeypatch.setattr(tools, "build_mcp_tool_interceptors", build_interceptors)
monkeypatch.setattr(interceptors, "build_mcp_tool_interceptors", build_interceptors)
monkeypatch.setattr(agent, "create_chat_model", create_model)
for recipient_credential in ("Bearer recipient", None):
credentials = {publisher: "Bearer publisher"}
if recipient_credential:
credentials[recipient] = recipient_credential
extensions = ExtensionsConfig.model_validate(
{
"mcpServers": {
"scoped": {"type": "http", "url": "https://fixture.example/scoped", "tool_name_prefix": False, "headers": {"Authorization": "Bearer discovery"}, "user_auth": {"users": credentials}},
"deployment": {"type": "http", "url": "https://fixture.example/deployment", "tool_name_prefix": False, "headers": {"Authorization": "Bearer deployment"}},
}
}
)
monkeypatch.setattr(ExtensionsConfig, "from_file", classmethod(lambda cls, *args, **kwargs: extensions))
loaded = await tools.get_mcp_tools(extensions)
assert {tool.name for tool in loaded} == {"scoped_probe", "deployment_probe", "denied_tool"}
monkeypatch.setattr(cache, "get_cached_mcp_tools", lambda: loaded)
requests.clear()
result = await ctx.client.post(f"/api/threads/{destination}/runs/wait", json={"input": {"messages": [{"type": "human", "content": "Continue the report"}]}}, headers=headers("bob"))
assert result.status_code == 200, result.text
assert "Recipient continuation finished." in result.text
assert set(requests) == {("deployment_probe", "Bearer deployment")} | ({("scoped_probe", recipient_credential)} if recipient_credential else set())
assert calls and all(call == (recipient, "user", None, None) for call in calls)
assert model_tool_names and all("denied_tool" not in names for names in model_tool_names)
assert "/personal" not in visited_paths
async def test_imported_nested_files_have_sandbox_write_permissions(share_app):
ctx = share_app
nested = ctx.output.parent / "reports" / "week"
nested.mkdir(parents=True)
(nested / "summary.txt").write_bytes(b"Report")
await ctx.state.aupdate(ctx.config, {"artifacts": ["/mnt/user-data/outputs/reports/week/summary.txt"]}, as_node="branch")
share = await make_share(ctx)
previous_umask = os.umask(0o077)
try:
result = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": str(uuid4())}, headers=headers("bob"))
finally:
os.umask(previous_umask)
assert result.status_code == 200, result.text
root = ctx.paths.sandbox_outputs_dir(result.json()["thread_id"], user_id=str(BOB.id))
output = root / "reports" / "week" / "summary.txt"
assert output.read_bytes() == b"Report"
assert stat.S_IMODE(output.stat().st_mode) == 0o666
for directory in (root, root / "reports", output.parent):
assert stat.S_IMODE(directory.stat().st_mode) == 0o777
async def test_unsafe_artifact_fails_without_publishing_partial_share(share_app):
ctx = share_app
ctx.output.unlink()
ctx.output.symlink_to(ctx.paths.base_dir / "secret.txt")
(ctx.paths.base_dir / "secret.txt").write_text("Do not share", encoding="utf-8")
response = await ctx.client.post("/api/threads/source/shares", json={}, headers=headers())
assert response.status_code == 422, response.text
assert (await ctx.client.get("/api/threads/source/shares", headers=headers())).json() == []
async def test_public_allowlist_never_grants_mutation_authority():
path = "/api/shares/" + "x" * 43
for method in ("POST", "PATCH", "PUT", "DELETE"):
assert not _is_public(path, method)
assert not _is_public(path + "/artifacts/report.html", method)
assert not _is_public(path + "/fork", "GET")
assert _is_public(path, "GET")
assert _is_public(path + "/artifacts/report.html", "HEAD")
async def test_failed_fork_rolls_back_metadata_checkpoint_history_and_files(share_app, monkeypatch):
ctx = share_app
share = await make_share(ctx)
reached = []
original = share_fork.write_history_seed
request_id = str(uuid4())
async def fail_seed(*args, **kwargs):
await original(*args, **kwargs)
reached.append(True)
raise RuntimeError("private filesystem diagnostic")
monkeypatch.setattr(share_fork, "write_history_seed", fail_seed)
response = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": request_id}, headers=headers("bob"))
assert reached == [True]
assert response.status_code == 500 and "private filesystem" not in response.text
assert await ctx.app.state.thread_store.search(user_id=str(BOB.id)) == []
assert list(ctx.paths.base_dir.rglob("report.html")) == [ctx.output]
monkeypatch.setattr(share_fork, "write_history_seed", original)
retry = await ctx.client.post(f"/api/shares/{share['token']}/fork", json={"request_id": request_id}, headers=headers("bob"))
assert retry.status_code == 200, retry.text
assert len(await ctx.app.state.run_event_store.list_messages(retry.json()["thread_id"], limit=1000, user_id=str(BOB.id))) == 208
async def test_disconnect_drains_capture_before_releasing_source(share_app, monkeypatch):
ctx = share_app
entered, release = asyncio.Event(), asyncio.Event()
original = shares.capture_share
async def slow_capture(*args, **kwargs):
entered.set()
await release.wait()
return await original(*args, **kwargs)
monkeypatch.setattr(shares, "capture_share", slow_capture)
task = asyncio.create_task(ctx.client.post("/api/threads/source/shares", json={}, headers=headers()))
try:
await asyncio.wait_for(entered.wait(), 5)
task.cancel()
# A competing writer must not enter while the cancelled request drains.
with pytest.raises(ConflictError):
async with ctx.app.state.run_manager.reserve_thread_operation("source", kind=ThreadOperationKind.artifact_write, user_id=str(ALICE.id)):
pytest.fail("capture released its reservation before finishing")
finally:
release.set()
await asyncio.gather(task, return_exceptions=True)
links = await ctx.client.get("/api/threads/source/shares", headers=headers())
assert len(links.json()) == 1

View File

@ -78,6 +78,7 @@ def _collect_thread_id_routes():
runs,
scheduled_tasks,
search,
shares,
skills,
subagent_batches,
suggestions,
@ -95,6 +96,7 @@ def _collect_thread_id_routes():
runs,
scheduled_tasks,
search,
shares,
skills,
subagent_batches,
suggestions,

Binary file not shown.

After

Width:  |  Height:  |  Size: 87 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 55 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 40 KiB

77
docs/guide/sharing.md Normal file
View File

@ -0,0 +1,77 @@
# Share a conversation snapshot
Open a conversation's sidebar menu, choose **Share**, select an expiry, then
choose **Create snapshot link**. Copy the generated `/share/<token>` URL. Anyone
with that link can read the snapshot without signing in. The dialog lists your
links and lets you revoke them. Revocation and expiry stop subsequent reads,
file downloads and imports; recipients keep copies they already downloaded or
imported.
![The conversation menu's Share action](assets/sharing/share-menu.png)
![Create, copy and revoke a snapshot link](assets/sharing/share-entry.png)
The snapshot contains the visible conversation history and the output files
presented as artifacts at creation time. Later messages and file edits do not
change it. Uploads, hidden messages, system prompts and execution metadata are
excluded. Tool result text is included, so review the conversation before
sharing it. A source conversation must have finished its current turn. Deleting
the source conversation also makes its links unavailable.
The shared page has no composer or source workspace controls. It previews
Markdown, text, CSV/TSV, images, PDF and static HTML; other files are available
for download. Text previews show at most 512 KiB. HTML scripts, forms and network
requests are blocked. The page rechecks availability on return and every
30 seconds while visible.
![A public read-only conversation snapshot](assets/sharing/share-preview.png)
Choose **Continue in my workspace** to create an independent conversation with
copied history and output files. Visitors sign in first and return to the shared
page. The copy uses their default agent and permissions. It does not inherit the
source's agent, project, sandbox, uploads, tool calls or credentials. Tool results
are imported as historical assistant text. The copied conversation remains
usable after the source link is revoked.
## Deployment and limits
Sharing requires SQLite or PostgreSQL persistence. The Gateway's startup
migration creates the share tables. Both the transcript and file bytes are
stored in the database, so backups include shared content and all Gateway workers
read the same frozen bytes. No separate public object store is required.
Links expire after 7 days by default; the supported range is 1 hour to 30 days.
A conversation can have 10 active links. Each snapshot allows 10,000 visible
history rows, 10 MiB of transcript JSON, and 64 artifacts, with limits of 20 MiB
per file and 100 MiB total. Exceeding a limit rejects the whole snapshot. Revoking
a link deletes its stored transcript and file bytes. You can revoke an expired
link from the same dialog to remove its stored copy.
Treat share URLs as access credentials, including in proxy access logs. The
server stores only a hash of the random token; a token is returned once on
creation. Public responses disable caching, referrers and search indexing.
## API contract
All paths below are relative to `/api`.
| Method and path | Access and behavior |
| --- | --- |
| `POST /threads/{id}/shares` | Owner with `threads:write` and `runs:read`; creates the snapshot. Body: optional `expires_in_seconds` and `checkpoint_id`. Returns `201`, token and snapshot coordinates. |
| `GET /threads/{id}/shares` | Owner with `threads:read`; returns up to 100 link summaries, with active links first and expired links awaiting cleanup before revoked links. Tokens are never listed. |
| `DELETE /threads/{id}/shares/{share_id}` | Owner with `threads:write`; revokes access and removes stored snapshot bytes. |
| `GET /shares/{token}` | Anonymous capability read; returns visible messages, artifact manifest, checkpoint and sequence. |
| `GET`, `HEAD /shares/{token}/artifacts/{path}` | Reads only a copied manifest entry. Supports byte ranges and `?download=true`. |
| `POST /shares/{token}/fork` | Signed-in visitor with `threads:write`; body contains a UUID `request_id`. Returns the new `thread_id`. Retrying the same request reuses a completed import. |
Creation captures the latest idle root checkpoint and its persisted visible
history under exclusive thread admission. `checkpoint_id`, when supplied, is
an expected-head guard: a stale value returns `409`. Older checkpoints cannot
be selected because the source filesystem does not retain file versions for
historical checkpoints. The response binds the snapshot to the captured
checkpoint and highest visible history sequence.
Expired, revoked and unknown tokens all return `404`. A running conversation or
stale checkpoint returns `409`; size limits return `413`; SQL-disabled servers
return `503`. Share tokens only authorize the two public read routes. They never
authorize thread mutation, execution, or access to the original file tree.

View File

@ -46,6 +46,10 @@ through `scripts/pnpm.py`.
## Conventions the tools do not enforce
- **Public shares stay outside the workspace auth layout.** Read snapshots
without session credentials, rewrite file links through their manifest, and
authenticate only when importing into a workspace. See
[sharing](../docs/guide/sharing.md).
- **`src/components/ui/` and `src/components/ai-elements/` are generated** from
component registries. Change them by regenerating, because a manual edit is
lost on the next update.

View File

@ -0,0 +1,10 @@
import { SharePage } from "@/components/shares/share-page";
export default async function Page({
params,
}: {
params: Promise<{ share_token: string }>;
}) {
const { share_token } = await params;
return <SharePage key={share_token} token={share_token} />;
}

View File

@ -0,0 +1,25 @@
import "katex/dist/katex.min.css";
import "streamdown/styles.css";
import type { Metadata } from "next";
import { I18nProvider } from "@/core/i18n/context";
import { detectLocaleServer } from "@/core/i18n/server";
export const metadata: Metadata = {
title: "Shared conversation · DeerFlow",
robots: { index: false, follow: false },
referrer: "no-referrer",
};
export default async function ShareLayout({
children,
}: {
children: React.ReactNode;
}) {
return (
<I18nProvider initialLocale={await detectLocaleServer()}>
{children}
</I18nProvider>
);
}

View File

@ -0,0 +1,115 @@
"use client";
import { useEffect, useState } from "react";
import { ArtifactTablePreview } from "@/components/workspace/artifacts/artifact-table-preview";
import { useI18n } from "@/core/i18n/hooks";
import { shareArtifactURL, type SharedArtifact } from "@/core/shares/api";
import { ShareMarkdown } from "./share-markdown";
export function ShareArtifact({
token,
artifact,
artifacts,
}: {
token: string;
artifact: SharedArtifact;
artifacts: SharedArtifact[];
}) {
const { t } = useI18n();
const url = shareArtifactURL(token, artifact.path);
const extension = artifact.path.split(".").at(-1)?.toLowerCase();
const isHTML = extension === "html" || extension === "htm";
const isImage = ["png", "jpg", "jpeg", "gif", "webp", "svg"].includes(
extension ?? "",
);
const isText =
!isHTML &&
(artifact.mime_type.startsWith("text/") ||
["json", "md", "csv", "tsv"].includes(extension ?? ""));
const [text, setText] = useState<string | null>(null);
const [error, setError] = useState(false);
useEffect(() => {
if (!isText) return;
const controller = new AbortController();
void globalThis
.fetch(url, {
credentials: "omit",
cache: "no-store",
referrerPolicy: "no-referrer",
headers: artifact.size > 0 ? { Range: "bytes=0-524287" } : undefined,
signal: controller.signal,
})
.then(async (response) => {
if (!response.ok) throw new Error();
return response.text();
})
.then((value) => {
if (!controller.signal.aborted) setText(value);
})
.catch(() => {
if (!controller.signal.aborted) setError(true);
});
return () => controller.abort();
}, [url, isText, artifact.size]);
if (isHTML || extension === "pdf")
return (
<div className="space-y-2">
<p className="text-muted-foreground text-sm">{t.sharing.previewNote}</p>
<iframe
className="h-[65vh] w-full rounded-md border bg-white"
sandbox={isHTML ? "" : undefined}
src={url}
referrerPolicy="no-referrer"
title={artifact.path}
/>
</div>
);
if (isImage)
return (
<img
className="max-h-[70vh] max-w-full object-contain"
src={url}
alt={artifact.path}
referrerPolicy="no-referrer"
/>
);
if (!isText) return <p>{t.sharing.noPreview}</p>;
if (error) return <p role="alert">{t.sharing.failed}</p>;
if (text === null) return <p role="status">{t.sharing.loading}</p>;
const truncated = artifact.size > 524288;
return (
<div className="space-y-3">
{truncated && (
<p className="text-muted-foreground text-sm">
{t.sharing.previewLimited}
</p>
)}
{extension === "csv" || extension === "tsv" ? (
<ArtifactTablePreview
content={text}
delimiter={extension === "tsv" ? "\t" : ","}
truncated={truncated}
identity={url}
/>
) : extension === "md" ? (
<ShareMarkdown
content={text}
token={token}
artifacts={artifacts}
basePath={
artifact.path.includes("/")
? artifact.path.slice(0, artifact.path.lastIndexOf("/") + 1)
: ""
}
/>
) : (
<pre className="overflow-x-auto text-sm whitespace-pre-wrap">
{text}
</pre>
)}
</div>
);
}

View File

@ -0,0 +1,97 @@
"use client";
import { useMemo } from "react";
import { visit } from "unist-util-visit";
import { shareArtifactURL, type SharedArtifact } from "@/core/shares/api";
import { resolveShareArtifact, sharedLink } from "@/core/shares/links";
import {
SafeMessageResponse,
streamdownPluginsWithoutRawHtml,
toStreamdownComponents,
} from "@/core/streamdown";
// Normalize only Markdown URL nodes before the existing sanitizer runs. Code
// blocks and ordinary prose retain their original text and schemes stay denied.
type MarkdownNode = { type: string; url?: string; children?: MarkdownNode[] };
function remarkOutputLinks() {
return (tree: MarkdownNode) => {
visit(tree, (node) => {
if (
(node.type === "link" ||
node.type === "image" ||
node.type === "definition") &&
node.url?.startsWith("sandbox:/mnt/user-data/outputs/")
) {
node.url = node.url.slice("sandbox:".length);
}
});
};
}
export function ShareMarkdown({
content,
token,
artifacts,
basePath = "",
}: {
content: string;
token: string;
artifacts: SharedArtifact[];
basePath?: string;
}) {
const components = useMemo(
() =>
toStreamdownComponents({
a: ({ href, children }) => {
const target = href
? sharedLink(href, token, artifacts, basePath)
: undefined;
return target ? (
<a
href={target}
target="_blank"
rel="noopener noreferrer"
referrerPolicy="no-referrer"
className="underline"
>
{children}
</a>
) : (
<span>{children}</span>
);
},
img: ({ src, alt }) => {
const artifact =
typeof src === "string"
? resolveShareArtifact(src, artifacts, basePath)
: undefined;
// Do not load third-party tracking images or live private file endpoints.
return artifact ? (
<img
src={shareArtifactURL(token, artifact.path)}
alt={alt ?? ""}
referrerPolicy="no-referrer"
className="max-w-full"
/>
) : (
<span>{alt}</span>
);
},
}),
[token, artifacts, basePath],
);
return (
<SafeMessageResponse
mode="static"
components={components}
{...streamdownPluginsWithoutRawHtml}
remarkPlugins={[
...(streamdownPluginsWithoutRawHtml.remarkPlugins ?? []),
remarkOutputLinks,
]}
>
{content}
</SafeMessageResponse>
);
}

View File

@ -0,0 +1,253 @@
"use client";
import Link from "next/link";
import { useRouter } from "next/navigation";
import { useEffect, useRef, useState } from "react";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogDescription,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { useI18n } from "@/core/i18n/hooks";
import {
forkShare,
readShare,
ShareError,
shareArtifactURL,
type SharedArtifact,
type SharedConversation,
} from "@/core/shares/api";
import { ShareArtifact } from "./share-artifact";
import { ShareMarkdown } from "./share-markdown";
export function SharePage({ token }: { token: string }) {
const { t, locale } = useI18n();
const labels = t.sharing;
const router = useRouter();
const [snapshot, setSnapshot] = useState<SharedConversation | null>(null);
const [error, setError] = useState("");
const [retry, setRetry] = useState(0);
const [selected, setSelected] = useState<SharedArtifact | null>(null);
const [forking, setForking] = useState(false);
const [forkError, setForkError] = useState("");
const forkController = useRef<AbortController | null>(null);
const forkRequest = useRef<string | null>(null);
useEffect(() => () => forkController.current?.abort(), []);
useEffect(() => {
const controller = new AbortController();
let pending = false;
async function load() {
if (pending) return;
pending = true;
try {
const data = await readShare(token, controller.signal);
if (!controller.signal.aborted) {
setSnapshot(data);
setError("");
}
} catch (cause) {
if (!controller.signal.aborted) {
setSnapshot(null);
setSelected(null);
setError(
cause instanceof ShareError && cause.status === 404
? labels.missing
: labels.failed,
);
}
} finally {
pending = false;
}
}
void load();
const refresh = () => {
if (document.visibilityState === "visible") void load();
};
const timer = setInterval(refresh, 30_000);
document.addEventListener("visibilitychange", refresh);
return () => {
controller.abort();
clearInterval(timer);
document.removeEventListener("visibilitychange", refresh);
};
}, [token, retry, labels.failed, labels.missing]);
useEffect(() => {
if (!snapshot) return;
const delay = Math.max(0, Date.parse(snapshot.expires_at) - Date.now());
// Polling will install a timer once a long-lived share enters this window.
if (delay > 2_147_483_647) return;
const timer = setTimeout(() => {
setSnapshot(null);
setSelected(null);
setError(labels.missing);
}, delay);
return () => clearTimeout(timer);
}, [snapshot, labels.missing]);
async function fork() {
if (forking) return;
const controller = new AbortController();
forkController.current = controller;
forkRequest.current ??= crypto.randomUUID();
setForking(true);
setForkError("");
try {
const result = await forkShare(
token,
forkRequest.current,
controller.signal,
);
if (!controller.signal.aborted)
router.push(`/workspace/chats/${encodeURIComponent(result.thread_id)}`);
} catch (cause) {
if (!controller.signal.aborted)
setForkError(
cause instanceof ShareError && cause.status === 404
? labels.missing
: labels.failed,
);
} finally {
if (!controller.signal.aborted) setForking(false);
}
}
return (
<main className="mx-auto min-h-screen max-w-5xl space-y-6 px-4 py-8">
<header className="flex flex-wrap items-center justify-between gap-4 border-b pb-5">
<div>
<Link href="/" className="font-serif text-2xl">
DeerFlow
</Link>
<p className="text-muted-foreground mt-1 text-sm">
{labels.readOnly}
</p>
</div>
{snapshot && (
<Button disabled={forking} onClick={() => void fork()}>
{forking ? labels.forking : labels.fork}
</Button>
)}
</header>
{forkError && (
<p role="alert" className="text-destructive">
{forkError}
</p>
)}
{error ? (
<div role="alert" className="space-y-4 rounded-lg border p-6">
<p>{error}</p>
<Button
variant="outline"
onClick={() => {
setError("");
setRetry((value) => value + 1);
}}
>
{labels.retry}
</Button>
</div>
) : !snapshot ? (
<p role="status">{labels.loading}</p>
) : (
<>
<div>
<h1 className="text-2xl font-semibold break-words">
{snapshot.title}
</h1>
<p className="text-muted-foreground mt-2 text-sm">
{new Date(snapshot.created_at).toLocaleString(locale)} ·{" "}
{labels.expires}{" "}
{new Date(snapshot.expires_at).toLocaleString(locale)}
</p>
</div>
{snapshot.artifacts.length > 0 && (
<section
aria-label={labels.artifacts}
className="space-y-3 rounded-lg border p-4"
>
<h2 className="font-medium">{labels.artifacts}</h2>
<div className="grid gap-2 sm:grid-cols-2">
{snapshot.artifacts.map((artifact) => (
<div
key={artifact.path}
className="bg-secondary/60 flex min-w-0 items-center justify-between gap-2 rounded-md p-3"
>
<button
className="min-w-0 text-left text-sm break-all underline"
onClick={() => setSelected(artifact)}
>
{artifact.path}
</button>
<a
className="shrink-0 text-sm underline"
href={`${shareArtifactURL(token, artifact.path)}?download=true`}
referrerPolicy="no-referrer"
>
{labels.download}
</a>
</div>
))}
</div>
</section>
)}
<section aria-label={labels.title} className="space-y-5">
{snapshot.messages.map((message) =>
message.type === "tool" ? (
<details key={message.id} className="rounded-lg border p-4">
<summary className="cursor-pointer text-sm">
{labels.tool}: {message.name}
</summary>
<pre className="mt-3 max-h-96 overflow-auto text-sm whitespace-pre-wrap">
{message.content}
</pre>
</details>
) : (
<article
key={message.id}
className={`space-y-3 rounded-lg p-5 ${message.type === "human" ? "bg-secondary/60" : "border"}`}
>
<h2 className="text-muted-foreground text-sm font-medium">
{message.type === "human" ? labels.you : labels.assistant}
</h2>
<ShareMarkdown
content={message.content}
token={token}
artifacts={snapshot.artifacts}
/>
</article>
),
)}
</section>
<Dialog
open={selected !== null}
onOpenChange={(open) => {
if (!open) setSelected(null);
}}
>
<DialogContent className="max-h-[90vh] overflow-auto sm:max-w-4xl">
<DialogHeader>
<DialogTitle className="break-all">
{selected?.path}
</DialogTitle>
<DialogDescription>{labels.preview}</DialogDescription>
</DialogHeader>
{selected && (
<ShareArtifact
key={selected.path}
token={token}
artifact={selected}
artifacts={snapshot.artifacts}
/>
)}
</DialogContent>
</Dialog>
</>
)}
</main>
);
}

View File

@ -49,7 +49,6 @@ import { ConversationExtensionActions } from "@/components/workspace/conversatio
import { getAPIClient } from "@/core/api";
import { useAuth } from "@/core/auth/AuthProvider";
import { hasPermission, PERMISSIONS } from "@/core/auth/permissions";
import { writeTextToClipboard } from "@/core/clipboard";
import { useI18n } from "@/core/i18n/hooks";
import { useProjects } from "@/core/projects";
import { useLocalSettings } from "@/core/settings";
@ -87,6 +86,7 @@ import {
} from "./thread-channel-source";
import { useThreadDeleteDialog } from "./thread-delete-dialog";
import { VirtualThreadList } from "./thread-list-virtualizer";
import { ThreadShareDialog } from "./thread-share-dialog";
import { useThreadArchiveAction } from "./use-thread-archive-action";
/**
@ -139,6 +139,7 @@ export function ThreadSidebarItem({
const [renameValue, setRenameValue] = useState("");
const [newProjectDialogOpen, setNewProjectDialogOpen] = useState(false);
const [exporting, setExporting] = useState(false);
const [shareOpen, setShareOpen] = useState(false);
const handleRenameSubmit = useCallback(() => {
if (renameValue.trim()) {
@ -177,24 +178,6 @@ export function ThreadSidebarItem({
);
}, [t.chats.pinChatFailed, thread, updatePinnedThread]);
const handleShare = useCallback(async () => {
// Link to this deployment, the only place the thread exists. Never
// rewrite the origin to another host: the thread id would leave the
// deployment and the link would not open the thread.
const shareUrl = `${window.location.origin}${pathOfThread(thread)}`;
try {
const didCopy = await writeTextToClipboard(shareUrl);
if (!didCopy) {
toast.error(t.clipboard.failedToCopyToClipboard);
return;
}
toast.success(t.clipboard.linkCopied);
} catch {
toast.error(t.clipboard.failedToCopyToClipboard);
}
}, [t, thread]);
const handleExport = useCallback(
async (format: ThreadExportFormat) => {
setExporting(true);
@ -325,7 +308,14 @@ export function ThreadSidebarItem({
<Pencil className="text-muted-foreground" />
<span>{t.common.rename}</span>
</DropdownMenuItem>
<DropdownMenuItem onSelect={() => void handleShare()}>
<DropdownMenuItem
disabled={
isStaticWebsiteOnly() ||
!hasPermission(user, PERMISSIONS.THREADS_WRITE) ||
!hasPermission(user, PERMISSIONS.RUNS_READ)
}
onSelect={() => setShareOpen(true)}
>
<Share2 className="text-muted-foreground" />
<span>{t.common.share}</span>
</DropdownMenuItem>
@ -374,6 +364,11 @@ export function ThreadSidebarItem({
</DropdownMenu>
)}
<ThreadShareDialog
threadId={thread.thread_id}
open={shareOpen}
onOpenChange={setShareOpen}
/>
{/* Rename Dialog */}
<Dialog open={renameDialogOpen} onOpenChange={setRenameDialogOpen}>
<DialogContent className="sm:max-w-[425px]">

View File

@ -0,0 +1,218 @@
"use client";
import { useEffect, useRef, useState } from "react";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogDescription,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { useAuth } from "@/core/auth/AuthProvider";
import { writeTextToClipboard } from "@/core/clipboard";
import { useI18n } from "@/core/i18n/hooks";
import {
createShare,
listShares,
revokeShare,
ShareError,
type ShareLink,
} from "@/core/shares/api";
export function ThreadShareDialog({
threadId,
open,
onOpenChange,
}: {
threadId: string;
open: boolean;
onOpenChange: (open: boolean) => void;
}) {
const { user } = useAuth();
const { t } = useI18n();
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="sm:max-w-lg">
<DialogHeader>
<DialogTitle>{t.sharing.title}</DialogTitle>
<DialogDescription>{t.sharing.description}</DialogDescription>
</DialogHeader>
{open && (
<ShareControls key={`${user?.id}:${threadId}`} threadId={threadId} />
)}
</DialogContent>
</Dialog>
);
}
function ShareControls({ threadId }: { threadId: string }) {
const { t, locale } = useI18n();
const labels = t.sharing;
const [links, setLinks] = useState<ShareLink[]>([]);
const [duration, setDuration] = useState(7 * 86400);
const [link, setLink] = useState("");
const [linkId, setLinkId] = useState("");
const [loading, setLoading] = useState(true);
const [busy, setBusy] = useState(false);
const [error, setError] = useState("");
const [copied, setCopied] = useState(false);
const controller = useRef<AbortController | null>(null);
useEffect(() => {
const request = new AbortController();
controller.current = request;
void listShares(threadId, request.signal)
.then((rows) => {
if (!request.signal.aborted) setLinks(rows);
})
.catch((cause: unknown) => {
if (!request.signal.aborted)
setError(
cause instanceof ShareError && cause.status === 503
? labels.unavailable
: labels.failed,
);
})
.finally(() => {
if (!request.signal.aborted) setLoading(false);
});
return () => request.abort();
}, [threadId, labels.failed, labels.unavailable]);
async function mutate(operation: (signal: AbortSignal) => Promise<void>) {
const request = controller.current;
if (!request || request.signal.aborted || busy) return;
setBusy(true);
setError("");
try {
await operation(request.signal);
} catch (cause) {
if (!request.signal.aborted)
setError(
cause instanceof ShareError && cause.status === 409
? labels.conflict
: cause instanceof ShareError && cause.status === 413
? labels.tooLarge
: labels.failed,
);
} finally {
if (!request.signal.aborted) setBusy(false);
}
}
async function copy() {
const success = await writeTextToClipboard(link);
if (!controller.current?.signal.aborted) {
setCopied(success);
if (!success) setError(t.clipboard.failedToCopyToClipboard);
}
}
return (
<div className="space-y-4">
<p className="text-muted-foreground text-sm">{labels.scope}</p>
<label className="flex items-center justify-between gap-4 text-sm">
{labels.expiry}
<select
className="border-input bg-background rounded-md border px-3 py-2"
value={duration}
onChange={(event) => setDuration(Number(event.target.value))}
disabled={busy}
>
<option value={3600}>{labels.hour}</option>
<option value={86400}>{labels.day}</option>
<option value={7 * 86400}>{labels.week}</option>
<option value={30 * 86400}>{labels.month}</option>
</select>
</label>
<Button
className="w-full"
disabled={busy || loading}
onClick={() =>
void mutate(async (signal) => {
const created = await createShare(threadId, duration, signal);
if (signal.aborted) return;
setLink(
`${window.location.origin}/share/${encodeURIComponent(created.token)}`,
);
setLinkId(created.share_id);
setCopied(false);
setLinks((rows) => [created, ...rows]);
})
}
>
{busy ? labels.working : labels.create}
</Button>
{link && (
<div className="flex gap-2">
<Input aria-label={labels.link} readOnly value={link} />
<Button variant="outline" onClick={() => void copy()}>
{copied ? labels.copied : labels.copy}
</Button>
</div>
)}
{error && (
<p role="alert" className="text-destructive text-sm">
{error}
</p>
)}
<div className="max-h-56 space-y-2 overflow-y-auto">
<h3 className="text-sm font-medium">{labels.links}</h3>
{loading ? (
<p role="status">{labels.loading}</p>
) : links.length === 0 ? (
<p className="text-muted-foreground text-sm">{labels.noLinks}</p>
) : (
links.map((item) => {
const expired = Date.parse(item.expires_at) <= Date.now();
return (
<div
key={item.share_id}
className="flex items-center justify-between gap-3 rounded-md border p-3 text-sm"
>
<div>
<time>
{new Date(item.created_at).toLocaleString(locale)}
</time>
<p className="text-muted-foreground">
{item.revoked_at
? labels.revoked
: expired
? labels.expired
: `${labels.expires} ${new Date(item.expires_at).toLocaleString(locale)}`}
</p>
</div>
{!item.revoked_at && (
<Button
size="sm"
variant="outline"
disabled={busy}
onClick={() =>
void mutate(async (signal) => {
await revokeShare(threadId, item.share_id, signal);
if (signal.aborted) return;
setLinks((rows) =>
rows.map((row) =>
row.share_id === item.share_id
? { ...row, revoked_at: new Date().toISOString() }
: row,
),
);
if (linkId === item.share_id) setLink("");
})
}
>
{labels.revoke}
</Button>
)}
</div>
);
})
)}
</div>
</div>
);
}

View File

@ -5,6 +5,7 @@ import type { User } from "./types";
* Kept in lockstep with `Permissions` in `backend/app/gateway/authz.py`.
*/
export const PERMISSIONS = {
THREADS_WRITE: "threads:write",
THREADS_DELETE: "threads:delete",
RUNS_CANCEL: "runs:cancel",
RUNS_CREATE: "runs:create",

View File

@ -14,6 +14,53 @@ import { bundledCopy } from "@/core/extensions/documentation";
import type { Translations } from "./types";
export const enUS: Translations = {
sharing: {
title: "Share conversation",
description:
"Anyone with the link can view this snapshot without signing in.",
scope:
"Includes visible conversation steps and presented output files. Later changes stay private. Revoke a link to stop future access; copies already downloaded or forked remain with their recipients.",
expiry: "Link expires after",
hour: "1 hour",
day: "1 day",
week: "7 days",
month: "30 days",
create: "Create snapshot link",
working: "Working…",
link: "Share link",
copy: "Copy link",
copied: "Copied",
links: "Created links",
noLinks: "No share links yet.",
revoke: "Revoke",
revoked: "Revoked",
expired: "Expired",
expires: "Expires",
loading: "Loading…",
failed: "Unable to complete this request. Try again.",
unavailable: "Sharing is unavailable on this server.",
conflict:
"Finish the current turn before sharing. If it is already finished, refresh or revoke an older link and try again.",
tooLarge:
"This conversation or its artifacts exceed the sharing size limit.",
readOnly: "Read-only snapshot",
fork: "Continue in my workspace",
forking: "Copying…",
missing: "This share link has expired, was revoked, or is unavailable.",
retry: "Retry",
artifacts: "Artifacts",
download: "Download",
preview: "Preview",
close: "Close preview",
you: "User",
assistant: "Assistant",
tool: "Tool result",
noPreview: "Download this file to open it.",
previewNote:
"Static preview. Interactive content may require downloading the file.",
previewLimited:
"Preview shows the first 512 KiB. Download for the complete file.",
},
usage: {
title: "Token usage",
description: "View input, output and cache-hit token usage.",

View File

@ -1,6 +1,47 @@
import type { LucideIcon } from "lucide-react";
export interface Translations {
sharing: {
title: string;
description: string;
scope: string;
expiry: string;
hour: string;
day: string;
week: string;
month: string;
create: string;
working: string;
link: string;
copy: string;
copied: string;
links: string;
noLinks: string;
revoke: string;
revoked: string;
expired: string;
expires: string;
loading: string;
failed: string;
unavailable: string;
conflict: string;
tooLarge: string;
readOnly: string;
fork: string;
forking: string;
missing: string;
retry: string;
artifacts: string;
download: string;
preview: string;
close: string;
you: string;
assistant: string;
tool: string;
noPreview: string;
previewNote: string;
previewLimited: string;
};
usage: {
title: string;
description: string;

View File

@ -14,6 +14,49 @@ import { bundledCopy } from "@/core/extensions/documentation";
import type { Translations } from "./types";
export const zhCN: Translations = {
sharing: {
title: "分享会话",
description: "任何持有链接的人都可以免登录查看此快照。",
scope:
"包含可见对话、工具执行结果和已展示的产物文件。后续修改不会同步到快照。撤销可停止后续访问,已下载或复制的内容仍由接收方保留。",
expiry: "链接有效期",
hour: "1 小时",
day: "1 天",
week: "7 天",
month: "30 天",
create: "创建快照链接",
working: "正在处理…",
link: "分享链接",
copy: "复制链接",
copied: "已复制",
links: "已创建的链接",
noLinks: "尚未创建分享链接。",
revoke: "撤销",
revoked: "已撤销",
expired: "已过期",
expires: "到期时间",
loading: "加载中…",
failed: "操作未完成,请重试。",
unavailable: "此服务器暂不支持分享。",
conflict:
"请在当前任务结束后分享。如果任务已经结束,请刷新页面或撤销旧链接后重试。",
tooLarge: "此会话或其产物超过分享大小限制。",
readOnly: "只读快照",
fork: "复制到我的工作区继续",
forking: "正在复制…",
missing: "此分享链接已过期、被撤销或无法访问。",
retry: "重试",
artifacts: "产物",
download: "下载",
preview: "预览",
close: "关闭预览",
you: "用户",
assistant: "助手",
tool: "工具执行结果",
noPreview: "请下载后打开此文件。",
previewNote: "静态预览。交互内容可能需要下载文件后查看。",
previewLimited: "预览仅显示前 512 KiB,下载可查看完整文件。",
},
usage: {
title: "Token 用量",
description: "查看输入、输出及缓存命中 Token 用量。",

View File

@ -0,0 +1,135 @@
import { z } from "zod";
import { fetch as authenticatedFetch } from "@/core/api/fetcher";
import { getBackendBaseURL } from "@/core/config";
export const shareLinkSchema = z.object({
share_id: z.string(),
checkpoint_id: z.string(),
seq: z.number(),
created_at: z.string(),
expires_at: z.string(),
revoked_at: z.string().nullable(),
});
export const sharedArtifactSchema = z.object({
path: z.string(),
mime_type: z.string(),
sha256: z.string(),
size: z.number(),
});
export const sharedConversationSchema = z.object({
title: z.string(),
checkpoint_id: z.string(),
seq: z.number(),
created_at: z.string(),
expires_at: z.string(),
messages: z.array(
z.object({
id: z.string(),
type: z.enum(["human", "ai", "tool"]),
content: z.string(),
name: z.string().optional(),
}),
),
artifacts: z.array(sharedArtifactSchema),
});
export type ShareLink = z.infer<typeof shareLinkSchema>;
export type SharedArtifact = z.infer<typeof sharedArtifactSchema>;
export type SharedConversation = z.infer<typeof sharedConversationSchema>;
export class ShareError extends Error {
constructor(readonly status: number) {
super(`Share request failed (${status})`);
}
}
async function checked(response: Response) {
if (!response.ok) throw new ShareError(response.status);
return response;
}
const sourceURL = (threadId: string) =>
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/shares`;
export const shareURL = (token: string) =>
`${getBackendBaseURL()}/api/shares/${encodeURIComponent(token)}`;
export const shareArtifactURL = (token: string, path: string) =>
`${shareURL(token)}/artifacts/${path.split("/").map(encodeURIComponent).join("/")}`;
export async function listShares(threadId: string, signal: AbortSignal) {
return z.array(shareLinkSchema).parse(
await (
await checked(
await authenticatedFetch(sourceURL(threadId), {
signal,
cache: "no-store",
}),
)
).json(),
);
}
export async function createShare(
threadId: string,
expiresInSeconds: number,
signal: AbortSignal,
) {
const response = await checked(
await authenticatedFetch(sourceURL(threadId), {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ expires_in_seconds: expiresInSeconds }),
signal,
}),
);
return shareLinkSchema
.extend({ token: z.string() })
.parse(await response.json());
}
export async function revokeShare(
threadId: string,
shareId: string,
signal: AbortSignal,
) {
await checked(
await authenticatedFetch(
`${sourceURL(threadId)}/${encodeURIComponent(shareId)}`,
{ method: "DELETE", signal },
),
);
}
export async function readShare(token: string, signal: AbortSignal) {
const response = await checked(
await globalThis.fetch(shareURL(token), {
credentials: "omit",
cache: "no-store",
referrerPolicy: "no-referrer",
signal,
}),
);
return sharedConversationSchema.parse(await response.json());
}
export async function forkShare(
token: string,
requestId: string,
signal: AbortSignal,
) {
// Authenticate before a CSRF-protected POST, so anonymous visitors reach sign-in.
await checked(
await authenticatedFetch(`${getBackendBaseURL()}/api/v1/auth/me`, {
cache: "no-store",
signal,
}),
);
const response = await checked(
await authenticatedFetch(`${shareURL(token)}/fork`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ request_id: requestId }),
signal,
}),
);
return z.object({ thread_id: z.string() }).parse(await response.json());
}

View File

@ -0,0 +1,55 @@
import { shareArtifactURL, type SharedArtifact } from "./api";
/** Markdown may name a copied output, never a live workspace/upload endpoint. */
export function resolveShareArtifact(
href: string,
artifacts: SharedArtifact[],
basePath = "",
) {
let path = href.replace(/^sandbox:/, "");
try {
path = decodeURIComponent(path);
} catch {
return undefined;
}
if (path.startsWith("/mnt/user-data/outputs/")) path = path.slice(23);
else if (path.startsWith("mnt/user-data/outputs/")) path = path.slice(22);
else if (
path.startsWith("/") ||
/^[a-zA-Z][\w+.-]*:/.test(path) ||
path.includes("\\")
)
return undefined;
else path = `${basePath}${path}`;
const normalized: string[] = [];
for (const part of path.split("/")) {
if (part === "..") {
if (!normalized.length) return undefined;
normalized.pop();
} else if (part && part !== ".") normalized.push(part);
}
return artifacts.find((artifact) => artifact.path === normalized.join("/"));
}
export function sharedLink(
href: string,
token: string,
artifacts: SharedArtifact[],
basePath = "",
) {
const artifact = resolveShareArtifact(href, artifacts, basePath);
if (artifact) return shareArtifactURL(token, artifact.path);
try {
const url = new URL(href);
if (
["https:", "http:"].includes(url.protocol) &&
url.origin !== window.location.origin &&
!url.username &&
!url.password
)
return url.href;
} catch {
/* Non-URL text remains visible without navigation. */
}
return undefined;
}

View File

@ -0,0 +1,273 @@
import { expect, test } from "@playwright/test";
import { MOCK_THREAD_ID, mockLangGraphAPI } from "./utils/mock-api";
const token = "S".repeat(43);
const created = {
share_id: "share-1",
checkpoint_id: "cp-1",
seq: 3,
created_at: new Date().toISOString(),
expires_at: new Date(Date.now() + 86400000).toISOString(),
revoked_at: null,
};
const snapshot = {
...created,
title: "Quarterly report",
messages: [
{ id: "human", type: "human", content: "Compare quarterly revenue." },
{
id: "answer",
type: "ai",
content:
"Revenue rose. [Report](sandbox:/mnt/user-data/outputs/report.html) ![tracking](https://tracker.invalid/pixel) [private](/workspace/chats/secret)",
},
{ id: "tool", type: "tool", name: "python", content: "Revenue: 42" },
],
artifacts: [
{
path: "report.html",
mime_type: "text/html",
size: 50,
sha256: "a".repeat(64),
},
{
path: "numbers.csv",
mime_type: "text/csv",
size: 23,
sha256: "b".repeat(64),
},
],
};
test("anonymous snapshot previews files without entering the private workspace", async ({
page,
}, testInfo) => {
const privateRequests: string[] = [];
page.on("request", (request) => {
if (
/\/api\/(threads|langgraph|v1\/auth)/.test(request.url()) ||
request.url().includes("tracker.invalid")
)
privateRequests.push(request.url());
});
let revoked = false;
await page.route(`**/api/shares/${token}`, (route) =>
route.fulfill({
status: revoked ? 404 : 200,
json: revoked ? {} : snapshot,
}),
);
await page.route(`**/api/shares/${token}/artifacts/*`, (route) =>
route.fulfill({
contentType: route.request().url().endsWith(".csv")
? "text/csv"
: "text/html",
headers: {
"Content-Security-Policy":
"sandbox; default-src 'none'; style-src 'unsafe-inline'",
},
body: route.request().url().endsWith(".csv")
? "Quarter,Revenue\nQ1,42\n"
: "<h1>Frozen report</h1><script>parent.document.body.innerHTML='unsafe'</script>",
}),
);
await page.goto(`/share/${token}`);
await expect(
page.getByRole("heading", { name: "Quarterly report" }),
).toBeVisible();
await expect(
page.getByText("Read-only snapshot", { exact: true }),
).toBeVisible();
await expect(page.locator("textarea, [contenteditable=true]")).toHaveCount(0);
await expect(
page.getByRole("link", { name: "private", exact: true }),
).toHaveCount(0);
await expect(
page.getByRole("link", { name: "Report", exact: true }),
).toHaveAttribute("href", `/api/shares/${token}/artifacts/report.html`);
await page.getByRole("button", { name: "report.html", exact: true }).click();
await expect(page.locator("iframe")).toHaveAttribute("sandbox", "");
await expect(
page.frameLocator("iframe").getByRole("heading", { name: "Frozen report" }),
).toBeVisible();
await page
.getByRole("dialog")
.getByRole("button", { name: "Close", exact: true })
.click();
await page.getByRole("button", { name: "numbers.csv", exact: true }).click();
await expect(
page.getByRole("dialog").getByText("42", { exact: true }),
).toBeVisible();
await page
.getByRole("dialog")
.getByRole("button", { name: "Close", exact: true })
.click();
expect(privateRequests).toEqual([]);
await expect(page.getByRole("dialog")).toBeHidden();
await page.screenshot({
path: testInfo.outputPath("share-preview.png"),
fullPage: true,
});
revoked = true;
await page.reload();
await expect(
page.getByText(
"This share link has expired, was revoked, or is unavailable.",
{ exact: true },
),
).toBeVisible();
await expect(
page.getByRole("button", { name: "Continue in my workspace" }),
).toHaveCount(0);
});
test("anonymous Fork signs in and returns to the shared snapshot", async ({
page,
}) => {
await page.route(`**/api/shares/${token}`, (route) =>
route.fulfill({ json: snapshot }),
);
await page.route("**/api/v1/auth/me", (route) =>
route.fulfill({ status: 401, json: { detail: "Sign in" } }),
);
await page.route("**/login**", (route) =>
route.fulfill({ contentType: "text/html", body: "Sign in" }),
);
let imports = 0;
await page.route(`**/api/shares/${token}/fork`, (route) => {
imports++;
return route.fulfill({ json: { thread_id: "copy" } });
});
await page.goto(`/share/${token}`);
await page.getByRole("button", { name: "Continue in my workspace" }).click();
await expect(page).toHaveURL(new RegExp(`/login\\?next=.*share.*${token}`));
expect(imports).toBe(0);
});
test("shared PDFs keep Chromium's native viewer enabled", async ({ page }) => {
const stream = "BT /F1 12 Tf 20 100 Td (Shared PDF report) Tj ET";
const objects = [
"<< /Type /Catalog /Pages 2 0 R >>",
"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 200 200] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
`<< /Length ${stream.length} >>\nstream\n${stream}\nendstream`,
];
let pdf = "%PDF-1.4\n";
const offsets = objects.map((object, index) => {
const offset = pdf.length;
pdf += `${index + 1} 0 obj\n${object}\nendobj\n`;
return offset;
});
const xref = pdf.length;
pdf += `xref\n0 6\n0000000000 65535 f \n${offsets.map((offset) => `${String(offset).padStart(10, "0")} 00000 n \n`).join("")}trailer\n<< /Size 6 /Root 1 0 R >>\nstartxref\n${xref}\n%%EOF\n`;
await page.route(`**/api/shares/${token}`, (route) =>
route.fulfill({
json: {
...snapshot,
artifacts: [
{
path: "report.pdf",
mime_type: "application/pdf",
size: pdf.length,
sha256: "c".repeat(64),
},
],
},
}),
);
await page.route(`**/api/shares/${token}/artifacts/report.pdf`, (route) =>
route.fulfill({
contentType: "application/pdf",
headers: { "X-Content-Type-Options": "nosniff" },
body: pdf,
}),
);
await page.goto(`/share/${token}`);
const response = page.waitForResponse((response) =>
response.url().endsWith("/artifacts/report.pdf"),
);
await page.getByRole("button", { name: "report.pdf", exact: true }).click();
expect((await response).headers()["content-type"]).toBe("application/pdf");
await expect(page.getByRole("dialog").locator("iframe")).toBeVisible();
await expect(page.getByRole("dialog").locator("iframe")).not.toHaveAttribute(
"sandbox",
);
});
test("authenticated Fork opens its new workspace conversation", async ({
page,
}) => {
mockLangGraphAPI(page, {
threads: [{ thread_id: MOCK_THREAD_ID, title: "Imported report" }],
});
await page.route(`**/api/shares/${token}`, (route) =>
route.fulfill({ json: snapshot }),
);
let requestId: string | undefined;
await page.route(`**/api/shares/${token}/fork`, (route) => {
requestId = (route.request().postDataJSON() as { request_id: string })
.request_id;
return route.fulfill({ json: { thread_id: MOCK_THREAD_ID } });
});
await page.goto(`/share/${token}`);
await page.getByRole("button", { name: "Continue in my workspace" }).click();
await expect(page).toHaveURL(`/workspace/chats/${MOCK_THREAD_ID}`, {
timeout: 15000,
});
expect(requestId).toMatch(/^[0-9a-f-]{36}$/);
});
test("owner creates and revokes a snapshot from the sidebar", async ({
page,
}, testInfo) => {
mockLangGraphAPI(page, {
threads: [{ thread_id: MOCK_THREAD_ID, title: "Owner report" }],
});
await page.route(`**/api/threads/${MOCK_THREAD_ID}/shares`, (route) => {
if (route.request().method() === "GET") return route.fulfill({ json: [] });
expect(route.request().postDataJSON()).toEqual({
expires_in_seconds: 3600,
});
return route.fulfill({
status: 201,
json: {
...created,
token,
expires_at: new Date(Date.now() + 3600000).toISOString(),
},
});
});
await page.route(
`**/api/threads/${MOCK_THREAD_ID}/shares/share-1`,
(route) => {
expect(route.request().method()).toBe("DELETE");
return route.fulfill({ status: 204 });
},
);
await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`);
const row = page.locator('[data-sidebar="menu-item"]').filter({
has: page.getByRole("link", { name: "Owner report", exact: true }),
});
await row.hover();
await row.getByRole("button", { name: "More", exact: true }).click();
await page.screenshot({
path: testInfo.outputPath("share-menu.png"),
fullPage: true,
});
await page.getByRole("menuitem", { name: "Share", exact: true }).click();
const dialog = page.getByRole("dialog");
await dialog.getByRole("combobox").selectOption("3600");
await dialog.getByRole("button", { name: "Create snapshot link" }).click();
await expect(dialog.getByLabel("Share link")).toHaveValue(
new RegExp(`/share/${token}$`),
);
await page.screenshot({
path: testInfo.outputPath("share-entry.png"),
fullPage: true,
});
await dialog.getByRole("button", { name: "Revoke", exact: true }).click();
await expect(dialog.getByText("Revoked", { exact: true })).toBeVisible();
await expect(dialog.getByLabel("Share link")).toHaveCount(0);
});

View File

@ -0,0 +1,116 @@
import { afterEach, expect, rs, test } from "@rstest/core";
import {
act,
cleanup,
fireEvent,
render,
screen,
waitFor,
} from "@testing-library/react";
rs.mock("next/navigation", () => ({
useRouter: () => ({ push: rs.fn(), replace: rs.fn(), refresh: rs.fn() }),
usePathname: () => "/workspace/chats",
}));
import { ThreadShareDialog } from "@/components/workspace/thread-share-dialog";
import { AuthProvider, type User } from "@/core/auth/AuthProvider";
import { I18nProvider } from "@/core/i18n/context";
const user = {
id: "owner",
email: "owner@example.test",
system_role: "admin",
} as User;
const link = {
share_id: "one",
checkpoint_id: "cp",
seq: 1,
created_at: new Date().toISOString(),
expires_at: new Date(Date.now() + 86400000).toISOString(),
revoked_at: null,
token: "s".repeat(43),
};
function dialog(threadId: string, open = true) {
return (
<I18nProvider initialLocale="en-US">
<AuthProvider initialUser={user}>
<ThreadShareDialog
threadId={threadId}
open={open}
onOpenChange={rs.fn()}
/>
</AuthProvider>
</I18nProvider>
);
}
afterEach(() => {
cleanup();
rs.restoreAllMocks();
});
test("changing the selected conversation aborts a create and discards its late result", async () => {
let finish: ((response: Response) => void) | undefined;
let createSignal: AbortSignal | null | undefined;
rs.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => {
if (init?.method === "POST") {
createSignal = init.signal;
return new Promise<Response>((resolve) => {
finish = resolve;
});
}
return new Response("[]");
});
const view = render(dialog("first"));
await waitFor(() =>
expect(
screen
.getByRole("button", { name: "Create snapshot link" })
.getAttribute("disabled"),
).toBeNull(),
);
fireEvent.click(screen.getByRole("button", { name: "Create snapshot link" }));
await waitFor(() => expect(createSignal).toBeDefined());
view.rerender(dialog("second"));
expect(createSignal?.aborted).toBe(true);
await act(async () => {
finish?.(new Response(JSON.stringify(link)));
});
expect(screen.queryByLabelText("Share link")).toBeNull();
await waitFor(() =>
expect(screen.getByText("No share links yet.")).toBeDefined(),
);
});
test("closing the dialog aborts its pending list request", async () => {
let listSignal: AbortSignal | null | undefined;
rs.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => {
listSignal = init?.signal;
return new Promise<Response>(() => {
/* The request remains pending until unmount. */
});
});
const view = render(dialog("first"));
await waitFor(() => expect(listSignal).toBeDefined());
view.rerender(dialog("first", false));
expect(listSignal?.aborted).toBe(true);
});
test("an expired link can be revoked to remove the stored snapshot", async () => {
const deleted: string[] = [];
const expired = { ...link, expires_at: "2000-01-01T00:00:00Z" };
rs.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
if (init?.method === "DELETE") {
deleted.push(url instanceof Request ? url.url : url.toString());
return new Response(null, { status: 204 });
}
return Response.json([expired]);
});
render(dialog("first"));
await screen.findByText("Expired");
fireEvent.click(screen.getByRole("button", { name: "Revoke" }));
await screen.findByText("Revoked");
expect(deleted).toHaveLength(1);
expect(deleted[0]).toContain("/api/threads/first/shares/one");
expect(screen.queryByRole("button", { name: "Revoke" })).toBeNull();
});

View File

@ -30,6 +30,7 @@ type HappyDOMWindow = Window & {
};
const THREAD_ID = "83d5133d-f8aa-4095-9bba-2aca03f8f61c";
const SHARE_TOKEN = "s".repeat(43);
const writeText = rs.mocked(writeTextToClipboard);
const initialURL = window.location.href;
@ -65,10 +66,34 @@ function renderItem(metadata: Record<string, unknown> = {}) {
/** Opens the row's "more" menu and selects Share. Radix opens on pointerdown. */
async function share(): Promise<string> {
rs.spyOn(globalThis, "fetch").mockImplementation(
async (_url, init) =>
new Response(
JSON.stringify(
init?.method === "POST"
? {
share_id: "one",
checkpoint_id: "cp",
seq: 1,
token: SHARE_TOKEN,
created_at: new Date().toISOString(),
expires_at: new Date(Date.now() + 86400000).toISOString(),
revoked_at: null,
}
: [],
),
),
);
const trigger = await screen.findByRole("button", { name: /more/i });
fireEvent.pointerDown(trigger, { button: 0, pointerType: "mouse" });
fireEvent.click(trigger);
fireEvent.click(await screen.findByText("Share"));
const create = await screen.findByRole("button", {
name: "Create snapshot link",
});
await waitFor(() => expect(create.getAttribute("disabled")).toBeNull());
fireEvent.click(create);
fireEvent.click(await screen.findByRole("button", { name: "Copy link" }));
await waitFor(() => expect(writeText).toHaveBeenCalledTimes(1));
return writeText.mock.calls[0]![0];
}
@ -76,6 +101,7 @@ async function share(): Promise<string> {
afterEach(() => {
cleanup();
writeText.mockReset();
rs.restoreAllMocks();
(window as unknown as HappyDOMWindow).happyDOM.setURL(initialURL);
});
@ -97,18 +123,16 @@ describe("ThreadSidebarItem Share", () => {
);
writeText.mockResolvedValue(true);
renderItem();
expect(await share()).toBe(`${origin}/workspace/chats/${THREAD_ID}`);
expect(await share()).toBe(`${origin}/share/${SHARE_TOKEN}`);
});
}
it("copies the custom agent route on the current origin", async () => {
it("creates the same public snapshot route for a custom agent conversation", async () => {
(window as unknown as HappyDOMWindow).happyDOM.setURL(
"http://localhost:2026/workspace",
);
writeText.mockResolvedValue(true);
renderItem({ agent_name: "research bot" });
expect(await share()).toBe(
`http://localhost:2026/workspace/agents/research%20bot/chats/${THREAD_ID}`,
);
expect(await share()).toBe(`http://localhost:2026/share/${SHARE_TOKEN}`);
});
});

View File

@ -0,0 +1,90 @@
import { afterEach, describe, expect, it, rs } from "@rstest/core";
import { forkShare, readShare, shareArtifactURL } from "@/core/shares/api";
import { resolveShareArtifact, sharedLink } from "@/core/shares/links";
const artifact = {
path: "charts/a b.svg",
mime_type: "image/svg+xml",
size: 10,
sha256: "hash",
};
const token = "x".repeat(43);
afterEach(() => {
rs.restoreAllMocks();
});
describe("share boundaries", () => {
it("only resolves copied outputs, including relative markdown references", () => {
expect(
resolveShareArtifact("sandbox:/mnt/user-data/outputs/charts/a%20b.svg", [
artifact,
]),
).toEqual(artifact);
expect(
resolveShareArtifact("../charts/a b.svg", [artifact], "reports/"),
).toEqual(artifact);
for (const path of [
"/api/threads/private/artifacts/file",
"/mnt/user-data/uploads/private",
"../../charts/a b.svg",
"javascript:alert(1)",
"https://tracker.example/a.svg",
"%zz",
]) {
expect(resolveShareArtifact(path, [artifact])).toBeUndefined();
}
expect(shareArtifactURL(token, artifact.path)).toContain(
`/shares/${token}/artifacts/charts/a%20b.svg`,
);
expect(
sharedLink(
`${window.location.origin}/workspace/chats/private`,
token,
[],
),
).toBeUndefined();
expect(sharedLink("https://example.org/report", token, [])).toBe(
"https://example.org/report",
);
expect(sharedLink("https://secret@example.org", token, [])).toBeUndefined();
});
it("reads anonymously without forwarding session credentials or caching", async () => {
const signal = new AbortController().signal;
const fetch = rs
.spyOn(globalThis, "fetch")
.mockResolvedValue(new Response("{}", { status: 404 }));
await expect(readShare(token, signal)).rejects.toMatchObject({
status: 404,
});
expect(fetch.mock.calls[0]?.[1]).toMatchObject({
credentials: "omit",
cache: "no-store",
referrerPolicy: "no-referrer",
signal,
});
});
it("authenticates before sending an idempotent, CSRF-protected fork", async () => {
document.cookie = "csrf_token=fixture; path=/";
const fetch = rs
.spyOn(globalThis, "fetch")
.mockResolvedValueOnce(new Response("{}"))
.mockResolvedValueOnce(
new Response(JSON.stringify({ thread_id: "copy" })),
);
await expect(
forkShare(token, "request-1", new AbortController().signal),
).resolves.toEqual({ thread_id: "copy" });
expect(fetch.mock.calls[0]?.[0]).toContain("/api/v1/auth/me");
expect(fetch.mock.calls[1]?.[0]).toContain(`/shares/${token}/fork`);
expect(fetch.mock.calls[1]?.[1]).toMatchObject({
method: "POST",
body: JSON.stringify({ request_id: "request-1" }),
});
expect(
new Headers(fetch.mock.calls[1]?.[1]?.headers).get("x-csrf-token"),
).toBe("fixture");
});
});