Repository navigation
Conversation
Co-authored-by: Heng Pan <pan@flower.ai>
Co-authored-by: Heng Pan <pan@flower.ai>
There was a problem hiding this comment.
🟡 Changes recommended
Confirmed regressions in AgentApp validation and SuperNode reply delivery can prevent callers from receiving responses.
3 open issues
What changed in this PR
Extends the destination-oriented delivery work from #8409 to simulation, aligning InMemoryGrid and VCE with the shared message lifecycle while retaining the included production changes.
Changes:
- Switches message routing to destination-based storage, retrieval, acknowledgment, and cancellation.
- Preserves full simulation messages and adds coverage for correlation filters, multiple replies, and enqueue failures.
- Adds legacy metadata migration and updates production delivery policy and identity checks.
The delivery changes need correction before approval. No separate readability cleanup is needed; completing the shared lifecycle transition is the priority. Keeping the production foundation in #8409 separate from the simulation wiring would simplify review and validation.
| File | Description |
|---|---|
| framework/py/flwr/supernode/start_client_internal.py | Moves inbound and outbound transfers to destination delivery. |
| framework/py/flwr/supernode/start_client_internal_test.py | Updates transfer and verification-error tests. |
| framework/py/flwr/supernode/servicer/runtime/runtime_handlers.py | Updates ClientApp delivery, source checks, and acknowledgments. |
| framework/py/flwr/supernode/servicer/runtime/runtime_handlers_test.py | Updates delivery mocks and source-validation coverage. |
| framework/py/flwr/supernode/nodestate/nodestate.py | Removes the legacy session-expiry hook. |
| framework/py/flwr/supernode/nodestate/nodestate_test.py | Updates storage, cancellation, and timing tests. |
| framework/py/flwr/supernode/nodestate/in_memory_nodestate.py | Updates message preregistration and timing cleanup. |
| framework/py/flwr/superlink/servicer/runtime/runtime_handlers.py | Adopts destination delivery and authenticated identity checks. |
| framework/py/flwr/superlink/servicer/runtime/runtime_handlers_test.py | Covers policy, filtering, and acknowledgment behavior. |
| framework/py/flwr/superlink/servicer/fleet/fleet_handlers.py | Updates Fleet storage, retrieval, and acknowledgments. |
| framework/py/flwr/superlink/servicer/fleet/fleet_handlers_test.py | Covers source checks and accepted-message accounting. |
| framework/py/flwr/superlink/servicer/control/control_servicer_test.py | Retrieves AgentApp prompts through destination delivery. |
| framework/py/flwr/superlink/servicer/control/control_handlers_test.py | Updates prompt-persistence assertions. |
| framework/py/flwr/superlink/routers/fleet/router.py | Updates acknowledgment-handler arguments. |
| framework/py/flwr/superlink/grid/inmemory_grid.py | Wires simulation Grid to shared delivery APIs. |
| framework/py/flwr/superlink/grid/inmemory_grid_test.py | Covers multiple replies, filters, and rejected positions. |
| framework/py/flwr/superlink/grid/http_grid.py | Updates validation, missing-object handling, and reply waits. |
| framework/py/flwr/superlink/grid/http_grid_test.py | Updates missing-object and timeout expectations. |
| framework/py/flwr/supercore/state/alembic/versions/rev_2026_10_22_migrate_legacy_message_metadata.py | Copies legacy metadata into the shared message table. |
| framework/py/flwr/supercore/state/alembic/utils_test.py | Tests migration preservation and repeatability. |
| framework/py/flwr/supercore/error/catalog.py | Defines the Fleet source-mismatch response. |
| framework/py/flwr/supercore/error/base.py | Adds the Fleet source-mismatch error code. |
| framework/py/flwr/supercore/corestate/sql_corestate.py | Cancels deliveries when upload sessions expire. |
| framework/py/flwr/supercore/corestate/in_memory_corestate.py | Updates locks and expired-session cancellation. |
| framework/py/flwr/supercore/corestate/corestate.py | Removes the obsolete expiry callback. |
| framework/py/flwr/supercore/corestate/corestate_test.py | Stabilizes lifecycle clock mocking. |
| framework/py/flwr/simulation/run_simulation.py | Initializes the local primary task as pending. |
| framework/py/flwr/simulation/app_test.py | Tests local Grid message exchange. |
| framework/py/flwr/server/superlink/linkstate/sql_linkstate.py | Uses shared storage and avoids restoring cleared bodies. |
| framework/py/flwr/server/superlink/linkstate/linkstate.py | Applies run and federation policy to delivery. |
| framework/py/flwr/server/superlink/linkstate/linkstate_test.py | Covers policy, duplicate uploads, and cleanup. |
| framework/py/flwr/server/superlink/linkstate/in_memory_linkstate.py | Updates in-memory storage and prompt insertion. |
| framework/py/flwr/server/superlink/fleet/vce/vce_api.py | Acknowledges full messages after enqueueing. |
| framework/py/flwr/server/superlink/fleet/vce/vce_api_test.py | Covers enqueue failures and rejects legacy API use. |
| framework/py/flwr/server/superlink/fleet/grpc_rere/fleet_servicer.py | Updates acknowledgment-handler arguments. |
| framework/py/flwr/server/superlink/fleet/grpc_rere/fleet_servicer_test.py | Updates delivery-policy and acknowledgment tests. |
Review effort: Balanced
github-actions
Bot
added
the
Maintainer
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1640e771b5
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1d3643b5e6
ℹ️ 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".
| messages = self.state.get_messages( | ||
| dst_node_id=self.node.node_id, | ||
| run_id=cast(Run, self._run).run_id, | ||
| reply_to_message_ids=( | ||
| set(reply_to_message_ids) if reply_to_message_ids is not None else None | ||
| ), | ||
| ) |
There was a problem hiding this comment.
When VCE acknowledges an instruction but its ClientApp never produces a reply—for example, the instruction expires or the destination becomes unavailable—this destination-inbox query can only return replies that were actually stored. The former get_message_res path synthesized correlated TTL/node-unavailable error messages, allowing the pending ID to be removed; now send_and_receive with its default timeout=None keeps polling forever, which can hang simulations after a client failure. Preserve a terminal error for these instructions or otherwise retire their pending correlation IDs.
AGENTS.md reference: framework/AGENTS.md:L118-L120
Useful? React with 👍 / 👎.
Issue
Description
Simulation still routes messages through legacy instruction/reply storage, while the production message path uses destination-oriented delivery. This prevents simulation from receiving messages independently of correlation metadata and retaining multiple replies to one message.
Related issues/PRs
Based on #8409. This branch includes its production delivery changes; this PR adds the simulation wiring on top.
Proposal
Explanation
store_messages,get_messages, andacknowledge_messagesAPIs. Grid queries its destination and run with optional correlation filters; VCE retains its per-nodelimit=1polling and ACKs only after a full Message is successfully queued.Checklist
Validation and scope
Passed the Grid, VCE (including Ray), and simulation app tests, plus targeted isort, Black, Ruff, Mypy, Pylint, and
git diff --check.The legacy API/storage removal is deferred. InMemoryGrid's SQLite body-delivery path is outside this simulation scope. No task lifecycle or scheduling redesign is included. Delivery metadata remains retained after ACK; missing replies can leave
send_and_receive(timeout=None)waiting indefinitely. Full framework tests and deployment E2E were not run.