fix(tests): wait for the final event on the tape, not the status row (#256)

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"
+    )

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

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.

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"
     ]