Fix #1995: thread cursor through BRC wait endpoint to close wait→wait race

docs/reference/agent-tools.md

@@ -106,8 +106,8 @@ that requires the handler docstring to explain why no CLI exists.
 | `mcp__brc__confirm` | Signal CONFIRMED — producer acknowledges all reviewer ACKs. | `handlers.brc.brc_confirm` | `egg-orch consensus confirmed` |
 | `mcp__brc__get_state` | Full structured consensus state (JSON; accepts `verbose: bool`). | `handlers.brc.brc_get_state` | — *(no CLI; CLI `egg-orch consensus status` prints text — this tool returns the dict)* |
 | `mcp__brc__list_blocking` | Return the list of agent roles currently blocking consensus (derived view). | `handlers.brc.brc_list_blocking` | — *(no CLI; new capability)* |
-| `mcp__brc__wait_for_event` | Block until a typed message (e.g. `CONSENSUS_ACK`, `CONSENSUS_NACK`) arrives for this agent. Event-driven alternative to polling in a Bash loop. | `handlers.message.message_wait` | `egg-orch message wait` |
-| `mcp__brc__wait_loop` | Loop `wait_for_event` until a match arrives or `max_iterations` trips; rides through timeouts and short transient gateway errors. | `handlers.message.message_wait_loop` | `egg-orch message wait-loop` |
+| `mcp__brc__wait_for_event` | Block until a typed message (e.g. `CONSENSUS_ACK`, `CONSENSUS_NACK`) arrives for this agent. Event-driven alternative to polling in a Bash loop. Returns `cursor` — thread into the next call's `since` to close the wait→wait race (#1995). | `handlers.message.message_wait` | `egg-orch message wait` |
+| `mcp__brc__wait_loop` | Loop `wait_for_event` until a match arrives or `max_iterations` trips; rides through timeouts and short transient gateway errors. Threads `cursor` internally between iterations and surfaces the final `cursor` for chaining across calls (#1995). | `handlers.message.message_wait_loop` | `egg-orch message wait-loop` |
 | `mcp__brc__send_heartbeat` | Emit a structured `HEARTBEAT` (schema-validated, per-role deduped, rate-limited) to the dedicated `/heartbeat` endpoint. Use `state=WAITING_ON_ROLE` + `waiting_on=<peer>` while blocking on BRC. | `handlers.message.message_heartbeat` | `egg-orch message heartbeat` |
 | `mcp__brc__read_peer_artifact` | Read entries from `.egg-state/brc-history/<pipeline_id>-<phase>.json` filtered by `peer_role`, with `limit`/`cursor` pagination (default `limit=50`). `pipeline_id` is resolved server-side from `EGG_PIPELINE_ID` / `EGG_ISSUE_NUMBER` (agents cannot pass an arbitrary id; path-traversal hardening). Returns `{items: [...], next_cursor: <str|None>, skipped_malformed: <int>}`. | `handlers.brc.read_peer_artifact` | — *(no CLI; reviewer-forensics helper that reads local files; operators inspect the files directly)* |
 

docs/reference/agent-wait-patterns.md

@@ -247,6 +247,47 @@ With `--since`, the wait starts at the named ID (exclusive) and
 includes any event — your own send, a peer's, or one that arrived
 in the window — that arrived after the anchor.
 
+### Cursor threading across waits (issue #1995)
+
+Every `egg-orch message wait` / `wait-loop` response and every
+`mcp__brc__wait_for_event` / `mcp__brc__wait_loop` return dict now
+carries a `cursor` field that callers should thread into the `since`
+parameter of the next call. Without threading, events that arrive
+between a returning wait and the subsequent wait call are missed —
+the same wait→wait race window §7 closes on the host side.
+
+The `cursor` is opaque from the caller's perspective:
+
+| Wait outcome | `cursor` value |
+|--------------|----------------|
+| Match (one or more messages returned) | ID of the **last** delivered message |
+| Timeout (no messages) | Current stream tip at server response time |
+| Stream empty on timeout | `null` / unset — caller may keep its prior cursor or omit `since` |
+
+`wait-loop` already threads the cursor internally between its own
+iterations, so a cursor-less call that rides through several timeouts
+before matching does not reopen the race. The surfaced cursor on the
+outer return closes the same race across successive `wait-loop`
+invocations by the agent.
+
+**Recommended pattern (BRC producer loop):**
+
+```python
+cursor = None
+while not consensus_done:
+    resp = mcp__brc__wait_loop(
+        for_types=["CONSENSUS_ACK", "CONSENSUS_NACK", "CONSENSUS_RE_REVIEW", "OVERSEER_ALERT"],
+        since=cursor,
+    )
+    cursor = resp.get("cursor", cursor)
+    for msg in resp["messages"]:
+        ...  # process ACK / NACK / re-review / alert
+```
+
+Legacy callers that ignore `cursor` and don't pass `since` keep their
+pre-#1995 behaviour — still correct for the common "wait once, exit"
+shape, still vulnerable to the wait→wait race for multi-ACK loops.
+
 ## 4. `HEARTBEAT` Message Type
 
 `HEARTBEAT` is a typed message agents emit on state transitions so the

orchestrator/routes/messages.py

@@ -426,12 +426,23 @@ def wait_messages(pipeline_id: str) -> tuple[Response, int]:
 
     messages = _apply_delphi_filter(pipeline_id, role, messages)
 
+    # Cursor returned on every response so callers can thread it into the
+    # next ``since_id=`` and avoid missing events that arrive between
+    # successive wait calls (issue #1995). On match: the last delivered
+    # message ID. On timeout: the current stream tip so the next call
+    # resumes strictly after what this call would have seen.
+    if messages:
+        cursor: str | None = messages[-1].id
+    else:
+        cursor = message_store.get_latest_id(pipeline_id)
+
     return _make_success(
         "Wait completed",
         data={
             "messages": [m.to_dict() for m in messages],
             "count": len(messages),
             "matched": bool(messages),
+            "cursor": cursor,
         },
     )
 

orchestrator/tests/test_messages.py

@@ -1339,6 +1339,187 @@ def test_wait_invalid_pipeline_id_returns_400(self, client, app):
                 )
                 assert resp.status_code == 400
 
