[Feature]: BackgroundAgentsProvider cannot tell a task that is running on another process from one whose owner died

[antsok]

Description

Problem. BackgroundAgentsProvider splits a background task into two halves. The record (BackgroundTaskInfo: id, status, description, result_text, error_text) is serialized into session.state, so it travels with the session to whichever process loads it. The runtime (_RuntimeState.in_flight_tasks, the asyncio.Task objects, and _RuntimeState.background_sessions, the child AgentSessions) lives in self._runtime, a process-local dict keyed by session_id. The comment in BackgroundAgentsProvider.__init__ says this runtime "cannot survive process restarts. If the provider instance is lost, _refresh_task_state() marks orphaned tasks as LOST." _refresh_task_state does exactly that for any RUNNING record without a local in-flight task, and then saves the LOST status back into session state. before_run calls _refresh_task_state on every turn, before the model sees anything.

That is correct when the process that started the task is dead. It is wrong when that process is alive and simply is not the one serving this turn, and the provider has no way to tell the two apart. Tasks outliving a run are also a case the provider supports: before_run refreshes "to get accurate statuses for any tasks that completed between turns". So the question is not whether a task can span turns, but whether the next turn lands on the same process.

Our deployment: a web host running two replicas behind a load balancer without session affinity, with conversation state in Redis. Any turn can land on either replica. agent-framework-core: 1.19.0 (tracking main).

Scenarios that fail today

