fix(web): carry the clarification round id on waiting task-state frames

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 = [

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]