feat: add durable task command transport

[qinxuye]

Summary

• add a durable database-backed inbox for message, pause, resume, and cancel commands
• commit commands before acknowledging clients, with stable idempotency keys and strict per-task FIFO ordering
• route commands to the active task-lease owner, then allow another worker to reclaim them after owner/claim expiry
• keep deferred message handoffs pending until the runtime injection is durably dispatched or failed
• recover unfinished commands on worker startup and reject stale control commands after a run-id change
• atomically claim and manage Telegram/Feishu task leases, closing the cross-process race tracked in fix(channels): telegram/feishu task claims bypass the DB-level task lease #856
• add the task-command migration, frontend control-command IDs, and cross-session/crash-recovery regression coverage

Why

P1 established run_id, state_version, and process-local command serialization, but an accepted command could still be lost if its receiving worker exited, and commands received by different workers were not durably ordered. This phase makes command acceptance and ownership database-backed so reconnects, retries, and worker failover preserve the user's intent without duplicate execution or stale-run mutations.

Validation

• 187 focused backend tests covering P1/P2 control state, leases, WebSocket/A2A entry points, migration, ordering, idempotency, and crash recovery
• 8 focused frontend WebSocket tests
• frontend TypeScript type-check
• commit hooks: Ruff, mypy, codespell, Alembic single-head check, frontend lint, and frontend type-check

Closes #856

[gemini-code-assist]

Code Review

This pull request introduces a durable, cross-worker transport for task execution commands, including a new database model, migration, and dispatcher service. It refactors WebSocket handlers and A2A endpoints to enqueue pause, resume, message, and cancel commands durably with idempotency keys, and updates Feishu and Telegram integrations to use a new managed task lease service. The review feedback highlights several critical and medium issues: returning a detached ORM object in load_task_command which can cause runtime errors, a race condition in the dispatcher's wakeup event clearing, missing exception handling in the heartbeat loop, converting None to a string in delivery status loading, and a missing database commit during task cancellation.

[qinxuye]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces a durable, cross-worker transport for task execution commands (such as pause, resume, cancel, and chat messages) to replace process-local serialization. It adds a new task_execution_commands database table, a command dispatcher service, and a managed task lease lifecycle. These changes are integrated into the WebSocket handlers, A2A API endpoints, and the Feishu and Telegram bots. Feedback on the changes suggests adding defensive checks when parsing agent_id from the command payload to prevent unhandled KeyError or TypeError exceptions.

[rogercloud]

Summary

This PR closes #856 by making task-control commands (message / pause / resume / cancel) durable and process-safe instead of best-effort in-memory dispatches. It introduces a DB-backed command inbox (TaskExecutionCommand + migration 20260711_add_task_execution_commands.py), a transport that commits each command with an idempotency key before ack and dispatches them in per-task FIFO order (src/xagent/web/services/task_command_transport.py), and a managed-lease wrapper (src/xagent/web/services/managed_task_lease.py) so Telegram/Feishu inline runs claim task leases through the same DB-atomic predicated UPDATE as the WS/API path. Commands are routed to the current lease owner, become reclaimable after owner/claim expiry, are recovered on worker startup, and are rejected as stale after a run_id rotation. The net effect is that the telegram/feishu check-then-act lease race that allowed cross-process double-execution is replaced by a single atomic claim, and control commands survive a worker crash rather than being silently dropped.

Re-review disclosure: This body layers a fresh, independent Round 0/1/2 pass on top of verification of the 8 prior review passes (6 inline findings from gemini-code-assist) — all 6 are re-confirmed resolved (5 fixed, 1 correctly waived) and recapped at the bottom.

Design verdict: acceptable-with-reservations

The core design is sound and fits the existing architecture: it reuses task_lease_service (adding a new_run predicate rather than a parallel locking mechanism), keeps task_execution_controller as the process-local gate, and layers the durable inbox cleanly on top. The #856 fix is real and proven — both channel bots now claim via the same DB-atomic predicated UPDATE, demonstrated by test_new_run_claim_is_atomic_across_sessions.

Reservations worth the author's attention:

