fix(gateway): return 404 on cross-owner thread_id instead of 500 (#8057)

* fix(gateway): return 404 on cross-owner thread_id instead of 500

When a caller-supplied thread_id already belongs to another user, the SQL-backed store rejects the duplicate primary key with IntegrityError. The route's recovery path resolved the surviving row through the requesting owner's scope, so a foreign row could not be read back and the route surfaced a 500 instead of the documented 404 ownership-conflict response ("Thread not found").

Read the row back bypassing the owner scope before collapsing to a 500 so the cross-owner case mirrors ThreadOwnershipConflictError.

Fix #8052

Signed-off-by: inchang-ing <197932532+inchang-ing@users.noreply.github.com>

* test(gateway): pin thread-ownership 404 and real-failure 500

Address review feedback on the cross-owner 404 recovery:
- Update test_insert_race_concurrent_trusted_claims_have_one_winner: the
  losing request now resolves the surviving unscoped row as a 404 ownership
  conflict instead of a 500, while the single-winner assertions and the
  non-disclosing detail are preserved.
- Add test_create_thread_keeps_real_integrity_failure_as_500 to pin the
  fall-through branch where a duplicate-key IntegrityError coincides with no
  surviving row, which must still surface as a 500 rather than collapse into
  the ownership-conflict 404.
- Document the user-facing error-surface change in the Unreleased Fixed
  section.

Signed-off-by: inchang-ing <197932532+inchang-ing@users.noreply.github.com>

---------

Signed-off-by: inchang-ing <197932532+inchang-ing@users.noreply.github.com>
Co-authored-by: inchang-ing <197932532+inchang-ing@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -60,6 +60,13 @@ This section accumulates work toward the **2.3.0** milestone
   returns a 500 and keeps the thread, so the owner can delete it again.
   ([#7992])
 
+- **gateway:** Creating a thread with a caller-supplied `thread_id` that
+  already belongs to another user now returns a non-disclosing 404
+  (`Thread not found`) instead of a 500. The duplicate-key recovery path reads
+  the surviving row back bypassing the owner scope so the cross-owner case
+  mirrors `ThreadOwnershipConflictError`; a duplicate key with no surviving row
+  is still surfaced as a 500. ([#8057])
+
 - **middleware:** With PII redaction enabled, a long run of text without
   spaces (such as a pasted hex dump) no longer stalls the Gateway. The email
   detector was quadratic in the length of such a run, and redaction re-scans
@@ -16211,3 +16218,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7998]: https://github.com/bytedance/deer-flow/issues/7998
 [#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

backend/app/gateway/routers/threads.py

@@ -1043,7 +1043,15 @@ async def _create_thread_locked(body: ThreadCreateRequest, request: Request, thr
         existing_record = await _resolve_existing_thread(thread_store, thread_id, thread_owner_user_id, thread_owner_kwargs)
         if existing_record is not None:
             return _existing_thread_response(thread_id, existing_record)
-        # A duplicate-key error with no row we can read back is a real failure.
+        # The same-owner reconciliation above found nothing, yet the insert hit
+        # a duplicate primary key — so the surviving row belongs to *another*
+        # user. That is the ownership-conflict case, which the route already
+        # maps ``ThreadOwnershipConflictError`` to: a non-disclosing 404. Read
+        # the row back bypassing the owner scope to confirm a foreign row
+        # exists before collapsing to a 500.
+        if await thread_store.get(thread_id, user_id=None) is not None:
+            raise HTTPException(status_code=404, detail="Thread not found") from None
+        # A duplicate-key error with no surviving row is a real failure.
         logger.exception("Failed to write thread_meta for %s", sanitize_log_param(thread_id))
         raise HTTPException(status_code=500, detail="Failed to create thread")
     except Exception:

backend/tests/test_thread_create_cross_owner_conflict.py

@@ -0,0 +1,87 @@
+"""Cross-owner thread_id creation returns 404, never 500 (SQL backend).
+
+Regression for issue #8052: when a caller-supplied ``thread_id`` already
+belongs to another user, the SQL-backed store rejects the duplicate primary
+key with ``IntegrityError`` (the memory store raises the app-level
+``ThreadOwnershipConflictError`` instead). The route's recovery path resolved
+the surviving row through the *requesting* owner's scope, so a foreign row
+could not be read back and the route surfaced a 500 instead of the documented
+404 ownership-conflict response ("Thread not found").
+
+The memory store does not have this bug, so the regression is exercised on the
+sqlite backend only.
+"""
+
+from uuid import UUID
+
+import httpx
+import pytest
+from _router_auth_helpers import make_authed_test_app
+from langgraph.checkpoint.memory import InMemorySaver
+
+from app.gateway.auth.models import User
+from app.gateway.routers import threads
+from deerflow.persistence.thread_meta import ThreadMetaRepository
+
+USER_A = User(
+    id=UUID("00000000-0000-0000-0000-0000000000a1"),
+    email="owner-a@example.com",
+    password_hash="x",
+    system_role="admin",
+)
+USER_B = User(
+    id=UUID("00000000-0000-0000-0000-0000000000b2"),
+    email="owner-b@example.com",
+    password_hash="x",
+    system_role="admin",
+)
+
+
+@pytest.fixture
+async def sql_thread_store(tmp_path):
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+    await init_engine(
+        "sqlite",
+        url=f"sqlite+aiosqlite:///{tmp_path / 'threads.db'}",
+        sqlite_dir=str(tmp_path),
+    )
+    try:
+        yield ThreadMetaRepository(get_session_factory())
+    finally:
+        await close_engine()
+
+
+def app_for(store, checkpointer, user):
+    app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True)
+    app.state.thread_store = store
+    app.state.checkpointer = checkpointer
+    app.include_router(threads.router)
+    return app
+
+
+@pytest.mark.anyio
+async def test_cross_owner_thread_id_returns_404(sql_thread_store):
+    saver = InMemorySaver()
+    app_a = app_for(sql_thread_store, saver, USER_A)
+    app_b = app_for(sql_thread_store, saver, USER_B)
+
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app_a), base_url="http://test") as client_a:
+        created = await client_a.post(
+            "/api/threads",
+            json={"thread_id": "shared-client-id", "metadata": {"marker": "owner-a"}},
+        )
+    assert created.status_code == 200
+
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app_b), base_url="http://test") as client_b:
+        conflict = await client_b.post(
+            "/api/threads",
+            json={"thread_id": "shared-client-id", "metadata": {"marker": "owner-b"}},
+        )
+
+    # The documented non-disclosing ownership-conflict response, not a 500.
+    assert conflict.status_code == 404
+    assert conflict.json()["detail"] == "Thread not found"
+    # A's metadata is untouched and B cannot read the foreign row.
+    assert await sql_thread_store.get("shared-client-id", user_id=str(USER_A.id)) is not None
+    assert await sql_thread_store.get("shared-client-id", user_id=str(USER_B.id)) is None

backend/tests/test_threads_router.py

@@ -1358,7 +1358,12 @@ def test_insert_race_concurrent_trusted_claims_have_one_winner(monkeypatch) -> N
     assert winners == [final_record["user_id"]]
     assert final_record["user_id"] in owners
     assert len(failures) == 1
-    assert failures[0].status_code == 500
+    # The losing request now hits the unscoped duplicate-key recovery path and
+    # resolves the surviving (legacy, owner-less) row as a 404 ownership
+    # conflict rather than a 500, while the single-winner assertions above and
+    # the non-disclosing detail below are preserved.
+    assert failures[0].status_code == 404
+    assert failures[0].detail == "Thread not found"
     assert all(not table._entries_by_loop for table in worker_locks)
 
 
@@ -1400,6 +1405,42 @@ def test_create_thread_does_not_swallow_non_integrity_errors() -> None:
     assert response.status_code == 500, response.text
 
 
+def test_create_thread_keeps_real_integrity_failure_as_500() -> None:
+    """A duplicate-key ``IntegrityError`` with no surviving row stays a 500.
+
+    The recovery path rescues the idempotency race only when a row now exists;
+    an ``IntegrityError`` that coincides with no readable row — a genuine
+    constraint violation unrelated to the caller-chosen id — must still surface
+    as a 500 rather than being collapsed into the ownership-conflict 404. This
+    pins the fall-through branch that the unscoped duplicate-key read would
+    otherwise silently convert into a 404.
+    """
+    from sqlalchemy.exc import IntegrityError
+
+    app, store, _checkpointer = _build_thread_app()
+
+    class _IntegrityFailureNoSurvivingRowStore(_PermissiveThreadMetaStore):
+        async def create(self, thread_id, *, assistant_id=None, user_id=None, display_name=None, metadata=None, project_id=None):  # type: ignore[override]
+            raise IntegrityError(
+                "INSERT INTO threads_meta",
+                {},
+                Exception("UNIQUE constraint failed: threads_meta.thread_id"),
+            )
+
+        async def get(self, thread_id, *, user_id=None):  # type: ignore[override]
+            # No row survives the failed insert, so the unscoped duplicate-key
+            # read returns nothing and the route must not treat this as a
+            # cross-owner conflict.
+            return None
+
+    app.state.thread_store = _IntegrityFailureNoSurvivingRowStore(store)
+
+    with TestClient(app) as client:
+        response = client.post("/api/threads", json={"thread_id": "integrity-no-row", "metadata": {}})
+
+    assert response.status_code == 500, response.text
+
+
 def test_put_goal_creates_missing_thread_checkpoint_and_returns_goal() -> None:
     app, _store, _checkpointer = _build_thread_app()