Move background queuing logic into dedicated container

[akavi]

What does this PR do?

The event queuing logic for background tasks in LlmAgent was getting a bit unwieldy, so I pulled it into a seperate class.

BackgroundQueue[T]
    subscribe(source)  — register an AsyncIterable as a source
    get_nowait() -> List[T]  — drain all buffered items (non-blocking)
    get() -> T | None  — wait for next item; None means all sources done + empty
    is_active -> bool  — True if items buffered or sources still running
    wait() -> None  — wait for all sources to complete; raises on source error

Also: cleaned up some dead code

Type of change

• Bug fix
• New feature
• Breaking change
• Documentation
• Other: Refactor

Testing

Unit tests

Checklist

• I have read the contributing guidelines
• I have added tests that prove my fix is effective or that my feature works
• I have formatted my code with make format

---

Note

Medium Risk
Refactors background tool event delivery and cancellation behavior in LlmAgent, which is concurrency-sensitive and could affect loopback timing or dropped events if edge cases are missed.

Overview
Extracts background tool event buffering into a new BackgroundQueue that runs each async source as its own task, preserves event ordering (including queued errors), and is cancellation-safe for consumers.

Updates LlmAgent to replace the prior asyncio.Queue + chained _background_task + _maybe_await_background_event() logic with BackgroundQueue (subscribe/get/get_nowait/is_active/wait) and waits for background sources in cleanup().

Removes dead History._append_local_with_event_id code and adds focused unit tests for BackgroundQueue, plus tightens an existing cancellation regression test to wait deterministically for background completion via agent._background_queue.wait().

^{Written by Cursor Bugbot for commit ad9d63e. This will update automatically on new commits. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

[cursor]

wait() doesn't remove error from buffer, causing double-raise

Medium Severity

The wait() method scans _buffer with a for loop and raises on the first _Error found, but never removes that _Error from the buffer. After wait() raises, the same error persists in the buffer and will be raised again by subsequent get() or get_nowait() calls. This is exactly the concern raised in the PR discussion — the error needs to be consumed or cleared after being raised to avoid duplicate error propagation.

 

[cursor]

Cancelling wait() cancels source tasks via gather

Medium Severity

The wait() method uses asyncio.gather() to await source tasks. If wait() is cancelled, gather() propagates the cancellation to all source tasks, destroying the in-flight background work. The old code used asyncio.shield() to protect background tasks from cancellation. This regression means cancelling cleanup() (e.g., during CallEnded handling) now kills background tool execution instead of letting it complete.

Additional Locations (1)

• line/llm_agent/llm_agent.py#L720-L721