Repository navigation
feat(server): release parked interrupted tasks after a configurable timeout - #1338
Open
christophstach wants to merge 1 commit into
Open
christophstach wants to merge 1 commit into
christophstach wants to merge 1 commit into
Conversation
…imeout A task in input-required or auth-required keeps its ActiveTask (producer, consumer, two dispatchers, registry entry) until a follow-up arrives, which may never happen. Add `interrupted_task_timeout` to DefaultRequestHandlerV2: after that many seconds with no request running and no subscriber attached, the ActiveTask is finished. The stored task is not modified; a follow-up recreates the ActiveTask from the task store. The default, None, keeps the current behavior. The registry now replaces a finished ActiveTask that is still registered, because its removal runs asynchronously and a follow-up arriving in between was rejected with "already completed". Removal only deletes the instance it was scheduled for, so it cannot drop the replacement. Part of a2aproject#1296
🧪 Code Coverage (vs
|
| Base | PR | Delta | |
|---|---|---|---|
| src/a2a/server/agent_execution/active_task.py | 95.14% | 95.46% | 🟢 +0.33% |
| src/a2a/server/agent_execution/active_task_registry.py | 97.37% | 97.65% | 🟢 +0.28% |
| src/a2a/server/cluster/database_event_stream.py | 96.94% | 92.86% | 🔴 -4.08% |
| Total | 93.16% | 93.15% | 🔴 -0.01% |
Generated by coverage-comment.yml
christophstach
marked this pull request as ready for review
October 14, 2026 12:29
Iwaniukooo11
added
the
component: server
Iwaniukooo11
requested changes
Oct 19, 2026
| task_store=self._versioned_store, | ||
| push_sender=self._push_sender, | ||
| event_stream=self._event_stream, | ||
| interrupted_task_timeout=interrupted_task_timeout, |
Member
There was a problem hiding this comment.
Please validate that timeout value>0
| name=f'idle-release:{self._task_id}', | ||
| ) | ||
|
|
||
| def _cancel_idle_release(self) -> None: |
| self._idle_release_task.cancel() | ||
| self._idle_release_task = None | ||
|
|
||
| async def _release_when_idle(self, timeout: float) -> None: |
Member
There was a problem hiding this comment.
Please introduce try/catch to handle self._task_manager.get_task() and asyncio.CancelledError (in fact asyncio.sleep handles that, but it's good practise to put async threads in try/catch to ensure proper cancellation)
| # set, which is the on_message_send create path establishing | ||
| # ownership. Masked as not-found so existence is not leaked. Done | ||
| # outside _lock because the store read is I/O and the miss-path | ||
| # check runs outside the lock too. |
Member
There was a problem hiding this comment.
Here _cancel_idle_release should happen. Since we want to cancel timeout every time some action is taken (which requires calling get_or_create), placing cancel here ensures that timeout is cleaned
| f'Task {self._task_id} is already completed.' | ||
| ) | ||
| self._reference_count += 1 | ||
| self._cancel_idle_release() |
Member
There was a problem hiding this comment.
Should be moved above async lock
| result = await _send(handler, 'Friday', task) | ||
|
|
||
| assert result.status.state == TaskState.TASK_STATE_COMPLETED | ||
| await handler.aclose() |
Member
There was a problem hiding this comment.
Please consider testing edge case scenario, when timeout is extremely close to 0
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_subscriber_keeps_interrupted_task_active() -> None: |
Member
There was a problem hiding this comment.
Ensure each test has a docstring
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Second half of #1296, following the direction agreed in #1296 (comment) and the maintainer's reply.
A task in
input-requiredorauth-requiredkeeps itsActiveTask(producer, consumer, two dispatchers, registry entry) until a follow-up arrives, which may never happen.Change
DefaultRequestHandlerV2(..., interrupted_task_timeout: float | None = None). The defaultNonekeeps today's behavior.ActiveTaskstarts a timer once a request finishes or the last subscriber leaves. A new request or subscriber cancels it. When it fires, theActiveTaskis released only if all of these still hold:ActiveTaskfrom the store.0releases as soon as the task is idle.Registry fix
Removing a finished
ActiveTaskfrom the registry runs asynchronously. A follow-up that arrived in between was rejected withTask … is already completed.get_or_createnow replaces a finished entry, and the scheduled removal only deletes the instance it was scheduled for. Side effect without a timeout: a follow-up to a task that just ended now getsis in terminal state: …, the same error as outside that window.Tests
Seven new tests in
test_default_request_handler_v2.py:ActiveTask(this one fails onmain).Checks
./scripts/lint.sh: passed (the 3tywarnings already exist onmain).uv run pytest: 2285 passed. The new tests passed 10/10 repeated runs and also pass on Python 3.14.event_stream).Part of #1296