[runtime][java] Make embedded Python interpreters thread-confined

[wenjin272]

Linked issue: Closes #1087

Purpose of change

Concurrent cross-language calls no longer enter one embedded PythonInterpreter from multiple threads, avoiding the native JVM crashes observed in async ChatModel and Mem0 paths without globally serializing calls.

Runtime flow

1. Each Python-enabled operator subtask creates an owner interpreter and a PythonInterpreterManager.
2. Every Java-to-Python action, resource, adapter, and conversion call goes through that manager.
3. The mailbox owner and managed Java async workers execute inline on their own thread-local interpreters. Other callers use a stable lane in a bounded callback-worker pool.
4. Multi-step operations retaining Python objects complete within one interpreter operation.
5. Operator shutdown stops the managed async workers first. Each managed or callback worker closes its interpreter on the owning thread, then the owner interpreter and Python environment are closed.

Key decisions

• Use hybrid routing instead of a global lock because a global lock would serialize all bridge calls and negate num-async-threads. Managed Java workers can safely own an interpreter, while creating one on a CPython-created thread can attach a second Python thread state to that native thread; only the latter calls are forwarded.
• Give each unmanaged source thread a stable callback lane so related opaque Python handles stay on one interpreter. Lane count reuses num-async-threads; threads and interpreters remain lazy.
• Reuse Python resource instances across thread-confined interpreters rather than initializing one resource graph per worker.
• Keep the fix inside Flink Agents; no Pemja or Flink release change is required.

Behavioral Semantics

Interaction decisions

Python bridge present	Calling context	Result
No	Any	No interpreter manager or callback worker is created.
Yes	Mailbox/owner thread	Execute inline with the owner interpreter.
Yes	Managed Java async worker	Lazily create and reuse that worker's interpreter; execute inline.
Yes	Python-created or other unmanaged thread	Submit synchronously to that caller's stable callback lane.
Yes	Callback worker re-entering Java-to-Python	Reuse its bound interpreter inline; do not submit recursively.

Behavioral contracts

1. Bridge calls on different execution threads do not concurrently use one PythonInterpreter.
2. Independent managed-worker and callback-lane calls can overlap; the fix does not globally serialize Python bridge traffic.
3. Python-to-Java-to-Python callbacks use bounded interpreter-owning workers and preserve caller-to-lane affinity.
4. Python object conversion, invocation, result retention, and nested Java-to-Python callbacks stay on the selected interpreter.
5. Interpreters are closed on their owning worker threads, accepted work completes before teardown, and calls after manager closure are rejected.
6. Plans without Python actions, Python resources, or Mem0 retain the existing no-Python-runtime path.

Failure behavior

• A callback operation or interpreter initialization failure is propagated; this layer does not retry or fall back to sharing another interpreter. A partially initialized interpreter is closed before the failure escapes.
• An interrupted caller finishes an accepted callback before its interrupt status is restored, preventing premature native-state teardown.
• Closing the manager from a non-owner thread, configuring a non-positive callback-worker count, or invoking after close raises a managed exception.
• Shutdown attempts all interpreter cleanup and reports later cleanup failures as suppressed exceptions. External chat, embedding, and vector-store failures keep their existing behavior.

Tests

Contract	Tests
1. Per-thread isolation	Manager isolation test; Java concurrent ChatModel E2E
2. Calls remain concurrent	Manager overlap test; Java and Python concurrent ChatModel E2Es
3. Bounded, affine callbacks	Unmanaged routing, bounded-reuse, and callback-reentry tests; Python E2E
4. Same-interpreter conversion	Python action and Java chat-message adapter tests
5. Owning-thread lifecycle	Managed/callback close, post-close rejection, and thread-factory tests
6. No-Python path stays lazy	Python bridge no-op test

Coverage targets the native-crash risks: concurrent interpreter entry, Python-created-thread re-entry, opaque-object affinity, and shutdown. The prerequisite-gated Mem0 E2E was also verified manually with its external dependencies available.

Not verified: none identified for the changed behavior.

Implementation invariants and supporting evidence

• The owner thread is captured when the manager is constructed; only that thread may close the manager.
• Managed-worker identity is scoped to AsyncExecutorThreadFactory execution and removed after its exit cleanup.
• Callback lanes are single-thread executors. Executor termination is followed by an actual worker-thread join so interpreter cleanup finishes before Python runtime teardown.
• The lifecycle read lock covers accepted operations; close first excludes new work, then drains callback executors and aggregates cleanup failures.
• Each newly created interpreter imports the same bridge modules before use.

API

No public API or configuration key is added. Existing callers require no source changes. num-async-threads now also bounds the per-subtask callback lanes; unused lanes create neither threads nor interpreters. Java-only and no-Python plans retain their existing execution behavior, apart from stricter async-worker shutdown waiting.

Documentation

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

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

• Yes
• No

Generated-by: Codex 0.144.5 (GPT-5)