mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-04 20:08:40 +00:00
* fix(gateway): unify thread ID validation at the API boundary
Thread ID entry points accepted arbitrary strings while downstream
consumers (filesystem paths, Kubernetes Provisioner, JSONL event store)
each enforced different character restrictions, so invalid IDs were
persisted first and only failed later during sandbox/workspace init.
Centralize validation in deerflow.utils.thread_id (pattern
^[A-Za-z0-9_-]{1,64}$): validate at routers, RunCreateRequest,
scheduler dispatch, paths.py, JSONL store, embedded client, and align
the Provisioner pattern (pinned by a parity test). UUIDs are still
generated only when no ID is supplied; caller-supplied opaque IDs stay
supported.
Deliberate exceptions: DELETE /threads/{id} keeps str as the legacy
cleanup escape hatch (filesystem cleanup guarded), read-only
client.get_thread stays unvalidated, and scheduler rows with legacy
invalid IDs record a failed dispatch instead of raising out of the
poll loop.
* docs: document canonical thread ID contract
README: caller-supplied thread IDs need not be UUIDs; the canonical
pattern and per-endpoint behavior. AGENTS.md: the shared
deerflow.utils.thread_id contract, its enforcement boundaries, and the
legacy-ID escape hatches.
* fix(gateway): close thread ID validation gaps at remaining entry points
Follow-up to the canonical thread ID contract: a full audit found the
uniform-422 coverage only reached about half of the thread_id surfaces.
- routers: 18 routes still took a bare thread_id: str — 13 in
thread_runs.py (including the five messages/events/workspace-changes
reads that returned 500 on the JSONL event store vs 404/empty on the
DB store), 4 read routes in threads.py, and the suggestions route
flagged in review. DELETE /api/threads/{id} keeps str as the declared
legacy-cleanup escape hatch.
- client: upload_files/delete_upload/list_uploads/get_artifact now
validate up front, fulfilling the RFC's 'all mutating entry points'
clause (get_thread stays unvalidated as the declared legacy read path).
- tui: the /resume literal-ref fallback validates against the canonical
contract and reports a descriptive error instead of failing deep in
the client.
- scripts/support_bundle.py: replace the drifted dot-allowing pattern
with a byte-identical copy of THREAD_ID_PATTERN (kept local so the
script still runs with a broken venv).
* test(gateway): guard the canonical thread ID contract against regressions
- test_thread_id_route_contract.py: static AST sweep asserting every
route handler with a thread_id parameter annotates ThreadId
(whitelist: the DELETE escape hatch), plus a runtime sweep hitting
all 44 thread_id routes with a non-canonical ID and asserting a 422
that names thread_id, plus a websocket upgrade-rejection case.
- test_thread_id_validation.py: client entry-point validation,
support_bundle pattern parity, and TUI literal-ref fallback tests.
- Align two tests that encoded the old contract (dotted IDs).
190 lines
6.5 KiB
Python
190 lines
6.5 KiB
Python
"""Feedback endpoints — create, list, stats, delete.
|
|
|
|
Allows users to submit thumbs-up/down feedback on runs,
|
|
optionally scoped to a specific message.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
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"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Request / response models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class FeedbackCreateRequest(BaseModel):
|
|
rating: int = Field(..., description="Feedback rating: +1 (positive) or -1 (negative)")
|
|
comment: str | None = Field(default=None, description="Optional text feedback")
|
|
message_id: str | None = Field(default=None, description="Optional: scope feedback to a specific message")
|
|
|
|
|
|
class FeedbackUpsertRequest(BaseModel):
|
|
rating: int = Field(..., description="Feedback rating: +1 (positive) or -1 (negative)")
|
|
comment: str | None = Field(default=None, description="Optional text feedback")
|
|
|
|
|
|
class FeedbackResponse(BaseModel):
|
|
feedback_id: str
|
|
run_id: str
|
|
thread_id: str
|
|
user_id: str | None = None
|
|
message_id: str | None = None
|
|
rating: int
|
|
comment: str | None = None
|
|
created_at: str = ""
|
|
|
|
|
|
class FeedbackStatsResponse(BaseModel):
|
|
run_id: str
|
|
total: int = 0
|
|
positive: int = 0
|
|
negative: int = 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@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: ThreadId,
|
|
run_id: str,
|
|
body: FeedbackUpsertRequest,
|
|
request: Request,
|
|
) -> dict[str, Any]:
|
|
"""Create or update feedback for a run (idempotent)."""
|
|
if body.rating not in (1, -1):
|
|
raise HTTPException(status_code=400, detail="rating must be +1 or -1")
|
|
|
|
user_id = await get_current_user(request)
|
|
|
|
run_store = get_run_store(request)
|
|
run = await run_store.get(run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
if run.get("thread_id") != thread_id:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found in thread {thread_id}")
|
|
|
|
feedback_repo = get_feedback_repo(request)
|
|
return await feedback_repo.upsert(
|
|
run_id=run_id,
|
|
thread_id=thread_id,
|
|
rating=body.rating,
|
|
user_id=user_id,
|
|
comment=body.comment,
|
|
)
|
|
|
|
|
|
@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: ThreadId,
|
|
run_id: str,
|
|
request: Request,
|
|
) -> dict[str, bool]:
|
|
"""Delete the current user's feedback for a run."""
|
|
user_id = await get_current_user(request)
|
|
feedback_repo = get_feedback_repo(request)
|
|
deleted = await feedback_repo.delete_by_run(
|
|
thread_id=thread_id,
|
|
run_id=run_id,
|
|
user_id=user_id,
|
|
)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail="No feedback found for this run")
|
|
return {"success": True}
|
|
|
|
|
|
@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: ThreadId,
|
|
run_id: str,
|
|
body: FeedbackCreateRequest,
|
|
request: Request,
|
|
) -> dict[str, Any]:
|
|
"""Submit feedback (thumbs-up/down) for a run."""
|
|
if body.rating not in (1, -1):
|
|
raise HTTPException(status_code=400, detail="rating must be +1 or -1")
|
|
|
|
user_id = await get_current_user(request)
|
|
|
|
# Validate run exists and belongs to thread
|
|
run_store = get_run_store(request)
|
|
run = await run_store.get(run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
if run.get("thread_id") != thread_id:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found in thread {thread_id}")
|
|
|
|
feedback_repo = get_feedback_repo(request)
|
|
return await feedback_repo.create(
|
|
run_id=run_id,
|
|
thread_id=thread_id,
|
|
rating=body.rating,
|
|
user_id=user_id,
|
|
message_id=body.message_id,
|
|
comment=body.comment,
|
|
)
|
|
|
|
|
|
@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: ThreadId,
|
|
run_id: str,
|
|
request: Request,
|
|
) -> list[dict[str, Any]]:
|
|
"""List all feedback for a run."""
|
|
feedback_repo = get_feedback_repo(request)
|
|
return await feedback_repo.list_by_run(thread_id, run_id)
|
|
|
|
|
|
@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: ThreadId,
|
|
run_id: str,
|
|
request: Request,
|
|
) -> dict[str, Any]:
|
|
"""Get aggregated feedback stats (positive/negative counts) for a run."""
|
|
feedback_repo = get_feedback_repo(request)
|
|
return await feedback_repo.aggregate_by_run(thread_id, run_id)
|
|
|
|
|
|
@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: ThreadId,
|
|
run_id: str,
|
|
feedback_id: str,
|
|
request: Request,
|
|
) -> dict[str, bool]:
|
|
"""Delete a feedback record."""
|
|
feedback_repo = get_feedback_repo(request)
|
|
# Verify feedback belongs to the specified thread/run before deleting
|
|
existing = await feedback_repo.get(feedback_id)
|
|
if existing is None:
|
|
raise HTTPException(status_code=404, detail=f"Feedback {feedback_id} not found")
|
|
if existing.get("thread_id") != thread_id or existing.get("run_id") != run_id:
|
|
raise HTTPException(status_code=404, detail=f"Feedback {feedback_id} not found in run {run_id}")
|
|
deleted = await feedback_repo.delete(feedback_id)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail=f"Feedback {feedback_id} not found")
|
|
return {"success": True}
|