wenjin272 opened a new issue, #1302: URL: https://github.com/apache/flink-agents/issues/1302
### Search before asking - [x] I searched the existing issues. #1301 tracks incorrect divergence detection for normal multi-Action execution; #1084 tracks event-occurrence key collisions. Neither restores the original input ordering across upstream channels. ### Description ActionStateStore reuse currently depends on replaying each business key's inputs in the same order. The operator assigns seqNum from a checkpointed per-key counter when it starts a top-level InputEvent; seqNum is not a stable source-record identifier. When the same business key arrives from multiple upstream parallel instances, keyBy routes those inputs to the same downstream instance but does not establish a replay-stable order across the incoming channels. A failure can therefore assign a different seqNum to the same logical input. This is a practical limitation for agents consuming merged streams or records distributed across multiple source partitions/upstream instances under the same agent key. Correcting #1301 avoids false invalidation for normal multi-Action execution, but does not eliminate this limitation. The [current consistency documentation](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/operations/consistency.md#conditions-for-reusing-saved-records) already requires stable per-key input order. This issue requests support for the broader topology and a precise user-facing recovery contract, rather than claiming that the existing documentation guarantees it. ### Example and impact Take checkpoint C0 after seqNum=9 for key K. Two upstream instances then emit distinct records for K: ```text upstream 1: A, C upstream 2: B, D original execution: seq=10: A seq=11: B seq=12: C seq=13: D completed ActionStates are persisted; failure before the next checkpoint restore C0, with a different channel interleaving: seq=10: B seq=11: A seq=12: C seq=13: D ``` Both deliveries preserve each upstream channel's order, yet the per-key merged sequence differs. The lookup for B at seq=10 cannot find B's completion saved at seq=11. Cache-miss/divergence handling can invalidate the remaining saved suffix, causing completed work to run again. Even when C returns to its old sequence number, blindly reusing its result is unsafe if its output or Memory updates depended on the earlier processing order. For example, A and B can set the same Memory field to different values, which C then reads. - Model/tool calls can repeat despite having saved completions. - If order changes at the first replayed input, reuse for that key's affected replay window can be lost. - Changes only in the interleaving of different keys do not cause this issue. - Parallelism or keyBy alone is not sufficient to trigger it: a key with a replay-stable ordered input path can still reuse saved state. - This is about ActionState reuse and repeated external work; it is not a claim that all Flink checkpointed state or transactional sinks lose their guarantees. Flink's [dataflow documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/learn-flink/overview/) explains that redistribution preserves ordering within each sending/receiving subtask pair, not across all senders. ### Proposed investigation Define how agents should recover when same-key inputs originate from multiple channels. Possible directions include durably recording the accepted per-key input order and restoring it, or introducing stable input identity together with the state-dependency semantics required for safe reuse. A stable input ID alone is insufficient for stateful Actions: recovering a result by ID must not replay Memory changes or outputs computed against a different preceding history. Likewise, disabling divergence protection would mask the mismatch rather than solve it. A substantial recovery-model change should receive a separate design discussion before implementation. ### Acceptance criteria - Add a controlled recovery test with at least two upstream instances emitting distinct inputs for the same agent key, deliberately reverse their interleaving after failure, and measure actual Action/tool invocations. - Define and implement a supported replay contract that preserves safe reuse of saved completed work for that topology. If ordering restoration is selected, verify that the accepted history is restored before applying dependent Memory updates. - Test stateful Actions whose results depend on preceding same-key inputs, not just stateless input-to-output functions. - Cover repeated failures and checkpoints taken while recovery is progressing; validate both Kafka and Fluss backends. - Keep different keys isolated and retain the existing ordered-input behavior. - Document supported topologies, ordering prerequisites, and fallback/re-execution behavior for inputs outside the supported contract. Until broader support exists, explain concretely that keyBy alone does not satisfy the ordering prerequisite. - Do not weaken the genuine divergence safeguards tracked in #1301. ### Version and validation status Flink Agents 0.4-SNAPSHOT; source analysis used audit baseline 47daff57b5c4189c80dd7037660f2a7832270879, including OperatorStateManager.initOrIncSequenceNumber and the Kafka/Fluss lookup paths. The sequence-number dependency and invalidation paths were inspected. The separate #1301 prototype exercised controlled changed-input histories through real operator snapshot/restore with mocked broker transport. A live multi-upstream network-shuffle failover reproduction for this issue has not yet been run. ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
