fix(web): expose structured terminal command outcomes on agent_error frames

src/xagent/web/api/websocket.py

@@ -165,6 +165,7 @@
     TerminalTaskEventDraft,
     TerminalTaskEventMessageCode,
     bind_terminal_event_draft,
+    first_party_message_terminal_text,
     is_external_cancel_command,
     terminal_event_draft_for_error,
 )
@@ -462,11 +463,24 @@ def client_safe_task_command_failure(
     picked by what the terminal exception proves -- non-application is
     asserted only when it is established, uncertainty otherwise -- and
     needs no task status, so the caller does not read the task for it.
+
+    A first-party MESSAGE drops the prefix for the same proof rule: its
+    sender is deciding whether to resend a durably accepted reply, so the
+    sentence comes from the bound terminal-event draft instead of the
+    exception text (#1500).
     """
     if is_external_cancel_command(kind=kind.value, scope=scope):
         return external_cancel_exhausted_message(task_status)
     if scope == EXTERNAL_COMMAND_SCOPE and kind == TaskCommandKind.MESSAGE:
         return external_input_terminal_message(error)
+    if kind == TaskCommandKind.MESSAGE:
+        # A first-party MESSAGE follows the external rule above rather than
+        # the generic fallback: restating the deferral's last wait condition
+        # under a "failed" prefix tells the sender nothing about whether the
+        # accepted reply was applied (#1500). The sentence is derived from
+        # the bound terminal-event draft, so it asserts non-application only
+        # when the persisted outcome proves it.
+        return first_party_message_terminal_text(terminal_event_draft_for_error(error))
     # kind.value in the text is safe only while every external-scope kind is
     # handled above; a new external-scope kind needs its own branch first.
     return f"Task command {kind.value} failed: {client_safe_error_message(error)}"
@@ -9759,6 +9773,16 @@ async def _broadcast_terminal_command_error(
             command.task_id,
         )
         return
+    # ``outcome``/``resend_safe``/``message_code`` expose the persisted
+    # terminal disposition structurally (#1500), so the sender can decide
+    # whether resending the command is safe without parsing ``message``.
+    # The field names match the durable terminal-event projection (#1904).
+    # Values come from the draft the dispatcher binds before broadcasting;
+    # a missing draft degrades to the unsafe/unknown reading. Only this
+    # identity-bearing frame carries them: the two external frames above
+    # deliberately expose nothing the anonymous audience cannot act on,
+    # and a retry decision needs the ``command_id`` they withhold.
+    draft = terminal_event_draft_for_error(error)
     await manager.broadcast_to_task(
         {
             "type": "agent_error",
@@ -9770,6 +9794,13 @@ async def _broadcast_terminal_command_error(
                 error,
                 scope=scope,
             ),
+            "outcome": "failed",
+            "resend_safe": draft is not None and bool(draft.resend_safe),
+            "message_code": (
+                draft.message_code.value
+                if draft is not None and draft.message_code is not None
+                else None
+            ),
             "command_kind": command.kind.value,
             "task_id": command.task_id,
             "command_id": command.command_id,

src/xagent/web/services/task_command_terminal_events.py

@@ -66,6 +66,37 @@ def terminal_event_draft_for_error(
     return draft if isinstance(draft, TerminalTaskEventDraft) else None
 
 
+# Wording for a first-party MESSAGE command that reached a terminal
+# disposition after the sender's reply was durably accepted. The split
+# mirrors ``external_task_input.external_input_terminal_message``: the
+# categorical "not applied" sentence is reserved for outcomes whose draft
+# proves non-application, and every other terminal gets the sentence that
+# asserts only uncertainty, because a worker may have injected the message
+# before crashing and the reclaiming attempt cannot know.
+FIRST_PARTY_MESSAGE_NOT_APPLIED_MESSAGE = (
+    "This message was not applied to the task. It is safe to send it again."
+)
+FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE = (
+    "We could not confirm whether this message was applied to the task. "
+    "Review the conversation before sending it again."
+)
+
+
+def first_party_message_terminal_text(draft: TerminalTaskEventDraft | None) -> str:
+    """Wording for a terminal first-party MESSAGE outcome, by what is provable.
+
+    Deriving from the draft rather than the exception keeps the sentence
+    aligned with the persisted terminal event: ``resend_safe`` is set only
+    when the failed handoff proved the command never reached the downstream
+    operation. A missing draft yields the uncertain sentence, the safe
+    direction for a duplicate-send decision.
+    """
+
+    if draft is not None and draft.resend_safe:
+        return FIRST_PARTY_MESSAGE_NOT_APPLIED_MESSAGE
+    return FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE
+
+
 def stage_terminal_event(
     db: Session,
     *,

tests/web/api/test_terminal_command_outcome_fields.py

@@ -0,0 +1,183 @@
+"""Structured command-outcome fields on terminal ``agent_error`` frames.
+
+Issue #1500: a client deciding whether an accepted clarification reply is
+safe to resend must read a structured, command-correlated outcome instead of
+parsing human-readable error text. These tests pin the wire contract of the
+live terminal broadcast: the identity-bearing first-party frame carries
+``outcome``, ``resend_safe`` and ``message_code``, the anonymous external
+frames keep their pinned minimal shape, and a first-party MESSAGE terminal
+states what is provable about application instead of restating the condition
+the command was last waiting on.
+
+Field names deliberately match the durable terminal-event projection planned
+in #1904 (``outcome``/``resend_safe``), so the frontend contract survives the
+switch from this live broadcast to durable delivery.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from xagent.web.api import websocket as websocket_api
+from xagent.web.api.websocket import execute_durable_task_command
+from xagent.web.services.external_task_input import (
+    EXTERNAL_INPUT_NOT_APPLIED_MESSAGE,
+)
+from xagent.web.services.task_command_terminal_events import (
+    FIRST_PARTY_MESSAGE_NOT_APPLIED_MESSAGE,
+    FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE,
+)
+from xagent.web.services.task_command_transport import (
+    MAX_COMMAND_FAILURES,
+    ClaimedTaskCommand,
+    TaskCommandDeferred,
+    TaskCommandKind,
+    TaskCommandRejected,
+    max_command_defers,
+)
+
+
+def _message_command(
+    *,
+    payload: dict | None = None,
+    defer_count: int = 0,
+    failure_count: int = 0,
+    kind: TaskCommandKind = TaskCommandKind.MESSAGE,
+) -> ClaimedTaskCommand:
+    return ClaimedTaskCommand(
+        id=1,
+        task_id=7,
+        actor_user_id=None,
+        command_id="clarification-reply-1",
+        kind=kind,
+        payload=payload if payload is not None else {},
+        target_run_id="run-1",
+        attempt_count=defer_count + 1,
+        failure_count=failure_count,
+        defer_count=defer_count,
+    )
+
+
+async def _run_terminal(command: ClaimedTaskCommand, error: BaseException) -> dict:
+    """Drive one exhausted execution and return the broadcast frame."""
+
+    with (
+        patch.object(
+            websocket_api,
+            "_execute_durable_task_command",
+            new=AsyncMock(side_effect=error),
+        ),
+        patch.object(
+            websocket_api.manager,
+            "broadcast_to_task",
+            new=AsyncMock(),
+        ) as broadcast,
+    ):
+        with pytest.raises(type(error)):
+            await execute_durable_task_command(command)
+        broadcast.assert_awaited_once()
+        frame, task_id = broadcast.await_args.args
+        assert task_id == command.task_id
+        return frame
+
+
+@pytest.mark.asyncio
+async def test_exhausted_resend_safe_deferral_broadcasts_not_applied_outcome() -> None:
+    command = _message_command(defer_count=max_command_defers() - 1)
+    error = TaskCommandDeferred(
+        "Message clarification-reply-1 is waiting for the live-control resume slot",
+        resend_safe=True,
+    )
+
+    frame = await _run_terminal(command, error)
+
+    assert frame["type"] == "agent_error"
+    assert frame["outcome"] == "failed"
+    assert frame["resend_safe"] is True
+    assert frame["message_code"] == "task_command_deferred"
+    assert frame["command_id"] == "clarification-reply-1"
+    assert frame["command_kind"] == "message"
+    assert frame["message"] == FIRST_PARTY_MESSAGE_NOT_APPLIED_MESSAGE
+
+
+@pytest.mark.asyncio
+async def test_exhausted_unsafe_deferral_broadcasts_unconfirmed_outcome() -> None:
+    command = _message_command(defer_count=max_command_defers() - 1)
+    error = TaskCommandDeferred(
+        "Message clarification-reply-1 is waiting for runtime injection",
+        resend_safe=False,
+    )
+
+    frame = await _run_terminal(command, error)
+
+    assert frame["outcome"] == "failed"
+    assert frame["resend_safe"] is False
+    assert frame["message_code"] == "task_command_deferred"
+    assert frame["message"] == FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE
+
+
+@pytest.mark.asyncio
+async def test_message_terminal_wording_never_restates_the_wait_reason() -> None:
+    """The exhaustion notice must not read "failed: ... is waiting for ..."."""
+
+    command = _message_command(defer_count=max_command_defers() - 1)
+    error = TaskCommandDeferred(
+        "Message clarification-reply-1 is waiting for the live-control resume slot",
+        resend_safe=False,
+    )
+
+    frame = await _run_terminal(command, error)
+
+    assert "waiting" not in frame["message"]
+    assert "resume slot" not in frame["message"]
+
+
+@pytest.mark.asyncio
+async def test_generic_message_failure_broadcasts_unconfirmed_outcome() -> None:
+    command = _message_command(failure_count=MAX_COMMAND_FAILURES - 1)
+    error = RuntimeError("worker exploded mid-injection")
+
+    frame = await _run_terminal(command, error)
+
+    assert frame["outcome"] == "failed"
+    assert frame["resend_safe"] is False
+    assert frame["message_code"] == "task_command_failed"
+    assert frame["message"] == FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE
+    assert "exploded" not in frame["message"]
+
+
+@pytest.mark.asyncio
+async def test_non_message_terminal_frame_gains_fields_keeps_wording() -> None:
+    command = _message_command(
+        kind=TaskCommandKind.PAUSE,
+        failure_count=MAX_COMMAND_FAILURES - 1,
+    )
+    error = RuntimeError("boom")
+
+    frame = await _run_terminal(command, error)
+
+    assert frame["outcome"] == "failed"
+    assert frame["resend_safe"] is False
+    assert frame["message_code"] == "task_command_failed"
+    assert frame["message"].startswith("Task command pause failed:")
+
+
+@pytest.mark.asyncio
+async def test_external_scope_terminal_frame_stays_identity_free() -> None:
+    """The anonymous external frame gains no structured outcome fields.
+
+    A retry decision needs the ``command_id`` the external frames withhold,
+    so the structured fields would be undecidable noise there; the frame
+    keeps its pinned minimal shape and its proof-aware wording.
+    """
+
+    command = _message_command(payload={"scope": "external"})
+    error = TaskCommandRejected(
+        "principal revoked",
+        reason="revoked_principal",
+    )
+
+    frame = await _run_terminal(command, error)
+
+    assert set(frame) == {"type", "message", "task_id", "timestamp"}
+    assert frame["message"] == EXTERNAL_INPUT_NOT_APPLIED_MESSAGE

tests/web/services/test_task_command_terminal_events.py

@@ -30,7 +30,11 @@
 from xagent.web.models.user import User
 from xagent.web.services import task_command_transport as task_command_transport_module
 from xagent.web.services.task_command_terminal_events import (
+    FIRST_PARTY_MESSAGE_NOT_APPLIED_MESSAGE,
+    FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE,
     TerminalTaskEventDraft,
+    TerminalTaskEventMessageCode,
+    first_party_message_terminal_text,
     stage_terminal_event,
 )
 from xagent.web.services.task_command_transport import (
@@ -1022,3 +1026,30 @@ def test_terminal_event_draft_uses_the_command_disposition_outcome(db_session) -
         .one()
     )
     assert event.outcome == "failed"
+
+
+def test_first_party_message_wording_asserts_non_application_only_on_proof() -> None:
+    proven = TerminalTaskEventDraft(
+        message_code=TerminalTaskEventMessageCode.TASK_COMMAND_DEFERRED,
+        resend_safe=True,
+    )
+    assert (
+        first_party_message_terminal_text(proven)
+        == FIRST_PARTY_MESSAGE_NOT_APPLIED_MESSAGE
+    )
+
+    unproven = TerminalTaskEventDraft(
+        message_code=TerminalTaskEventMessageCode.TASK_COMMAND_DEFERRED,
+        resend_safe=False,
+    )
+    assert (
+        first_party_message_terminal_text(unproven)
+        == FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE
+    )
+
+
+def test_first_party_message_wording_treats_a_missing_draft_as_unknown() -> None:
+    assert (
+        first_party_message_terminal_text(None)
+        == FIRST_PARTY_MESSAGE_UNCONFIRMED_MESSAGE
+    )