From f17e0504d0b99d22ddedbde1726a8e4cd4456996 Mon Sep 17 00:00:00 2001 From: kongxin Date: Mon, 9 Nov 2026 15:47:01 -0800 Subject: [PATCH] fix(feedback): return 409 instead of 500 when feedback is POSTed twice (#8064) * fix(feedback): return 409 instead of 500 when feedback is POSTed twice POST /api/threads/{id}/runs/{rid}/feedback did a plain INSERT. A second POST by the same user for the same run (a retry, or feedback on another message of that run) hit uq_feedback_thread_run_user and the uncaught IntegrityError surfaced as HTTP 500. FeedbackRepository.create now rolls back and raises FeedbackAlreadyExistsError on that collision, and the route maps it to 409 pointing callers at PUT, which already upserts. * fix(feedback): only map the run/user unique key to 409; cover concurrent POSTs Address review on #8064: - create() now re-reads the (thread, run, user) row after an IntegrityError and raises FeedbackAlreadyExistsError only when it exists; any other violation (e.g. a feedback_id collision) propagates. - Add a commit-barrier test where two first POSTs race: one 200, one 409, and the winner's row is kept. - Add CHANGELOG.md / CHANGELOG_zh.md Fixed entries. --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 9 ++ CHANGELOG_zh.md | 6 ++ backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/feedback.py | 23 +++-- .../deerflow/persistence/feedback/__init__.py | 4 +- .../deerflow/persistence/feedback/sql.py | 32 ++++++- backend/tests/test_feedback.py | 37 +++++++- backend/tests/test_feedback_router.py | 85 +++++++++++++++++++ 8 files changed, 184 insertions(+), 14 deletions(-) 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"])]