diff --git a/docs/reference/agent-tools.md b/docs/reference/agent-tools.md index 3d004ba90f..ec498a0164 100644 --- a/docs/reference/agent-tools.md +++ b/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=` while blocking on BRC. | `handlers.message.message_heartbeat` | `egg-orch message heartbeat` | | `mcp__brc__read_peer_artifact` | Read entries from `.egg-state/brc-history/-.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: , skipped_malformed: }`. | `handlers.brc.read_peer_artifact` | — *(no CLI; reviewer-forensics helper that reads local files; operators inspect the files directly)* | diff --git a/docs/reference/agent-wait-patterns.md b/docs/reference/agent-wait-patterns.md index 96a03914dc..c6fa3117f6 100644 --- a/docs/reference/agent-wait-patterns.md +++ b/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 diff --git a/orchestrator/routes/messages.py b/orchestrator/routes/messages.py index 5b5c329b88..b99294d70d 100644 --- a/orchestrator/routes/messages.py +++ b/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, }, ) diff --git a/orchestrator/tests/test_messages.py b/orchestrator/tests/test_messages.py index 4b91207d2f..235091dc66 100644 --- a/orchestrator/tests/test_messages.py +++ b/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= 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``. diff --git a/sandbox/egg_agent_tools/handlers/message.py b/sandbox/egg_agent_tools/handlers/message.py index ec072f7bb9..000366a029 100644 --- a/sandbox/egg_agent_tools/handlers/message.py +++ b/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: } + { ok: True, matched: bool, messages: list, cursor: str | None, + role: str | None, for_types: list[str], raw: } + + ``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,6 +141,14 @@ 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`` @@ -140,9 +156,9 @@ def message_wait_loop(req: dict[str, Any]) -> dict[str, Any]: 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: , ... }. + { ok: True, matched: False, cursor, iterations: , ... }. 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 diff --git a/sandbox/egg_agent_tools/tools/message.py b/sandbox/egg_agent_tools/tools/message.py index fc150c2ebc..102a9fc47b 100644 --- a/sandbox/egg_agent_tools/tools/message.py +++ b/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", diff --git a/tests/sandbox/egg_agent_tools/test_handlers_message.py b/tests/sandbox/egg_agent_tools/test_handlers_message.py index 4055c214f6..d58d7f4df0 100644 --- a/tests/sandbox/egg_agent_tools/test_handlers_message.py +++ b/tests/sandbox/egg_agent_tools/test_handlers_message.py @@ -95,6 +95,63 @@ def test_gateway_error_propagates(self): with pytest.raises(GatewayError): message.message_wait({"pipeline_id": "p", "for_types": ["X"]}) + def test_cursor_surfaced_on_match(self): + """Issue #1995: server cursor is threaded through the handler.""" + server = { + "success": True, + "data": { + "matched": True, + "messages": [{"id": "m-7", "message_type": "CONSENSUS_ACK"}], + "cursor": "m-7", + }, + } + with patch( + "egg_agent_tools.handlers.message.orchestrator_request", + return_value=server, + ): + resp = message.message_wait({"pipeline_id": "p", "for_types": ["CONSENSUS_ACK"]}) + assert resp["cursor"] == "m-7" + + def test_cursor_surfaced_on_timeout(self): + """Issue #1995: even on timeout the server reports the stream tip.""" + server = { + "success": True, + "data": {"matched": False, "messages": [], "cursor": "tip-12"}, + } + with patch( + "egg_agent_tools.handlers.message.orchestrator_request", + return_value=server, + ): + resp = message.message_wait({"pipeline_id": "p", "for_types": ["X"]}) + assert resp["matched"] is False + assert resp["cursor"] == "tip-12" + + def test_cursor_defaults_to_none_when_server_omits(self): + """Older orchestrators that don't emit ``cursor`` must not crash.""" + server = {"success": True, "data": {"matched": False, "messages": []}} + with patch( + "egg_agent_tools.handlers.message.orchestrator_request", + return_value=server, + ): + resp = message.message_wait({"pipeline_id": "p", "for_types": ["X"]}) + assert resp["cursor"] is None + + def test_since_param_forwarded_to_endpoint(self): + server = {"success": True, "data": {"matched": False, "messages": []}} + with patch( + "egg_agent_tools.handlers.message.orchestrator_request", + return_value=server, + ) as req: + message.message_wait( + { + "pipeline_id": "p", + "for_types": ["X"], + "since": "m-3", + } + ) + endpoint = req.call_args.args[0] + assert "since_id=m-3" in endpoint + class TestMessageWaitLoop: def test_matches_on_first_iteration(self): @@ -182,6 +239,83 @@ def fake_wait(req): assert resp["matched"] is False assert resp["iterations"] == 2 + def test_cursor_threaded_between_iterations(self): + """Issue #1995: each timeout hands its cursor to the next call. + + Without this, an event that lands on the bus between iteration N + returning (timeout) and iteration N+1 starting would be invisible + because from_tip=True would snap to a new tip past it. + """ + observed_since: list[str | None] = [] + responses = [ + {"ok": True, "matched": False, "messages": [], "cursor": "tip-1"}, + {"ok": True, "matched": False, "messages": [], "cursor": "tip-2"}, + { + "ok": True, + "matched": True, + "messages": [{"id": "m-final"}], + "cursor": "m-final", + }, + ] + + def fake_wait(req): + observed_since.append(req.get("since")) + return responses.pop(0) + + with patch("egg_agent_tools.handlers.message.message_wait", side_effect=fake_wait): + resp = message.message_wait_loop( + {"pipeline_id": "p", "for_types": ["CONSENSUS_ACK"], "max_iterations": 5} + ) + assert resp["matched"] is True + assert resp["cursor"] == "m-final" + # First call: caller passed no ``since``. + # Subsequent calls: handler must thread the cursor from the + # prior server response so the gap between iterations is closed. + assert observed_since == [None, "tip-1", "tip-2"] + + def test_cursor_from_initial_since_preserved_if_server_returns_none(self): + """Stream empty → server sends cursor=None. Handler must not + overwrite the caller-supplied ``since`` with None — otherwise the + next iteration would re-scan from start / tip.""" + observed_since: list[str | None] = [] + responses = [ + {"ok": True, "matched": False, "messages": [], "cursor": None}, + {"ok": True, "matched": True, "messages": [{"id": "m-x"}], "cursor": "m-x"}, + ] + + def fake_wait(req): + observed_since.append(req.get("since")) + return responses.pop(0) + + with patch("egg_agent_tools.handlers.message.message_wait", side_effect=fake_wait): + message.message_wait_loop( + { + "pipeline_id": "p", + "for_types": ["X"], + "since": "m-caller", + "max_iterations": 5, + } + ) + assert observed_since == ["m-caller", "m-caller"] + + def test_cursor_surfaced_on_safety_cap(self): + """When the safety cap trips, the last seen cursor must still + be surfaced so the caller can resume cleanly.""" + responses = [ + {"ok": True, "matched": False, "messages": [], "cursor": "tip-a"}, + {"ok": True, "matched": False, "messages": [], "cursor": "tip-b"}, + ] + + def fake_wait(req): + return responses.pop(0) + + with patch("egg_agent_tools.handlers.message.message_wait", side_effect=fake_wait): + resp = message.message_wait_loop( + {"pipeline_id": "p", "for_types": ["X"], "max_iterations": 2} + ) + assert resp["matched"] is False + assert resp["cursor"] == "tip-b" + class TestMessageHeartbeat: def test_happy_path(self):