diff --git a/README.md b/README.md index 273f34c7c..eaa4e0f1f 100644 --- a/README.md +++ b/README.md @@ -1048,6 +1048,18 @@ DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI- DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted: +Thread IDs may be supplied by callers and do not have to be UUIDs. Explicit +IDs must contain 1–64 ASCII letters, digits, hyphens, or underscores +(`^[A-Za-z0-9_-]{1,64}$`). DeerFlow generates a UUID only when `thread_id` is +omitted or `None`; an explicitly supplied empty string is invalid. +Existing route-addressable threads created under older, looser rules remain +readable and deletable, but cannot start new runs or create new filesystem or +sandbox state. Legacy deletion skips local path cleanup when the ID is not +safe under the canonical contract. For canonical legacy threads whose +conversation exists only in LangGraph checkpoints, DeerFlow seeds an empty +run-event feed from the checkpoint before the first new run so +`/messages/page` keeps both the old and new turns. + ```python from deerflow.client import DeerFlowClient diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 61a492ba5..bd9b64876 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -501,6 +501,19 @@ Localhost persistence deliberately reads the direct request `Host` and ignores ` | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | | **GitHub Event-Driven Agents** | Custom agents can declare a `github:` block in their `config.yaml` to bind to repos and event triggers. Webhook fan-out publishes one `InboundMessage` per matching binding to the channel bus; `GitHubChannel` routes those messages through `ChannelManager`. The response `dispatch` summarizes matched/fired/skipped agents. | +Thread identifiers use the shared `deerflow.utils.thread_id` contract +`^[A-Za-z0-9_-]{1,64}$`. Caller-provided opaque IDs remain supported; UUIDs +are generated only for `None`, while explicit empty strings fail validation. +Gateway creation and state-producing request boundaries, embedded-client +entry points, filesystem/upload/event-store consumers, scheduled launches, +and the standalone Provisioner enforce the same contract before persistence +or workspace initialization. Route-addressable legacy IDs remain accepted by +pure reads and cleanup/control endpoints. Deleting a noncanonical legacy ID +best-effort removes its metadata and checkpoints but deliberately skips local +filesystem cleanup, so the raw value is never interpolated into a host path; +new runs, workspace/sandbox operations, and other state-producing mutations +remain blocked. + **Workspace change review**: `packages/harness/deerflow/workspace_changes/` captures a pre-run and post-run snapshot of the thread-owned `workspace` and `outputs` directories. `runtime/runs/worker.py` performs the filesystem scan via diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py index 35019b97d..512d38eaf 100644 --- a/backend/app/gateway/routers/artifacts.py +++ b/backend/app/gateway/routers/artifacts.py @@ -23,6 +23,7 @@ from deerflow.config.paths import make_safe_user_id from deerflow.runtime import ConflictError, ThreadOperationKind from deerflow.runtime.user_context import get_effective_user_id from deerflow.sandbox.sandbox_provider import get_sandbox_provider +from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -271,7 +272,7 @@ def _read_artifact_payload(actual_path: Path, path: str, download: bool) -> tupl description="Retrieve an artifact file generated by the AI agent. Text and binary files can be viewed inline, while active web content is always downloaded.", ) @require_permission("threads", "read", owner_check=True) -async def get_artifact(thread_id: str, path: str, request: Request, download: bool = False) -> Response: +async def get_artifact(thread_id: ThreadId, path: str, request: Request, download: bool = False) -> Response: """Get an artifact file by its path. The endpoint automatically detects file types and returns appropriate content types. @@ -385,7 +386,7 @@ async def get_artifact(thread_id: str, path: str, request: Request, download: bo ) @require_permission("threads", "write", owner_check=True, require_existing=True) async def update_artifact( - thread_id: str, + thread_id: ThreadId, path: str, body: ArtifactUpdateRequest, request: Request, diff --git a/backend/app/gateway/routers/browser.py b/backend/app/gateway/routers/browser.py index ce8e264c8..2220b3076 100644 --- a/backend/app/gateway/routers/browser.py +++ b/backend/app/gateway/routers/browser.py @@ -10,6 +10,7 @@ from app.gateway.authz import require_permission from app.gateway.browser_capability import browser_capability from deerflow.config.paths import get_paths from deerflow.runtime.user_context import get_effective_user_id, reset_current_user, set_current_user +from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -76,7 +77,7 @@ async def _browser_thread_owned_by(thread_store, thread_id: str, user_id: str) - description="Steer the thread's live browser session to a URL from the UI and capture a screenshot.", ) @require_permission("threads", "write", owner_check=True, require_existing=True) -async def navigate_browser(thread_id: str, body: BrowserNavigateRequest, request: Request) -> BrowserNavigateResponse: +async def navigate_browser(thread_id: ThreadId, body: BrowserNavigateRequest, request: Request) -> BrowserNavigateResponse: user_id = str(request.state.auth.user.id) thread_store = getattr(request.app.state, "thread_store", None) if thread_store is None or not await _browser_thread_owned_by(thread_store, thread_id, user_id): @@ -176,7 +177,7 @@ def _ws_origin_allowed(websocket: WebSocket) -> bool: @router.websocket("/threads/{thread_id}/browser/stream") -async def browser_stream(websocket: WebSocket, thread_id: str) -> None: +async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None: """Bidirectional live browser stream. Server → client: JSON ``{"type":"frame","data":""}`` frames diff --git a/backend/app/gateway/routers/feedback.py b/backend/app/gateway/routers/feedback.py index ca5c1d406..c86a9c32c 100644 --- a/backend/app/gateway/routers/feedback.py +++ b/backend/app/gateway/routers/feedback.py @@ -14,6 +14,7 @@ from pydantic import BaseModel, Field from app.gateway.authz import require_permission from app.gateway.deps import get_current_user, get_feedback_repo, get_run_store +from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/threads", tags=["feedback"]) @@ -61,7 +62,7 @@ class FeedbackStatsResponse(BaseModel): @router.put("/{thread_id}/runs/{run_id}/feedback", response_model=FeedbackResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) async def upsert_feedback( - thread_id: str, + thread_id: ThreadId, run_id: str, body: FeedbackUpsertRequest, request: Request, @@ -92,7 +93,7 @@ async def upsert_feedback( @router.delete("/{thread_id}/runs/{run_id}/feedback") @require_permission("threads", "delete", owner_check=True, require_existing=True) async def delete_run_feedback( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, ) -> dict[str, bool]: @@ -112,7 +113,7 @@ async def delete_run_feedback( @router.post("/{thread_id}/runs/{run_id}/feedback", response_model=FeedbackResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) async def create_feedback( - thread_id: str, + thread_id: ThreadId, run_id: str, body: FeedbackCreateRequest, request: Request, @@ -145,7 +146,7 @@ async def create_feedback( @router.get("/{thread_id}/runs/{run_id}/feedback", response_model=list[FeedbackResponse]) @require_permission("threads", "read", owner_check=True) async def list_feedback( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, ) -> list[dict[str, Any]]: @@ -157,7 +158,7 @@ async def list_feedback( @router.get("/{thread_id}/runs/{run_id}/feedback/stats", response_model=FeedbackStatsResponse) @require_permission("threads", "read", owner_check=True) async def feedback_stats( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, ) -> dict[str, Any]: @@ -169,7 +170,7 @@ async def feedback_stats( @router.delete("/{thread_id}/runs/{run_id}/feedback/{feedback_id}") @require_permission("threads", "delete", owner_check=True, require_existing=True) async def delete_feedback( - thread_id: str, + thread_id: ThreadId, run_id: str, feedback_id: str, request: Request, diff --git a/backend/app/gateway/routers/runs.py b/backend/app/gateway/routers/runs.py index 7ea06ca85..dc80a60f1 100644 --- a/backend/app/gateway/routers/runs.py +++ b/backend/app/gateway/routers/runs.py @@ -8,7 +8,6 @@ is reused so that conversation history is preserved across calls. from __future__ import annotations import logging -import uuid from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse @@ -19,6 +18,7 @@ from app.gateway.pagination import trim_run_message_page from app.gateway.run_models import RunCreateRequest from app.gateway.services import build_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion from deerflow.runtime import serialize_channel_values_for_api +from deerflow.utils.thread_id import resolve_thread_id logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/runs", tags=["runs"]) @@ -27,9 +27,7 @@ router = APIRouter(prefix="/api/runs", tags=["runs"]) def _resolve_thread_id(body: RunCreateRequest) -> str: """Return the thread_id from the request body, or generate a new one.""" thread_id = ((body.config or {}).get("configurable") or {}).get("thread_id") - if thread_id: - return str(thread_id) - return str(uuid.uuid4()) + return resolve_thread_id(thread_id) @router.post("/stream") diff --git a/backend/app/gateway/routers/scheduled_tasks.py b/backend/app/gateway/routers/scheduled_tasks.py index 35cd74bf1..da1b5d130 100644 --- a/backend/app/gateway/routers/scheduled_tasks.py +++ b/backend/app/gateway/routers/scheduled_tasks.py @@ -23,6 +23,7 @@ from deerflow.scheduler.schedules import ( normalize_cron_expression, validate_timezone, ) +from deerflow.utils.thread_id import ThreadId router = APIRouter(prefix="/api", tags=["scheduled-tasks"]) @@ -36,7 +37,7 @@ def _ensure_task_mutable(task: dict[str, Any]) -> None: class ScheduledTaskCreateRequest(BaseModel): - thread_id: str | None = None + thread_id: ThreadId | None = None context_mode: str = "fresh_thread_per_run" title: str = Field(min_length=1) prompt: str = Field(min_length=1) @@ -47,7 +48,7 @@ class ScheduledTaskCreateRequest(BaseModel): class ScheduledTaskUpdateRequest(BaseModel): context_mode: str | None = None - thread_id: str | None = None + thread_id: ThreadId | None = None title: str | None = Field(default=None, min_length=1) prompt: str | None = Field(default=None, min_length=1) schedule_spec: dict[str, Any] | None = None @@ -302,7 +303,7 @@ async def list_scheduled_task_runs( @router.get("/threads/{thread_id}/scheduled-tasks") @require_permission("threads", "read", owner_check=True) -async def list_thread_scheduled_tasks(thread_id: str, request: Request): +async def list_thread_scheduled_tasks(thread_id: ThreadId, request: Request): repo = get_scheduled_task_repo(request) user = await get_optional_user_from_request(request) if user is None: diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index d1f315182..82d063874 100644 --- a/backend/app/gateway/routers/skills.py +++ b/backend/app/gateway/routers/skills.py @@ -31,6 +31,7 @@ from deerflow.skills.security_static_scanner import ( ) from deerflow.skills.storage import SkillStorage, get_or_new_user_skill_storage from deerflow.skills.types import SKILL_MD_FILE, SkillCategory +from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -65,7 +66,7 @@ class SkillUpdateRequest(BaseModel): class SkillInstallRequest(BaseModel): """Request model for installing a skill from a .skill file.""" - thread_id: str = Field(..., description="The thread ID where the .skill file is located") + thread_id: ThreadId = Field(..., description="The thread ID where the .skill file is located") path: str = Field(..., description="Virtual path to the .skill file (e.g., mnt/user-data/outputs/my-skill.skill)") diff --git a/backend/app/gateway/routers/suggestions.py b/backend/app/gateway/routers/suggestions.py index ce9e38f1b..59dc2cb3f 100644 --- a/backend/app/gateway/routers/suggestions.py +++ b/backend/app/gateway/routers/suggestions.py @@ -10,6 +10,7 @@ from app.gateway.deps import get_config from deerflow.config.app_config import AppConfig from deerflow.config.suggestions_config import DEFAULT_MAX_SUGGESTIONS, MAX_SUGGESTIONS_LIMIT from deerflow.utils.oneshot_llm import run_oneshot_llm +from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -102,7 +103,7 @@ async def get_suggestions_config( ) @require_permission("threads", "read", owner_check=True) async def generate_suggestions( - thread_id: str, + thread_id: ThreadId, body: SuggestionsRequest, request: Request, config: AppConfig = Depends(get_config), diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index de2dbb0b7..ec658fbc0 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -43,6 +43,7 @@ from deerflow.agents.middlewares.dynamic_context_middleware import strip_injecte from deerflow.runtime import CancelOutcome, RunRecord, RunStatus, serialize_channel_values_for_api from deerflow.runtime.secret_context import redact_config_secrets, redact_metadata_secrets from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text +from deerflow.utils.thread_id import ThreadId from deerflow.workspace_changes import get_workspace_changes_response logger = logging.getLogger(__name__) @@ -813,7 +814,7 @@ async def _default_history_hidden_run_ids(run_mgr: Any, thread_id: str, *, user_ @router.post("/{thread_id}/runs/regenerate/prepare", response_model=RegeneratePrepareResponse) @require_permission("runs", "create", owner_check=True, require_existing=True) async def prepare_regenerate_run( - thread_id: str, + thread_id: ThreadId, body: RegeneratePrepareRequest, request: Request, ) -> RegeneratePrepareResponse: @@ -824,7 +825,7 @@ async def prepare_regenerate_run( @router.post("/{thread_id}/runs/edit-regenerate/prepare", response_model=EditRegeneratePrepareResponse) @require_permission("runs", "create", owner_check=True, require_existing=True) async def prepare_edit_regenerate_run( - thread_id: str, + thread_id: ThreadId, body: EditRegeneratePrepareRequest, request: Request, ) -> EditRegeneratePrepareResponse: @@ -834,7 +835,7 @@ async def prepare_edit_regenerate_run( @router.post("/{thread_id}/runs", response_model=RunResponse) @require_permission("runs", "create", owner_check=True, require_existing=True) -async def create_run(thread_id: str, body: RunCreateRequest, request: Request) -> RunResponse: +async def create_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> RunResponse: """Create a background run (returns immediately).""" record = await start_run(body, thread_id, request) return _record_to_response(record) @@ -842,7 +843,7 @@ async def create_run(thread_id: str, body: RunCreateRequest, request: Request) - @router.post("/{thread_id}/runs/stream") @require_permission("runs", "create", owner_check=True, require_existing=True) -async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) -> StreamingResponse: +async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> StreamingResponse: """Create a run and stream events via SSE. The response includes a ``Content-Location`` header with the run's @@ -870,7 +871,7 @@ async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) - @router.post("/{thread_id}/runs/wait", response_model=dict) @require_permission("runs", "create", owner_check=True, require_existing=True) -async def wait_run(thread_id: str, body: RunCreateRequest, request: Request) -> dict: +async def wait_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> dict: """Create a run and block until it completes, returning the final state.""" bridge = get_stream_bridge(request) run_mgr = get_run_manager(request) @@ -899,7 +900,7 @@ async def wait_run(thread_id: str, body: RunCreateRequest, request: Request) -> @router.get("/{thread_id}/runs", response_model=list[RunResponse]) @require_permission("runs", "read", owner_check=True) -async def list_runs(thread_id: str, request: Request) -> list[RunResponse]: +async def list_runs(thread_id: ThreadId, request: Request) -> list[RunResponse]: """List all runs for a thread.""" run_mgr = get_run_manager(request) user_id = await get_current_user(request) @@ -909,7 +910,7 @@ async def list_runs(thread_id: str, request: Request) -> list[RunResponse]: @router.get("/{thread_id}/runs/{run_id}", response_model=RunResponse) @require_permission("runs", "read", owner_check=True) -async def get_run(thread_id: str, run_id: str, request: Request) -> RunResponse: +async def get_run(thread_id: ThreadId, run_id: str, request: Request) -> RunResponse: """Get details of a specific run.""" run_mgr = get_run_manager(request) user_id = await get_current_user(request) @@ -922,7 +923,7 @@ async def get_run(thread_id: str, run_id: str, request: Request) -> RunResponse: @router.post("/{thread_id}/runs/{run_id}/cancel") @require_permission("runs", "cancel", owner_check=True, require_existing=True) async def cancel_run( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, wait: bool = Query(default=False, description="Block until run completes after cancel"), @@ -981,7 +982,7 @@ async def cancel_run( @router.get("/{thread_id}/runs/{run_id}/join") @require_permission("runs", "read", owner_check=True) -async def join_run(thread_id: str, run_id: str, request: Request) -> StreamingResponse: +async def join_run(thread_id: ThreadId, run_id: str, request: Request) -> StreamingResponse: """Join an existing run's SSE stream.""" run_mgr = get_run_manager(request) record = await run_mgr.get(run_id) @@ -1010,7 +1011,7 @@ async def join_run(thread_id: str, run_id: str, request: Request) -> StreamingRe @router.post("/{thread_id}/runs/{run_id}/stream", response_model=None) @require_permission("runs", "read", owner_check=True) async def stream_existing_run( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, action: Literal["interrupt", "rollback"] | None = Query(default=None, description="Cancel action"), @@ -1083,7 +1084,7 @@ async def stream_existing_run( @router.get("/{thread_id}/messages") @require_permission("runs", "read", owner_check=True) async def list_thread_messages( - thread_id: str, + thread_id: ThreadId, request: Request, limit: int = Query(default=50, ge=1, le=200), before_seq: int | None = Query(default=None, ge=1), @@ -1337,7 +1338,7 @@ async def _enrich_thread_message_page( @router.get("/{thread_id}/messages/page", response_model=ThreadMessagesPageResponse) @require_permission("runs", "read", owner_check=True) async def list_thread_messages_page( - thread_id: str, + thread_id: ThreadId, request: Request, limit: int = Query(default=50, ge=1, le=200), before_seq: int | None = Query(default=None, ge=1), @@ -1365,7 +1366,7 @@ async def list_thread_messages_page( @router.get("/{thread_id}/runs/{run_id}/messages") @require_permission("runs", "read", owner_check=True) async def list_run_messages( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, limit: int = Query(default=50, le=200, ge=1), @@ -1400,7 +1401,7 @@ async def list_run_messages( @router.get("/{thread_id}/runs/{run_id}/events") @require_permission("runs", "read", owner_check=True) async def list_run_events( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, event_types: str | None = Query(default=None), @@ -1437,7 +1438,7 @@ async def list_run_events( @router.get("/{thread_id}/runs/{run_id}/workspace-changes") @require_permission("runs", "read", owner_check=True) async def get_run_workspace_changes( - thread_id: str, + thread_id: ThreadId, run_id: str, request: Request, include_files: bool = Query(default=True), @@ -1457,7 +1458,7 @@ async def get_run_workspace_changes( @router.get("/{thread_id}/token-usage", response_model=ThreadTokenUsageResponse) @require_permission("threads", "read", owner_check=True) async def thread_token_usage( - thread_id: str, + thread_id: ThreadId, request: Request, include_active: bool = Query(default=False, description="Include running run progress snapshots"), ) -> ThreadTokenUsageResponse: diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index df45306a9..8fa7b1205 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -68,6 +68,7 @@ from deerflow.runtime.runs.worker import valid_duration_entry from deerflow.runtime.secret_context import redact_metadata_secrets from deerflow.runtime.user_context import get_effective_user_id from deerflow.utils.file_io import run_file_io +from deerflow.utils.thread_id import ThreadId, resolve_thread_id, validate_thread_id from deerflow.utils.time import coerce_iso, now_iso logger = logging.getLogger(__name__) @@ -372,7 +373,7 @@ class ThreadResponse(_MetadataRedactingResponse): class ThreadCreateRequest(BaseModel): """Request body for creating a thread.""" - thread_id: str | None = Field(default=None, description="Optional thread ID (auto-generated if omitted)") + thread_id: ThreadId | None = Field(default=None, description="Optional thread ID (auto-generated if omitted)") assistant_id: str | None = Field(default=None, description="Associate thread with an assistant") metadata: dict[str, Any] = Field(default_factory=dict, description="Initial metadata") @@ -627,8 +628,18 @@ async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteRe """ from app.gateway.deps import get_thread_store - # Clean local filesystem - response = _delete_thread_data(thread_id, user_id=get_effective_user_id()) + # Legacy IDs may predate the canonical filesystem-safe contract. They can + # still be removed from metadata/checkpoint stores, but must never be + # interpolated into a host path during cleanup. + try: + validate_thread_id(thread_id) + except ValueError: + response = ThreadDeleteResponse( + success=True, + message="Skipped local data cleanup for legacy thread ID", + ) + else: + response = _delete_thread_data(thread_id, user_id=get_effective_user_id()) # Remove checkpoints (best-effort) checkpointer = getattr(request.app.state, "checkpointer", None) @@ -707,7 +718,7 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe checkpointer = get_checkpointer(request) thread_store = get_thread_store(request) - thread_id = body.thread_id or str(uuid.uuid4()) + thread_id = resolve_thread_id(body.thread_id) now = now_iso() thread_owner_user_id = get_trusted_internal_owner_user_id(request) thread_owner_kwargs = {"user_id": thread_owner_user_id} if thread_owner_user_id else {} @@ -774,7 +785,7 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe @router.post("/{thread_id}/branches", response_model=ThreadBranchResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) -async def branch_thread(thread_id: str, body: ThreadBranchRequest, request: Request) -> ThreadBranchResponse: +async def branch_thread(thread_id: ThreadId, body: ThreadBranchRequest, request: Request) -> ThreadBranchResponse: """Create a new main-thread branch from a completed assistant turn.""" from app.gateway.deps import get_thread_store @@ -977,7 +988,7 @@ async def search_threads(body: ThreadSearchRequest, request: Request) -> list[Th @router.patch("/{thread_id}", response_model=ThreadResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) -async def patch_thread(thread_id: str, body: ThreadPatchRequest, request: Request) -> ThreadResponse: +async def patch_thread(thread_id: ThreadId, body: ThreadPatchRequest, request: Request) -> ThreadResponse: """Merge metadata into a thread record.""" from app.gateway.deps import get_thread_store @@ -1010,7 +1021,7 @@ async def patch_thread(thread_id: str, body: ThreadPatchRequest, request: Reques @router.get("/{thread_id}", response_model=ThreadResponse) @require_permission("threads", "read", owner_check=True) -async def get_thread(thread_id: str, request: Request) -> ThreadResponse: +async def get_thread(thread_id: ThreadId, request: Request) -> ThreadResponse: """Get thread info from metadata plus the graph's materialized state.""" from app.gateway.deps import get_thread_store @@ -1063,7 +1074,7 @@ async def get_thread(thread_id: str, request: Request) -> ThreadResponse: @router.get("/{thread_id}/goal", response_model=ThreadGoalResponse) @require_permission("threads", "read", owner_check=True) -async def get_thread_goal(thread_id: str, request: Request) -> ThreadGoalResponse: +async def get_thread_goal(thread_id: ThreadId, request: Request) -> ThreadGoalResponse: """Return the active Claude-style goal for a thread, if any.""" checkpointer = get_checkpointer(request) try: @@ -1076,7 +1087,7 @@ async def get_thread_goal(thread_id: str, request: Request) -> ThreadGoalRespons @router.put("/{thread_id}/goal", response_model=ThreadGoalResponse) @require_permission("threads", "write", owner_check=True) -async def set_thread_goal(thread_id: str, body: ThreadGoalRequest, request: Request) -> ThreadGoalResponse: +async def set_thread_goal(thread_id: ThreadId, body: ThreadGoalRequest, request: Request) -> ThreadGoalResponse: """Set or replace the active goal for a thread. ``/chats/new`` pages already hold a generated UUID before the first run, so @@ -1102,7 +1113,7 @@ async def set_thread_goal(thread_id: str, body: ThreadGoalRequest, request: Requ @router.delete("/{thread_id}/goal", response_model=ThreadGoalResponse) @require_permission("threads", "write", owner_check=True) -async def clear_thread_goal(thread_id: str, request: Request) -> ThreadGoalResponse: +async def clear_thread_goal(thread_id: ThreadId, request: Request) -> ThreadGoalResponse: """Clear the active goal for a thread.""" checkpointer = get_checkpointer(request) try: @@ -1133,7 +1144,7 @@ def _thread_compact_response(result: ThreadCompactionResult) -> ThreadCompactRes @router.post("/{thread_id}/compact", response_model=ThreadCompactResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) -async def compact_thread(thread_id: str, body: ThreadCompactRequest, request: Request) -> ThreadCompactResponse: +async def compact_thread(thread_id: ThreadId, body: ThreadCompactRequest, request: Request) -> ThreadCompactResponse: """Manually summarize old thread context while preserving the visible history.""" # Compaction writes only base-schema channels (messages + summary_text); # every other channel — including middleware-contributed ones — is carried @@ -1180,7 +1191,7 @@ async def compact_thread(thread_id: str, body: ThreadCompactRequest, request: Re # --------------------------------------------------------------------------- @router.get("/{thread_id}/state", response_model=ThreadStateResponse) @require_permission("threads", "read", owner_check=True) -async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse: +async def get_thread_state(thread_id: ThreadId, request: Request) -> ThreadStateResponse: """Get the latest materialized graph state for a thread.""" # Resolve through the thread's assistant so custom middleware channels # appear in the response instead of being dropped by the default schema. @@ -1222,7 +1233,7 @@ async def get_thread_state(thread_id: str, request: Request) -> ThreadStateRespo @router.post("/{thread_id}/state", response_model=ThreadStateResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) -async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse: +async def update_thread_state(thread_id: ThreadId, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse: """Replace selected thread-state fields through the materialized graph.""" from app.gateway.deps import get_thread_store @@ -1322,7 +1333,7 @@ def _checkpoint_run_durations(metadata: Any) -> dict[str, int]: @router.post("/{thread_id}/history", response_model=list[HistoryEntry]) @require_permission("threads", "read", owner_check=True) async def get_thread_history( - thread_id: str, + thread_id: ThreadId, body: ThreadHistoryRequest, request: Request, background_tasks: BackgroundTasks, diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py index f9b29fed5..7c5628273 100644 --- a/backend/app/gateway/routers/uploads.py +++ b/backend/app/gateway/routers/uploads.py @@ -35,6 +35,7 @@ from deerflow.uploads.manager import ( ) from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown from deerflow.utils.file_io import run_file_io +from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -299,7 +300,7 @@ def _auto_convert_documents_enabled(app_config: AppConfig) -> bool: @router.post("", response_model=UploadResponse) @require_permission("threads", "write", owner_check=True, require_existing=False) async def upload_files( - thread_id: str, + thread_id: ThreadId, request: Request, files: list[UploadFile] = File(...), config: AppConfig = Depends(get_config), @@ -445,7 +446,7 @@ async def upload_files( @router.get("/limits", response_model=UploadLimits) @require_permission("threads", "read", owner_check=True) async def get_upload_limits( - thread_id: str, + thread_id: ThreadId, request: Request, config: AppConfig = Depends(get_config), ) -> UploadLimits: @@ -455,7 +456,7 @@ async def get_upload_limits( @router.get("/list", response_model=UploadListResponse) @require_permission("threads", "read", owner_check=True) -async def list_uploaded_files(thread_id: str, request: Request) -> UploadListResponse: +async def list_uploaded_files(thread_id: ThreadId, request: Request) -> UploadListResponse: """List all files in a thread's uploads directory.""" try: result = await run_file_io(_list_uploaded_files_for_thread, thread_id, get_effective_user_id()) @@ -467,7 +468,7 @@ async def list_uploaded_files(thread_id: str, request: Request) -> UploadListRes @router.delete("/{filename}") @require_permission("threads", "delete", owner_check=True, require_existing=True) -async def delete_uploaded_file(thread_id: str, filename: str, request: Request) -> dict: +async def delete_uploaded_file(thread_id: ThreadId, filename: str, request: Request) -> dict: """Delete a file from a thread's uploads directory.""" try: return await run_file_io(_delete_uploaded_file_for_thread, thread_id, filename, get_effective_user_id()) diff --git a/backend/app/gateway/run_models.py b/backend/app/gateway/run_models.py index 08e648501..f476a8ee0 100644 --- a/backend/app/gateway/run_models.py +++ b/backend/app/gateway/run_models.py @@ -4,10 +4,11 @@ from __future__ import annotations from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator +from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator from pydantic_core import PydanticCustomError from deerflow.runtime.stream_modes import RunStreamMode, UnsupportedStreamModeError, normalize_stream_modes +from deerflow.utils.thread_id import validate_thread_id class RunCreateRequest(BaseModel): @@ -36,6 +37,19 @@ class RunCreateRequest(BaseModel): if_not_exists: Literal["create"] = Field(default="create", description="Compatibility default; missing threads are created") feedback_keys: None = Field(default=None, description="Compatibility placeholder; feedback key collection is not supported") + @model_validator(mode="after") + def validate_configurable_thread_id(self) -> RunCreateRequest: + """Validate the stateless-run thread selector inside RunnableConfig.""" + if not isinstance(self.config, dict): + return self + configurable = self.config.get("configurable") + if not isinstance(configurable, dict) or "thread_id" not in configurable: + return self + thread_id = configurable["thread_id"] + if thread_id is not None: + validate_thread_id(thread_id) + return self + @field_validator( "webhook", "on_completion", diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 6d30f494e..553221a32 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -71,6 +71,7 @@ from deerflow.runtime.secret_context import ( from deerflow.runtime.stream_modes import normalize_stream_modes from deerflow.runtime.user_context import reset_current_user, set_current_user from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY +from deerflow.utils.thread_id import validate_thread_id logger = logging.getLogger(__name__) @@ -1062,6 +1063,11 @@ async def start_run( request : Request FastAPI request — used to retrieve singletons from ``app.state``. """ + try: + validate_thread_id(thread_id) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + body_config = getattr(body, "config", None) config_metadata = body_config.get("metadata") if isinstance(body_config, dict) else None try: diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 9bcfc268a..2a6574f03 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -12,6 +12,7 @@ from fastapi import HTTPException from deerflow.persistence.scheduled_task_runs import ActiveScheduledRunConflict from deerflow.runtime import ConflictError, RunRecord from deerflow.scheduler.schedules import next_run_at +from deerflow.utils.thread_id import validate_thread_id logger = logging.getLogger(__name__) @@ -105,8 +106,41 @@ class ScheduledTaskService: trigger: str, ) -> dict[str, Any]: execution_thread_id = task.get("thread_id") - if task.get("context_mode") == "fresh_thread_per_run" or not execution_thread_id: + if task.get("context_mode") == "fresh_thread_per_run" or execution_thread_id is None: execution_thread_id = str(uuid.uuid4()) + try: + validate_thread_id(execution_thread_id) + except ValueError as exc: + # Rows persisted before the thread-id contract was centralized may + # hold IDs that were valid then (dots, unlimited length) but fail + # the canonical pattern now. Route through the normal failure + # bookkeeping instead of raising: an uncaught ValueError here would + # surface as HTTP 500 on manual trigger and, in the poller, abort + # the rest of the claimed batch every cycle while the task itself + # is never marked with last_error. + task_status = self._task_status_for_failure(task, trigger=trigger) + await self._task_repo.update_after_launch( + task["id"], + status=task_status, + next_run_at=next_run_at( + task["schedule_type"], + task["schedule_spec"], + task["timezone"], + now=now, + ), + last_run_at=now, + last_run_id=None, + last_thread_id=execution_thread_id, + last_error=str(exc), + increment_run_count=False, + ) + return { + "outcome": "failed", + "task_run_id": None, + "run_id": None, + "thread_id": execution_thread_id, + "error": str(exc), + } # "skip" must hold for fresh-thread runs too, where every run gets a new # thread and the same-thread multitask ConflictError below can never # fire. Checked before creating this dispatch's own run row so the row diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index cf667a359..508764a15 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -72,6 +72,7 @@ from deerflow.uploads.manager import ( upload_artifact_url, upload_virtual_path, ) +from deerflow.utils.thread_id import resolve_thread_id, validate_thread_id logger = logging.getLogger(__name__) @@ -525,6 +526,7 @@ class DeerFlowClient: def get_goal(self, thread_id: str) -> dict: """Return the active goal for a thread, if any.""" + validate_thread_id(thread_id) checkpointer = self._get_thread_checkpointer() goal = _run_async_from_sync(read_thread_goal(checkpointer, thread_id)) return {"goal": goal} @@ -537,6 +539,7 @@ class DeerFlowClient: max_continuations: int = DEFAULT_MAX_GOAL_CONTINUATIONS, ) -> dict: """Set or replace a thread-scoped goal.""" + validate_thread_id(thread_id) checkpointer = self._get_thread_checkpointer() goal = build_goal_state(objective, max_continuations=max_continuations) @@ -549,6 +552,7 @@ class DeerFlowClient: def clear_goal(self, thread_id: str) -> dict: """Clear the active goal for a thread.""" + validate_thread_id(thread_id) checkpointer = self._get_thread_checkpointer() async def _clear_goal() -> None: @@ -803,8 +807,7 @@ class DeerFlowClient: Tool results also include ``"artifact"`` when the source ToolMessage has a non-None artifact. - type="end" data={"usage": {"input_tokens": int, "output_tokens": int, "total_tokens": int}} """ - if thread_id is None: - thread_id = str(uuid.uuid4()) + thread_id = resolve_thread_id(thread_id) config = self._get_runnable_config(thread_id, **kwargs) inject_checkpoint_mode(config, self._checkpoint_channel_mode) @@ -1481,6 +1484,7 @@ class DeerFlowClient: FileNotFoundError: If any file does not exist. ValueError: If any supplied path exists but is not a regular file. """ + validate_thread_id(thread_id) from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown # Validate all files upfront to avoid partial uploads. @@ -1584,6 +1588,7 @@ class DeerFlowClient: Dict with "files" and "count" keys, matching the Gateway API ``list_uploaded_files`` response. """ + validate_thread_id(thread_id) uploads_dir = get_uploads_dir(thread_id) result = list_files_in_dir(uploads_dir) return enrich_file_listing(result, thread_id) @@ -1603,6 +1608,7 @@ class DeerFlowClient: FileNotFoundError: If the file does not exist. PermissionError: If path traversal is detected. """ + validate_thread_id(thread_id) from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS uploads_dir = get_uploads_dir(thread_id) @@ -1626,6 +1632,7 @@ class DeerFlowClient: FileNotFoundError: If the artifact does not exist. ValueError: If the path is invalid. """ + validate_thread_id(thread_id) try: actual = get_paths().resolve_virtual_path(thread_id, path, user_id=get_effective_user_id()) except ValueError as exc: diff --git a/backend/packages/harness/deerflow/config/paths.py b/backend/packages/harness/deerflow/config/paths.py index 495ecafed..670e24906 100644 --- a/backend/packages/harness/deerflow/config/paths.py +++ b/backend/packages/harness/deerflow/config/paths.py @@ -6,11 +6,11 @@ import shutil from pathlib import Path, PureWindowsPath from deerflow.config.runtime_paths import runtime_home +from deerflow.utils.thread_id import validate_thread_id # Virtual path prefix seen by agents inside the sandbox VIRTUAL_PATH_PREFIX = "/mnt/user-data" -_SAFE_THREAD_ID_RE = re.compile(r"^[A-Za-z0-9_\-]+$") _SAFE_USER_ID_RE = re.compile(r"^[A-Za-z0-9_\-]+$") _SAFE_INTEGRATION_ID_RE = re.compile(r"^[A-Za-z0-9_.\-]+$") _UNSAFE_USER_ID_CHAR_RE = re.compile(r"[^A-Za-z0-9_\-]") @@ -26,9 +26,7 @@ def _default_local_base_dir() -> Path: def _validate_thread_id(thread_id: str) -> str: """Validate a thread ID before using it in filesystem paths.""" - if not _SAFE_THREAD_ID_RE.match(thread_id): - raise ValueError(f"Invalid thread_id {thread_id!r}: only alphanumeric characters, hyphens, and underscores are allowed.") - return thread_id + return validate_thread_id(thread_id) def _validate_user_id(user_id: str) -> str: diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index a1af143b6..105ce894d 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -32,6 +32,7 @@ from typing import Any from deerflow.runtime.events.store.base import RunEventStore from deerflow.runtime.user_context import AUTO, _AutoSentinel +from deerflow.utils.thread_id import validate_thread_id logger = logging.getLogger(__name__) @@ -56,7 +57,7 @@ class JsonlRunEventStore(RunEventStore): return value def _thread_dir(self, thread_id: str) -> Path: - self._validate_id(thread_id, "thread_id") + validate_thread_id(thread_id) return self._base_dir / "threads" / thread_id / "runs" def _run_file(self, thread_id: str, run_id: str) -> Path: diff --git a/backend/packages/harness/deerflow/tui/session.py b/backend/packages/harness/deerflow/tui/session.py index e1effe32a..478712ee6 100644 --- a/backend/packages/harness/deerflow/tui/session.py +++ b/backend/packages/harness/deerflow/tui/session.py @@ -39,18 +39,29 @@ class Session: Matches an existing thread by id first, then by exact title. Falls back to the literal ref (treated as an id) when nothing matches, so an unknown id - still continues/creates that namespace. + still continues/creates that namespace — provided it satisfies the + canonical thread ID contract. """ try: threads = self.client.list_threads(limit=100).get("thread_list", []) except Exception: # noqa: BLE001 - resolution is best-effort - return ref + return self._validated_literal_ref(ref) if any(t.get("thread_id") == ref for t in threads): return ref for thread in threads: if (thread.get("title") or "") == ref: - return thread.get("thread_id") or ref - return ref + return thread.get("thread_id") or self._validated_literal_ref(ref) + return self._validated_literal_ref(ref) + + @staticmethod + def _validated_literal_ref(ref: str) -> str: + """Validate a literal ref before it is adopted as a thread id.""" + from deerflow.utils.thread_id import validate_thread_id + + try: + return validate_thread_id(ref) + except ValueError as exc: + raise ValueError(f"Thread reference {ref!r} matches no existing thread and is not a valid thread id (expected 1-64 ASCII letters, digits, hyphens, or underscores).") from exc def recent_threads(self, limit: int = 20) -> list[dict]: return self.client.list_threads(limit=limit).get("thread_list", []) diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index 691cbc478..58d6b3ad2 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -7,13 +7,13 @@ Both Gateway and Client delegate to these functions. import errno import logging import os -import re import stat from pathlib import Path from urllib.parse import quote from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths from deerflow.runtime.user_context import get_effective_user_id +from deerflow.utils.thread_id import validate_thread_id class PathTraversalError(ValueError): @@ -26,22 +26,10 @@ class UnsafeUploadPathError(ValueError): logger = logging.getLogger(__name__) -# thread_id must be alphanumeric, hyphens, underscores, or dots only. -_SAFE_THREAD_ID = re.compile(r"^[a-zA-Z0-9._-]+$") UPLOAD_STAGING_PREFIX = ".upload-" UPLOAD_STAGING_SUFFIX = ".part" -def validate_thread_id(thread_id: str) -> None: - """Reject thread IDs containing characters unsafe for filesystem paths. - - Raises: - ValueError: If thread_id is empty or contains unsafe characters. - """ - if not thread_id or not _SAFE_THREAD_ID.match(thread_id): - raise ValueError(f"Invalid thread_id: {thread_id!r}") - - def get_uploads_dir(thread_id: str, *, user_id: str | None = None) -> Path: """Return the uploads directory path for a thread (no side effects).""" validate_thread_id(thread_id) diff --git a/backend/packages/harness/deerflow/utils/thread_id.py b/backend/packages/harness/deerflow/utils/thread_id.py new file mode 100644 index 000000000..1bae01348 --- /dev/null +++ b/backend/packages/harness/deerflow/utils/thread_id.py @@ -0,0 +1,37 @@ +"""Canonical thread identifier validation shared across DeerFlow backends.""" + +from __future__ import annotations + +import re +import uuid +from typing import Annotated + +from pydantic import AfterValidator, StringConstraints + +THREAD_ID_PATTERN = r"^[A-Za-z0-9_-]{1,64}$" +_THREAD_ID_RE = re.compile(THREAD_ID_PATTERN) + + +def validate_thread_id(thread_id: str) -> str: + """Return a valid thread ID or raise ``ValueError``. + + Thread IDs are caller-defined opaque identifiers, not necessarily UUIDs, + but they must be safe for every persistence and filesystem backend. + """ + if not isinstance(thread_id, str) or _THREAD_ID_RE.fullmatch(thread_id) is None: + raise ValueError("Invalid thread_id: expected 1-64 ASCII letters, digits, hyphens, or underscores") + return thread_id + + +def resolve_thread_id(thread_id: str | None) -> str: + """Validate a supplied ID, generating a UUID only when it is ``None``.""" + if thread_id is None: + return str(uuid.uuid4()) + return validate_thread_id(thread_id) + + +ThreadId = Annotated[ + str, + StringConstraints(min_length=1, max_length=64, pattern=THREAD_ID_PATTERN), + AfterValidator(validate_thread_id), +] diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 9d283a3b6..0d204b7dc 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -2748,3 +2748,17 @@ async def test_run_agent_full_mode_checks_selected_checkpoint_before_graph(): RunStatus.error, error="Thread requires delta mode; materialize and convert its checkpoints before using full mode.", ) + + +@pytest.mark.asyncio +async def test_start_run_rejects_invalid_thread_id_before_resolving_dependencies(): + from fastapi import HTTPException + + from app.gateway.run_models import RunCreateRequest + from app.gateway.services import start_run + + with pytest.raises(HTTPException) as exc_info: + await start_run(RunCreateRequest(), "thread.with.dot", SimpleNamespace()) + + assert exc_info.value.status_code == 422 + assert "Invalid thread_id" in exc_info.value.detail diff --git a/backend/tests/test_provisioner_request_threading.py b/backend/tests/test_provisioner_request_threading.py index 4c5c2c8be..c0e74ea6f 100644 --- a/backend/tests/test_provisioner_request_threading.py +++ b/backend/tests/test_provisioner_request_threading.py @@ -15,6 +15,33 @@ from blockbuster import BlockBuster from kubernetes.client.rest import ApiException +def test_provisioner_thread_id_pattern_matches_gateway_contract(provisioner_module) -> None: + from deerflow.utils.thread_id import THREAD_ID_PATTERN + + assert provisioner_module.SAFE_THREAD_ID_PATTERN == THREAD_ID_PATTERN + + +@pytest.mark.parametrize("thread_id", ["", "thread.with.dot", "../escape", "x" * 65]) +def test_provisioner_rejects_noncanonical_thread_ids(provisioner_module, thread_id: str) -> None: + from pydantic import ValidationError + + with pytest.raises(ValidationError): + provisioner_module.CreateSandboxRequest( + sandbox_id="sandbox-validation", + thread_id=thread_id, + ) + + +@pytest.mark.parametrize("thread_id", ["a", "A1_b-2", "x" * 64]) +def test_provisioner_accepts_canonical_thread_ids(provisioner_module, thread_id: str) -> None: + request = provisioner_module.CreateSandboxRequest( + sandbox_id="sandbox-validation", + thread_id=thread_id, + ) + + assert request.thread_id == thread_id + + class _RecordingCoreV1: def __init__( self, diff --git a/backend/tests/test_run_event_store.py b/backend/tests/test_run_event_store.py index 41df9cac1..00b543cb4 100644 --- a/backend/tests/test_run_event_store.py +++ b/backend/tests/test_run_event_store.py @@ -725,6 +725,20 @@ class TestMakeRunEventStore: class TestJsonlRunEventStore: + @pytest.mark.anyio + @pytest.mark.parametrize("thread_id", ["", "thread.with.dot", "../escape", "x" * 65]) + async def test_rejects_noncanonical_thread_ids(self, tmp_path, thread_id): + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + store = JsonlRunEventStore(base_dir=tmp_path / "jsonl") + with pytest.raises(ValueError, match="Invalid thread_id"): + await store.put( + thread_id=thread_id, + run_id="r1", + event_type="human_message", + category="message", + ) + @pytest.mark.anyio async def test_basic_crud(self, tmp_path): from deerflow.runtime.events.store.jsonl import JsonlRunEventStore diff --git a/backend/tests/test_runs_api_endpoints.py b/backend/tests/test_runs_api_endpoints.py index 12ab7f1dd..6dacf6a28 100644 --- a/backend/tests/test_runs_api_endpoints.py +++ b/backend/tests/test_runs_api_endpoints.py @@ -324,6 +324,19 @@ def test_resolve_thread_id_handles_null_configurable(): assert runs._resolve_thread_id(RunCreateRequest(config={"configurable": {"thread_id": "t1"}})) == "t1" +@pytest.mark.parametrize( + "thread_id", + ["", "thread.with.dot", "../escape", "x" * 65, 123], +) +def test_run_request_rejects_invalid_configurable_thread_id(thread_id): + from pydantic import ValidationError + + from app.gateway.run_models import RunCreateRequest + + with pytest.raises(ValidationError): + RunCreateRequest(config={"configurable": {"thread_id": thread_id}}) + + def test_build_run_config_handles_null_configurable(): """A null ``configurable`` must also survive ``build_run_config``. diff --git a/backend/tests/test_scheduled_task_router_behavior.py b/backend/tests/test_scheduled_task_router_behavior.py index 70e516752..a2c055860 100644 --- a/backend/tests/test_scheduled_task_router_behavior.py +++ b/backend/tests/test_scheduled_task_router_behavior.py @@ -7,6 +7,31 @@ import pytest from app.gateway.routers import scheduled_tasks +@pytest.mark.parametrize( + "model", + [ + scheduled_tasks.ScheduledTaskCreateRequest, + scheduled_tasks.ScheduledTaskUpdateRequest, + ], +) +@pytest.mark.parametrize("thread_id", ["", "thread.with.dot", "../escape", "x" * 65]) +def test_scheduled_task_models_reject_invalid_thread_ids(model, thread_id): + from pydantic import ValidationError + + kwargs = {"thread_id": thread_id} + if model is scheduled_tasks.ScheduledTaskCreateRequest: + kwargs.update( + title="Task", + prompt="Prompt", + schedule_type="cron", + schedule_spec={"cron": "0 * * * *"}, + timezone="UTC", + ) + + with pytest.raises(ValidationError): + model(**kwargs) + + class _Repo: def __init__(self) -> None: self.created = [] diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index a071ad298..cf185dd36 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/backend/tests/test_scheduled_task_service.py @@ -278,6 +278,112 @@ async def test_manual_overlap_conflict_returns_conflict(): assert run_repo.updated[-1][1]["status"] == "failed" +@pytest.mark.asyncio +async def test_dispatch_task_records_failure_for_legacy_invalid_thread_id(): + """Rows persisted before the thread-id contract was centralized may store + IDs that fail the canonical pattern (dots, >64 chars). Dispatch must record + the failure through normal bookkeeping instead of raising — an uncaught + ValueError surfaces as HTTP 500 on manual trigger and, in the poller, + aborts the rest of the claimed batch every cycle.""" + + async def fake_launch(**_kwargs): + raise AssertionError("launch_run must not be called for an invalid thread_id") + + task_repo = DummyTaskRepo( + [ + { + "id": "task-legacy", + "user_id": "user-1", + "thread_id": "thread.with.dot", + "context_mode": "reuse_thread", + "assistant_id": "lead_agent", + "prompt": "Summarize thread", + "schedule_type": "cron", + "schedule_spec": {"cron": "0 9 * * *"}, + "timezone": "UTC", + "status": "enabled", + } + ] + ) + run_repo = DummyRunRepo() + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + result = await service.dispatch_task( + task_repo.rows[0], + now=datetime.now(UTC), + trigger="scheduled", + ) + + assert result["outcome"] == "failed" + assert result["task_run_id"] is None + assert result["run_id"] is None + assert "Invalid thread_id" in result["error"] + assert run_repo.created is None + assert task_repo.updated[1]["last_error"] == result["error"] + assert task_repo.updated[1]["last_thread_id"] == "thread.with.dot" + assert task_repo.updated[1]["increment_run_count"] is False + + +@pytest.mark.asyncio +async def test_run_once_continues_batch_after_invalid_thread_id(): + """A poison legacy row must not prevent later claimed tasks from dispatching.""" + launched = [] + + async def fake_launch(**kwargs): + launched.append(kwargs) + return {"run_id": "run-ok", "thread_id": kwargs["thread_id"]} + + task_repo = DummyTaskRepo( + [ + { + "id": "task-legacy", + "user_id": "user-1", + "thread_id": "thread.with.dot", + "context_mode": "reuse_thread", + "assistant_id": "lead_agent", + "prompt": "Summarize thread", + "schedule_type": "cron", + "schedule_spec": {"cron": "0 9 * * *"}, + "timezone": "UTC", + "status": "enabled", + }, + { + "id": "task-valid", + "user_id": "user-1", + "thread_id": "thread-ok", + "context_mode": "reuse_thread", + "assistant_id": "lead_agent", + "prompt": "Summarize thread", + "schedule_type": "cron", + "schedule_spec": {"cron": "0 9 * * *"}, + "timezone": "UTC", + "status": "enabled", + }, + ] + ) + run_repo = DummyRunRepo() + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + await service.run_once(now=datetime.now(UTC)) + + assert len(launched) == 1 + assert launched[0]["thread_id"] == "thread-ok" + + @pytest.mark.asyncio async def test_handle_run_completion_persists_success(): task_repo = DummyTaskRepo( diff --git a/backend/tests/test_support_bundle.py b/backend/tests/test_support_bundle.py index 1e08dd4e3..7bcc721ea 100644 --- a/backend/tests/test_support_bundle.py +++ b/backend/tests/test_support_bundle.py @@ -715,7 +715,17 @@ def test_validate_thread_id_rejects_dot_traversal(thread_id): def test_validate_thread_id_accepts_safe_ids(): support_bundle._validate_thread_id("thread-123") - support_bundle._validate_thread_id("a.b_c-1") + support_bundle._validate_thread_id("ab_c-1") + + +def test_validate_thread_id_rejects_noncanonical_ids(): + """The script's pattern is pinned byte-identical to the canonical + ``THREAD_ID_PATTERN`` (see test_thread_id_validation.py) — dotted or + over-length IDs are rejected even though they are not traversals.""" + with pytest.raises(ValueError, match="Invalid thread_id"): + support_bundle._validate_thread_id("a.b_c-1") + with pytest.raises(ValueError, match="Invalid thread_id"): + support_bundle._validate_thread_id("x" * 65) def test_main_reports_invalid_thread_id_without_traceback(tmp_path, capsys): diff --git a/backend/tests/test_thread_id_route_contract.py b/backend/tests/test_thread_id_route_contract.py new file mode 100644 index 000000000..adb91feb7 --- /dev/null +++ b/backend/tests/test_thread_id_route_contract.py @@ -0,0 +1,162 @@ +"""Canonical thread ID contract: every HTTP route and embedded-client entry +point that takes a ``thread_id`` must enforce ``ThreadId`` validation. + +Two complementary guards: + +1. A static sweep (AST over ``app/gateway/routers/*.py``) asserting every + route handler that declares a ``thread_id`` parameter annotates it + ``ThreadId`` — this is what prevents new routes from silently landing + with a raw ``str`` again (the suggestions/thread_runs/threads gaps). +2. A runtime sweep hitting every ``{thread_id}`` route with a non-canonical + ID and asserting a 422 whose error location names ``thread_id``. + +Deliberate exceptions (RFC #4588): +- ``DELETE /api/threads/{thread_id}`` keeps ``thread_id: str`` as the + legacy-cleanup escape hatch. +- The browser websocket stream validates on upgrade; covered separately. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path + +import pytest +from _router_auth_helpers import make_authed_test_app +from fastapi.testclient import TestClient + +ROUTERS_DIR = Path(__file__).resolve().parent.parent / "app" / "gateway" / "routers" + +# (handler name) route handlers deliberately allowed to keep ``thread_id: str``. +STATIC_WHITELIST = {"delete_thread_data"} + +# (method, path) routes deliberately excluded from the runtime 422 sweep. +RUNTIME_WHITELIST = { + ("DELETE", "/api/threads/{thread_id}"), # legacy-cleanup escape hatch +} + +BAD_THREAD_ID = "bad.thread.id" + +_ROUTE_DECORATOR_RE = re.compile(r"router\.(get|post|delete|put|patch|websocket)") + + +def _iter_route_handlers(path: Path): + """Yield (handler_name, has_thread_id_param, annotation) for route handlers.""" + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if not any(isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute) and isinstance(dec.func.value, ast.Name) and _ROUTE_DECORATOR_RE.fullmatch(f"{dec.func.value.id}.{dec.func.attr}") for dec in node.decorator_list): + continue + for arg in (*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs): + if arg.arg == "thread_id": + annotation = ast.unparse(arg.annotation) if arg.annotation else None + yield node.name, annotation + + +def test_every_thread_id_route_handler_uses_canonical_type(): + """Static guard: no route handler may declare a bare ``str`` thread_id.""" + violations = [] + for path in sorted(ROUTERS_DIR.glob("*.py")): + for handler, annotation in _iter_route_handlers(path): + if handler in STATIC_WHITELIST: + continue + if annotation != "ThreadId": + violations.append(f"{path.name}:{handler} -> {annotation!r}") + assert not violations, "route handlers with non-canonical thread_id:\n" + "\n".join(violations) + + +def _collect_thread_id_routes(): + """Import every gateway router and collect (method, full_path) with {thread_id}.""" + from app.gateway.routers import ( + artifacts, + browser, + feedback, + runs, + scheduled_tasks, + skills, + suggestions, + thread_runs, + threads, + uploads, + ) + + routers = [artifacts, browser, feedback, runs, scheduled_tasks, skills, suggestions, thread_runs, threads, uploads] + cases = [] + for module in routers: + for route in module.router.routes: + path = getattr(route, "path", "") + if "{thread_id}" not in path: + continue + methods = getattr(route, "methods", None) + if methods is None: + continue # websocket routes — covered by the dedicated test below + for method in sorted(methods): + if (method, path) in RUNTIME_WHITELIST: + continue + cases.append((module.__name__.rsplit(".", 1)[-1], method, path)) + return cases + + +_THREAD_ID_ROUTES = _collect_thread_id_routes() + + +def test_browser_websocket_rejects_noncanonical_thread_id(): + """The browser stream websocket validates thread_id on upgrade.""" + from starlette.websockets import WebSocketDisconnect + + from app.gateway.routers import browser + + app = make_authed_test_app() + app.include_router(browser.router) + + with TestClient(app, raise_server_exceptions=False) as client: + with pytest.raises(WebSocketDisconnect): + with client.websocket_connect(f"/api/threads/{BAD_THREAD_ID}/browser/stream"): + pass + + +def test_sweep_covers_expected_surface(): + """Sanity: the sweep must actually see the known thread_id routes.""" + assert len(_THREAD_ID_ROUTES) >= 30 + assert any("suggestions" in name for name, _, _ in _THREAD_ID_ROUTES) + + +@pytest.mark.parametrize( + ("router_name", "method", "path"), + _THREAD_ID_ROUTES, + ids=[f"{name}:{method}:{path}" for name, method, path in _THREAD_ID_ROUTES], +) +def test_noncanonical_thread_id_gets_422(router_name, method, path): + """Runtime guard: a non-canonical thread_id yields 422 naming thread_id.""" + import importlib + from unittest.mock import MagicMock + + from app.gateway.deps import get_config + + module = importlib.import_module(f"app.gateway.routers.{router_name}") + app = make_authed_test_app() + app.include_router(module.router) + # get_config 503s when no config.yaml exists (CI), and dependency solving + # precedes path-param validation — override it so the 422 contract is + # exercised regardless of the environment. + app.dependency_overrides[get_config] = MagicMock() + + url = path.replace("{thread_id}", BAD_THREAD_ID) + # Other path params get a harmless canonical placeholder. + url = re.sub(r"\{(\w+)(?::path)?\}", "x", url) + + with TestClient(app, raise_server_exceptions=False) as client: + if method == "GET": + response = client.get(url) + elif method == "DELETE": + response = client.delete(url) + elif method in {"POST", "PUT", "PATCH"}: + response = client.request(method, url, json={}) + else: # WEBSOCKET etc. — not expected in the sweep + pytest.skip(f"unsupported method {method}") + + assert response.status_code == 422, f"{method} {path} -> {response.status_code}: {response.text[:300]}" + detail = response.json()["detail"] + assert any("thread_id" in str(err.get("loc", ())) for err in detail), f"422 did not name thread_id: {detail}" diff --git a/backend/tests/test_thread_id_validation.py b/backend/tests/test_thread_id_validation.py new file mode 100644 index 000000000..cb987360a --- /dev/null +++ b/backend/tests/test_thread_id_validation.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import pytest + + +@pytest.mark.parametrize( + "thread_id", + [ + "a", + "A1_b-2", + "x" * 64, + ], +) +def test_validate_thread_id_accepts_canonical_ids(thread_id: str) -> None: + from deerflow.utils.thread_id import validate_thread_id + + assert validate_thread_id(thread_id) == thread_id + + +@pytest.mark.parametrize( + "thread_id", + [ + "", + "x" * 65, + "thread.with.dot", + "../escape", + "has space", + "line\nbreak", + "线程", + ], +) +def test_validate_thread_id_rejects_noncanonical_ids(thread_id: str) -> None: + from deerflow.utils.thread_id import validate_thread_id + + with pytest.raises(ValueError, match="Invalid thread_id"): + validate_thread_id(thread_id) + + +@pytest.mark.parametrize("thread_id", [1, {}, []]) +def test_validate_thread_id_rejects_non_strings(thread_id: object) -> None: + from deerflow.utils.thread_id import validate_thread_id + + with pytest.raises(ValueError, match="Invalid thread_id"): + validate_thread_id(thread_id) # type: ignore[arg-type] + + +@pytest.mark.parametrize("method_name", ["upload_files", "list_uploads", "delete_upload", "get_artifact"]) +def test_client_mutating_and_read_entry_points_validate_thread_id(method_name: str) -> None: + """RFC #4588: the embedded client validates on all mutating entry points. + + Validation is the first statement of each method, so a bare ``__new__`` + instance is enough — no config, sandbox, or event loop is touched. + """ + from deerflow.client import DeerFlowClient + + client = DeerFlowClient.__new__(DeerFlowClient) + method = getattr(client, method_name) + args = {"upload_files": (["x"],), "list_uploads": (), "delete_upload": ("f.txt",), "get_artifact": ("mnt/user-data/outputs/f.txt",)}[method_name] + + with pytest.raises(ValueError, match="Invalid thread_id"): + method("bad.thread.id", *args) + + +def test_support_bundle_thread_id_pattern_matches_canonical() -> None: + """scripts/support_bundle.py keeps its own copy of the pattern (it must + run with a broken venv) — pin it byte-identical to the canonical one.""" + import importlib.util + from pathlib import Path + + from deerflow.utils.thread_id import THREAD_ID_PATTERN + + script = Path(__file__).resolve().parents[2] / "scripts" / "support_bundle.py" + spec = importlib.util.spec_from_file_location("support_bundle", script) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + assert module.SAFE_THREAD_ID_RE.pattern == THREAD_ID_PATTERN + + +def test_tui_resume_literal_ref_validated() -> None: + """The TUI /resume fallback adopts a literal ref as thread id only when + it satisfies the canonical contract.""" + from deerflow.tui.session import Session + + session = Session.__new__(Session) + session.client = type("StubClient", (), {"list_threads": lambda self, limit: {"thread_list": []}})() + + assert session.resolve_ref("valid-id_1") == "valid-id_1" + with pytest.raises(ValueError, match="not a valid thread id"): + session.resolve_ref("bad.thread.id") diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index 9c2269925..dd07023ef 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -410,18 +410,66 @@ def test_delete_thread_route_rejects_invalid_thread_id(tmp_path): assert response.status_code == 404 -def test_delete_thread_route_returns_422_for_route_safe_invalid_id(tmp_path): - paths = Paths(tmp_path) +def test_delete_thread_route_cleans_legacy_metadata_without_resolving_unsafe_path(): + app, store, _checkpointer = _build_thread_app() + legacy_thread_id = "legacy.thread" - app = make_authed_test_app() - app.include_router(threads.router) + asyncio.run( + store.aput( + THREADS_NS, + legacy_thread_id, + { + "thread_id": legacy_thread_id, + "status": "idle", + "created_at": "", + "updated_at": "", + "metadata": {}, + }, + ) + ) - with patch("app.gateway.routers.threads.get_paths", return_value=paths): - with TestClient(app) as client: - response = client.delete("/api/threads/thread.with.dot") + with ( + patch( + "app.gateway.routers.threads._delete_thread_data", + side_effect=AssertionError("legacy thread ID must not reach filesystem cleanup"), + ), + TestClient(app) as client, + ): + response = client.delete(f"/api/threads/{legacy_thread_id}") + + assert response.status_code == 200 + assert "Skipped local data cleanup" in response.json()["message"] + assert asyncio.run(store.aget(THREADS_NS, legacy_thread_id)) is None + + +def test_legacy_thread_metadata_mutation_is_rejected(): + app, store, _checkpointer = _build_thread_app() + legacy_thread_id = "legacy.thread" + + asyncio.run( + store.aput( + THREADS_NS, + legacy_thread_id, + { + "thread_id": legacy_thread_id, + "status": "idle", + "created_at": "", + "updated_at": "", + "metadata": {"original": True}, + }, + ) + ) + + with TestClient(app) as client: + response = client.patch( + f"/api/threads/{legacy_thread_id}", + json={"metadata": {"mutated": True}}, + ) assert response.status_code == 422 - assert "Invalid thread_id" in response.json()["detail"] + record = asyncio.run(store.aget(THREADS_NS, legacy_thread_id)) + assert record is not None + assert record.value["metadata"] == {"original": True} def test_delete_thread_data_returns_generic_500_error(tmp_path): @@ -490,6 +538,31 @@ def test_create_thread_returns_iso_timestamps() -> None: assert body["created_at"] == body["updated_at"] +@pytest.mark.parametrize( + "thread_id", + ["", "thread.with.dot", "../escape", "x" * 65], +) +def test_create_thread_rejects_invalid_explicit_thread_id_before_persistence(thread_id: str) -> None: + app, store, checkpointer = _build_thread_app() + + with TestClient(app) as client: + response = client.post("/api/threads", json={"thread_id": thread_id}) + + assert response.status_code == 422 + assert asyncio.run(store.asearch(THREADS_NS)) == [] + assert not checkpointer.storage + + +def test_create_thread_preserves_valid_explicit_thread_id() -> None: + app, _store, _checkpointer = _build_thread_app() + + with TestClient(app) as client: + response = client.post("/api/threads", json={"thread_id": "caller_thread-1"}) + + assert response.status_code == 200 + assert response.json()["thread_id"] == "caller_thread-1" + + def test_create_thread_returns_existing_when_insert_loses_race() -> None: """A concurrent create that loses the INSERT race stays idempotent. diff --git a/docker/provisioner/app.py b/docker/provisioner/app.py index 0f81aa32e..1a6c81367 100644 --- a/docker/provisioner/app.py +++ b/docker/provisioner/app.py @@ -100,7 +100,7 @@ if not (1 <= SANDBOX_CONTAINER_PORT <= 65535): raise RuntimeError(f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT}; expected a value in [1, 65535]") if SANDBOX_SERVICE_TYPE not in {"NodePort", "ClusterIP"}: raise RuntimeError(f"Invalid SANDBOX_SERVICE_TYPE={SANDBOX_SERVICE_TYPE!r}; expected 'NodePort' or 'ClusterIP'") -SAFE_THREAD_ID_PATTERN = r"^[A-Za-z0-9_\-]+$" +SAFE_THREAD_ID_PATTERN = r"^[A-Za-z0-9_-]{1,64}$" SAFE_USER_ID_PATTERN = r"^[A-Za-z0-9_\-]+$" DEFAULT_USER_ID = "default" MAX_EXTRA_MOUNTS = 9 diff --git a/scripts/support_bundle.py b/scripts/support_bundle.py index fbc29954e..d73624759 100644 --- a/scripts/support_bundle.py +++ b/scripts/support_bundle.py @@ -63,7 +63,11 @@ SECRET_FLAG_RE = re.compile(r"(?i)^--?[\w.-]*(?:api[_-]?key|token|secret|passwor HEADER_KEY_RE = re.compile(r"(?i)header") POSIX_HOME_RE = re.compile(r"(?