fix: replay terminal command outcomes

[codeacme17]

Summary

• replay durable terminal task-command outcomes across worker restart and reconnect boundaries
• authorize subscriptions by stable task ownership and preserve immutable task/run/command/outcome correlation
• isolate slow or failing sinks, bound hostile cursors to the database integer domain, and redact external command identity
• persist reconnect cursors by the validated event task and adopt real task_id_updated frames before reconnecting

Dependency graph

This PR is the join point for two prerequisite arms:

#1901 durable persistence ─> #1933 contention deferral (supersedes #1926) ─> unblocks #1500
#1901 ─────────────────────────────────┐
                                      ├─> #1904 durable delivery/replay (#1858's remaining half)
#1902 task writer ─> #1903 UI writers ┘

Updated 2026-08-30: #1926 has been superseded by #1933, which lands the #1469 contention-deferral layer directly on #1901. This PR is no longer on the #1469 → #1500 critical path; it remains the implementation of #1858's durable delivery/replay half. #1933 ports the small _terminal_command_event_draft builder ahead of this PR — once #1933 merges, the rebase collapses that hunk out of this layer's diff.

This PR depends on both #1901 and #1903. It remains Draft until both arms land; it will then be rebased onto main before becoming Ready for review.

Exact review range for this layer: dfdaca42...57be5a7e

The temporary parent dfdaca42 is a merge of the current #1901 and #1903 prerequisite heads. Until those PRs land, GitHub's full diff against main includes both prerequisite arms; use the exact range above to review only #1904.

Verification

• focused backend persistence, delivery, API, transport, and client-safe guard suites
• 227 focused frontend WebSocket/AppProvider tests
• frontend type checking
• Ruff lint and git diff --check

PostgreSQL-parametrized tests were skipped locally because XAGENT_TEST_POSTGRES_URL is not configured.

Scope

Closes #1858 when merged. Existing related work tracked in #1500, #1736, and #1859 remains out of scope.

• ref(web): deduplicate public terminal-event subscription lifecycle #1928 — deduplicate the public widget/share terminal-event subscription lifecycle (non-blocking P2 refactor)

[gemini-code-assist]

Code Review

This pull request introduces a durable, cross-worker delivery mechanism for terminal task-command outcomes by persisting append-only events in a new task_command_terminal_events table and fanning them out to local sockets via a poller hub. It also adds connection-owned serialization for outbound WebSocket frames to prevent cross-socket blocking. The review feedback highlights three critical improvements: limiting retries in the event delivery loop to prevent infinite loops on closed sockets, disconnecting stale WebSockets immediately if terminal event delivery fails, and using .populate_existing() in SQLAlchemy queries to avoid reading stale in-memory command states.

[gemini-code-assist]

The delivery loop retries indefinitely upon any Exception. For WebSocket connections, write failures are terminal and cannot be recovered on the same socket instance. Retrying indefinitely causes an infinite loop of failing writes, sleeping, and flooding the logs with warnings if a connection becomes stale or is abruptly closed. We should limit the number of retries (e.g., max 3 attempts) before propagating the exception to prevent infinite loops and log flooding.

                attempts = 0
                while authorized:
                    try:
                        await subscriber.sink(event)
                    except asyncio.CancelledError:
                        raise
                    except Exception:
                        attempts += 1
                        if attempts >= 3:
                            logger.error(
                                "Terminal task event delivery failed after max retries; "
                                "aborting subscription event_id=%s task_id=%s",
                                event.event_id,
                                event.task_id,
                                exc_info=True,
                            )
                            raise
                        logger.warning(
                            "Terminal task event delivery failed; retrying "
                            "event_id=%s task_id=%s (attempt %s)",
                            event.event_id,
                            event.task_id,
                            attempts,
                            exc_info=True,
                        )
                        await asyncio.sleep(self._poll_interval_seconds)
                    else:
                        break

[gemini-code-assist]

If the terminal event delivery fails, the connection is likely dead. We should catch exceptions from _send_terminal_task_event and call manager.disconnect(websocket) to ensure the stale connection is cleaned up and the subscription is detached immediately, preventing futile retries.

    async def send(event: TerminalTaskEvent) -> None:
        if manager.is_connection_registered(websocket, task_id):
            try:
                await _send_terminal_task_event(websocket, event)
            except Exception as e:
                logger.warning(
                    f"Terminal event delivery failed, disconnecting websocket: {e}"
                )
                manager.disconnect(websocket)
                raise

[gemini-code-assist]

If the TaskExecutionCommand instance was already loaded in the session's identity map, SQLAlchemy will return the stale in-memory instance without refreshing its attributes from the database (since synchronize_session=False was used in the update). This can cause command.status to still be 'processing' and raise a ValueError stating that the command is not terminal. Use .populate_existing() on the query to guarantee that the fresh, updated state of the command is always read from the database.

    snapshot = (
        db.query(TaskExecutionCommand, Task)
        .join(Task, Task.id == TaskExecutionCommand.task_id)
        .filter(TaskExecutionCommand.id == command_db_id)
        .populate_existing()
        .one_or_none()
    )

References

1. When querying a database row by primary key in SQLAlchemy, do not replace db.get(Model, id) with db.query(Model).filter(Model.id == id).first() to prevent stale attribute reads. If the row still exists, first() will still return the same identity-mapped instance without refreshing its attributes. To force a refresh of stale attributes, use populate_existing() or an explicit refresh/expire.