• No general cross-worker fencing. A worker whose claim has expired keeps executing even after another worker reclaims and starts the same command (task_command_transport.py:224-289, :495-561). For the one concrete side effect this could double-apply — a MESSAGE turn injection — the PR is saved by an orthogonal DB unique constraint (uq_task_chat_messages_task_role_turn_id on (task_id, role, turn_id), enforced via claim_user_message_delivery in chat_history_service.py and gated at websocket.py:3735-3738): only the DB-race winner injects. So messages are not corruptible, but the protection is incidental to this PR, not a fencing token. The important nuance is that the same idempotency guarantee does not hold for CANCEL (see Major #2). Recommend confirming PAUSE/RESUME/CANCEL are each idempotent under concurrent replay before relying on the inbox as a general control channel.
• Post-takeover routing has an escape hatch. When a lease is reclaimed after expiry without a run rotation (only runner_id changes, run_id is preserved via coalesce(Task.run_id, ...)), a command routed to the old owner becomes claimable by any third worker through the Task.runner_id != target_runner_id predicate (task_command_transport.py:231-234, task_lease_service.py:116-122), and the run-id staleness recheck cannot catch it because run_id never changed. PAUSE degrades safely here (misrouted pause_execution_by_id just returns False and sends an error, no DB write); CANCEL does not (Major #2).

Findings

Major

1. Deferred MESSAGE can head-of-line-block a task's CANCEL indefinitely — src/xagent/web/services/task_command_transport.py:204-229 (claimable predicate), :449-479 (defer_task_command). defer_task_command resets a deferred MESSAGE back to pending without ever consulting MAX_COMMAND_ATTEMPTS — that cap is only enforced in fail_task_command, never on the defer path. A MESSAGE that keeps returning DELIVERY_PENDING can defer forever, and because the claimable predicate requires all earlier commands for a task_id to be terminal, every later command for that task — including a user's CANCEL — is permanently blocked. Nothing else sweeps or times these out. Recommend bounding the defer path (attempt/time budget) so a stuck delivery cannot wedge the per-task queue.

2. Misrouted CANCEL marks the task terminal in the DB while the real execution keeps running (zombie execution) — src/xagent/web/api/a2a.py:904-939, with the routing predicate at task_command_transport.py:231-234 / task_lease_service.py:116-122. Via the reclaim-without-rotation escape hatch above, a CANCEL can be claimed by a worker C that does not own the running task. On C, background_task_manager.running_tasks is process-local, so cancel_task() no-ops (websocket.py:2481-2482) — but _cancel_task_unserialized never checks whether the in-process cancel actually found and stopped anything; it unconditionally marks the task CANCELED/FAILED and commits. The DB then reports canceled/failed while the true owner's execution continues untouched. This is state corruption, not just a dropped request. Recommend gating the DB status write on the cancel having actually interrupted a local task, or fencing the CANCEL to the confirmed owner.

Minor

3. attempt_count conflates claims with failures — task_command_transport.py:281-283 (increment on every claim), :449-479 (defer never compensates), :428-430 (>= MAX_COMMAND_ATTEMPTS check). claim_task_command bumps attempt_count on every claim including benign defers, but defer_task_command's docstring promises it releases "without consuming the failure budget." Net: a command deferred 4 times with zero real failures, then hitting one genuine error, trips attempt_count >= 5 and is terminally FAILED — contradicting the documented intent. Track claims and failures separately, or decrement on defer.

4. Dead conditional — websocket.py:3203-3205: if not result.payload_matches: return result is immediately followed by return result; both branches return the identical value, so the guard is a no-op. Remove it.

5. Duplicated command-id regex — COMMAND_ID_PATTERN = re.compile(r"[A-Za-z0-9._:-]{1,64}") at task_command_transport.py:39 is re-inlined verbatim as r"[A-Za-z0-9._:-]{1,64}" in _client_message_id at websocket.py:169, with no shared import. Export and reuse the one constant.

6. Detached fire-and-forget dispatch task — dispatch_task_command_promptly (task_command_transport.py:564-580) spawns via create_task and on timeout returns without retaining a reference or an add_done_callback. An exception raised before the coroutine's own try/except (e.g. a DB error in claim_task_command) surfaces only as "Task exception was never retrieved" at GC time, and app.py's shutdown handler awaits only the long-running dispatcher loop, never these ad-hoc tasks. Retain the handle or attach a done-callback that logs.

7. Confirm control-command idempotency under replay (architecture note) — following the fencing discussion above, CANCEL just failed exactly the "is this safe if replayed on the wrong worker?" check that MESSAGE happens to pass. Recommend auditing PAUSE / RESUME / CANCEL each for idempotency/fencing rather than relying on the message path's incidental DB constraint as evidence the channel is safe generally.

Test coverage

Good coverage of idempotency, FIFO serial ordering, deferred retry, startup recovery, detached-snapshot correctness, heartbeat transient-error resilience, the wakeup race, and frontend pause/resume idempotency keys. 82 of 82 targeted new/changed backend+frontend tests pass locally.

Confirmed gaps:

• No test where two workers/threads race to claim the same command row (the lease primitive has test_new_run_claim_is_atomic_across_sessions; the command inbox has no equivalent).
• No test for MAX_COMMAND_ATTEMPTS reaching terminal COMMAND_FAILED.
• No test for a stuck/deferred command head-of-line-blocking a later CANCEL (Major #1).
• No test for a MESSAGE applied against a task whose run_id changed since enqueue.
• No test for two workers genuinely racing the same command after claim expiry (existing expiry+reclaim tests are always sequential).

#856 regression test — PARTIAL. Issue #856 asked for a test that races a channel-bot inline run against a WS/API command on the same task_id. The PR adds test_channel_and_web_claims_are_cross_process_exclusive (tests/web/services/test_task_orchestrator.py), which races acquire_task_lease_isolated(new_run=True) against _begin_turn_atomic_sync under a Barrier(2) + ThreadPoolExecutor and asserts a single winner — a real improvement proving the underlying exclusivity. But it does not drive the actual bot handlers (claim_managed_task_lease in managed_task_lease.py, called from channels/telegram/bot.py:553 and channels/feishu/bot.py:267), so the race is not exercised at the level #856 specifically named.

Testing note (not a required fix): tests/web/api/test_websocket_owner_actor.py::test_running_chat_message_is_persisted_before_resume failed inside a large combined suite but passed in isolation and when its own file ran alone — this looks like pre-existing cross-file test-ordering pollution, not a regression from this PR. Worth a look if the author has capacity.

Simplification opportunities

• src/xagent/web/models/task_command.py:72-73: yagni: task/actor ORM relationships on TaskExecutionCommand are never dereferenced (zero hits across src/ and tests/, no back_populates). Remove both relationship(...) lines.
• src/xagent/web/services/task_command_transport.py:568: yagni: handoff_seconds param on dispatch_task_command_promptly is never passed a non-default at any of its 3 call sites or in tests. Drop the param, inline the 0.05 default.
• src/xagent/web/services/task_lease_service.py:166: yagni: new_run on acquire_task_lease_isolated has no production caller (the real new-run path claim_managed_task_lease calls the non-isolated acquire_task_lease); only a unit test exercises it. Remove new_run from the isolated variant, or delete the untriggered test path.
• src/xagent/migrations/versions/20260711_add_task_execution_commands.py: shrink: ix_task_execution_commands_task_id is a strict prefix of the composite ix_task_commands_task_order (task_id, id) created two lines later, with no UNIQUE/FK requiring the narrower index. Drop the single-column index.

net: -15 to -20 lines possible.

Prior findings status

• 1. Detached ORM object in load_task_command (critical) — FIXED. Now db.expunge(command) before return (task_command_transport.py:~629-643); caller at a2a.py:884-889 reads scalar fields off the detached snapshot; test_load_task_command_returns_an_explicit_detached_snapshot asserts sa_inspect(stored).detached. Test re-run passes.
• 2. Dispatcher wakeup race (medium) — FIXED. _dispatcher_wakeup.clear() now runs unconditionally at loop top (:594) before dispatch_one_task_command (:595); test_dispatcher_does_not_erase_wakeup_during_empty_claim covers it. Verified passing.
• 3. Heartbeat crash on transient DB error (medium) — FIXED. _claim_heartbeat (:340-365) wraps renew_task_command_claim in try/except, warns, and continues; genuine lease loss still stops via explicit renewed=False. test_claim_heartbeat_survives_transient_database_error passes.
• 4. str(None) → "None" for legacy delivery_status (medium) — FIXED. _load_command_message_delivery_status (websocket.py:5677-5695) uses isinstance(..., str) and returns Python None otherwise; test_legacy_delivery_status_preserves_none asserts is None. Passes.
• 5. Missing db.commit() after _cancel_task_unserialized (medium) — WAIVED, correctly. _cancel_task_unserialized (a2a.py:904-939) already commits the same session passed in from execute_durable_task_command's managed block; a second commit is redundant. Author's rejection is correct.
• 6. Missing agent_id validation on cancel payload (medium) — FIXED. websocket.py:5649-5662 now rejects missing/None and non-numeric agent_id with a descriptive ValueError that propagates to fail_task_command (graceful). 3 parametrized regression tests pass.

[qinxuye]

Addressed the actionable findings in 18018e78:

• Bounded deferred commands at 60 claims; an exhausted deferred MESSAGE becomes terminal and releases FIFO so a later CANCEL can proceed.
• Split real failure accounting into failure_count, so benign defers no longer consume the 5-failure budget.
• Routed reassigned commands only to the current live lease owner and added an execution-time foreign-owner recheck before CANCEL can mutate task state.
• Retained and observed timed-out prompt dispatch tasks, including shutdown cleanup and late-exception logging.
• Removed the dead payload branch, reused the shared command-id regex, removed unused ORM relationships, and dropped the redundant single-column task index.
• Added regressions for head-of-line unblock, current-owner takeover routing, foreign-owner CANCEL deferral, same-row concurrent claim exclusivity, run rotation rejection, bounded failures, and late detached-task errors.

I kept acquire_task_lease_isolated(new_run=...): it is the cross-session concurrency seam used by the #856 race regression, even though the production channel helper wraps the non-isolated primitive. A general fencing-token redesign remains broader than this PR; the concrete CANCEL corruption path is now fenced by owner-only routing plus the pre-apply ownership recheck.

Validation: command transport + migration tests, full A2A/owner-actor tests, full task orchestrator tests, pre-commit, SQLite migration upgrade, and PostgreSQL upgrade/idempotence/incremental/downgrade all pass locally.

[rogercloud]

Update summary

Since d0403905, the substantive change landed in 18018e78 ("fix: bound and fence durable command replay"), with a small hardening precursor in a839da93. It touches task_command_transport.py, websocket.py, models/task_command.py, the 20260711_add_task_execution_commands migration, and test_task_command_transport.py. Net effect: the deferred-command retry loop is now bounded (MAX_COMMAND_DEFERS=60) with a separate failure_count/MAX_COMMAND_FAILURES=5 budget; the _claimable_query routing predicate was tightened to Task.runner_id == runner_id and CANCEL is now fenced behind a live-foreign-runner check; several dead/duplicated snippets were removed; fire-and-forget prompt-dispatch tasks are now tracked and awaited on shutdown; and 9 new regression tests were added (including a real concurrent-claim race). a2a.py and task_lease_service.py are unchanged.

Design verdict

Acceptable with reservations (unchanged). The durable command inbox + lease-reclaim model is coherent and #856 is genuinely fixed (acquire_task_lease(new_run=True)'s atomic predicated UPDATE, covered by test_channel_and_web_claims_are_cross_process_exclusive / test_new_run_lease_claim_rejects_a_second_claim). Remaining reservations, now refined:

• There are three overlapping "who owns this" mechanisms (task lease, command claim, message-delivery claim). Their interactions are subtle, but each layer's downstream side effects are independently and atomically re-guarded, so the overlap reads as defense-in-depth rather than a gap.
• Strict per-task FIFO means a slowly-deferring MESSAGE can head-of-line-block a following CANCEL/PAUSE for up to ~60s. This is intentional and tested, but worth flagging as a design trade-off.

Prior findings — status

1. ✅ FIXED — Major, deferred MESSAGE HOL-blocks CANCEL. defer_task_command now bounds retries via MAX_COMMAND_DEFERS=60; exhaustion goes terminal COMMAND_FAILED, unblocking later commands. Covered by test_deferred_message_eventually_fails_and_unblocks_cancel. (src/xagent/web/services/task_command_transport.py)
2. ✅ FIXED — Major, misrouted CANCEL zombie execution. _claimable_query's escape hatch replaced with Task.runner_id == runner_id, plus a task_has_live_foreign_runner() fence on the CANCEL branch. Covered by test_reassigned_command_routes_only_to_the_current_live_owner, test_cancel_command_defers_on_a_live_foreign_owner, test_live_foreign_runner_is_rechecked_before_cancel. Narrow residual: a heartbeat lagging past lease TTL while still executing (see new finding A). (task_command_transport.py, src/xagent/web/api/websocket.py)
3. ✅ FIXED — Minor, attempt_count conflates claims with failures. New failure_count column tracked independently; fail/defer now use separate budgets. Covered by test_real_failures_use_a_separate_bounded_budget, test_deferred_handoff_retries_without_consuming_failure_budget. (src/xagent/web/models/task_command.py)
4. ✅ FIXED — Minor, dead conditional. The redundant if not result.payload_matches: return result guard was removed. (websocket.py)
5. ✅ FIXED — Minor, duplicated command-id regex. websocket.py now imports COMMAND_ID_PATTERN from task_command_transport. (websocket.py)
6. ✅ FIXED — Minor, detached fire-and-forget dispatch task. Timed-out tasks retained in _prompt_dispatch_tasks with a done-callback; stop_task_command_dispatcher cancels and awaits them. Covered by test_prompt_dispatch_observes_late_task_failure. (websocket.py)
7. ⚠️ PARTIALLY ADDRESSED (now a narrower note) — control-command idempotency under replay. The post-takeover routing hatch is now closed for all kinds via the _claimable_query fix; CANCEL adds an explicit execution-time re-check, while PAUSE/RESUME/MESSAGE rely on the tightened predicate plus their own independent atomic downstream guards. The asymmetry (only CANCEL has an explicit foreign-runner check) is real but each path has its own atomic guard, so this is defense-in-depth, not a gap. Non-blocking: document the asymmetry so a future reader doesn't mistake it for an oversight.

New findings

Major

A. Non-atomic read-modify-write in fail_task_command / defer_task_command causes a real lost-update race
src/xagent/web/services/task_command_transport.py (fail_task_command ~404–445, defer_task_command ~448–493)
Unlike claim_task_command / finish_task_command / renew_task_command_claim (single predicated UPDATE ... WHERE id AND status AND claimed_by + rowcount check), these two do .filter(...).first() then mutate the ORM object and db.commit(). Verified with a repro: SQLAlchemy keys the flush UPDATE on primary key only (no version_id_col on TaskExecutionCommand), so it does not re-assert the status/claimed_by seen at SELECT. If worker A's claim expires and worker B reclaims the row before A commits, A's commit unconditionally overwrites the row back to pending/FAILED, silently erasing B's claim while B may still be processing — a real duplicate-processing bug, not a benign updated_at race. Pre-dates 18018e78 (introduced with the file in 0e971e62) but in-scope since the file is wholly new to this PR.
Suggestion: convert both to the same single predicated-UPDATE-with-rowcount pattern used by claim/finish/renew.

B. A CANCEL that reaches terminal COMMAND_FAILED permanently blocks all future cancels for that run
src/xagent/web/api/a2a.py (cancel_task, command_identity construction ~853–901)
The identity cancel:{task_id}:{run_id} is stable for a run's lifetime. Once the command goes terminal COMMAND_FAILED (defer-budget or MAX_COMMAND_FAILURES exhaustion), a re-POSTed cancel matches idempotency (payload_matches=True, no 409), but _claimable_query only matches PENDING or expired-claim PROCESSING rows — never FAILED — so it is never reclaimed. The polling loop then deterministically returns HTTP 500 for every subsequent cancel on that run, with no client recovery path.
Suggestion: mint a fresh command_identity per attempt once terminal-failed, or add a path to re-enqueue/reset a terminal-failed CANCEL.

Minor

C. _cancel_task_unserialized can force-FAIL a task that just legitimately completed
src/xagent/web/api/a2a.py:904–939
Task completion does not rotate run_id (only begin_turn/new-run paths do), so the expected_run_id optimistic check shared by the completion-status write and the forced-cancel write cannot distinguish "just completed" from "still running" — last commit wins. Pre-existing (the old in-process lock never covered the background completion writer either) and the window is milliseconds; not worsened by this PR. Worth tracking alongside B since both touch this function.

D. Defer budget shares attempt_count with benign reclaims
src/xagent/web/services/task_command_transport.py
claim_task_command increments attempt_count on every successful claim (including non-defer reclaims after crash/graceful-shutdown), while defer_task_command checks the same attempt_count against MAX_COMMAND_DEFERS. New in 18018e78 — the failure_count separation applied to the fail path was not applied to defer. Low real-world likelihood (hitting 60 reclaims purely from churn needs sustained crash-looping) but a genuine inconsistency.
Suggestion: track a separate defer_count, mirroring failure_count.

E. _load_command_actor is required unconditionally, including for CANCEL which never uses it
src/xagent/web/api/websocket.py (execute_durable_task_command)
actor_user_id is ON DELETE SET NULL, so a deleted actor makes CANCEL fail/retry/terminal-fail even though CANCEL never reads the actor. Reachable: an admin can cancel another user's task (acting-principal ≠ owner is explicitly supported); if that admin is deleted while the CANCEL is queued/deferred/replayed, the command is blocked purely by the missing actor lookup.
Suggestion: load the actor conditionally, only for kinds that use it (MESSAGE/PAUSE/RESUME).

F. claim_task_command's confirming UPDATE doesn't re-assert the routing predicate
src/xagent/web/services/task_command_transport.py
_claimable_query checks routing/ownership, but the follow-up UPDATE only checks id + status. Verified NOT independently exploitable today: PAUSE no-ops without a local adapter; RESUME is gated by task.status + the controller's atomic expected_run_id transition; MESSAGE falls through to acquire_task_lease's fully-repredicated atomic UPDATE.
Suggestion (defense-in-depth only): have the confirming UPDATE join Task and repeat the routing predicate, for robustness against future changes that remove a downstream guard.

G. Durably-accepted commands that fail via unhandled exceptions or defer-budget exhaustion produce no client-visible signal
src/xagent/web/api/websocket.py; task_command_transport.py ~471–489
The busy/not-found/validation-error branches in _handle_chat_message_unserialized do broadcast agent_error/flip task status (despite the intentional commit-before-ack suppression of the direct delivery-ack), but the catch-all except Exception handlers and defer_task_command's terminal-exhaustion branch produce neither a broadcast nor a status flip.
Suggestion: add a broadcast (or at least a terminal status flip) in these two silent paths for parity with the other failure branches.

Trivial / fragility note

H. The User object passed into command handlers is explicitly db.expunge'd (detached)
src/xagent/web/api/websocket.py (_load_command_actor)
Every current access (directly and via get_agent_for_task/create_default_tools/WebToolConfig/SkillScopeContext) only reads scalar columns (.id, .is_admin), so there is no reachable DetachedInstanceError today. Flag only so a future caller doesn't dereference a lazy relationship on this object.

Other trivial nits

• _prompt_dispatch_tasks is a bare module-global set with no event-loop key; harmless today since stop_task_command_dispatcher clears it, but relevant if multiple loops are ever in play.
• defer_task_command unconditionally calls notify_task_command_dispatcher() even though claim_expires_at is 1s in the future — a harmless spurious wakeup.
• enqueue_task_command for pause/resume stores command_id redundantly inside the JSON payload as well as the dedicated column.
• _client_message_id(...) is recomputed 5 times within handle_chat_message — cosmetic.
• The alembic test's negative assertion ("ix_task_execution_commands_task_id" not in indexes) is an unusual but consistent style — no issue.

Test coverage

Prior gaps:

• ✅ Two workers racing the same command row → FILLED (test_same_command_row_has_a_single_concurrent_claim_winner, real threads + independent DB sessions + threading.Barrier, asserts a single winner).
• ✅ Terminal COMMAND_FAILED via the failure budget → FILLED (test_real_failures_use_a_separate_bounded_budget).
• ✅ Stuck deferred command HOL-blocking a later CANCEL → FILLED (test_deferred_message_eventually_fails_and_unblocks_cancel).
• ✅ Two workers racing the same command after claim expiry → FILLED by the concurrent-claim test above.
• ⚠️ MESSAGE applied against a task whose run_id changed since enqueue → STILL PARTIALLY OPEN. test_command_is_rejected_after_run_rotation covers PAUSE, and the rejection path (websocket.py: if command.kind != TaskCommandKind.MESSAGE and command.target_run_id is not None: raise TaskCommandRejected(...)) deliberately excludes MESSAGE (it is treated as applying to a new turn). This is an intentional branch, but MESSAGE's specific run-rotation behavior remains untested. Follow-up test recommended, not blocking.

All 22 tests in tests/web/services/test_task_command_transport.py pass (.venv/bin/python -m pytest tests/web/services/test_task_command_transport.py -v → 22 passed).

Simplification opportunities

Applied since last review: task/actor unused ORM relationships removed; unused handoff_seconds param removed (hardcoded 0.05s); redundant ix_task_execution_commands_task_id index dropped.

Still open (non-blocking):

• new_run param on acquire_task_lease_isolated (src/xagent/web/services/task_lease_service.py, unchanged since last review). The only production caller (task_orchestrator.py) never passes it; the real new-run path (managed_task_lease.py) uses the separate non-isolated acquire_task_lease. Only a unit test exercises new_run=True on the isolated variant. Removing the param plus its dead branch would drop roughly 10–15 lines.

net: -10 to -15 lines possible.

A fresh simplification-lens pass over the full current diff found no new over-engineering beyond this carried-over item.

[qinxuye]

Addressed the actionable findings in 4ff94da3:

• fenced claim confirmation and every claim-generation mutation (heartbeat, finish, fail, defer) with the current attempt_count, including a stale same-runner reclaim regression
• reset terminal-failed A2A CANCEL commands on an explicit retry while preserving the stable idempotency key
• re-read task state after background cancellation so a concurrent legitimate completion is not overwritten
• split defer_count from claim/failure counters and migrated/tested the new column
• made actor loading conditional so CANCEL survives actor deletion
• broadcast only final unhandled/defer-exhaustion transport failures, avoiding false alarms for intermediate retries
• added explicit coverage that committed MESSAGE intent is recovered across run rotation

Also tightened finish/heartbeat with the same generation fence as fail/defer; otherwise an old handler sharing the same process runner id could still mutate or renew a reclaimed row.

Validation: 27 command-transport tests, A2A + migration tests, related lease/orchestrator/owner-actor/startup tests, and the full pre-commit suite all pass. upstream/main is unchanged at c4d16fde, and the branch is rebased on it.

The detached-user note and the remaining cosmetic/simplification nits are non-blocking and do not justify widening this PR.

[rogercloud]

Re-review — commit 4ff94da3 (follow-up round 2)

Summary

This round adds exactly one new commit, 4ff94da3 "fix: close durable command retry races", on top of the previously-reviewed 18018e78. It hardens the durable command-transport layer against a set of concurrency and terminal-state defects raised in the last review: fail_task_command/defer_task_command now use predicated, snapshotted optimistic-concurrency UPDATEs (no more read-modify-write lost updates); a new retry_failed_task_command helper lets a2a.py's cancel_task recover a terminal-COMMAND_FAILED row instead of returning HTTP 500 forever; a dedicated defer_count column (model + migration) decouples the defer budget from attempt_count/failure_count; the CANCEL path skips the actor load; claim_task_command's confirming UPDATE re-asserts the full routing/availability predicate; and durable executor failures now broadcast a client-visible agent_error. It also adds an a2a.py +21 change and reworks websocket.py control-command handling. Tests are expanded (test_a2a_api.py +124, test_task_command_transport.py +225).

Design verdict

Sound (acceptable-with-reservations) — unchanged from the prior round. The core design correctly fixes #856: Telegram/Feishu now claim via the same DB-atomic predicated UPDATE as WS/API (acquire_task_lease(new_run=True)). Carried-forward concerns:

1. Dual serialization models. The durable executor runs pause/resume/message/cancel without the task_execution_controller.command() process-lock, relying entirely on DB single-claim + expected_run_id optimistic guards; the legacy path still holds the process-lock. Correctness now hinges on every handler using run-id-guarded transitions — defensible, but the responsibility is scattered.
2. Strict per-task FIFO. A deferring MESSAGE can still block a later PAUSE/CANCEL for the same task for up to ~60s. Intentional and tested, but a real UX trade-off worth documenting.
3. Cross-worker dispatch latency floor. notify_task_command_dispatcher() is in-process only; a command routed to a different worker waits for that worker's 0.5s idle poll. Minor, currently undocumented.

Prior findings — status

1. ✅ Major A — FIXED. fail_task_command/defer_task_command now snapshot failure_count/defer_count/attempt_count then issue a single predicated UPDATE re-asserting status, claimed_by, and the snapshotted counters (rowcount==1 check) — matching the existing claim/finish/renew pattern. New test test_stale_attempt_cannot_mutate_reclaimed_command covers it.
2. ✅ Major B — FIXED. New retry_failed_task_command atomically resets a terminal-FAILED row to PENDING (clearing failure_count/defer_count/error/claimed_by), guarded by WHERE id=... AND status==COMMAND_FAILED. a2a.py's cancel_task calls it when the enqueue lookup finds an existing FAILED row for the same identity. New test test_cancel_retries_a_previous_terminal_transport_failure confirms 200/TASK_STATE_CANCELED instead of 500.
3. ⚠️ Minor C — PARTIAL. _cancel_task_unserialized now adds db.expire(task) + db.refresh(task) + a second terminal-status check immediately after await background_task_manager.cancel_task(...), closing the main (whole-async-operation) race window. The root cause — task completion not rotating run_id, so expected_run_id can't distinguish "just completed" from "still running" — is not fixed. A much smaller synchronous window remains between the refresh/recheck and the final apply_task_control_transition UPDATE, now exploitable only by a genuinely concurrent writer landing in that few-line gap. Pre-existing, not worsened; residual risk materially smaller but not architecturally closed.
4. ✅ Minor D — FIXED. New independent defer_count column (model + migration); defer_task_command tracks/checks its own defer_count against MAX_COMMAND_DEFERS, fully decoupled from attempt_count/failure_count. Tests test_deferred_message_eventually_fails_and_unblocks_cancel, test_real_failures_use_a_separate_bounded_budget, test_failed_command_can_be_reset_for_explicit_retry assert the independence.
5. ✅ Minor E — FIXED. Actor load is now gated behind if command.kind != TaskCommandKind.CANCEL, so a deleted actor no longer blocks CANCEL.
6. ✅ Minor F — FIXED (hardened beyond the ask). claim_task_command's confirming UPDATE now joins Task via an exists() subquery repeating the full _command_routing_predicate, and additionally re-checks _claim_availability_predicate and ~_unfinished_earlier_command() — previously only checked at the _claimable_query read.
7. ✅ Minor G — FIXED. execute_durable_task_command is now a thin wrapper that, on terminal failure/defer-exhaustion (failure_count+1 >= MAX_COMMAND_FAILURES / defer_count+1 >= MAX_COMMAND_DEFERS), broadcasts an agent_error via _broadcast_terminal_command_error before re-raising, so the underlying terminal transition still happens but the client is notified.
8. ℹ️ Trivial H — STILL ACCURATE (unchanged). _load_command_actor is byte-for-byte unchanged; the detached (db.expunge'd) User remains a latent risk, but all access is still scalar-only (.id, .is_admin). The only change is its call site becoming conditional (see Minor E), which doesn't affect this note.

New findings

Major

I. MESSAGE commands without client_message_id break the durable delivery guarantee.
src/xagent/web/api/websocket.py — _enqueue_websocket_task_command synthesizes command_id = f"{kind.value}:{uuid.uuid4()}" when the caller omits client_message_id (optional server-side; no validation requires it). Separately, _handle_chat_message_unserialized derives turn_id = client_message_id or str(uuid.uuid4()) — an independent, freshly-generated UUID unrelated to the command's command_id. The retry recheck _load_command_message_delivery_status(command.task_id, command.command_id) looks up TaskChatMessage.turn_id == command.command_id, which is never written (the row's actual turn_id is the other UUID), so the lookup always returns None. Consequence: a genuinely PENDING/FAILED delivery is silently treated as success (defeating the "keep deferred handoffs pending" guarantee), and each retry regenerates yet another turn_id, inserting a duplicate TaskChatMessage row per attempt. Blast radius is limited today because the official frontend (use-websocket.ts) always sends client_message_id; this only affects non-frontend WS API callers that omit it.
Suggestion: thread the enqueue-time command_id/resolved_command_id through as the turn_id (or require the two to be the same value), so delivery-status lookups match the row they check.

II. Pause/resume handlers swallow RuntimeError, so failed control commands are silently recorded as COMPLETED.
src/xagent/web/api/websocket.py — _handle_pause_task_unserialized and _handle_resume_task_unserialized both catch RuntimeError (reporting to the socket) without re-raising; only a bare Exception is re-raised. StaleTaskRunError (raised by task_execution_controller.transition(...) on a stale-run race, reachable from both pause and resume) is a subclass of RuntimeError. Since the handler returns normally, _execute_durable_task_command falls through to a static success-shaped result dict, and the dispatcher's else branch calls finish_task_command, marking the command COMPLETED — even though the transition never applied. There is no alternate signal the handler emits, and finish_task_command never inspects result for an embedded failure marker. This directly undermines the "durable, retriable control commands" guarantee for PAUSE/RESUME.
Suggestion: either stop swallowing RuntimeError in these two handlers (let it propagate so the dispatcher fails/retries the command), or have the handler return an explicit success/failure signal that _execute_durable_task_command checks before calling finish_task_command.

Minor

III. Backstop dispatcher loop is strictly serial per worker.
src/xagent/web/services/task_command_transport.py — run_task_command_dispatcher's main loop awaits dispatch_one_task_command(executor) to completion (no create_task) before claiming the next command. A slow command (e.g. a CANCEL awaiting background_task_manager.cancel_task) stalls recovery of every other task's commands on that worker. Live traffic is unaffected — dispatch_task_command_promptly (the WS-triggered path) spawns a detached per-command task — but a startup backlog or a hung cancel serializes all backstop-path recovery.
Suggestion: spawn each recovered command as its own task (with bounded concurrency) rather than awaiting serially in the backstop loop.

IV. A2A cancel maps a benign stale-run rejection to HTTP 500.
src/xagent/web/api/a2a.py cancel_task (unchanged by this commit) — when a run rotates between enqueue and dispatch, execute_durable_task_command raises TaskCommandRejected (a deliberate "no-op, run already ended" signal), caught alongside genuine internal exceptions and marked COMMAND_FAILED with no field distinguishing "rejected because stale" from "failed for real." cancel_task's polling loop maps any COMMAND_FAILED uniformly to a2a_error("internal_error", ..., 500). So a single first-attempt cancel against an already-ended run gets a misleading 500 instead of an idempotent success/409. Distinct from Major B (a stuck terminal-FAILED row blocking all future attempts); this is the first attempt's response code being wrong on a benign outcome.
Suggestion: have the polling loop (or TaskCommandRejected handling) map stale-run rejections to a 200/409 "already ended" response rather than 500.

V. Fragile string-literal coupling between delivery-status and command-status constants.
src/xagent/web/api/websocket.py — handle_chat_message branches on enqueued.status == DELIVERY_FAILED, but enqueued.status is populated from TaskExecutionCommand.status (the COMMAND_* family) while DELIVERY_FAILED is defined independently in chat_history_service.py as part of the unrelated DELIVERY_* family. The comparison only works because both constants happen to equal the literal "failed"; if either family's literal changes independently, the comparison silently stops matching (no type error), misrouting the retry-prompt UX.
Suggestion: compare against COMMAND_FAILED (not DELIVERY_FAILED), or add a comment pinning the invariant.

(No new over-engineering beyond what's tracked: the detached-User risk and the _canonical_payload json.dumps(default=str) hash were both examined and dropped — the former duplicates Trivial H with all consumers verified scalar-only, the latter is theoretical with no reachable non-JSON-safe payload shape.)

Test coverage

All 53 tests in tests/web/services/test_task_command_transport.py + tests/web/api/test_a2a_api.py pass (full run, 100% green, exit code 0). The new/changed tests directly exercise the fixed findings: test_stale_attempt_cannot_mutate_reclaimed_command (A), test_cancel_retries_a_previous_terminal_transport_failure (B), test_deferred_message_eventually_fails_and_unblocks_cancel, test_real_failures_use_a_separate_bounded_budget, test_failed_command_can_be_reset_for_explicit_retry (D).

Open gaps (not newly introduced by this commit):

• No test for a MESSAGE command lacking client_message_id (relates to new Major I).
• No test asserting a swallowed pause/resume RuntimeError still correctly fails/retries the durable command (relates to new Major II).
• MESSAGE behavior under run-rotation remains untested — test_command_is_rejected_after_run_rotation covers PAUSE only (carried forward).

Simplification opportunities

• src/xagent/web/services/task_command_transport.py — drop EnqueuedTaskCommand.created. Set at 4 construction sites, zero production reads (only asserted in one test). (~a few lines)
• src/xagent/web/services/task_command_transport.py — collapse defer_task_command/fail_task_command. Now near-duplicates (same snapshot query, expected_attempt_count guard, optimistic-count UPDATE, commit, notify); only the counter column, terminal threshold, backoff expression, and error text differ. Parameterize into one _release_claim(counter_col, max_count, backoff, ...) helper. (~40-50 lines)
• Carried forward: remove unused new_run param on acquire_task_lease_isolated in task_lease_service.py. No production caller; only exercised by a unit test. (~10-15 lines)

net: -60 lines possible

[qinxuye]

Addressed the new review in 1ada58bb (rebased onto current upstream/main):

• MESSAGE now uses the durable command id as client_message_id/turn id whenever the client omits or supplies an invalid id, so delivery lookup and retries address the same persisted row.
• PAUSE/RESUME now propagate retryable RuntimeError failures instead of being recorded as completed. Explicit handled failures also set a durable rejection marker, so unsupported/invalid control operations cannot silently complete.
• The recovery dispatcher now uses a bounded four-worker pool, preserving per-task FIFO in the DB while allowing unrelated task backlogs to recover concurrently.
• Stale-run rejection is persisted as structured command result metadata; A2A cancel maps it to 409 with retry guidance instead of a misleading 500.
• Command status checks now use COMMAND_FAILED, not the unrelated delivery-status constant.
• Closed the prior cancellation residual as well: A2A cancellation now fences its terminal write with both run_id and state_version, preventing a completion committed after the refresh from being overwritten.

Regression coverage includes missing message ids, pause/resume stale-run propagation, structured stale rejection + A2A mapping, concurrent recovery of unrelated tasks, and stale state-version rejection. Focused command/A2A/owner/controller tests, related lease/orchestrator/startup/migration tests, and full pre-commit all pass.

I did not apply the suggested helper/field removals: they are non-blocking cleanup and would widen an already concurrency-sensitive PR without improving correctness.

[qinxuye]

CI follow-up: Pytest Fast (web) exposed a timing-only assertion in test_running_chat_message_is_persisted_before_resume. With the bounded recovery pool, a background worker may legitimately claim the durable command before the request-local prompt dispatcher, so handle_chat_message returning is not a completion barrier. dc1f3ebc updates the test to await the durable dispatch event instead of asserting synchronously.

The affected owner-actor file passes locally with the CI xdist settings (-n 4 --dist=loadscope). The rerun is now fully green: PR checks (including 2,091 web tests), SQLite/PostgreSQL migrations, and Real RAG all passed.

[rogercloud]

Round 3 Re-review — commits 1ada58bb, dc1f3ebc

Summary of this round's update

This update closes out the remaining open items from round 2. Commit 1ada58bb hardens durable command recovery: it threads an expected_state_version CAS through apply_task_control_transition so a cancel can no longer silently overwrite a task that legitimately completed in the refresh→UPDATE window; it makes durable pause/resume raise on RuntimeError/StaleTaskRunError instead of swallowing the failure (so failed control commands are no longer falsely marked COMPLETED); it reworks the backstop dispatcher from a strictly-serial loop into a bounded worker pool (DISPATCHER_CONCURRENCY = 4) while preserving per-task FIFO via an in-claim _unfinished_earlier_command() re-check; it unifies MESSAGE command_id/turn_id so delivery-status lookups match; and it tags stale-run rejections with a reason so the A2A cancel path returns a 409 retry hint instead of a generic 500. Commit dc1f3ebc stabilizes the new tests by waiting for durable dispatch. Net: 5 previously-open items (both prior Majors plus 3 Minors) are now resolved.

Design verdict

Sound. Every open round-2 item is closed with minimal, well-targeted mechanisms that reuse existing primitives (state-version CAS, reason-tagged rejections, command_id/turn_id unification, RuntimeError propagation, worker-pool dispatch) rather than inventing new machinery. The one caveat: tightening pause/resume failure semantics (Major II's fix) removed a safety net that used to accidentally paper over a specific lease-expiry race, and the retry-worthy task_has_live_foreign_runner defer guard that CANCEL has was never extended to pause/resume — a real, narrow gap (new Major below), not a flaw in the overall approach.

Prior findings status

1. Major A (lost-update in fail/defer_task_command) — FIXED, unchanged. Snapshot + predicated-UPDATE pattern untouched; test_stale_attempt_cannot_mutate_reclaimed_command passes.
2. Major B (terminal COMMAND_FAILED permanently 500ing cancel) — FIXED, unchanged. retry_failed_task_command and its a2a.py call site untouched; test_cancel_retries_a_previous_terminal_transport_failure passes.
3. Minor C (cancel race between refresh and final UPDATE) — NOW FULLY FIXED. apply_task_control_transition in src/xagent/web/services/task_execution_controller.py gains expected_state_version; _cancel_task_unserialized in src/xagent/web/api/a2a.py passes expected_state_version=int(task.state_version or 0) (captured right after db.refresh(task)) into the final transition, so a legitimate completion in the window bumps state_version via release_task_lease/release_current_runner_task_lease and the CAS raises StaleTaskRunError instead of overwriting. Covered by test_transition_rejects_a_stale_state_version and test_cancel_maps_stale_run_rejection_to_conflict. Residual minor caveat: when _cancel_task_unserialized raises StaleTaskRunError directly (not via the durable-command replay path) it is not tagged reason="stale_run", so it surfaces as a generic 500 internal_error rather than the 409 invalid_request the durable path now returns for the equivalent condition — a two-path inconsistency, not a correctness bug.
4. Minor D (defer_count coupled to failure_count) — FIXED, unchanged. Independent defer_count untouched; defer/real-failure tests pass.
5. Minor E (actor load blocking CANCEL for deleted actor) — FIXED, unchanged. CANCEL-gated load untouched.
6. Minor F (claim_task_command confirming UPDATE hardening) — FIXED, unchanged. Hardened claim predicate untouched.
7. Minor G (durable executor failures not broadcast to client) — FIXED, unchanged, holds for the new paths. A re-raised RuntimeError from pause/resume is not caught by the specific TaskCommandDeferred/TaskCommandRejected handlers and correctly falls through to the generic except Exception branch that calls _broadcast_terminal_command_error.
8. Trivial H (detached User from _load_command_actor) — STILL ACCURATE, unchanged. _load_command_actor byte-for-byte identical; still harmless (scalar-only access).
9. Major I (MESSAGE commands without client_message_id break delivery-status lookup) — FIXED. _enqueue_websocket_task_command now backfills payload["client_message_id"] = resolved_command_id for MESSAGE commands when the caller omits/sends an invalid id, so _handle_chat_message_unserialized's turn_id equals the durable command_id and _load_command_message_delivery_status's lookup matches the persisted row. test_chat_without_client_id_uses_durable_command_id_as_turn_id asserts command.payload["client_message_id"] == command.command_id and passes.
10. Major II (pause/resume swallow RuntimeError → falsely COMPLETED) — FIXED, both handlers. _handle_pause_task_unserialized and _handle_resume_task_unserialized now raise after logging/reporting (verified absent in the parent commit), so StaleTaskRunError propagates and the command is retried/failed rather than marked COMPLETED. test_durable_pause_propagates_stale_run_error and test_durable_resume_propagates_stale_run_error pass.
11. Minor III (backstop dispatcher strictly serial per worker) — FIXED. Reworked into a worker pool (DISPATCHER_CONCURRENCY = 4, _run_task_command_dispatcher_worker) with per-task FIFO preserved via _unfinished_earlier_command() re-checked inside the atomic claim UPDATE. test_dispatcher_recovers_unrelated_tasks_concurrently (which would hang under the old serial loop) passes.
12. Minor IV (A2A cancel maps benign stale-run rejection to 500) — FIXED. TaskCommandRejected carries a reason; the stale_run case is tagged and persisted via fail_task_command(result={"rejection_reason": ...}); a2a.py's cancel-polling loop checks rejection_reason == "stale_run" and returns 409 invalid_request. test_cancel_maps_stale_run_rejection_to_conflict passes. (Only covers the durable-replay path — see Minor C's caveat for the direct-exception path.)
13. Minor V (fragile DELIVERY_FAILED vs COMMAND_FAILED string coupling) — FIXED. handle_chat_message now compares against COMMAND_FAILED explicitly instead of the coincidentally-equal DELIVERY_FAILED.

New findings

Major — Pause/resume lack CANCEL's "defer if a foreign runner is still live" guard, so Major II's fix created a new terminal-failure trap.
src/xagent/web/api/websocket.py — In _execute_durable_task_command, the CANCEL branch checks task_has_live_foreign_runner(command.task_id) and raises TaskCommandDeferred (safe retry) before attempting the operation; PAUSE and RESUME have no equivalent check. Before this update, a pause/resume claimed by a non-owner worker during a lease-expiry race failed internally but the bare except RuntimeError (no raise) swallowed it and marked the command COMPLETED — a silent no-op. Now that both handlers raise on RuntimeError (including StaleTaskRunError) and _durable_command_error is set on more paths, a pause/resume that hits the same lease-race window is durably, terminally marked COMMAND_FAILED (force_terminal once retries exhaust) with no defer/retry, unlike CANCEL. In other words, closing "durable commands silently succeed when they didn't" (Major II) removed the net that used to accidentally paper over this race for pause/resume, and the intentional defer path was never extended to them. Net effect: a pause/resume the true owner could have applied a moment later can now be permanently rejected instead of retried.
Suggestion: add the same task_has_live_foreign_runner check (raising TaskCommandDeferred) to the PAUSE and RESUME branches in _execute_durable_task_command, alongside CANCEL's existing check.

Minor — Resume-reservation leak on a stale RESUME_REQUESTED transition is now retried repeatedly instead of failing silently once.
src/xagent/web/api/websocket.py, _handle_resume_task_unserialized — background_task_manager.reserve_resume(task_id) succeeds, then task_execution_controller.transition(..., RESUME_REQUESTED) can raise StaleTaskRunError from a call site that sits outside the try/except BaseException block that releases the reservation via release_resume_reservation. The leak itself pre-dates this update, but this update's raise (Major II's fix) now makes the durable dispatcher retry the failing resume (up to MAX_COMMAND_FAILURES = 5 with backoff), and every retry re-enters the handler, re-calls reserve_resume, and is rejected as "already in progress" because the reservation was never released from the first failure. Previously the leak existed but the single failure was swallowed with no retry storm. Not a new leak, but this update makes its consequences materially worse (5 wasted retries + backoff before terminal, vs one quiet no-op).
Suggestion: widen the try (or add a finally) around the transition(..., RESUME_REQUESTED) call so release_resume_reservation runs on any exception from it, not just exceptions from the block that currently wraps it.

Examined and dropped

• Resume's "already in progress" branch doesn't set _durable_command_error (deduped concurrent resume marked COMPLETED, not FAILED): intentional. This is a single-flight de-dup guard; the in-flight resume that won the race will actually resume the task, so marking the loser COMPLETED correctly reflects that the user's intent is fulfilled. Not comparable to pause's analogous branch — pause has no de-dup mutex, so its False return is a genuine no-op.
• fail_task_command writes result unconditionally, even on the non-terminal retry branch: benign. The only caller that can pass a non-None result (TaskCommandRejected handling) always uses force_terminal=True, and the only caller reachable on the non-terminal branch always passes result=None; no consumer reads result outside COMMAND_FAILED state. Low-priority robustness nit: this holds by convention between the two current call sites, not by a structural guarantee in fail_task_command.

Test coverage

All targeted suites pass — tests/web/services/test_task_command_transport.py, tests/web/api/test_a2a_api.py, tests/web/api/test_websocket_owner_actor.py (new file), tests/web/services/test_task_execution_controller.py — full run, 100% green, exit code 0. New tests directly exercise every fixed finding: state_version CAS, stale_run 409 mapping, pause/resume RuntimeError propagation, concurrent dispatcher recovery, and command_id/turn_id unification.

Gaps: no test for the new pause/resume foreign-runner-deferral gap (new Major) or the resume-reservation-leak-under-retry scenario (new Minor). Both are concurrent/multi-worker edge cases requiring a simulated lease-expiry race, which the suite currently attempts only for cancel (per round 2's carried-forward gap note).

Simplification opportunities

New (this update's diff only):

• src/xagent/web/api/websocket.py — the pause/resume failure strings (e.g. "No live execution found to pause", "Task is not paused and cannot be resumed.") are duplicated across ~6 new branches, written both into message_data["_durable_command_error"] and inline into the send_personal_message payload. Bind each to a local once, or add a small _fail_durable(message_data, error_text) helper reused by all six sites. (shrink, a few lines)
• src/xagent/web/api/a2a.py — the stale-run detection builds an intermediate rejection_reason variable purely for a single comparison; inline to if isinstance(stored.result, dict) and stored.result.get("rejection_reason") == "stale_run":. (shrink, 1-2 lines)

Round 2's carried-forward items remain open, unchanged by this update: drop the unused EnqueuedTaskCommand.created field; collapse the near-duplicate fail_task_command/defer_task_command into one parameterized helper; remove the unused new_run param on acquire_task_lease_isolated.

Estimated net: roughly −15 to −25 lines across the new and carried-forward simplifications combined.

[qinxuye]

Addressed the round-3 findings in 800f3854, rebased onto current upstream/main (41f5867f):

• PAUSE, RESUME, and CANCEL now share the same execution-time live-foreign-runner fence. A command caught in the lease handoff window raises TaskCommandDeferred and remains retryable instead of entering the failure budget on the wrong worker.
• The live-control RESUME reservation now wraps the initial RESUME_REQUESTED transition in the same cleanup scope as task creation/registration. Any exception releases the reservation; rollback to PAUSED/WAITING_FOR_USER only runs when the first transition actually succeeded.
• A direct StaleTaskRunError from the A2A cancel state-version CAS is normalized to TaskCommandRejected(reason="stale_run"), keeping both stale-cancel paths on the same structured 409 mapping.

Regression coverage now includes PAUSE/RESUME deferral under a live foreign owner and verifies reservation release when the initial transition is stale. Command-transport + owner-actor tests, the full related A2A/controller/orchestrator/lease/startup suites, and pre-commit all pass.

The cleanup-only simplification suggestions remain intentionally out of scope.

[rogercloud]

Round 4 Re-Review — #857

Summary of this round's change

Since round 3 exactly one commit landed, 800f3854 "fix: defer misrouted control commands", touching only src/xagent/web/api/websocket.py (+39/-29) plus two test files. It closes round 3's open Major by lifting CANCEL's task_has_live_foreign_runner → TaskCommandDeferred guard out of the CANCEL branch into a single shared pre-dispatch guard covering PAUSE, RESUME, and CANCEL, and deletes CANCEL's now-redundant inline copy. As a result, a pause/resume claimed by a non-owner worker during a lease-expiry race is now deferred (via the defer_count budget) instead of being durably marked COMMAND_FAILED.

Design verdict: Sound

The commit does exactly what round 3 recommended, and slightly better. Rather than copy-pasting CANCEL's guard into the PAUSE and RESUME branches, it consolidates the check into one shared pre-dispatch guard (src/xagent/web/api/websocket.py:5549-5560) over {PAUSE, RESUME, CANCEL} and removes the duplicated CANCEL copy — the correct fix and a net simplification. MESSAGE is correctly excluded, since it has its own DELIVERY_PENDING-based defer path. The guard's placement — after the target_run_id stale-run check and before the actor-dependent handlers — preserves CANCEL's prior ordering.

Round-3 findings status

Major (defer-if-foreign-runner guard missing for pause/resume): FIXED. The shared guard at src/xagent/web/api/websocket.py:5549-5560 now defers PAUSE/RESUME/CANCEL uniformly when task_has_live_foreign_runner is true, raising TaskCommandDeferred. That exception is routed by dispatch_one_task_command (src/xagent/web/services/task_command_transport.py:692-699) to defer_task_command — the defer_count / MAX_COMMAND_DEFERS=60 budget — rather than the terminal failure_count / MAX_COMMAND_FAILURES=5 path that unmapped exceptions hit. The guard runs before the actual PAUSE/RESUME dispatch, not cosmetically. The new parametrized test test_control_command_defers_on_a_live_foreign_owner (tests/web/services/test_task_command_transport.py:336-355, covering both PAUSE and RESUME) plus the existing CANCEL variant pass: 3 passed, 28 deselected.

Minor (resume-reservation leak): FIXED (incidentally, by the same commit). In _handle_resume_task_unserialized (src/xagent/web/api/websocket.py), the first transition(..., RESUME_REQUESTED) call (lines 5398-5402) is now inside the try: block opened at line 5397, guarded by a new resume_snapshot: Any | None = None (line 5395). The except BaseException: handler (line 5414) now unconditionally calls release_resume_reservation(task_id) (line 5417) before re-raising (line 5430), and the resume_snapshot is not None check (line 5418) avoids a NoneType error in the rollback-transition path when the failure occurred on the very first transition. The new assertion bg_mgr.release_resume_reservation.assert_called_once_with(int(task.id)) in test_durable_resume_propagates_stale_run_error (tests/web/api/test_websocket_owner_actor.py:610) proves this directly and passes: 1 passed.

New findings

Minor — residual check-then-act asymmetry between CANCEL and PAUSE/RESUME. The shared guard (src/xagent/web/api/websocket.py:5549-5560) is a single point-in-time check. Between it passing and PAUSE/RESUME's own transition(..., expected_run_id=...) executing (_handle_pause_task_unserialized does user-auth / DB / get_agent_for_task work before its transition at ~line 5210; _handle_resume_task_unserialized similarly at ~lines 5398-5402 / 5436-5440), a foreign runner can still acquire the lease and cause that transition to raise StaleTaskRunError. This commit adds a CANCEL-specific mapping for exactly this residual race (src/xagent/web/api/websocket.py:5601-5608: except StaleTaskRunError as exc: raise TaskCommandRejected(str(exc), reason="stale_run") from exc), but PAUSE/RESUME have no equivalent, so their StaleTaskRunError (a RuntimeError subclass) falls through to the dispatcher's generic except Exception (src/xagent/web/services/task_command_transport.py:712-725), which calls fail_task_command without force_terminal=True.

The practical effect is the opposite of a naive read: CANCEL's mapping to TaskCommandRejected uses force_terminal=True (src/xagent/web/services/task_command_transport.py:700-711), so a CANCEL hitting this residual race terminates immediately on the first occurrence with zero retries. PAUSE/RESUME, lacking the mapping, take the generic non-terminal failure path, which allows up to MAX_COMMAND_FAILURES=5 retries with backoff before going terminal — and because the shared guard will catch the now-stale foreign-runner state on the next attempt and defer properly, PAUSE/RESUME will very likely self-heal within that budget. So the real inconsistency is that CANCEL is stricter (zero-retry termination) while PAUSE/RESUME are more lenient (self-healing retries) — not the reverse. This is not a re-opening of the original terminal-failure Major; it is a narrow, self-healing, low-severity asymmetry, worth cleaning up for consistency (arguably CANCEL's zero-retry termination on a single race hit is the worse UX of the two).

Suggestion: pick one behavior and apply it uniformly — either add the same StaleTaskRunError → TaskCommandRejected(reason="stale_run") mapping to PAUSE/RESUME (all three terminal-on-first-hit for this residual race), or relax CANCEL's mapping to defer via TaskCommandDeferred instead of force_terminal=True (all three self-healing) — rather than leaving CANCEL and PAUSE/RESUME on different footing.

Test flakiness note (not attributable to this commit). One run of tests/web/api/test_websocket_owner_actor.py::test_chat_admin_append_to_other_users_task_claims_as_owner failed intermittently ("Expected mock to have been awaited once. Awaited 0 times") but passed on every subsequent run (isolated and combined). This is a MESSAGE-path test, untouched by the new guard (MESSAGE is excluded from the guard's kind-set), and the parent commit was "test: wait for durable message dispatch" — this looks like pre-existing timing flakiness in durable-message-dispatch tests, not a regression from this commit. Mentioned for visibility only; no action attributed to this PR.

Simplification

The commit is itself a simplification: it consolidates what would have been two more copy-pasted task_has_live_foreign_runner checks (one each for PAUSE and RESUME) into a single shared guard extending the existing CANCEL check, net removing duplication. No further simplification opportunities were found in this diff.

Test coverage

tests/web/api/test_websocket_owner_actor.py + tests/web/services/test_task_command_transport.py: 53 passed (up from 51 on the parent commit — 2 new tests added), reproduced across two clean runs. The full 4-file suite (test_task_command_transport.py, test_a2a_api.py, test_websocket_owner_actor.py, test_task_execution_controller.py) also passes in full, exit code 0.

[qinxuye]

Addressed the intermittent admin-append test race in dcac0e08: the test now keeps the mock active until the durable worker dispatches (with a bounded timeout) before asserting. Production behavior is unchanged. Also rebased onto latest upstream/main; the full test file passes locally (22 tests).