diff --git a/CHANGELOG.md b/CHANGELOG.md index a60d8a1bf6..62b6dbaf1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,13 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **gateway:** Posting feedback twice for the same run now returns a 409 + instead of a 500. `POST /api/threads/{id}/runs/{run_id}/feedback` allows one + feedback row per user and run; a retry, or feedback on another message of the + same run, hit that unique key and the Gateway failed with an unhandled + database error. The 409 points callers at `PUT` on the same route, which + updates the existing feedback. ([#8064]) + - **sandbox:** AIO commands from independent subagents now run concurrently when they carry per-call environment variables, including Lark CLI commands. Commands within one subagent remain ordered and client cleanup drains them. @@ -16224,3 +16231,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#8006]: https://github.com/bytedance/deer-flow/pull/8006 [#8043]: https://github.com/bytedance/deer-flow/pull/8043 [#8057]: https://github.com/bytedance/deer-flow/pull/8057 +[#8064]: https://github.com/bytedance/deer-flow/pull/8064 + diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index f365016a81..b6c0a688da 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -26,6 +26,11 @@ ### 修复 +- **网关：** 同一用户对同一次运行重复提交反馈时，现在返回 409 而不是 500。 + `POST /api/threads/{id}/runs/{run_id}/feedback` 每个用户每次运行只允许一条反馈；此前重试， + 或对同一次运行的另一条消息提交反馈，会触发该唯一约束，Gateway 因未处理的数据库错误而失败。 + 409 会提示调用方改用同一路由的 `PUT` 更新已有反馈。([#8064]) + - **网关：** 删除线程时不再在对话仍可读取的情况下返回成功。此前 `DELETE /api/threads/{id}` 在检查点存储失败时只记录 DEBUG 日志，仍删除线程元数据行并 返回成功；之后任何已登录用户都能通过 `GET /threads/{id}`、`/state`、`/history` 和 @@ -12848,3 +12853,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7993]: https://github.com/bytedance/deer-flow/pull/7993 [#8006]: https://github.com/bytedance/deer-flow/pull/8006 [#8043]: https://github.com/bytedance/deer-flow/pull/8043 +[#8064]: https://github.com/bytedance/deer-flow/pull/8064 diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index e5e02f76ea..624077b6e3 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -149,7 +149,7 @@ owner-scoped assistant version selection remains enabled. | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | | **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, treating a protocol-valid hidden `human_input_response` as confirmed replay input while rejecting other hidden/control humans, preserving that response's hidden/correlation metadata, and carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - newest 100 runs as an array; `GET /page` - keyset history page `{data, has_more, next_before_created_at, next_before_run_id}`; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. | -| **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | +| **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback (409 if the user already rated this run; use `PUT` to change it); `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | | **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback | | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | | **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. | diff --git a/backend/app/gateway/routers/feedback.py b/backend/app/gateway/routers/feedback.py index c86a9c32c9..b6d6e241c6 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.persistence.feedback import FeedbackAlreadyExistsError from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -133,14 +134,20 @@ async def create_feedback( 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, - ) + try: + 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, + ) + except FeedbackAlreadyExistsError: + raise HTTPException( + status_code=409, + detail="Feedback already exists for this run; use PUT to update it", + ) from None @router.get("/{thread_id}/runs/{run_id}/feedback", response_model=list[FeedbackResponse]) diff --git a/backend/packages/harness/deerflow/persistence/feedback/__init__.py b/backend/packages/harness/deerflow/persistence/feedback/__init__.py index ee958b0275..ef880a5edf 100644 --- a/backend/packages/harness/deerflow/persistence/feedback/__init__.py +++ b/backend/packages/harness/deerflow/persistence/feedback/__init__.py @@ -1,6 +1,6 @@ """Feedback persistence — ORM and SQL repository.""" from deerflow.persistence.feedback.model import FeedbackRow -from deerflow.persistence.feedback.sql import FeedbackRepository +from deerflow.persistence.feedback.sql import FeedbackAlreadyExistsError, FeedbackRepository -__all__ = ["FeedbackRepository", "FeedbackRow"] +__all__ = ["FeedbackAlreadyExistsError", "FeedbackRepository", "FeedbackRow"] diff --git a/backend/packages/harness/deerflow/persistence/feedback/sql.py b/backend/packages/harness/deerflow/persistence/feedback/sql.py index dec5514c6a..2264f1f1d6 100644 --- a/backend/packages/harness/deerflow/persistence/feedback/sql.py +++ b/backend/packages/harness/deerflow/persistence/feedback/sql.py @@ -19,6 +19,10 @@ from deerflow.utils.time import coerce_iso _UPSERT_MAX_ATTEMPTS = 3 +class FeedbackAlreadyExistsError(Exception): + """The user already has feedback for this run (``uq_feedback_thread_run_user``).""" + + class FeedbackRepository: def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: self._sf = session_factory @@ -42,7 +46,11 @@ class FeedbackRepository: message_id: str | None = None, comment: str | None = None, ) -> dict: - """Create a feedback record. rating must be +1 or -1.""" + """Create a feedback record. rating must be +1 or -1. + + Raises ``FeedbackAlreadyExistsError`` when the user already has + feedback for this run; use ``upsert`` to replace it. + """ if rating not in (1, -1): raise ValueError(f"rating must be +1 or -1, got {rating}") resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.create") @@ -58,7 +66,27 @@ class FeedbackRepository: ) async with self._sf() as session: session.add(row) - await session.commit() + try: + await session.commit() + except IntegrityError as exc: + await session.rollback() + # Only a committed row for this (thread, run, user) key means + # "already exists"; any other violation propagates unchanged. + # A NULL user_id never conflicts on that key. + if resolved_user_id is None: + raise + existing = await session.scalar( + select(FeedbackRow.feedback_id) + .where( + FeedbackRow.thread_id == thread_id, + FeedbackRow.run_id == run_id, + FeedbackRow.user_id == resolved_user_id, + ) + .limit(1) + ) + if existing is None: + raise + raise FeedbackAlreadyExistsError(f"Feedback already exists for run {run_id}") from exc await session.refresh(row) return self._row_to_dict(row) diff --git a/backend/tests/test_feedback.py b/backend/tests/test_feedback.py index 9d4ef4f802..af3a9d7b97 100644 --- a/backend/tests/test_feedback.py +++ b/backend/tests/test_feedback.py @@ -7,7 +7,7 @@ import pytest from _repository_race_helpers import miss_select_factory from sqlalchemy.exc import IntegrityError -from deerflow.persistence.feedback import FeedbackRepository +from deerflow.persistence.feedback import FeedbackAlreadyExistsError, FeedbackRepository from deerflow.persistence.feedback.sql import _UPSERT_MAX_ATTEMPTS @@ -67,6 +67,41 @@ class TestFeedbackRepository: assert record["user_id"] == "user-1" await _cleanup() + @pytest.mark.anyio + async def test_create_twice_for_same_run_and_user_raises_already_exists(self, tmp_path): + """A second create for one (thread, run, user) raises a domain error, not a raw IntegrityError.""" + repo = await _make_feedback_repo(tmp_path) + try: + first = await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="u1", message_id="msg-a") + with pytest.raises(FeedbackAlreadyExistsError): + await repo.create(run_id="r1", thread_id="t1", rating=-1, user_id="u1", message_id="msg-b") + + rows = await repo.list_by_run("t1", "r1", user_id="u1") + assert [(row["feedback_id"], row["rating"]) for row in rows] == [(first["feedback_id"], 1)] + # The session was rolled back, so the repository keeps working. + other = await repo.create(run_id="r1", thread_id="t1", rating=-1, user_id="u2") + assert other["user_id"] == "u2" + finally: + await _cleanup() + + @pytest.mark.anyio + async def test_create_reraises_integrity_errors_other_than_the_run_user_key(self, tmp_path, monkeypatch): + """A violation that is not the (thread, run, user) key, e.g. a feedback_id collision, is not reported as 'already exists'.""" + import uuid + from types import SimpleNamespace + + from deerflow.persistence.feedback import sql as feedback_sql + + fixed_id = uuid.UUID("00000000-0000-4000-8000-000000000001") + monkeypatch.setattr(feedback_sql, "uuid", SimpleNamespace(uuid4=lambda: fixed_id)) + repo = await _make_feedback_repo(tmp_path) + try: + await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="u1") + with pytest.raises(IntegrityError): + await repo.create(run_id="r2", thread_id="t1", rating=1, user_id="u2") + finally: + await _cleanup() + @pytest.mark.anyio async def test_create_invalid_rating_zero(self, tmp_path): repo = await _make_feedback_repo(tmp_path) diff --git a/backend/tests/test_feedback_router.py b/backend/tests/test_feedback_router.py index 4288d9080d..990766d74e 100644 --- a/backend/tests/test_feedback_router.py +++ b/backend/tests/test_feedback_router.py @@ -112,3 +112,88 @@ async def test_concurrent_first_feedback_puts_both_succeed_with_one_row(feedback assert len(rows) == 1 # The later commit wins, as with sequential submissions. assert rows[0]["rating"] in (1, -1) + + +class _CommitBarrierSession: + """Hold each session at commit until every writer has staged its INSERT. + + Both POSTs then commit at once and the database's unique key decides the + winner, which is the interleaving two concurrent first submissions produce. + """ + + def __init__(self, session, barrier: asyncio.Barrier): + self._session = session + self._barrier = barrier + self._waited = False + + def __getattr__(self, name): + return getattr(self._session, name) + + async def commit(self): + if not self._waited: + self._waited = True + async with asyncio.timeout(5): + await self._barrier.wait() + return await self._session.commit() + + async def __aenter__(self): + await self._session.__aenter__() + return self + + async def __aexit__(self, *args): + return await self._session.__aexit__(*args) + + +def _feedback_post_app(feedback_repo): + app = make_authed_test_app(user_factory=_user) + + @app.middleware("http") + async def _session_auth_source(request, call_next): + request.state.auth_source = AUTH_SOURCE_SESSION + return await call_next(request) + + run_store = MagicMock() + run_store.get = AsyncMock(return_value={"run_id": RUN_ID, "thread_id": THREAD_ID}) + app.state.run_store = run_store + app.state.feedback_repo = feedback_repo + app.include_router(feedback.router) + return app + + +def _client(app) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test") + + +FEEDBACK_URL = f"/api/threads/{THREAD_ID}/runs/{RUN_ID}/feedback" + + +@pytest.mark.anyio +async def test_second_feedback_post_for_same_run_returns_409_not_500(feedback_repo): + async with _client(_feedback_post_app(feedback_repo)) as client: + first = await client.post(FEEDBACK_URL, json={"rating": 1, "message_id": "msg-a"}) + second = await client.post(FEEDBACK_URL, json={"rating": -1, "message_id": "msg-b"}) + + assert first.status_code == 200, first.text + assert second.status_code == 409, second.text + assert "PUT" in second.json()["detail"] + rows = await feedback_repo.list_by_run(THREAD_ID, RUN_ID, user_id=str(_user().id)) + assert [(row["feedback_id"], row["rating"]) for row in rows] == [(first.json()["feedback_id"], 1)] + + +@pytest.mark.anyio +async def test_concurrent_first_feedback_posts_return_one_200_and_one_409(feedback_repo): + barrier = asyncio.Barrier(2) + real_factory = feedback_repo._sf + feedback_repo._sf = lambda: _CommitBarrierSession(real_factory(), barrier) + + async with _client(_feedback_post_app(feedback_repo)) as client: + up, down = await asyncio.gather( + client.post(FEEDBACK_URL, json={"rating": 1}), + client.post(FEEDBACK_URL, json={"rating": -1}), + ) + + assert sorted([up.status_code, down.status_code]) == [200, 409], (up.text, down.text) + winner = up if up.status_code == 200 else down + feedback_repo._sf = real_factory + rows = await feedback_repo.list_by_run(THREAD_ID, RUN_ID, user_id=str(_user().id)) + assert [(row["feedback_id"], row["rating"]) for row in rows] == [(winner.json()["feedback_id"], winner.json()["rating"])]