diff --git a/src/xagent/web/api/websocket.py b/src/xagent/web/api/websocket.py index abc7bfa125..d422df149e 100644 --- a/src/xagent/web/api/websocket.py +++ b/src/xagent/web/api/websocket.py @@ -18,6 +18,7 @@ TYPE_CHECKING, Any, Dict, + Iterable, Iterator, List, Literal, @@ -988,6 +989,57 @@ def _is_duplicate_user_message_turn( return False +def _clarification_request_id(result: Any) -> Optional[str]: + """The waiting round's stable identity, read off a normalized result. + + The runtime mints one ``event_id`` per ask and threads it through the + clarification draft (``core/agent/clarification.py``: "event_id is the + clarification's stable identity"). Frames that assert WAITING_FOR_USER + carry it as ``request_id`` so a client can bind replies - and its + retry gate (#1500) - to the exact round, across live delivery, resume, + replay, and restore. Tolerates both the draft dataclass and its + ``to_dict`` form; anything else reads as "no identity", never a throw, + because these calls sit on broadcast paths. + """ + + if not isinstance(result, dict): + return None + draft = result.get("clarification_draft") + if isinstance(draft, dict): + event_id = draft.get("event_id") + else: + event_id = getattr(draft, "event_id", None) + if isinstance(event_id, str) and event_id: + return event_id + return None + + +def _latest_ask_event_id(trace_events: Iterable[Any]) -> Optional[str]: + """The most recent persisted ask's ``event_id``, for id-less surfaces. + + Restore/reassert paths have no normalized result in hand; the persisted + ask trace row (an ``agent_message`` with ``expect_response``) is the + durable copy of the same identity. The scan stops at the newest ask so + an older round's id can never label a newer question. + + Delegated-child asks cannot shadow the parent round: only the top-level + runtime's outbound handler persists ``expect_response`` rows, and a + waiting child is hard-classified as an unsupported nested interaction + before it could ever produce one (agent_tool.py). + """ + + latest: Optional[str] = None + for event in trace_events: + data = getattr(event, "data", None) + if not isinstance(data, dict): + continue + if data.get("expect_response") is not True: + continue + event_id = data.get("event_id") + latest = event_id if isinstance(event_id, str) and event_id else None + return latest + + def create_stream_event( event_type: str, task_id: Union[int, str], @@ -2897,6 +2949,18 @@ async def execute_task_background( "agent_id": broadcast_agent_meta["agent_id"], "agent_name": broadcast_agent_meta["agent_name"], "agent_logo_url": broadcast_agent_meta["agent_logo_url"], + # The waiting round's identity (#1500) - see + # _clarification_request_id. Key present only + # when an id exists, matching the replay frames. + **( + {"request_id": waiting_request_id} + if ( + waiting_request_id := _clarification_request_id( + result + ) + ) + else {} + ), **control_event_state, }, broadcast_meta["updated_at"] or None, @@ -3892,6 +3956,20 @@ async def mark_deferred_delivery_failed() -> bool: "agent_id": task_agent_id, "agent_name": agent_name, "agent_logo_url": agent_logo_url, + # The waiting round's identity (#1500) - see + # _clarification_request_id. Key present only when + # an id exists (never for "interrupted"), matching + # the replay frames. + **( + {"request_id": resume_waiting_request_id} + if status == "waiting_for_user" + and ( + resume_waiting_request_id := _clarification_request_id( + result + ) + ) + else {} + ), **control_event_state, }, ), @@ -7874,6 +7952,19 @@ def _load_historical_stream_snapshot_sync( .all() ) + # The waiting round's identity for this replay (#1500): the + # persisted ask trace row is the durable copy of the id the live + # ask carried. Backfill the task_info built above - the trace + # rows were not loaded yet at that point - and stamp the + # reassertion frame below from the same source. + replay_ask_request_id = ( + _latest_ask_event_id(trace_events) + if task.status == TaskStatus.WAITING_FOR_USER + else None + ) + if replay_ask_request_id and isinstance(task_event.get("data"), dict): + task_event["data"]["request_id"] = replay_ask_request_id + # DAG execution info is now directly provided by DAG plan-execute trace events # DAG execution events are now directly sent by DAG plan-execute, no need to rebuild @@ -8226,6 +8317,11 @@ def sort_key(x: dict[str, Any]) -> datetime: status_event["question"] = question_message if isinstance(question_interactions, list): status_event["interactions"] = question_interactions + if replay_ask_request_id: + # The reasserted round keeps its identity (#1500), so a + # reloading client can rebind its reply - and its retry + # gate - to the same ask. + status_event["request_id"] = replay_ask_request_id cached_stream_events.append(status_event) detached_events = [ diff --git a/tests/web/api/test_clarification_round_id_emission.py b/tests/web/api/test_clarification_round_id_emission.py new file mode 100644 index 0000000000..59775c653c --- /dev/null +++ b/tests/web/api/test_clarification_round_id_emission.py @@ -0,0 +1,202 @@ +"""The clarification round id (``request_id``) on task-state frames. + +Issue #1500: a client gating clarification retries must bind replies to the +exact ask. The runtime mints one ``event_id`` per ask; these tests pin that +every surface a waiting round reaches the client through carries it as +``request_id`` — the live/resume waiting ``task_info``, the history-replay +``task_info``, and the history-replay ``task_waiting_for_user`` reassertion. +(The question-less lease-restore corrective broadcast deliberately stays +id-less; the frontend preserves a known id across it.) +""" + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from tests.web.api.conftest import _direct_db_session, _test_db + +__all__ = ["_test_db"] +from xagent.web.api import websocket as websocket_api +from xagent.web.api.websocket import ( + _clarification_request_id, + _latest_ask_event_id, +) +from xagent.web.models.task import Task, TaskStatus, TraceEvent +from xagent.web.models.user import User + + +def test_request_id_reads_a_dict_draft() -> None: + result = {"clarification_draft": {"event_id": "evt-1"}} + assert _clarification_request_id(result) == "evt-1" + + +def test_request_id_reads_a_dataclass_draft() -> None: + result = {"clarification_draft": SimpleNamespace(event_id="evt-2")} + assert _clarification_request_id(result) == "evt-2" + + +@pytest.mark.parametrize( + "result", + [ + None, + "waiting", + {}, + {"clarification_draft": None}, + {"clarification_draft": {}}, + {"clarification_draft": {"event_id": ""}}, + {"clarification_draft": SimpleNamespace(event_id=None)}, + ], + ids=[ + "non-dict-none", + "non-dict-str", + "no-draft", + "none-draft", + "empty-dict-draft", + "empty-id", + "none-id-attr", + ], +) +def test_request_id_degrades_to_none(result: object) -> None: + assert _clarification_request_id(result) is None + + +def _trace_row(event_id: str | None, *, expect_response: bool = True) -> object: + data: dict[str, object] = {"expect_response": expect_response} + if event_id is not None: + data["event_id"] = event_id + return SimpleNamespace(data=data) + + +def test_latest_ask_wins_over_an_older_round() -> None: + rows = [_trace_row("evt-old"), SimpleNamespace(data=None), _trace_row("evt-new")] + assert _latest_ask_event_id(rows) == "evt-new" + + +def test_an_id_less_newest_ask_yields_no_identity() -> None: + # The scan must not reach past the newest ask into an older round's id: + # a stale identity on a newer question is worse than none. + rows = [_trace_row("evt-old"), _trace_row(None)] + assert _latest_ask_event_id(rows) is None + + +def test_non_ask_rows_are_ignored() -> None: + rows = [ + _trace_row("evt-ask"), + SimpleNamespace(data={"event_id": "evt-progress"}), + SimpleNamespace(data={"expect_response": False, "event_id": "evt-no"}), + ] + assert _latest_ask_event_id(rows) == "evt-ask" + assert _latest_ask_event_id([]) is None + + +def _waiting_task_with_asks(ask_event_ids: list[str | None]) -> tuple[int, int]: + db = _direct_db_session() + try: + user = User(username="round-id-replay-user", password_hash="hash") + db.add(user) + db.flush() + task = Task( + user_id=int(user.id), + title="Round id replay", + description="Round id replay", + status=TaskStatus.WAITING_FOR_USER, + ) + db.add(task) + db.flush() + base = datetime.now(timezone.utc) - timedelta(minutes=10) + for index, ask_event_id in enumerate(ask_event_ids): + data: dict[str, object] = { + "expect_response": True, + "message": f"Question {index}?", + "metadata": { + "interactions": [ + {"type": "text_input", "field": "answer", "label": "Answer"} + ] + }, + } + if ask_event_id is not None: + data["event_id"] = ask_event_id + db.add( + TraceEvent( + task_id=int(task.id), + event_id=ask_event_id or f"row-{index}", + event_type="agent_message", + timestamp=base + timedelta(minutes=index), + data=data, + ) + ) + db.commit() + return int(task.id), int(user.id) + finally: + db.close() + + +async def _replay( + monkeypatch: pytest.MonkeyPatch, task_id: int, user_id: int +) -> list[dict]: + sent: list[dict] = [] + + async def send_personal_message(event: dict, _websocket: object) -> None: + sent.append(event) + + monkeypatch.setattr(websocket_api, "cache_get", lambda _key: None) + monkeypatch.setattr(websocket_api, "cache_set", lambda *_a, **_k: None) + monkeypatch.setattr( + websocket_api.manager, "send_personal_message", send_personal_message + ) + await websocket_api.send_historical_data_as_stream( + websocket=object(), + task_id=task_id, + user=SimpleNamespace(id=user_id, is_admin=False), + ) + return sent + + +@pytest.mark.asyncio +async def test_replay_carries_the_newest_ask_id_on_both_waiting_frames( + monkeypatch: pytest.MonkeyPatch, + _test_db: None, +) -> None: + task_id, user_id = _waiting_task_with_asks(["evt-round-1", "evt-round-2"]) + + sent = await _replay(monkeypatch, task_id, user_id) + + task_infos = [ + event + for event in sent + if event.get("event_type") == "task_info" + and isinstance(event.get("data"), dict) + ] + assert task_infos, "replay produced no task_info" + assert task_infos[0]["data"]["request_id"] == "evt-round-2" + + reasserts = [ + event for event in sent if event.get("type") == "task_waiting_for_user" + ] + assert reasserts, "replay produced no waiting reassertion" + assert reasserts[0]["request_id"] == "evt-round-2" + + +@pytest.mark.asyncio +async def test_replay_without_a_persisted_ask_stays_id_less( + monkeypatch: pytest.MonkeyPatch, + _test_db: None, +) -> None: + task_id, user_id = _waiting_task_with_asks([]) + + sent = await _replay(monkeypatch, task_id, user_id) + + task_infos = [ + event + for event in sent + if event.get("event_type") == "task_info" + and isinstance(event.get("data"), dict) + ] + assert task_infos + assert "request_id" not in task_infos[0]["data"] + reasserts = [ + event for event in sent if event.get("type") == "task_waiting_for_user" + ] + assert reasserts + assert "request_id" not in reasserts[0]