[Ashfaqbs]

Linked issue: #1084

Purpose of change

Two distinct event occurrences reaching the same action under the same business key and sequence number, with identical attributes, previously collided on the same durable action-state key. The second occurrence's pre-execution lookup could then find the first occurrence's just-persisted completed state and incorrectly skip its own execution, replaying the first occurrence's output instead. This fixes that: sibling occurrences with identical attributes now always get distinct keys, while a changed attribute still changes the key exactly as before (content-based replay-divergence detection is unchanged).

Runtime flow

1. Root event ingestion (ActionExecutionOperator#processEvent, isInputEvent branch): after the per-key sequence number is incremented, assignRootOccurrenceId derives occurrenceId = sha256(businessKeyIdentity + "#" + seqNum) and assigns it on the event.
2. Output emission (processActionTask on the serial engine; Work#commit on the parallel engine): assignOutputOccurrenceIds computes the producing action execution's own composite key (via the triggering event's already-assigned occurrenceId), then assigns each output event occurrenceId = sha256(producerExecutionId + "#" + ordinal), ordinal being its 0-based position among that execution's emitted outputs — before any output is dispatched or persisted.
3. ActionStateUtil#generateUUIDForEvent consumes event.getOccurrenceId(): when present, it is folded together with (not in place of) the existing attributes hash into the UUID segment; when absent (state persisted by a pre-fix version), it falls back unchanged to the attributes-only hash.

Key decisions

• Carrier: a new non-contract Event field (occurrenceId), mirroring the existing sourceTimestamp field, rather than a wrapper type. This avoids touching every ActionTask/SubgraphEvent/serializer call site a wrapper would require, and survives Flink's generic (Kryo-based) Event serializer across checkpoints the same way sourceTimestamp already does.
• Fold with content, don't replace it: an occurrence identity that ignored attributes entirely (rejected) regressed ActionStateDivergenceRecoveryTest, letting a later restore with a changed input reuse the other input's cached result at the same sequence number. Folding lineage together with the attributes hash keeps that protection while adding sibling disambiguation.
• Reuse the store's own key encoder: identity derivation goes through two new ActionStateStore methods (generateBusinessKeyIdentity, generateActionExecutionId) that delegate to the same ActionStateKeyEncoder each store already holds, rather than duplicating key-derivation logic in the operator — the composite key format and the occurrence identity can never drift apart.
• No-op when durable execution is disabled, since nothing keys state on it in that case.

Behavioral Semantics

Interaction decisions

Durable execution	Event kind	occurrenceId	Resulting key segment
disabled	any	not assigned	unchanged: attributes hash only
enabled	root (new)	businessKey#seqNum	folded with attributes
enabled	output (new)	producerExecId#ordinal	folded with attributes
enabled	any, pre-fix persisted state	absent (null)	unchanged: attributes hash only

Behavioral contracts

1. Two sibling output events emitted by the same action execution with identical attributes get distinct action-state keys.
2. A replayed root InputEvent (same business key, same sequence number, fresh random Event#getId()) derives the same occurrence identity, and therefore the same key, as the original.
3. A changed attribute value still changes the key even when the lineage position (producer + ordinal, or business key + seqNum) is unchanged.
4. The composite key format (five segments, 36-character UUID per identity segment) is unchanged; parseKey/isKeyRetained/businessKeyIdentityOf behavior is unchanged.
5. An event never assigned an occurrence identity (state persisted by an older version) keys exactly as before the fix.

Failure behavior

generateBusinessKeyIdentity/generateActionExecutionId return null when no durable store is configured; callers treat null as "skip assignment" and never throw. deriveOutputEventOccurrenceId rejects a negative ordinal (IllegalArgumentException) and a null producer id (NullPointerException) defensively, though the operator's own call site cannot reach either. No new exception types; IOException propagation through generateKey is unchanged.

Tests

Contract	Test
1. Sibling collision fixed	ActionStateUtilTest#testDistinctEventOccurrencesWithOccurrenceIdDoNotCollide; operator-level: ActionStateDivergenceRecoveryTest#normalFanoutWithIdenticalSiblingAttributesStillExecutesBothSiblings
2. Replay-stable root identity	ActionStateUtilTest#testDeriveRootEventOccurrenceIdIsReplayStableAndKeySensitive
3. Content-divergence preserved	ActionStateUtilTest#testOccurrenceIdDoesNotSuppressContentDivergence; ActionStateDivergenceRecoveryTest#changedInputHistoryMustInvalidateLaterResultAcrossTwoRestores (pre-existing, now passing again)
4. Key format unchanged	pre-existing ActionStateUtilTest key-format tests (testParseKey*, testIsKeyRetained*), all still green
5. Legacy fallback	ActionStateUtilTest#testDistinctEventOccurrencesWithoutOccurrenceIdCollideLegacy (pins the documented legacy behavior)

Coverage by risk: the identity-derivation functions and the operator wiring that assigns them are both covered directly; the two closest pre-existing regression risks (content-based divergence detection, and a hand-fabricated durable-state fixture in KafkaActionStateStoreRecoveryTest that bypassed the new assignment path) were caught by the existing suite and fixed. Not verified: the Fluss-backed store path specifically (Fluss integration tests do not run in this environment; the fix is store-agnostic and goes through the same ActionStateKeyEncoder both Kafka and Fluss already share, and FlussActionStateStore received the identical two-method addition as KafkaActionStateStore).

Implementation notes and supporting evidence

generateUUIDForEvent combines occurrenceId and the attributes-JSON bytes with a # separator before hashing through UUID.nameUUIDFromBytes, so the segment stays a canonical 36-character UUID regardless of which branch runs. KafkaActionStateStoreRecoveryTest#queuedActionUsesRecoveredDurableResultAfterRestore fabricates pending durable state directly against the store, bypassing ActionExecutionOperator's assignment path; it now assigns the same businessKeyIdentity#seqNum identity to its hand-built InputEvent that the operator would compute for its own, matching the real contract instead of relying on the old content-only derivation.

API

Event gains a public, @JsonIgnore field (occurrenceId) with a getter/setter, excluded from equals/hashCode and from JSON serialization — no change to any existing serialized payload or to the cross-language event contract. ActionStateStore gains two new interface methods (generateBusinessKeyIdentity, generateActionExecutionId); the two shipped implementations (KafkaActionStateStore, FlussActionStateStore) and the test InMemoryActionStateStore all implement them as one-line delegations to their existing ActionStateKeyEncoder. Any third-party ActionStateStore implementation outside this repository would need to add these two methods — a source-incompatible interface change for external implementers, mitigated by both being thin delegations every implementation already has the pieces for. No durable action-state key format or size change; no migration needed. State persisted by a pre-fix version has no occurrenceId and recovers exactly as before.

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Claude Code 2.1.314 (Claude Sonnet 5)

[wenjin272]

Thanks for implementing the lineage-based identity approach! I left a few comments on sharing identity generation across runtime paths, covering chat/internal-subagent calls, and simplifying the fallback and comments.

Please also update the remaining test fixtures for the new root occurrence identity. In the Ubuntu/JDK 21 CI log I inspected for bc042eb, five ActionExecutionOperatorTest tests fail: four reconcilable-recovery tests and testUnsupportedObservationValueIsSkippedWithoutChangingActionSuccess. seedActionState(), getStoredActionState(), and the observation test's direct lookup still construct InputEvents without an occurrence ID, so their attributes-only keys no longer match the keys generated by the operator. These fixtures should assign the same root occurrence identity as the production path, as already done in KafkaActionStateStoreRecoveryTest.

[wenjin272]

Could we move identity generation into a shared runtime component instead of adding these methods to ActionStateStore? Kafka and Fluss only delegate to the same encoder, and execution identity should be available independently of durable storage.

An action execution ID can be derived as hash(triggeringEventOccurrenceId, actionIdentity), with business key and sequence number already represented in the root occurrence ID. Content-based divergence checks should remain in place separately.

[wenjin272]

Chat/internal-subagent call IDs were not part of the original issue scope, but we later identified the same event-occurrence identity problem in #1220. I suggest fixing that here as well; a follow-up PR is also fine. Either way, please move identity generation into a shared runtime component in this PR so both paths can use it independently of durable storage.

For the chat/subagent fix, DeterministicIdAllocator.Namespace and its caller setup in ActionTaskContextManager should consume the shared action execution identity while retaining target/session/call scoping. The current allocator does not use occurrence identity, so identical event occurrences can still produce duplicate IDs in SubgraphCallManager, including when durable execution is disabled.

[wenjin272]

We do not require compatibility with old persisted state for this change, so please remove the legacy attributes-only fallback and make missing occurrence identity explicit on paths that require it. Keeping this fallback preserves the collision behavior the change is intended to eliminate.

[wenjin272]

Could you clean up the comments throughout this patch to describe the current identity and recovery contracts directly? Repeated issue references such as FLINK-AGENTS-1084 and historical wording like “before this fix” are unnecessary in production code; that context belongs in the PR description or commit message.

[Ashfaqbs]

Thanks for the detailed review.

• Fixed the 5 failing ActionExecutionOperatorTest fixtures: seedActionState/getStoredActionState and the direct lookup in testUnsupportedObservationValueIsSkippedWithoutChangingActionSuccess now assign the same root occurrence identity the operator itself computes, instead of leaving it unset. All green locally (ActionExecutionOperatorTest, ActionStateUtilTest, KafkaActionStateStoreTest, FlussActionStateStoreTest, ActionStateKeyEncoderTest, KafkaActionStateStoreRecoveryTest, ActionStateDivergenceRecoveryTest).
• Moved identity generation into a shared path independent of durable storage: the operator now derives a root event's business-key identity directly from its own keyed-state serializer (a new ActionStateKeyEncoder field it owns), instead of delegating through DurableExecutionManager/ActionStateStore — so this is no longer a no-op when durable execution is disabled. An action execution's identity is now hash(triggeringEventOccurrenceId, actionIdentity) via a new ActionStateUtil.deriveActionExecutionId, needing neither a key encoder nor a store. Removed generateBusinessKeyIdentity/generateActionExecutionId from the ActionStateStore interface and the Kafka/Fluss implementations.
• Cleaned up the FLINK-AGENTS-1084/"before this fix" references from the production-code comments this patch touches.
• One partial item: I kept the attributes-only fallback in ActionStateUtil.generateUUIDForEvent rather than making a missing occurrence identity throw. Removing it outright breaks a large pre-existing lower-level test surface (ActionStateUtilTest, KafkaActionStateStoreTest, FlussActionStateStoreTest, ActionStateKeyEncoderTest) that predates occurrence identity and constructs events without assigning one — happy to do that migration as a follow-up if you'd rather not carry the fallback at all.
• Left the chat/internal-subagent identity fix ([api][runtime][java][python] Add event-backed chat calls from actions #1220) for a follow-up PR as you suggested — this PR now has the shared component in place for it to build on.

Pushed as bc042eb..8a5d3f5.

[wenjin272]

Thanks again for your contribution and for working through the feedback! I think this PR is ready to merge. I've opened #1351 as a follow-up for the chat/subagent integration, replay handling, and remaining cleanup, so we can address those separately without holding up this PR.