[wenjin272]

Linked issue: #1301
Fixes #1301

Purpose of change

Normal fan-out and cascades now retain later saved Action completions during recovery. Previously, both stores treated multiple ActionStates for one key/sequence as divergence, causing later inputs to rerun unnecessarily.

Runtime flow

Kafka and Fluss rebuild a shared indexed cache incrementally from their ordered logs. Each lookup checks input fingerprints and, where available, completed producers' output Event IDs before returning a saved state. Conflicts invalidate the affected history; ordinary misses retain the existing conservative suffix invalidation. Kafka removes corresponding recovery offsets. Fluss refreshes cached rounds in sequence order before checkpointing so a subsequent rebuild preserves recovered future work.

Key decisions

Action counts do not identify a fork. Root matching keeps the existing attribute fingerprint because restored InputEvents can have new random IDs. Internal occurrence checks require persisted completed-producer evidence; comparing every random Event ID would incorrectly reject regenerated subgraph bootstraps. A shared per-key, ordered-sequence index avoids whole-cache scans in lookup and log replay.

Behavioral Semantics

Interaction decisions

All invalidation below is restricted to the same business key.

Evidence and lookup result	Behavior
Multiple Actions, one input fingerprint, matching candidate	Reuse the candidate; retain later rounds.
Conflicting input fingerprints at the same sequence	Remove the current round and all later rounds; execute the requested Action.
Candidate has a different triggering Event ID, and the current internal Event ID appears in a completed producer's outputs for that round	Remove that candidate and later rounds; retain other states in the current round.
Different Event ID without completed-producer evidence	ID difference alone does not invalidate the candidate.
No candidate and no input conflict	Keep other current-round states; invalidate later rounds.
Earlier-round record encountered during ordered log rebuild	Discard the previously reconstructed suffix; also discard the current round if its input conflicts.

Behavioral contracts

• Normal multi-Action histories reuse all matching completed Actions across recovery.
• Conflicting input histories and evidenced stale child occurrences cannot reuse the invalidated current/suffix results, including after another failure or checkpoint.
• Other keys remain isolated, and regenerated subgraph bootstraps retain their existing replay behavior.
• Checkpoint recovery markers and Fluss refresh ordering preserve the surviving cached work.

Failure behavior

Detected conflicts and misses fall back to execution rather than failing recovery. The existing suffix invalidation remains conservative even when a miss does not prove divergence. Backend write/rebuild failures retain the stores' existing exception behavior; failed writes do not add new indexed producer evidence. No new configuration or backend schema is introduced.

Tests

Contract	Tests
Normal fan-out/cascade/repeated Action reuse	ActionStateDivergenceRecoveryTest normal-history cases; ActionStatePythonDivergenceIT
Input changes and stale same-round children cannot reuse incompatible results	ActionStateDivergenceRecoveryTest changed-input, changed-root and truncated-replay cases; ActionStateDivergenceIntegrationTest.changedRootCannotReuseChildComputedAgainstAnotherMemoryHistory
Another failure/checkpoint cannot revive invalidated history	ActionStateDivergenceRecoveryTest.checkpointAfterDivergenceMustNotReviveOldSuffix and repeated-restore cases
Other keys and regenerated bootstraps remain valid	ActionStateCacheTest cross-key/boundary cases; regenerated-bootstrap recovery case
Surviving work remains recoverable through marker/refresh updates	KafkaActionStateStoreTest.divergenceRemovesOffsetsFromRecoveryMarkers; ActionStateDivergenceIntegrationTest.completedCascadeSurvivesCheckpointBeforeReplayingRecoveredInputs

Java 11: 380 regression tests plus 4 native Python/Pemja tests passed, with no skips. Java 21 clean rebuild: 394 passed, 14 skipped by existing non-continuation assumptions in ActionExecutionOperatorTest; those cases passed on Java 11. All new tests executed on both JDKs. The integration suite used an actual Kafka container and embedded Fluss cluster. Spotless (Java 11), Ruff, license checks and whitespace checks passed.

Not verified: the full repository suite, exhaustive retention/rescaling combinations, a throughput or heap benchmark, and a dedicated shared-Memory interleaving reproduction. Python/Pemja tests use simulated backend transport; live backend tests use Java Actions. This does not solve cross-channel input ordering (#1302), general same-attribute occurrence collisions (#1084), or arbitrary nondeterminism.

Index invariants and reproducible test selections

Each key has an ordered sequence map with state keys, input fingerprint counts and completed-output ID counts. Per-state evidence snapshots handle ActionState objects mutated in place before put. Put/remove/clear, tombstones and pruning maintain these indexes; map views cannot bypass maintenance.

Lookup costs O(log R), indexing O(log R + E), and invalidation visits affected rounds/states. R is cached rounds for one key; E is the record's output count. Fluss checkpoint refresh still enumerates surviving states. The NoScanMap regression prohibits whole-map traversal over 3,000 states across 100 keys. Nine cache tests cover evidence reference counts, mutation, deletion and sequence bounds.

Run from the checkout with the selected JDK and a supported Python on PATH. Set PYTHONPATH to the checkout's python directory, runtime/src/test/resources and the active Python's purelib directory. Local Mockito runs used a Byte Buddy 1.15.4 startup agent.

mvn -pl runtime -am clean test \
  '-Dtest=ActionState*Test,ActionStatePythonDivergenceIT,KafkaActionState*Test,FlussActionState*Test,DurableExecutionManagerTest,ActionExecutionOperatorTest,MemoryWriteBelowValueRecoveryTest,InternalSubagentRecoveryTest,SubgraphCallTest,DecisionModelActionRecoveryTest' \
  -Dsurefire.failIfNoSpecifiedTests=false

This is the Java 21 expanded selection. Java 11 ran the state-store suites, DurableExecutionManagerTest, ActionExecutionOperatorTest and MemoryWriteBelowValueRecoveryTest, plus the Python suite separately. The native Python test ends in IT and must be selected explicitly. Docker is required for Kafka integration coverage.

API

No public API, configuration, persisted key format or payload schema changes. Existing stores gain corrected cache invalidation and reuse behavior. Memory overhead increases for the indexes; recovery still loads the required log window into memory.

Documentation

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

Consistency now documents shared Memory ordering under Known limitations, alongside external side effects. The Actions chapter links to the combined section.

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

• Yes
• No

Generated-by: Codex 0.153.4 (GPT-6)