Python: fix(core): Distributed agent orchestration with pluggable BackgroundTaskRuntimeStore - #9040

Motivation & Context

BackgroundAgentsProvider tracks running tasks using only in-process asyncio.Task references. In any multi-replica deployment — where successive turns of the same session can land on different worker processes — this causes several cascading failures described in #8760:

1. Cross-process invisibility: A task running on Process A is invisible to Process B, which immediately marks it LOST with no path to recovery.
2. Sticky LOST state: _refresh_task_state skipped records already marked LOST, so once a task was marked lost, its result was permanently discarded even if the owning process finished successfully.
3. Result never delivered cross-process: There was no mechanism to publish a completed task's result out of the owning process. A non-owning process had no way to read the outcome.
4. No liveness heartbeat: No renewal mechanism existed. A long-running task's owner would be considered dead the moment the lease TTL elapsed.
5. Expired lease not surfaced: A task whose owner process died had no way to surface as LOST to another process.
6. release_session on non-owner is a silent no-op: The early-return guard prevented the cancel signal from reaching the owning process.
7. Wait tool blocks only on local tasks: background_agents_wait_for_first_completion had no store-polling path, so a non-owning process would immediately error instead of waiting.

Description & Review Guide

What are the major changes?

1. BackgroundTaskRuntimeStore protocol (_background_agents.py)

A new @experimental Protocol class with six async methods:

The default _InMemoryBackgroundTaskRuntimeStore reproduces the existing single-process behaviour exactly — no change to deployments that don't supply a store.

2. _run_agent_with_renewal wrapper

Replaces the bare asyncio.create_task(_run_agent(...)) call. Runs the agent coroutine as a sub-task while periodically calling store.renew() at ttl_seconds / 2. Cancellation (local or remote) propagates correctly via BaseException re-raise + work_task.cancel().

3. _refresh_task_state — fixed LOST recovery

The loop guard changed from:

if task_info.status != BackgroundTaskStatus.RUNNING:
    continue

to processing both RUNNING and LOST records. For tasks with no local asyncio.Task, the function now:

1. Calls store.is_alive() — if True, leaves status as RUNNING (cross-process task still healthy).
2. Calls store.fetch_outcome() — if a result is present, finalizes via _finalize_task_from_info().
3. Only falls back to LOST if neither check yields a result.

4. Done-callback via _make_done_callback

Attached to every asyncio.Task at start and continue time. Fires the instant the task settles and schedules store.publish_outcome() via loop.create_task(). The outcome is available to any process on the very next turn.

5. release_session — cancel signal fires before early-return guard

The store.request_cancel() call is now placed before runtime = self._runtime.get(session_id), so a non-owning process (which has no local runtime entry) still propagates the signal to the process that owns the tasks. The call is gated on cancel_running=True so release_session(..., cancel_running=False) does not issue a remote kill.

6. background_agents_wait_for_first_completion — store-polling fallback

Rewritten as a time-bounded poll loop. On each iteration it calls _refresh_task_state (which now queries the store), then either waits on local asyncio.Task objects or falls back to asyncio.sleep(1.0) when no local tasks are present — correctly handling the cross-process wait case.

What is the impact of these changes?

• Single-process deployments: Zero behavioural change. _InMemoryBackgroundTaskRuntimeStore is the default and replicates existing semantics exactly.
• Multi-replica deployments: Supply a shared BackgroundTaskRuntimeStore (e.g. Redis-backed) via BackgroundAgentsProvider(agents, runtime_store=my_store). All six failure scenarios described in [Feature]: BackgroundAgentsProvider cannot tell a task that is running on another process from one whose owner died #8760 are resolved.
• Non-breaking: The new runtime_store kwarg is optional and defaults to the in-memory implementation.

What do you want reviewers to focus on?

• _refresh_task_state loop guard change (the LOST-state recovery fix) — this is the highest-risk edit since it changes which records are processed on every refresh.
• _make_done_callback first-write-wins semantics — ensures a late or duplicate done-callback cannot overwrite an already-committed outcome.
• release_session ordering — request_cancel must fire before the early-return guard; confirm the placement is correct.
• _run_agent_with_renewal cancellation path — verify BaseException catch + work_task.cancel() + await work_task is the right cleanup sequence.

Related Issue

Fixes #8760

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• This is not a breaking change.