weiqingy opened a new issue, #998: URL: https://github.com/apache/flink-agents/issues/998
### Search before asking - [X] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description `Mem0LongTermMemory` is shared across partition keys and holds the current key in a mutable field. `switch_context` sets it on the mailbox thread before each action (`mem0_long_term_memory.py:310`), while `add`, `get`, `search`, `delete` and `delete_memory_set` can run on a pool thread when an action passes them to `durable_execute_async`. Those operations read the field themselves, so the key they see is whatever the mailbox thread last wrote, not the key of the action that submitted the work. The key is the isolation boundary, not just observability metadata: it is passed to Mem0 as `agent_id` (`mem0_long_term_memory.py:504` for `add`, `:457` for `delete_memory_set`). Two keys sharing a job id and a memory set name are separated by `agent_id` alone, so reading the wrong one means one key's items land in or are read from another key's set. The same applies to `_observation_id` and `_observation_suppressed`, which are read from the same shared state in all five operations (`:450-451, :488-489, :553-554, :608-609, :665-666`), so observations can be misattributed independently of the memory itself. Several keys really can be in flight at once. An action that awaits suspends without completing (`plan/function.py:436,451`), the operator re-queues it as a fresh mail (`ActionExecutionOperator.java:502-508,537`), and processing keys are tracked in a `ListState` rather than as a single key (`OperatorStateManager.java:83`). Nothing in `Mem0LongTermMemory` holds a lock or thread-local. Two things worth stating plainly: - **The window is narrow.** Each operation copies the field into a local on its first lines, so the exposure is from `executor.submit` (`flink_runner_context.py:159`) to that first statement, not the whole Mem0 round trip. - **No failure has been attributed to this.** It is a structural defect found by reading the code. #983 recorded one unexplained `assert 0 == 1` in the long-term memory e2e test, and this race would produce that symptom, but nothing links the two. #984 adds a cross-key leak check in both directions so a future occurrence is diagnosable. Java is affected through the same code. `Mem0LongTermMemory.java` is a bridge whose operations all call into the same Python object (`:80,100,111,128`), and the cross-language e2e agent already captures a `MemorySet` on the mailbox thread and calls `memorySet.add` inside `durableExecuteAsync` (`Mem0LongTermMemoryAgent.java:183-198`). Which thread the pemja call runs on for such a callable has not been verified, so a live Java race is not being claimed here, only that the fix has to cover the Java path. **Direction under discussion** (see the thread on #983): bind the partition key, observation id and suppression flag onto the `MemorySet` when it is constructed, and have the operations read them from the set instead of from shared state. `add`, `get`, `search` and `delete` already take `memory_set` as their first parameter (`api/memory/long_term_memory.py:172,192,213,223`), so no new plumbing is needed. Two consequences: `get_memory_set` must be called per action on the mailbox thread and not cached across actions, and the bridge helper that rebuilds the Python set from a name alone (`python_java_utils.py:388-393`) has to carry the key too. `delete_memory_set` takes a name rather than a `MemorySet`, so it needs separate handling. ### How to reproduce There is no deterministic reproduction. The race is established structurally rather than observationally, and the window is a few statements wide. The shape that exposes it, as in `python/flink_agents/e2e_tests/e2e_tests_integration/long_term_memory_test.py`: 1. Key a stream so at least two keys are processed by the same operator instance. 2. In an action, call `ctx.durable_execute_async(memory_set.add, items=...)` and await it. 3. The action suspends, the operator picks up another key and calls `switch_context` with that key. 4. If the submitted callable has not yet reached its first statement, it reads the second key and writes under that `agent_id`. Observing it requires the mailbox thread to run `switch_context` inside that window. Inserting a delay between `executor.submit` and the operation body makes it reachable, which is a way to confirm the mechanism rather than a reproduction of a real failure. ### Version and environment `main` at 6d7afb0f. Affects both the Python implementation and the Java bridge over it. Not specific to a Flink version, deployment mode or OS. ### Are you willing to submit a PR? - [X] 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]
