[Tech Debt] align action recovery model

[da-daken]

Linked issue: #1016

Purpose of change

Simplify the recovery matching model for durable execution from the two-field(functionId, argsDigest)to a single functionId field, making the Java and Python APIs consistent. On the Python side, by embedding the args digest inside the functionId string (in the format module.qualname::digest), the automatic deduplication capability based on function + args is preserved, maintaining the language's own characteristics. At the same time, assign a unique durable ID (tool-call:{llm_id}) to each tool call of ToolCallAction, avoiding recovery matching confusion between multiple tool calls.

Behavioral Semantics

Recovery Matching Model Change: Previously, CallResult used two independent fields, functionId and argsDigest, to match call sites during replay. Now, functionId itself carries the complete matching semantics—provided by DurableCallable.getId() on the Java side, and returned as a single string by durable_identity_for_call() on the Python side. This eliminates the complexity for callers of having to manage two identity fields separately.

Python's args digest preservation strategy: When there is no explicit durable_id, durable_identity_for_call() returns a single string in the format {function_id}::{args_digest}. This means the behavioral semantics on the Python side are completely equivalent to before (the same function + different args will still generate different identities), but the interface is simplified to a single str return type.

ToolCallAction Uniqueness: On the Java side, TOOL_CALL_DURABLE_ID is changed from the constant "tool-call" to "tool-call:" + id. On the Python side, all three durable_execute calls pass durable_id=f"tool-call:{execution.id}". The durable log entries for each tool call are now independent of each other, and the recovery of different tools in parallel tool calls (parallel batch) will not interfere with each other.

Forward Compatibility: Add @JsonIgnoreProperties(ignoreUnknown = true) to CallResult to ensure that serialized data containing the old argsDigest field will not throw an exception during deserialization.

Java-Python Bridge Layer: The arrays returned by getCallResultFieldsAt / getCurrentCallResultFields are reduced from a 5-tuple [functionId, argsDigest, status, result, exception] to a 4-tuple [functionId, status, result, exception]. On the Python side, _PersistedCallResult is synchronously unpacked into 4 elements.

Tests

• Java: All 6 migration-related test classes passed (~95 test cases), including ActionStateSerdeTest (20), ActionStateTest (17), DurableExecutionContextTest (17), JavaRunnerContextImplDurableExecuteAsyncTest (19), RunnerContextImplDurableExecuteTest (17), and ActionExecutionOperatorTest (56).
• Python: test_flink_runner_context_reconcilable.py (~50+) and test_tool_call_action.py (41) all passed. The Mock _Context's durable_execute / durable_execute_async performs kwargs.pop("durable_id", None) filtering to ensure the newly added durable_id keyword argument is not passed through to the tool function under test.

API

no API

Documentation

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

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

• Yes
• No

Generated-by: cursor (deepseek-v4-pro)

[da-daken]

@wenjin272 @pltbkd Would you mind taking a look at this PR at your convenience? The reason CI is red seems to be an environment issue.

[wenjin272]

Thanks for the contribution! This looks good to me. Let's wait for @pltbkd to take another look.