dangzitou opened a new issue, #1190: URL: https://github.com/apache/flink-agents/issues/1190
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description Python durable execution does not consistently recognize identical `ChatMessage` arguments after a process restart. `_compute_args_digest` hashes raw `cloudpickle.dumps((args, kwargs))` bytes. Pydantic messages include unordered `__pydantic_fields_set__` sets in their pickle state, whose iteration/serialization order can change with the Python process hash seed. As a result, the same function, the same message content and the same kwargs can produce different durable argument digests. The Java journal interprets this as a different call, clears the previously persisted result and invokes the callable again, instead of returning the saved result. Expected: a deterministic action issuing the same durable call with the same serializable arguments should replay the saved result across Python processes. Actual: changing only `PYTHONHASHSEED` from `1` to `5` causes a previously successful, persisted call to execute a second time. A fresh-process `1` to `1` control replays correctly. This affects the input shape used by the built-in chat action: [`chat_model_action.py`](https://github.com/apache/flink-agents/blob/99103da672f897e9c241b51922f2c31a8a728e43/python/flink_agents/plan/actions/chat_model_action.py#L443-L466) passes `List[ChatMessage]` and `prompt_args` into the durable API without an explicit `durable_id`. Repeating model calls can incur extra cost and produce different decisions during recovery. The probe below uses a local callable, so it demonstrates the replay failure without relying on a live model or claiming an observed production billing incident. The action satisfies the [documented same-arguments/same-order requirement](https://github.com/apache/flink-agents/blob/99103da672f897e9c241b51922f2c31a8a728e43/python/flink_agents/api/runner_context.py#L255-L266). It neither mutates the messages nor passes a user-created set. A fresh process is normal during TaskManager replacement; fixed seeds are used here only to make the two serialization orders reproducible. Relevant code: [`_compute_args_digest`](https://github.com/apache/flink-agents/blob/99103da672f897e9c241b51922f2c31a8a728e43/python/flink_agents/runtime/durable_execution.py#L82-L88) and [Java mismatch handling](https://github.com/apache/flink-agents/blob/99103da672f897e9c241b51922f2c31a8a728e43/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java#L1132-L1166). Related #1016 / #1155 concern durable identity representation. The no-explicit-ID argument digest remains based on cloudpickle and this is a distinct cross-process stability problem. Related #1189 concerns swallowed persistence errors in the same Python durable-execution path. Here, persistence succeeds; replay misses because the argument identity changes. ### How to reproduce A small first check needs no running JVM. In the checkout's Python environment, run the following in two fresh processes: ```bash export PYTHONPATH="$PWD/python:$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')" for seed in 1 5; do PYTHONHASHSEED="$seed" python - <<'PYTHON' from flink_agents.api.chat_message import ChatMessage from flink_agents.runtime.durable_execution import _compute_args_digest message = ChatMessage.user("hello") print(message.model_dump_json()) print(_compute_args_digest(([message],), {"prompt_args": None})) PYTHON done ``` Both inputs print: ```json {"role":"user","blocks":[{"type":"text","text":"hello"}],"tool_calls":[],"extra_args":{}} ``` But their digests are `736cc9b5ac74ba8b` (seed 1) and `22bee87e554379b8` (seed 5). The input JSON, function ID, package versions and code are identical. Reconstructing each message with `ChatMessage.model_validate_json(message.model_dump_json())` also reproduces the problem: `d2767ec414353032` versus `500492ee2f708e61`. To verify the effect beyond the digest calculation, the complete local integration probe below uses the real Python `FlinkRunnerContext`, real Java `RunnerContextImpl.DurableExecutionContext`, and production `ActionStateSerde`. A successful call is persisted as serialized bytes; a new Java context restores those bytes, and a **fresh Python process** makes the identical call. A local file counts actual callable invocations across processes. Observed in **4 independent journal instances per case**, for both directly constructed messages and JSON-reconstructed messages: | API mode | Initial → replay seed | Saved entries before recovery | Total callable invocations | Replay result | | --- | --- | --- | --- | --- | | sync | 1 → 1 (control) | 1 | 1 | `reply-1` (saved) | | sync | 1 → 5 | 1 | 2 | `reply-2` (re-executed) | | individually awaited async | 1 → 1 (control) | 1 | 1 | `reply-1` (saved) | | individually awaited async | 1 → 5 | 1 | 2 | `reply-2` (re-executed) | No persistence error is injected in these tests. All initial writes succeed. The Python probe asserts input JSON equality, function ID equality, digest comparison, saved entry count and invocation count. <details> <summary>Complete local Java journal fixture: JournalGateway.java</summary> ```java import java.net.InetAddress; import java.util.HashMap; import java.util.Map; import org.apache.flink.agents.api.InputEvent; import org.apache.flink.agents.plan.AgentPlan; import org.apache.flink.agents.runtime.actionstate.ActionState; import org.apache.flink.agents.runtime.actionstate.ActionStateSerde; import org.apache.flink.agents.runtime.context.RunnerContextImpl; import py4j.GatewayServer; /** Local test entry point: production Java durable context and serde, in-memory persistence. */ public class JournalGateway { private final Map<String, Journal> journals = new HashMap<>(); public Journal create(String name) { Journal journal = new Journal(); journals.put(name, journal); return journal; } public Journal get(String name) { return journals.get(name); } public static final class Journal { private final InputEvent event = new InputEvent("local-repro-input"); private byte[] persisted = ActionStateSerde.serialize(new ActionState(event)); private RunnerContextImpl context; private boolean failPersist; private int persistCalls; public Journal() { recover(); } public RunnerContextImpl getContext() { return context; } public void setFailPersist(boolean value) { failPersist = value; } public void setPersistFailure(boolean value) { failPersist = value; } public void resetFromPersistedSnapshot() { recover(); } public int getPersistCalls() { return persistCalls; } public int getPersistedCount() { return ActionStateSerde.deserialize(persisted).getCallResultCount(); } public String getPersistedJson() { return new String(persisted, java.nio.charset.StandardCharsets.UTF_8); } public void recover() { context = new RunnerContextImpl(null, () -> {}, new AgentPlan(new HashMap<>()), null, "local-repro"); context.setDurableExecutionContext(new RunnerContextImpl.DurableExecutionContext( "local-key", 1L, null, event, ActionStateSerde.deserialize(persisted), (key, seq, action, input, state) -> { persistCalls++; if (failPersist) { throw new RuntimeException("Failed to persist ActionState", new java.io.IOException("local test write failed")); } persisted = ActionStateSerde.serialize(state); })); } } public static void main(String[] args) throws Exception { int port = args.length == 0 ? 0 : Integer.parseInt(args[0]); GatewayServer server = new GatewayServer.GatewayServerBuilder(new JournalGateway()) .javaPort(port).javaAddress(InetAddress.getLoopbackAddress()).build(); server.start(); System.out.println("PORT=" + server.getListeningPort()); } } ``` </details> <details> <summary>Complete cross-process Python probe: repro_hash_seed_recovery.py</summary> ```python """Replay identical ChatMessage arguments in fresh processes using a real Java journal.""" import argparse import asyncio import json import os import subprocess import sys import tempfile from concurrent.futures import ThreadPoolExecutor from pathlib import Path from py4j.java_gateway import GatewayParameters, JavaGateway from flink_agents.api.chat_message import ChatMessage from flink_agents.runtime.durable_execution import durable_identity_for_call from flink_agents.runtime.flink_runner_context import FlinkRunnerContext PLAN = '{"actions":{},"resource_providers":{},"config":{"conf_data":{}}}' def model_request(messages, *, prompt_args): # A local invocation counter; no model server or external service is called. with COUNTER.open("a") as output: output.write("called\n") return "reply-" + str(len(COUNTER.read_text().splitlines())) async def consume(future): return await future def child(args): global COUNTER COUNTER = Path(args.counter) gateway = JavaGateway(gateway_parameters=GatewayParameters(port=args.port)) try: if args.phase == "first": journal = gateway.entry_point.create(args.name) else: journal = gateway.entry_point.get(args.name) journal.recover() messages = [ChatMessage.user("hello")] if args.roundtrip: messages = [ChatMessage.model_validate_json(messages[0].model_dump_json())] function_id, digest = durable_identity_for_call( model_request, (messages,), {"prompt_args": None} ) with ThreadPoolExecutor(max_workers=1) as executor: ctx = FlinkRunnerContext(journal.getContext(), PLAN, executor, None) try: if args.mode == "sync": result = ctx.durable_execute(model_request, messages, prompt_args=None) else: result = asyncio.run(consume(ctx.durable_execute_async( model_request, messages, prompt_args=None ))) finally: ctx.close() print(json.dumps(dict( seed=os.environ.get("PYTHONHASHSEED"), phase=args.phase, function_id=function_id, digest=digest, input_json=[message.model_dump_json() for message in messages], result=result, calls=len(COUNTER.read_text().splitlines()), persisted_count=journal.getPersistedCount(), ))) finally: gateway.close() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("port", type=int) parser.add_argument("--phase", choices=("first", "replay")) parser.add_argument("--name") parser.add_argument("--counter") parser.add_argument("--mode", choices=("sync", "async")) parser.add_argument("--roundtrip", action="store_true") args = parser.parse_args() if args.phase: child(args) else: with tempfile.TemporaryDirectory() as tmp: for repeat in range(1, 5): for mode in ("sync", "async"): for replay_seed in (1, 5): name = f"digest-{repeat}-{mode}-{replay_seed}" samples = [] for phase, seed in (("first", 1), ("replay", replay_seed)): output = subprocess.check_output( [sys.executable, __file__, str(args.port), "--phase", phase, "--name", name, "--counter", str(Path(tmp) / name), "--mode", mode] + (["--roundtrip"] if args.roundtrip else []), env={**os.environ, "PYTHONHASHSEED": str(seed)}, text=True, ) samples.append(json.loads(output)) first, replay = samples assert first["function_id"] == replay["function_id"] assert first["input_json"] == replay["input_json"] assert first["calls"] == first["persisted_count"] == 1 assert replay["persisted_count"] == 1 assert (first["digest"] == replay["digest"]) == (replay_seed == 1) assert replay["calls"] == (1 if replay_seed == 1 else 2) print(json.dumps(dict(repeat=repeat, mode=mode, samples=samples)), flush=True) ``` </details> With JDK 17 and the Python environment described below, save the two files in a separate directory and run these commands from the repository root: ```bash # Activate the Python environment; set JAVA_HOME to JDK 17. mvn -B -ntp -pl runtime -am test-compile jar:test-jar dependency:build-classpath \ -DskipTests -Dspotless.skip=true \ -Dmdep.outputFile=target/repro-classpath.txt -Dmdep.includeScope=test # Set REPRO_DIR to the directory containing the two files above. PY4J_JAR="$(python -c 'import sys; print(sys.prefix)')/share/py4j/py4j0.10.9.7.jar" CP="$REPRO_DIR:runtime/target/classes:plan/target/classes:api/target/classes:integrations/mcp/target/classes:$PY4J_JAR:$(cat runtime/target/repro-classpath.txt)" "$JAVA_HOME/bin/javac" -cp "$CP" "$REPRO_DIR/JournalGateway.java" "$JAVA_HOME/bin/java" -cp "$CP" JournalGateway 25339 # In a second terminal, activate the same Python environment, use the repository # root as cwd, and set REPRO_DIR again to the directory containing the probe: export PYTHONPATH="$PWD/python:$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')" python "$REPRO_DIR/repro_hash_seed_recovery.py" 25339 # Also verify messages reconstructed from their serialized JSON: python "$REPRO_DIR/repro_hash_seed_recovery.py" 25339 --roundtrip ``` The fixture binds to loopback. The hash seed is set at Python process launch, not modified after startup. ### Version and environment - Flink Agents `main`, commit `99103da672f897e9c241b51922f2c31a8a728e43` (`0.4-SNAPSHOT` / Python `0.4.dev0`), unmodified source. - macOS arm64; Python 3.12.14; Pydantic 2.11.4; cloudpickle 2.2.1; PyFlink 2.2.0 used for Python imports; Py4J 0.10.9.7; OpenJDK 17.0.20. - Java runtime and dependency modules built from the same checkout; Python imports resolve to this checkout. - This is a local cross-process Python-to-Java journal recovery test through Py4J. It is not a distributed Flink failover, a Pemja end-to-end test, or a live LLM test. - Existing focused Python durable/reconciler tests: 56 passed; Java durable-context tests: 34 passed. ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! AI disclosure: This report and reproducer were generated with OpenAI Codex. The observations above come from executing the included local probes. -- 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]
