github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-01 19:06:01 +00:00
Code Issues Packages Projects Releases Wiki Activity
deer-flow/backend/tests/test_thread_run_messages_pagination.py
Xinmin Zeng e56481d9e3
fix(runs): stamp elapsed duration once per run across message APIs (#4163) 
* fix(runs): show run duration once per run and label it as elapsed work

turn_duration is the run's wall-clock lifetime (updated_at - created_at),
but both message endpoints stamped it onto every AI message of the run
and the frontend rendered each stamp as 'Thought for X seconds' — the
same number repeated per message, and tool-wait time presented as model
thinking latency (#4152).

- share one stamping helper between list_messages and list_run_messages:
  only the run's final non-middleware AI message carries turn_duration
  (list_run_messages already tried to do this but iterated reversed()
  without stopping at the first match)
- completed-state trigger copy becomes 'Worked for X seconds' since the
  number includes tool execution, not just thinking

Fixes #4152

* fix(threads): stamp turn_duration once per run in /history replay too

get_thread_history stamped turn_duration on every AI message of a run
instead of only the last one, so a run with several AI messages (e.g.
a tool call followed by a final answer) showed the same duration badge
more than once on reload.

Rebasing onto main picked up #4118, which replaced the /history
duration path with a checkpoint-metadata fast path plus an
event-store/run-manager fallback for legacy checkpoints - both of
which stamped every AI message per run_id, reintroducing this same
bug on both paths. This commit folds the once-per-run stamping
(stamp_turn_duration_on_last_ai) into both paths instead of only the
legacy one, and narrows the fast-path/fallback boundary to a per-run
completeness check (turn_run_ids - checkpoint_run_durations) instead
of a per-message one, so a run whose duration is already in checkpoint
metadata never re-triggers the event-store correlation fallback just
because its non-final AI messages correctly have no turn_duration.

* fix(frontend): stop caching a client-measured turn_duration per message

A run can produce several standalone content-only AI messages (e.g. a
subagent handoff), each briefly becoming the newest message and each
mounting its own Reasoning timer via the shared turnStartTime. Wiring
onTurnDurationChange let that per-message timer cache and display a
"Worked for X seconds" badge the moment a later message superseded it
- with a different, premature number, since the timer only measured
that message's own streaming window, not the run's total elapsed
time. That reintroduced the #4152 duplicate-badge bug through a path
the backend fix (once-per-run stamping) doesn't cover.

cachedDuration now only ever mirrors a backend-confirmed
turn_duration (already handled by the other effect here), so a
superseded message can no longer display a stale or wrong duration.

* fix(frontend): use 'Worked for' framing even without a persisted duration

The completed-state copy diverged depending on whether turn_duration
had landed yet: "Thought for a few seconds" (no duration) vs. "Worked
for X seconds" (duration present). Both describe the same completed
run, just with or without a persisted number, so the framing
shouldn't disagree on what happened.

* docs(runs): note why the middleware skip is inert on /history, add test

stamp_turn_duration_on_last_ai's middleware-caller skip only has an
effect on the event-store message shape (metadata.caller); checkpoint
messages replayed on /history never carry that field. Document why
that's not a gap - middleware writes go to thread metadata, not the
messages channel, so no middleware message ever reaches a
checkpoint's messages list.

Also lock the middleware-skip contract on list_thread_messages with
its own regression test, mirroring the existing one for
list_run_messages - the thread feed only gained this skip through the
same shared helper, and previously stamped every AI message,
middleware included.

* chore(frontend): retain current run duration UI

The frontend behavior was superseded by #4348; keep the latest main implementation while retaining this PR's backend fix.

* test(frontend): retain current reasoning coverage

Align the old PR test with the frontend implementation already merged in #4348.

---------

Co-authored-by: fancyboi999 <fancyboi999@users.noreply.github.com>
2026-07-29 12:47:30 +08:00

441 lines
16 KiB
Python
Raw Permalink Blame History

"""Tests for paginated GET /api/threads/{thread_id}/runs/{run_id}/messages endpoint."""
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock, MagicMock
from _router_auth_helpers import make_authed_test_app
from _run_message_pagination_helpers import assert_run_message_page
from fastapi.testclient import TestClient
from app.gateway.routers import thread_runs
from deerflow.runtime import END_SENTINEL, MemoryStreamBridge, RunManager
from deerflow.runtime.runs.store.memory import MemoryRunStore
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_app(event_store=None, run_manager=None, stream_bridge=None):
"""Build a test FastAPI app with stub auth and mocked state."""
app = make_authed_test_app()
app.include_router(thread_runs.router)
app.state.stream_bridge = stream_bridge or MemoryStreamBridge()
if event_store is not None:
app.state.run_event_store = event_store
if run_manager is None:
run_manager = AsyncMock()
run_manager.get.return_value = None
app.state.run_manager = run_manager
return app
class _EndingCrossProcessBridge:
supports_cross_process = True
async def publish(self, run_id, event, data):
return None
async def publish_end(self, run_id):
return None
def subscribe(self, run_id, *, last_event_id=None, heartbeat_interval=15.0):
async def _events():
yield END_SENTINEL
return _events()
async def cleanup(self, run_id, *, delay=0):
return None
def _make_event_store(rows: list[dict]):
"""Return an AsyncMock event store whose list_messages_by_run() returns rows."""
store = MagicMock()
store.list_messages_by_run = AsyncMock(return_value=rows)
return store
def _make_message(seq: int) -> dict:
return {"seq": seq, "event_type": "ai_message", "category": "message", "content": f"msg-{seq}"}
def _make_store_only_run_manager() -> RunManager:
store = MemoryRunStore()
asyncio.run(
store.put(
"store-only-run",
thread_id="thread-store",
assistant_id="lead_agent",
status="running",
multitask_strategy="reject",
metadata={},
kwargs={},
created_at="2026-01-01T00:00:00+00:00",
)
)
return RunManager(store=store)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_returns_paginated_envelope():
"""GET /api/threads/{tid}/runs/{rid}/messages returns {data: [...], has_more: bool}."""
rows = [_make_message(i) for i in range(1, 4)]
app = _make_app(event_store=_make_event_store(rows))
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/runs/run-1/messages")
assert response.status_code == 200
body = response.json()
assert "data" in body
assert "has_more" in body
assert body["has_more"] is False
assert len(body["data"]) == 3
def test_has_more_true_when_extra_row_returned():
"""has_more=True when event store returns limit+1 rows."""
# Default limit is 50; provide 51 rows
rows = [_make_message(i) for i in range(1, 52)] # 51 rows
app = _make_app(event_store=_make_event_store(rows))
with TestClient(app) as client:
response = client.get("/api/threads/thread-2/runs/run-2/messages")
assert response.status_code == 200
body = response.json()
assert body["has_more"] is True
assert len(body["data"]) == 50 # trimmed to limit
assert [m["seq"] for m in body["data"]] == list(range(2, 52))
def test_default_page_keeps_newest_messages_when_extra_row_returned():
"""Default latest-page trimming drops the older sentinel row, not the newest message."""
rows = [_make_message(i) for i in range(16, 67)]
app = _make_app(event_store=_make_event_store(rows))
with TestClient(app) as client:
assert_run_message_page(
client,
"/api/threads/thread-2/runs/run-2/messages",
expected_seq=list(range(17, 67)),
)
def test_before_seq_page_keeps_newest_side_when_extra_row_returned():
"""Backward pagination trims the older sentinel so adjacent pages do not miss the boundary message."""
rows = [_make_message(i) for i in range(1, 18)]
app = _make_app(event_store=_make_event_store(rows))
with TestClient(app) as client:
assert_run_message_page(
client,
"/api/threads/thread-2/runs/run-2/messages?before_seq=18&limit=16",
expected_seq=list(range(2, 18)),
)
def test_after_seq_page_keeps_oldest_side_when_extra_row_returned():
"""Forward pagination still trims the newer sentinel row."""
rows = [_make_message(i) for i in range(11, 62)]
app = _make_app(event_store=_make_event_store(rows))
with TestClient(app) as client:
assert_run_message_page(
client,
"/api/threads/thread-2/runs/run-2/messages?after_seq=10",
expected_seq=list(range(11, 61)),
)
def test_after_seq_forwarded_to_event_store():
"""after_seq query param is forwarded to event_store.list_messages_by_run."""
rows = [_make_message(10)]
event_store = _make_event_store(rows)
app = _make_app(event_store=event_store)
with TestClient(app) as client:
response = client.get("/api/threads/thread-3/runs/run-3/messages?after_seq=5")
assert response.status_code == 200
event_store.list_messages_by_run.assert_awaited_once_with(
"thread-3",
"run-3",
limit=51, # default limit(50) + 1
before_seq=None,
after_seq=5,
)
def test_before_seq_forwarded_to_event_store():
"""before_seq query param is forwarded to event_store.list_messages_by_run."""
rows = [_make_message(3)]
event_store = _make_event_store(rows)
app = _make_app(event_store=event_store)
with TestClient(app) as client:
response = client.get("/api/threads/thread-4/runs/run-4/messages?before_seq=10")
assert response.status_code == 200
event_store.list_messages_by_run.assert_awaited_once_with(
"thread-4",
"run-4",
limit=51,
before_seq=10,
after_seq=None,
)
def test_custom_limit_forwarded_to_event_store():
"""Custom limit is forwarded as limit+1 to the event store."""
rows = [_make_message(i) for i in range(1, 6)]
event_store = _make_event_store(rows)
app = _make_app(event_store=event_store)
with TestClient(app) as client:
response = client.get("/api/threads/thread-5/runs/run-5/messages?limit=10")
assert response.status_code == 200
event_store.list_messages_by_run.assert_awaited_once_with(
"thread-5",
"run-5",
limit=11, # 10 + 1
before_seq=None,
after_seq=None,
)
def test_empty_data_when_no_messages():
"""Returns empty data list with has_more=False when no messages exist."""
app = _make_app(event_store=_make_event_store([]))
with TestClient(app) as client:
response = client.get("/api/threads/thread-6/runs/run-6/messages")
assert response.status_code == 200
body = response.json()
assert body["data"] == []
assert body["has_more"] is False
def test_get_run_hydrates_store_only_run():
"""GET /api/threads/{tid}/runs/{rid} should read historical store rows."""
app = _make_app(run_manager=_make_store_only_run_manager())
with TestClient(app) as client:
response = client.get("/api/threads/thread-store/runs/store-only-run")
assert response.status_code == 200
body = response.json()
assert body["run_id"] == "store-only-run"
assert body["thread_id"] == "thread-store"
assert body["status"] == "running"
def test_cancel_store_only_run_returns_409():
"""Store-only runs are readable but not cancellable by this worker."""
app = _make_app(run_manager=_make_store_only_run_manager())
with TestClient(app) as client:
response = client.post("/api/threads/thread-store/runs/store-only-run/cancel")
assert response.status_code == 409
assert "not active on this worker" in response.json()["detail"]
def test_join_store_only_run_returns_409():
"""join endpoint should return 409 for store-only runs (no local stream state)."""
app = _make_app(run_manager=_make_store_only_run_manager())
with TestClient(app) as client:
response = client.get("/api/threads/thread-store/runs/store-only-run/join")
assert response.status_code == 409
assert "not active on this worker" in response.json()["detail"]
def test_stream_store_only_run_returns_409():
"""stream endpoint (action=None) should return 409 for store-only runs."""
app = _make_app(run_manager=_make_store_only_run_manager())
with TestClient(app) as client:
response = client.get("/api/threads/thread-store/runs/store-only-run/stream")
assert response.status_code == 409
assert "not active on this worker" in response.json()["detail"]
def test_join_store_only_run_allowed_with_cross_process_bridge():
"""Redis-like bridges can stream store-only runs hydrated on another worker."""
app = _make_app(run_manager=_make_store_only_run_manager(), stream_bridge=_EndingCrossProcessBridge())
with TestClient(app) as client:
response = client.get("/api/threads/thread-store/runs/store-only-run/join")
assert response.status_code == 200
assert "event: end" in response.text
def test_list_run_messages_injects_turn_duration_only_on_last_ai():
"""#4152: turn_duration is the run's wall-clock elapsed and belongs to the run,
so only the run's final AI message carries it — not every AI message."""
from unittest.mock import AsyncMock
from deerflow.runtime import RunRecord
# Mock a run record that took exactly 5 seconds
mock_run = RunRecord(
run_id="run-1",
thread_id="thread-1",
assistant_id=None,
status="success",
on_disconnect="cancel",
created_at="2026-06-20T10:00:00Z",
updated_at="2026-06-20T10:00:05Z",
)
rows = [
{"seq": 1, "run_id": "run-1", "content": {"type": "human", "text": "Hello"}},
{"seq": 2, "run_id": "run-1", "content": {"type": "ai", "text": "Thinking..."}},
{"seq": 3, "run_id": "run-1", "content": {"type": "ai", "text": "Response"}},
]
event_store = _make_event_store(rows)
run_manager = AsyncMock()
run_manager.get.return_value = mock_run
app = _make_app(event_store=event_store, run_manager=run_manager)
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/runs/run-1/messages")
assert response.status_code == 200
data = response.json()["data"]
assert "turn_duration" not in data[0]["content"].get("additional_kwargs", {})
assert "turn_duration" not in data[1]["content"].get("additional_kwargs", {})
assert data[2]["content"]["additional_kwargs"]["turn_duration"] == 5
def test_list_run_messages_turn_duration_skips_middleware_tail():
"""The duration badge lands on the assistant's final answer, not on a
trailing middleware-caller message (e.g. title generation)."""
from unittest.mock import AsyncMock
from deerflow.runtime import RunRecord
mock_run = RunRecord(
run_id="run-1",
thread_id="thread-1",
assistant_id=None,
status="success",
on_disconnect="cancel",
created_at="2026-06-20T10:00:00Z",
updated_at="2026-06-20T10:00:05Z",
)
rows = [
{"seq": 1, "run_id": "run-1", "content": {"type": "ai", "text": "Response"}},
{"seq": 2, "run_id": "run-1", "content": {"type": "ai", "text": "Title"}, "metadata": {"caller": "middleware:title"}},
]
event_store = _make_event_store(rows)
run_manager = AsyncMock()
run_manager.get.return_value = mock_run
app = _make_app(event_store=event_store, run_manager=run_manager)
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/runs/run-1/messages")
assert response.status_code == 200
data = response.json()["data"]
assert data[0]["content"]["additional_kwargs"]["turn_duration"] == 5
assert "turn_duration" not in data[1]["content"].get("additional_kwargs", {})
def test_list_thread_messages_injects_turn_duration_once_per_run():
"""#4152: in the thread feed each run contributes exactly one duration badge —
on its last AI message — even when the run produced several AI messages and
other runs interleave in the same thread."""
from unittest.mock import AsyncMock
from deerflow.runtime import RunRecord
def _run(run_id: str, seconds: int) -> RunRecord:
return RunRecord(
run_id=run_id,
thread_id="thread-1",
assistant_id=None,
status="success",
on_disconnect="cancel",
created_at="2026-06-20T10:00:00Z",
updated_at=f"2026-06-20T10:00:{seconds:02d}Z",
)
rows = [
{"seq": 1, "run_id": "run-1", "content": {"type": "human", "text": "Hello"}},
{"seq": 2, "run_id": "run-1", "content": {"type": "ai", "text": "Before tools"}},
{"seq": 3, "run_id": "run-1", "content": {"type": "ai", "text": "Final answer"}},
{"seq": 4, "run_id": "run-2", "content": {"type": "human", "text": "Again"}},
{"seq": 5, "run_id": "run-2", "content": {"type": "ai", "text": "Second answer"}},
]
event_store = MagicMock()
event_store.list_messages = AsyncMock(return_value=rows)
run_manager = AsyncMock()
run_manager.list_by_thread = AsyncMock(return_value=[_run("run-1", 5), _run("run-2", 9)])
feedback_repo = MagicMock()
feedback_repo.list_by_thread_grouped = AsyncMock(return_value={})
app = _make_app(event_store=event_store, run_manager=run_manager)
app.state.feedback_repo = feedback_repo
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages")
assert response.status_code == 200
data = response.json()
assert "turn_duration" not in data[0].get("content", {}).get("additional_kwargs", {})
assert "turn_duration" not in data[1]["content"].get("additional_kwargs", {})
assert data[2]["content"]["additional_kwargs"]["turn_duration"] == 5
assert "turn_duration" not in data[3].get("content", {}).get("additional_kwargs", {})
assert data[4]["content"]["additional_kwargs"]["turn_duration"] == 9
def test_list_thread_messages_turn_duration_skips_middleware_tail():
"""The thread feed gained the middleware skip through the same
``stamp_turn_duration_on_last_ai`` helper as the per-run endpoint — before
that it stamped every AI message, middleware included. Lock the contract
on this path too, not just ``list_run_messages``."""
from unittest.mock import AsyncMock
from deerflow.runtime import RunRecord
mock_run = RunRecord(
run_id="run-1",
thread_id="thread-1",
assistant_id=None,
status="success",
on_disconnect="cancel",
created_at="2026-06-20T10:00:00Z",
updated_at="2026-06-20T10:00:05Z",
)
rows = [
{"seq": 1, "run_id": "run-1", "content": {"type": "ai", "text": "Response"}},
{"seq": 2, "run_id": "run-1", "content": {"type": "ai", "text": "Title"}, "metadata": {"caller": "middleware:title"}},
]
event_store = MagicMock()
event_store.list_messages = AsyncMock(return_value=rows)
run_manager = AsyncMock()
run_manager.list_by_thread = AsyncMock(return_value=[mock_run])
feedback_repo = MagicMock()
feedback_repo.list_by_thread_grouped = AsyncMock(return_value={})
app = _make_app(event_store=event_store, run_manager=run_manager)
app.state.feedback_repo = feedback_repo
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages")
assert response.status_code == 200
data = response.json()
assert data[0]["content"]["additional_kwargs"]["turn_duration"] == 5
assert "turn_duration" not in data[1]["content"].get("additional_kwargs", {})
Reference in New Issue View Git Blame Copy Permalink