1. Healthy task marked LOST, then duplicated. Turn N runs on process A; the model calls background_agents_start_task, and the run ends while the task is still running (see scenario 3 for how). Turn N+1 is routed to process B. B loads the session, before_run → _refresh_task_state finds the task RUNNING with no entry in B's in_flight_tasks, marks it LOST and persists that. The model's injected status block now reads Task 1 [lost]; background_agents_get_task_results returns "Task state was lost (reference unavailable)." The model does the reasonable thing and starts the same work again as task 2 on B. Meanwhile task 1 is still running on A. Two copies of the same delegated work now run concurrently against the same tools, possibly writing the same files, and the user pays for both.
2. The original's result is orphaned. When task 1 finishes on A, nothing records it: finalization only happens inside a run on A, and on A's next turn for this session _refresh_task_state skips the record because the persisted status is now LOST, not RUNNING (the loop continues on anything that isn't RUNNING). The completed work is never delivered. It sits in A's in_flight_tasks until release_session or process exit.
3. Runs end with live tasks despite the instruction. DEFAULT_BACKGROUND_AGENTS_INSTRUCTIONS says "Always wait for outstanding tasks to finish before you finish processing." That is a request to the model, and several paths end the run no matter what the model intends. (a) A tool that requires approval ends the run with a pending approval request, and the user may answer hours later, typically on a different process. (b) max_function_calls sets tool_choice="none" (_disable_tools_at_function_call_limit), and max_duration_seconds takes the same path, so the model cannot call background_agents_wait_for_first_completion even if it wants to. (c) Provider errors abort the run. (d) The user presses stop. Each of these leaves RUNNING records in session state, and on a multi-instance host each one becomes scenario 1 on the next turn with probability roughly (N-1)/N.
4. background_agents_continue_task fails across processes. Even when a task completed and was finalized on A (so its record is COMPLETED, result_text persisted, and results are readable anywhere), continuing it on B returns "Error: Session for task 1 is no longer available." The child AgentSession exists only in A's background_sessions.
5. Restarts and rolling deploys. Same sequence as scenario 1, except A is gone. Here LOST is the right answer. We are not asking to hide it. The point is that the provider currently reaches the same verdict for "owner dead" (correct) and "owner alive elsewhere" (wrong), because the only signal it has is its own process's memory.
6. release_session is process-local. A host that deletes a conversation calls release_session on whichever process handles the delete. If the tasks run on another process, it finds no runtime and returns without doing anything. The tasks keep running, and keep calling tools, for a conversation that no longer exists.
7. Waiting and listing cannot see other processes. On B, background_agents_wait_for_first_completion([1]) finds nothing in in_flight_tasks, refreshes (marking LOST), and returns "Task 1 is not running; current status: lost." background_agents_get_all_tasks reports the same. B cannot observe a task on A at all, so it cannot even choose to wait for it.

Expected behavior. A host with a shared store should be able to give the provider enough information to (a) mark LOST only when the owning process is actually gone, (b) report a task running on another instance as running, not lost, (c) deliver a task's result from whichever process serves the next turn, and ideally (d) signal cancellation to the owner. Hosts without a shared store keep today's single-process behavior unchanged.

Proposed shape (upstream's call). A small pluggable seam on BackgroundAgentsProvider, e.g. a BackgroundTaskRuntimeStore / liveness protocol with an in-memory default that reproduces current semantics:

• The host provides an opaque qualified session key (namespace + session identity) and the owner process records a lease per (qualified_session_key, task_id) when it starts or continues a task, renews it while the task runs, and clears it on finalization.
• _refresh_task_state asks the store before marking LOST: lease present and fresh → leave RUNNING (or a distinct status such as RUNNING_ELSEWHERE, which the tools render as "running on another instance"); lease missing or expired → LOST as today.
• When a task finishes, the owner writes its outcome (status, result_text/error_text) to the store from a done-callback, not only inside a later run. Any process's _refresh_task_state then picks it up, which fixes scenario 2 and makes wait_for_first_completion able to poll a remote task instead of returning immediately.
• Optionally, a cancel signal the store propagates to the owner, so release_session on one process cancels tasks owned by another (scenario 6).
• Continuing a task on a non-owner (scenario 4) is harder, since it needs the child session. Serializing the child AgentSession into the store on finalization would cover it where the child agent's session is serializable. Otherwise a clear "continue is only available on the instance that ran the task" error is still better than today's generic message.

Nothing here requires the framework to ship a distributed store. A protocol plus the in-memory default is enough; hosts bring Redis or anything else. A reference Redis implementation would fit naturally next to the existing agent-framework-redis package, but that is optional.

Alternatives considered

• Rely on "always wait before finishing". This is an instruction, not enforcement, and scenario 3 lists four paths where the model cannot comply. The provider itself already expects tasks to span turns.
• Enforce it: cancel or await outstanding tasks at run end. Awaiting blocks the run, and so the user, for the full task duration, which defeats the point of background work. Cancelling at an approval interrupt throws away delegated work right before the user approves the step that needed it.
• Session affinity at the load balancer. Fixes scenarios 1, 2, 4 and 7 while every process is up, but breaks exactly on restarts and rolling deploys, and many hosts (serverless containers, some managed ingresses) cannot pin reliably. It also does nothing for scenario 6 when the delete request lands elsewhere.
• Durable execution (Durable Agents). Solves durability for a different hosting model and is much heavier than a liveness lease. It does not integrate with BackgroundAgentsProvider, which is what harness-based hosts use.
• Host-side wrapper. Possible, but only by reaching into the private _runtime / _RuntimeState / _refresh_task_state, which every framework bump can break. The decision point (_refresh_task_state choosing LOST) is inside the provider, so the seam belongs there.

Code Sample

from typing import Protocol

class BackgroundTaskRuntimeStore(Protocol):
    async def acquire(self, qualified_session_key: str, task_id: int, *, ttl_seconds: float) -> None: ...
    async def renew(self, qualified_session_key: str, task_id: int, *, ttl_seconds: float) -> None: ...
    async def is_alive(self, qualified_session_key: str, task_id: int) -> bool: ...
    async def publish_outcome(self, qualified_session_key: str, task_id: int, info: BackgroundTaskInfo) -> None: ...
    async def fetch_outcome(self, qualified_session_key: str, task_id: int) -> BackgroundTaskInfo | None: ...
    async def request_cancel(self, qualified_session_key: str) -> None: ...

provider = BackgroundAgentsProvider(
    agents=[researcher, writer],
    runtime_store=MyRedisBackgroundTaskRuntimeStore(redis),  # default: in-memory, today's behavior
)

Language/SDK

Both

[karthik-0306]

Eduard van Valkenburg (@eavanvalkenburg) , I dug into this and can confirm the race exists exactly as described in _refresh_task_state (_background_agents.py, lines ~235–260): in_flight is looked up only in the process-local _RuntimeState.in_flight_tasks, so any RUNNING record with no local entry gets marked LOST regardless of whether the owning process is dead or just not the one serving this turn. I traced all seven scenarios against the current code and they all check out.

I'd like to take this on, implementing the BackgroundTaskRuntimeStore protocol as proposed:

class BackgroundTaskRuntimeStore(Protocol):
    async def acquire(self, qualified_session_key: str, task_id: int, *, ttl_seconds: float) -> None: ...
    async def renew(self, qualified_session_key: str, task_id: int, *, ttl_seconds: float) -> None: ...
    async def is_alive(self, qualified_session_key: str, task_id: int) -> bool: ...
    async def publish_outcome(self, qualified_session_key: str, task_id: int, info: BackgroundTaskInfo) -> None: ...
    async def fetch_outcome(self, qualified_session_key: str, task_id: int) -> BackgroundTaskInfo | None: ...
    async def request_cancel(self, qualified_session_key: str) -> None: ...

Implementation plan:

• BackgroundAgentsProvider.__init__ gains an optional runtime_store: BackgroundTaskRuntimeStore | None = None, defaulting to an in-memory implementation that reproduces today's single-process semantics exactly — no behavior change for hosts that don't configure one. This mirrors the existing SessionStore protocol already in _sessions.py, so a RedisBackgroundTaskRuntimeStore could sit alongside RedisStore in agent-framework-redis.
• background_agents_start_task calls acquire() on start; a wrapper coroutine around _run_agent(...) renews the lease on a configurable interval (defaulting to ttl_seconds / 2) for the life of the task, rather than a separate heartbeat task, so renewal is tied to the task's own lifecycle. A done-callback on the underlying asyncio.Task calls publish_outcome the instant the task settles, from whichever process owns it — today _finalize_task only runs inside a run on the owner, and never at all once a record's been marked LOST, so this is what actually delivers the orphaned result.
• _refresh_task_state calls is_alive() before marking LOST: lease present and fresh → leave RUNNING; lease missing/expired → check fetch_outcome() and finalize from that if present, LOST otherwise, same as today.
• release_session calls request_cancel() before the current early return at self._runtime.get(session_id) is None (lines 374–378) — today that early return is exactly why a delete on a non-owning process silently no-ops, so the cancel call has to happen ahead of it, not alongside it.
• request_cancel(qualified_session_key) is session-granular by design, matching release_session's own scope (cancel everything for a session, not one task). Happy to add an optional task_id: int | None = None if partial cancellation turns out to matter.
• For continue_task on a non-owner, since the child AgentSession isn't serialized into the store here, the error becomes "Task {task_id} was completed on another instance; continue is not available cross-process in this release." — more specific than today's generic "session no longer available" message.
• qualified_session_key will be session.session_id used as-is, with no built-in namespacing in the protocol — hosts running multiple namespaced environments would wrap or subclass the store rather than the protocol handling it. Flagging this as an open question below rather than assuming it.
• fetch_outcome/publish_outcome return and accept BackgroundTaskInfo directly, which couples any external store implementation (Redis, etc.) to importing that framework class. This is the same shape SessionStore already uses with AgentSession, so it's precedented, but worth naming since BackgroundTaskInfo is currently @experimental.

A few things I'd like alignment on before opening the draft PR:

1. qualified_session_key composition — I plan to use session.session_id directly, with no namespacing built into the protocol itself. Is that the right default, or should __init__ also accept something like session_key_prefix: str = "" or a key factory for hosts that need namespacing?
2. Lease renewal mechanism — I plan to implement renew via a wrapper coroutine around _run_agent(...) rather than a separate heartbeat task, so renewal lifecycle is tied to the task itself. Does that match what you had in mind, or would you rather see an independent heartbeat loop?
3. request_cancel granularity — the protocol as sketched cancels everything for a session key. release_session will call it ahead of the current early-return so the signal reaches the owner even when this process has no local runtime. Let me know if task-granular cancellation is needed instead.
4. TTL clock source — time.monotonic() is process-local and unusable across processes, so time.time() is the only semantically correct option here; flagging this as confirmation rather than an open preference, unless you'd rather the store manage its own timestamps.

Happy to open a draft PR once we're aligned on these four.