fix(framework): Avoid race condition in agent publishing

[charlesbvll]

Issue

Description

Related issues/PRs

Proposal

Explanation

Checklist

• Implement proposed change
• Write tests
• Update documentation
• Address LLM-reviewer comments, if applicable (e.g., GitHub Copilot)
• Make CI checks pass
• Ping maintainers on Slack (channel #contributions)

Any other comments?

[Copilot]

Pull request overview

This PR adjusts the shutdown semantics of RuntimeAgentEvents (the background Agent event publisher) to prevent premature worker termination and adds a unit test to validate that queued events are published on close.

Changes:

• Switch the worker loop from _closed-gated termination to a stop-sentinel–driven loop (while True + _EVENT_PUBLISH_STOP).
• Change close() to enqueue the stop sentinel using a blocking Queue.put() instead of put_nowait().
• Add a unit test asserting that close() drains/publishes queued events before the worker stops.

Critical issues

• close(timeout=...) can still hang indefinitely because Queue.put() is unbounded (no timeout) and occurs before join(), so the provided timeout may be ignored.
• A concurrent/in-flight emit() can enqueue an event after the stop sentinel (it may pass the _closed check before close() flips the flag), causing the worker to exit on the sentinel and leave a late event unpublished.

Simplicity/readability suggestions

• None.

Consistency concerns

• None.

Whether the PR should be split

• No.

Overall verdict

• Not ready as-is due to shutdown/timeout edge cases that can lead to hangs or dropped events.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
framework/py/flwr/supercore/task_process/agent/session.py	Changes worker shutdown behavior to use a stop sentinel rather than _closed loop termination.
framework/py/flwr/supercore/task_process/agent/session_test.py	Adds a test to verify queued events are published before shutdown completes.

Suppressed comments (1)

framework/py/flwr/supercore/task_process/agent/session.py:131

• With the stop-sentinel approach, an emit() that is already in progress can enqueue a TaskEvent after _EVENT_PUBLISH_STOP (it can pass the _closed check before close() flips the flag). In that case, the worker will return as soon as it reads the stop sentinel and the late event remains unpublished in the queue. Consider switching to a two-phase shutdown: once the stop sentinel is observed, drain any remaining queued events (for a short grace period) before exiting.

        while True:
            item = self._queue.get()
            if item is _EVENT_PUBLISH_STOP:
                return

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[panh99]

LGTM!

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.

Suppressed comments (1)

framework/py/flwr/supercore/task_process/agent/session.py:110

• timeout no longer bounds close(): this blocking put() happens before join(timeout). If all 256 slots are occupied while the publisher is stuck in PushTaskEvents, the close(1) used by the signal-exit path can hang indefinitely and never reach the timed join. Apply one deadline to both enqueueing the stop marker and joining the worker (using a timed put, translating Full to TimeoutError, then joining with the remaining time).

        self._queue.put(_EVENT_PUBLISH_STOP)

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: a74bf0e48f

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Honor the close timeout while enqueueing the stop marker

When the 256-entry queue is full because PushTaskEvents is slow or stalled, this blocking put occurs before join(timeout), so the timeout does not bound close at all. In particular, run_agentapp.py calls agent_events.close(1) during exit, but shutdown can instead wait for the Runtime HTTP request timeout (or indefinitely if the worker has terminated unexpectedly). Apply the same deadline to inserting the stop marker, or otherwise avoid blocking before the timed join.

Useful? React with 👍 / 👎.