From 6c5328b16b75dbf63988dac7dbf82dcb1f10a8c0 Mon Sep 17 00:00:00 2001 From: holo Date: Mon, 29 Jun 2026 21:34:46 +0800 Subject: [PATCH] fix(tests): wait for the final event on the tape, not the status row (#256) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `test_event_tape_replay_after_terminal` (and its same-family siblings `test_resume_from_seq_returns_tail_only` and the synth tape assert) trusted the task status row as a proxy for "the `final` event is on the event tape". But `TaskManager._run` flips the status row terminal *before* appending the `final` event (so a `final`-watching follower that calls `/result` always finds the row terminal). A `wait_task_terminal` wait followed by a `wait=0` tape read therefore races the trailing `progress` event onto the last slot, intermittently failing `events[-1]["type"] == "final"` with `'progress'`. Add a shared `wait_event_tape_final` conftest helper that polls the events endpoint until the tape's last event is `final` — the signal the asserts actually depend on — and route the three affected HTTP-level tests through it. This is the HTTP-layer analogue of the store-level waiters already in `test_task_manager.py` / `test_telemetry_tracing.py`. No product code changes. The helper fails fast on a non-200 from `/events` (so a real 500 isn't masked as a "never reached final" timeout) and pages through `has_more`/`next_from_seq` so the trailing `final` is found past the 1000-event page cap. New `test_event_tape_waiter.py` pins these behaviours on known-bad input. The sibling `test_task_span_marks_cancel_not_error` flake is NOT addressed here: it already waits for the `final` event, so its ~30s timeout is a manager-side cancel race (a different mechanism), left to a separate root-caused change rather than bundling a speculative product fix. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/server/conftest.py | 69 ++++++++++++- tests/server/test_event_tape_waiter.py | 131 +++++++++++++++++++++++++ tests/server/test_ingest_task.py | 25 ++--- tests/server/test_synth_tasks.py | 10 +- 4 files changed, 211 insertions(+), 24 deletions(-) create mode 100644 tests/server/test_event_tape_waiter.py diff --git a/tests/server/conftest.py b/tests/server/conftest.py index 7a01efcc..0a206b8d 100644 --- a/tests/server/conftest.py +++ b/tests/server/conftest.py @@ -147,9 +147,13 @@ async def wait_task_terminal( """Poll ``GET /v1/tasks/{id}`` until status is terminal; return the row. Shared by every HTTP-level task test that needs to wait for a - submitted runner to finish before asserting on the event tape or - final result. Default 10s timeout — synth/eval paths that need - more should pass an explicit value.""" + submitted runner to finish before asserting on the ``/result`` payload. + The status row is terminal the instant ``/result`` is consistent, but + NOT a sound proxy for "the ``final`` event is on the tape" — a test that + reads the event tape and asserts on its last entry must use + ``wait_event_tape_final`` instead (see its docstring for the race). + Default 10s timeout — synth/eval paths that need more should pass an + explicit value.""" import asyncio as _asyncio deadline = _asyncio.get_event_loop().time() + timeout @@ -163,3 +167,62 @@ async def wait_task_terminal( return r.json() await _asyncio.sleep(0.05) raise AssertionError(f"task {task_id} never reached a terminal state") + + +async def wait_event_tape_final( + client: httpx.AsyncClient, task_id: str, *, timeout: float = 10.0 +) -> list[dict[str, Any]]: + """Poll ``GET /v1/tasks/{id}/events`` until the *complete* tape's last + event is the terminal ``final`` envelope; return the full event list. + + ``TaskManager._run`` flips the task status row to its terminal state + *before* appending the ``final`` event to the tape (so a follower that + sees ``final`` and immediately calls ``/result`` always finds the row + terminal — the reverse order races ``task_not_terminal``). A test that + trusts ``wait_task_terminal`` (the status-row proxy) and then reads the + tape with ``wait=0`` therefore races the trailing ``progress`` event onto + the last slot, intermittently seeing ``'progress'`` where it expects + ``'final'``. Such tests must wait for the signal they actually depend on — + the ``final`` event itself. This is the HTTP-layer analogue of the + store-level waiter in ``test_task_manager.py`` / ``test_telemetry_tracing.py``. + + A ``wait>0`` long-poll cannot stand in here: the events endpoint only + enters its wait loop when the ``from_seq`` slice is empty, but these tests + read from ``from_seq=0`` where the tape already carries ``task_started`` + + ``progress`` events, so the handler returns immediately with whatever is on + the tape — possibly still missing ``final``. Polling the tape is the fix. + + Each poll reads the *whole* tape by following ``has_more`` / ``next_from_seq`` + (the endpoint caps a single page at 1000 events), so the trailing ``final`` + envelope is found even on a tape longer than one page rather than paging off + the end. A non-200 from the endpoint is a real failure (e.g. a 500 from a + store-read regression) and is surfaced immediately rather than masked as a + misleading "never reached final" timeout. + """ + import asyncio as _asyncio + + deadline = _asyncio.get_event_loop().time() + timeout + while _asyncio.get_event_loop().time() < deadline: + events: list[dict[str, Any]] = [] + from_seq = 0 + while True: + r = await client.get( + f"/v1/tasks/{task_id}/events", + params={"from_seq": from_seq, "limit": 1000, "wait": 0}, + ) + if r.status_code != 200: + raise AssertionError( + f"GET /v1/tasks/{task_id}/events returned " + f"{r.status_code}: {r.text}" + ) + page = r.json() + events.extend(page["events"]) + if not page["has_more"]: + break + from_seq = page["next_from_seq"] + if events and events[-1]["type"] == "final": + return events + await _asyncio.sleep(0.05) + raise AssertionError( + f"task {task_id} event tape never ended with a 'final' event" + ) diff --git a/tests/server/test_event_tape_waiter.py b/tests/server/test_event_tape_waiter.py new file mode 100644 index 00000000..79caa0c5 --- /dev/null +++ b/tests/server/test_event_tape_waiter.py @@ -0,0 +1,131 @@ +"""Unit tests for the ``wait_event_tape_final`` conftest waiter. + +The waiter de-flakes HTTP-level tape-tail asserts (issue #256). These tests +pin the three behaviours the integration callers exercise only by timing / +not at all, against a minimal fake client so each branch is deterministic: + + * surfaces a non-200 from ``/events`` immediately (a real 500 must not be + masked as a 10s "never reached final" timeout); + * pages through a multi-page tape via ``has_more`` / ``next_from_seq`` so the + trailing ``final`` is found even past the 1000-event page cap; + * keeps polling until the ``final`` envelope actually lands, then returns the + full tape. +""" + +from __future__ import annotations + +from typing import Any, cast + +import httpx +import pytest + +from .conftest import wait_event_tape_final + + +class _FakeResp: + def __init__( + self, status_code: int, payload: dict[str, Any] | None = None, text: str = "" + ) -> None: + self.status_code = status_code + self._payload = payload or {} + self.text = text + + def json(self) -> dict[str, Any]: + return self._payload + + +class _FakeClient: + """Serves a scripted list of responses in ``get`` call order. + + The waiter's call sequence is deterministic, so scripting responses by + order is enough; an exhausted script clamps to the last response. + """ + + def __init__(self, responses: list[_FakeResp]) -> None: + self._responses = responses + self.calls = 0 + + async def get( + self, url: str, params: dict[str, Any] | None = None + ) -> _FakeResp: + idx = min(self.calls, len(self._responses) - 1) + self.calls += 1 + return self._responses[idx] + + +def _ev(seq: int, type_: str, **extra: Any) -> dict[str, Any]: + return {"seq": seq, "type": type_, **extra} + + +@pytest.mark.asyncio +async def test_surfaces_non_200_immediately() -> None: + client = _FakeClient([_FakeResp(500, text="boom")]) + with pytest.raises(AssertionError, match="returned 500: boom"): + await wait_event_tape_final( + cast(httpx.AsyncClient, client), "t1", timeout=2.0 + ) + # Failed on the very first fetch — did not busy-poll to the deadline. + assert client.calls == 1 + + +@pytest.mark.asyncio +async def test_pages_past_one_page_to_find_final() -> None: + page1 = _FakeResp( + 200, + { + "events": [_ev(1, "task_started"), _ev(2, "progress")], + "has_more": True, + "next_from_seq": 3, + }, + ) + page2 = _FakeResp( + 200, + { + "events": [_ev(3, "progress"), _ev(4, "final", status="succeeded")], + "has_more": False, + "next_from_seq": 5, + }, + ) + client = _FakeClient([page1, page2]) + events = await wait_event_tape_final( + cast(httpx.AsyncClient, client), "t2", timeout=2.0 + ) + # Concatenated both pages and saw the trailing ``final``. + assert [e["type"] for e in events] == [ + "task_started", + "progress", + "progress", + "final", + ] + assert client.calls == 2 + + +@pytest.mark.asyncio +async def test_waits_until_final_lands() -> None: + not_yet = _FakeResp( + 200, + { + "events": [_ev(1, "task_started"), _ev(2, "progress")], + "has_more": False, + "next_from_seq": 3, + }, + ) + done = _FakeResp( + 200, + { + "events": [ + _ev(1, "task_started"), + _ev(2, "progress"), + _ev(3, "final", status="succeeded"), + ], + "has_more": False, + "next_from_seq": 4, + }, + ) + client = _FakeClient([not_yet, done]) + events = await wait_event_tape_final( + cast(httpx.AsyncClient, client), "t3", timeout=2.0 + ) + assert events[-1]["type"] == "final" + # One poll saw only progress, the next saw final. + assert client.calls == 2 diff --git a/tests/server/test_ingest_task.py b/tests/server/test_ingest_task.py index 69c7f8cf..4c6b3d20 100644 --- a/tests/server/test_ingest_task.py +++ b/tests/server/test_ingest_task.py @@ -23,6 +23,7 @@ import httpx import pytest +from .conftest import wait_event_tape_final as _wait_tape_final from .conftest import wait_task_terminal as _wait_terminal # ---- happy path --------------------------------------------------------- @@ -81,14 +82,11 @@ async def test_event_tape_replay_after_terminal( "/v1/ingest", json={"no_embed": True} ) task_id = submit.json()["task_id"] - await _wait_terminal(server_client, task_id) - - resp = await server_client.get( - f"/v1/tasks/{task_id}/events", - params={"from_seq": 0, "limit": 1000, "wait": 0}, - ) - assert resp.status_code == 200 - events = resp.json()["events"] + # Wait for the ``final`` event to land on the tape, not just the status + # row — the manager flips the row terminal *before* appending ``final``, + # so a bare ``wait=0`` read races the trailing ``progress`` event onto + # the last slot (see ``wait_event_tape_final``). + events = await _wait_tape_final(server_client, task_id) assert events[0]["type"] == "task_started" assert events[0]["op"] == "ingest" assert events[-1]["type"] == "final" @@ -112,14 +110,9 @@ async def test_resume_from_seq_returns_tail_only( "/v1/ingest", json={"no_embed": True} ) task_id = submit.json()["task_id"] - await _wait_terminal(server_client, task_id) - - # First read the full tape to learn the seq range. - full_resp = await server_client.get( - f"/v1/tasks/{task_id}/events", - params={"from_seq": 0, "limit": 1000, "wait": 0}, - ) - full = full_resp.json()["events"] + # Read the full tape once ``final`` has landed, to learn the seq range + # without racing the status-row/tape ordering (see ``wait_event_tape_final``). + full = await _wait_tape_final(server_client, task_id) last_seq = full[-1]["seq"] # Resume from the middle. diff --git a/tests/server/test_synth_tasks.py b/tests/server/test_synth_tasks.py index 7093e6b9..b36424fd 100644 --- a/tests/server/test_synth_tasks.py +++ b/tests/server/test_synth_tasks.py @@ -20,6 +20,7 @@ from dikw_core.server import synth_op as synth_op_module from ..fakes import FakeEmbeddings, FakeLLM +from .conftest import wait_event_tape_final as _wait_tape_final from .conftest import wait_task_terminal as _wait_terminal FIXTURES = Path(__file__).parent.parent / "fixtures" / "notes" @@ -136,11 +137,10 @@ async def test_synth_task_emits_per_source_progress_and_final_report( assert result["errors"] == 0 # Event tape carries one progress event per source, all phase=synth. - resp = await server_client.get( - f"/v1/tasks/{task_id}/events", - params={"from_seq": 0, "limit": 1000, "wait": 0}, - ) - events = resp.json()["events"] + # Wait for ``final`` to land on the tape — the status row (already + # terminal above) is flipped *before* ``final`` is appended, so a bare + # ``wait=0`` read races the trailing event (see ``wait_event_tape_final``). + events = await _wait_tape_final(server_client, task_id) synth_progress = [ e for e in events if e["type"] == "progress" and e["phase"] == "synth" ]