+    def test_wait_returns_cursor_on_match(self, client, app):
+        """Issue #1995: wait response includes the ID of the last
+        delivered message so the caller can thread it on the next call."""
+        import threading
+        import time as _t
+
+        store = MessageStore()
+
+        def _add_after_delay() -> None:
+            _t.sleep(0.15)
+            store.add_message(
+                Message(
+                    pipeline_id="test-pipeline",
+                    from_role="coder",
+                    to_role="all",
+                    message_type=MessageType.CONSENSUS_CONFIRMED,
+                    subject="done",
+                )
+            )
+
+        with app.test_request_context():
+            with (
+                patch("routes.messages.get_message_store", return_value=store),
+                patch(
+                    "routes.messages.get_state_store_for_pipeline"
+                ) as mock_get_store_for_pipeline,
+            ):
+                mock_get_store_for_pipeline.return_value = (
+                    MagicMock(),
+                    _make_pipeline_mock(),
+                )
+                t = threading.Thread(target=_add_after_delay)
+                t.start()
+                try:
+                    resp = client.get(
+                        "/api/v1/pipelines/test-pipeline/messages/wait"
+                        "?for=CONSENSUS_CONFIRMED&timeout=3"
+                    )
+                finally:
+                    t.join(timeout=2)
+                assert resp.status_code == 200
+                data = json.loads(resp.data)
+                assert data["data"]["matched"] is True
+                # Cursor must equal the ID of the last delivered message so
+                # the next wait can resume strictly after it.
+                assert data["data"]["cursor"] == data["data"]["messages"][-1]["id"]
+
+    def test_wait_returns_cursor_on_timeout(self, client, app):
+        """Issue #1995: on timeout with no match the server still returns
+        a cursor (current stream tip) so the next wait can pick up
+        anything that arrived while the caller was round-tripping."""
+        store = MessageStore()
+        # Seed one non-matching message so the stream has a tip.
+        seeded = store.add_message(
+            Message(
+                pipeline_id="test-pipeline",
+                from_role="documenter",
+                to_role="all",
+                message_type=MessageType.PROGRESS,
+                subject="ignore me",
+            )
+        )
+
+        with app.test_request_context():
+            with (
+                patch("routes.messages.get_message_store", return_value=store),
+                patch(
+                    "routes.messages.get_state_store_for_pipeline"
+                ) as mock_get_store_for_pipeline,
+            ):
+                mock_get_store_for_pipeline.return_value = (
+                    MagicMock(),
+                    _make_pipeline_mock(),
+                )
+                resp = client.get(
+                    "/api/v1/pipelines/test-pipeline/messages/wait"
+                    "?for=CONSENSUS_CONFIRMED&timeout=1"
+                )
+                assert resp.status_code == 200
+                data = json.loads(resp.data)
+                assert data["data"]["matched"] is False
+                assert data["data"]["cursor"] == seeded.id
+
+    def test_wait_returns_null_cursor_when_stream_empty(self, client, app):
+        """Stream has never had a message → cursor is null. Next call
+        may safely omit ``since_id`` and the server will snap to a fresh
+        tip as before."""
+        store = MessageStore()
+        with app.test_request_context():
+            with (
+                patch("routes.messages.get_message_store", return_value=store),
+                patch(
+                    "routes.messages.get_state_store_for_pipeline"
+                ) as mock_get_store_for_pipeline,
+            ):
+                mock_get_store_for_pipeline.return_value = (
+                    MagicMock(),
+                    _make_pipeline_mock(),
+                )
+                resp = client.get(
+                    "/api/v1/pipelines/test-pipeline/messages/wait"
+                    "?for=CONSENSUS_CONFIRMED&timeout=1"
+                )
+                assert resp.status_code == 200
+                data = json.loads(resp.data)
+                assert data["data"]["matched"] is False
+                assert data["data"]["cursor"] is None
+
+    def test_wait_cursor_threading_closes_between_call_race(self, client, app):
+        """Issue #1995 regression: reproduces the BRC deadlock scenario.
+
+        Timeline:
+          1. Reviewer A ACKs → wait #1 returns with messages=[ack_a],
+             cursor=ack_a.id.
+          2. Reviewer B ACKs *between* wait #1 returning and wait #2
+             starting. Without cursor threading, wait #2 would snap to
+             a new tip past ack_b and deadlock.
+          3. Producer threads cursor=ack_a.id as since_id on wait #2;
+             ack_b is delivered immediately.
+        """
+        store = MessageStore()
+
+        ack_a = store.add_message(
+            Message(
+                pipeline_id="test-pipeline",
+                from_role="reviewer_agent_design",
+                to_role="all",
+                message_type=MessageType.CONSENSUS_ACK,
+                subject="ack-a",
+            )
+        )
+
+        with app.test_request_context():
+            with (
+                patch("routes.messages.get_message_store", return_value=store),
+                patch(
+                    "routes.messages.get_state_store_for_pipeline"
+                ) as mock_get_store_for_pipeline,
+            ):
+                mock_get_store_for_pipeline.return_value = (
+                    MagicMock(),
+                    _make_pipeline_mock(),
+                )
+                # Wait #1: caller passes since_id=<pre-existing tip> so
+                # the pre-existing ack_a is delivered.  Here we thread
+                # a since_id of an unknown id so the server falls back
+                # to "return full history", which delivers ack_a.
+                resp1 = client.get(
+                    "/api/v1/pipelines/test-pipeline/messages/wait"
+                    "?for=CONSENSUS_ACK&timeout=1&since_id=unknown-sentinel"
+                )
+                assert resp1.status_code == 200
+                data1 = json.loads(resp1.data)
+                assert data1["data"]["matched"] is True
+                cursor1 = data1["data"]["cursor"]
+                assert cursor1 == ack_a.id
+
+                # Simulate the between-calls race: ack_b lands right now.
+                ack_b = store.add_message(
+                    Message(
+                        pipeline_id="test-pipeline",
+                        from_role="reviewer_refine",
+                        to_role="all",
+                        message_type=MessageType.CONSENSUS_ACK,
+                        subject="ack-b",
+                    )
+                )
+
+                # Wait #2: caller threads cursor1 as since_id. Without
+                # cursor threading this call would deadlock (bug #1995);
+                # with it, ack_b is delivered immediately.
+                resp2 = client.get(
+                    "/api/v1/pipelines/test-pipeline/messages/wait"
+                    f"?for=CONSENSUS_ACK&timeout=1&since_id={cursor1}"
+                )
+                assert resp2.status_code == 200
+                data2 = json.loads(resp2.data)
+                assert data2["data"]["matched"] is True
+                assert data2["data"]["messages"][0]["id"] == ack_b.id
+                assert data2["data"]["cursor"] == ack_b.id
+
     def test_wait_timeout_clamped_to_env_cap(self, client, app, monkeypatch):
         """``timeout`` is clamped by ``EGG_MESSAGE_POLL_MAX_WAIT``.
 

sandbox/egg_agent_tools/handlers/message.py

@@ -72,8 +72,14 @@ def message_wait(req: dict[str, Any]) -> dict[str, Any]:
         pipeline_id: override.
 
     Response:
-        { ok: True, matched: bool, messages: list, role: str | None,
-          for_types: list[str], raw: <server response> }
+        { ok: True, matched: bool, messages: list, cursor: str | None,
+          role: str | None, for_types: list[str], raw: <server response> }
+
+    ``cursor`` threads through the wait-endpoint's stream cursor so
+    callers can chain successive waits without losing events that arrive
+    between calls (issue #1995). On match it is the ID of the last
+    delivered message; on timeout it is the stream tip at server
+    response time; ``None`` only when the stream is empty.
 
     Raises:
         HandlerError: invalid arguments.
@@ -115,10 +121,12 @@ def message_wait(req: dict[str, Any]) -> dict[str, Any]:
     data = result.get("data", {}) if isinstance(result, dict) else {}
     messages = list(data.get("messages", []) or [])
     matched = bool(data.get("matched")) or bool(messages)
+    cursor = data.get("cursor")
     return {
         "ok": True,
         "matched": matched,
         "messages": messages,
+        "cursor": cursor,
         "role": role,
         "for_types": for_types,
         "raw": result,
@@ -133,16 +141,24 @@ def message_wait_loop(req: dict[str, Any]) -> dict[str, Any]:
     Permanent errors (HandlerError or GatewayError with 4xx non-408)
     propagate to the caller.
 
+    Threads the server-side ``cursor`` through successive iterations via
+    ``since_id`` so a message that arrives after one inner wait times
+    out (and before the next begins) is still delivered on the next
+    iteration. The final ``cursor`` is also surfaced in the response so
+    callers can chain successive ``wait_loop`` invocations across tool
+    boundaries without reopening the same race at the outer layer
+    (issue #1995).
+
     Request:
         Same as :func:`message_wait` plus:
         max_iterations (int): safety cap on outer iterations.  ``None``
             or non-positive means effectively unbounded (``sys.maxsize``),
             matching the CLI's "loops forever by default" contract.
 
     Response:
-        { ok: True, matched: True, messages, iterations, ... }
+        { ok: True, matched: True, messages, cursor, iterations, ... }
         Or — if the safety cap trips without a match —
-        { ok: True, matched: False, iterations: <cap>, ... }.
+        { ok: True, matched: False, cursor, iterations: <cap>, ... }.
 
     Raises:
         HandlerError / GatewayError on permanent failure (4xx non-408).
@@ -172,6 +188,14 @@ def message_wait_loop(req: dict[str, Any]) -> dict[str, Any]:
             backoff = min(backoff * 2, 5.0)
             continue
         last_resp = resp
+        # Thread the server cursor into the next wait's ``since`` so
+        # events that arrive between this response and the next call
+        # can't slip through the gap (issue #1995). A cursor of ``None``
+        # (empty stream) leaves ``inner["since"]`` unchanged so we keep
+        # whatever cursor the caller originally passed in, if any.
+        next_cursor = resp.get("cursor")
+        if next_cursor is not None:
+            inner["since"] = next_cursor
         if resp.get("matched"):
             resp_out = dict(resp)
             resp_out["iterations"] = i

sandbox/egg_agent_tools/tools/message.py

@@ -36,7 +36,14 @@
     },
     "role": {"type": "string", "description": "Filter for this receiver role"},
     "from_role": {"type": "string", "description": "Filter by sender role"},
-    "since": {"type": "string", "description": "Return messages after this ID"},
+    "since": {
+        "type": "string",
+        "description": (
+            "Return messages after this ID. Thread the ``cursor`` from the "
+            "previous wait_for_event / wait_loop response here to avoid "
+            "missing events that arrive between successive calls."
+        ),
+    },
     "limit": {"type": "integer", "description": "Max messages to return"},
     "timeout": {
         "type": "integer",