fix(agent-bus): establish a dispatched session task, and cancel a round cleanly

[zhanghanduo]

Summary

Found while migrating FrontierAgent onto the shared agent-bus kernel (ApodexAI/FrontierAgent#103): the product's _dispatch_session_task ended with an await asyncio.sleep(0) that the shared implementation does not have. Restoring it turned out to expose two real defects in cancel_for_parent.

1. A dispatched session task was not guaranteed to have started

submit_task_to_session is fire-and-return by contract — a host tool hands back a job id and its invocation scope closes. But nothing on the dispatch path is guaranteed to suspend: _emit_session_task_submitted returns before awaiting when no sink is installed, and a sink whose append never awaits does not yield either. So the coroutine could still be unstarted when the caller moved on.

Consequences:

• a host reporting job state right after dispatch (task board, status line) described a worker that had not begun as merely "submitted";
• every cancellation in that window depended on _mark_job_aborted standing in for a finally the coroutine never reached.

One trailing yield establishes the task, so cancellations land inside the job wrapper on their own.

2. cancel_for_parent started the tasks it was tearing down

Cancelling a running job runs its finally, which drains that session's queue. In a single interleaved pass that dispatched the very tasks the call was cancelling — each reached the LLM before being cancelled again a few iterations later. Queued jobs are now cleared in a first pass, so the drain finds nothing to start.

3. cancel_for_parent under-reported what it cancelled

The count only incremented for jobs that still needed the stand-in. A job that had actually started absorbs its own cancellation (_run_and_finalize flips the status itself), so a round that really did tear down every running sub-agent returned 0.

This is also why test_cancel_for_parent_clears_running_and_queued was passing: all three of its jobs had never started, which is only true without the yield in (1). Reproducible on main without any change to the bus — let the first job start with one await asyncio.sleep(0) before cancelling and the existing assert cancelled == 3 fails while every job still ends up correctly aborted.

Tests

tests/test_agent_bus_session_dispatch_handoff.py, both verified red without the corresponding fix:

• test_dispatch_leaves_the_job_running_not_merely_submitted — 'submitted' == 'running' without the yield;
• test_cancelling_a_round_never_lets_a_queued_task_reach_the_llm — ['T1', 'T2', 'T3'] == ['T1'] without the two-pass cancel, plus the count and final statuses.

test_cancel_for_parent_clears_running_and_queued now passes for the right reason rather than by accident.

Validation

ruff check agent_core tests clean, pyright agent_core 0 errors, pytest -q 911 passed, uv build OK.