[python] Verify and admit `bytes` in the Python memory value contract

[weiqingy]

Description

Follow-up to #723 / #839.

The set()-time memory value validator added in #839 (validate_memory_value in python/flink_agents/api/memory_object.py) accepts a recursively checkpoint-stable contract of None | bool | int | float | str | list[...] | dict[str, ...]. The accepted scalar set is exact-typed:

_CHECKPOINT_STABLE_SCALARS = (bool, int, float, str)

bytes was part of the contract originally proposed in #723:

None | bool | int | float | str | bytes | list[MemoryValue] | dict[str, MemoryValue]

but was intentionally excluded from the validator. The reason: the Python→Java bytes conversion through Pemja is unverified end-to-end, and accepting an unsafe value would defeat the validator's purpose (turning an early, debuggable rejection back into a checkpoint-time JVM crash on restore). Excluding it was the safe default. This issue tracks closing that gap.

Task

Verify whether bytes survives the full Pemja → Flink state → checkpoint → TaskManager-restart → restore path, the same real-job path used to reproduce #723 (not just direct Flink serializer behavior):

1. Confirm whether Pemja materializes a Python bytes value into a native, checkpoint-stable JVM type (e.g. byte[]) on the FlinkMemoryObject.set() path, rather than a PyObject/PyJObject wrapper.
2. If it does materialize safely:
   • Add bytes to _CHECKPOINT_STABLE_SCALARS.
   • Extend test_memory_value_validation.py to cover bytes accept (incl. nested in list/dict).
   • Update the Python memory value contract docs (added in [doc] Document the Python memory value contract and its divergence from Java #840) to include bytes.
3. If it does not materialize safely:
   • Keep bytes rejected, and document in the contract docs that bytes must be materialized before storing (e.g. base64-encode to str), with the rationale.

Notes

A true restore test can't run on the MiniCluster — in-place recovery does not recreate the JVM, so the Pemja conversion path isn't crossed. Verification needs the real restart path described in #723 (RocksDB + filesystem checkpointing + manual TaskManager kill/restart), or the e2e checkpoint-recovery harness tracked in #836.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[weiqingy]

Verification outcome: bytes is checkpoint-safe → admitting it

Verified the full question (does Python bytes survive the Pemja → Flink state path as a native, checkpoint-stable JVM type?) two independent ways, both agreeing:

1. Pemja source (the conversion logic). In Pemja 0.5.5 (src/main/c/pemja/core/pyutils.c), the Python→Java dispatch JcpPyObject_AsJObject routes PyBytes_CheckExact to JcpPyBytes_AsJObject, whose body is NewByteArray + SetByteArrayRegion — a genuine JVM byte[] with no native back-pointer. bytearray has no branch and falls through to the generic JcpPyObject_AsJPyObject wrapper (Py_INCREF + a process-local pointer) — the unsafe case that crashes on restore. The conversion is byte-for-byte identical in 0.5.5 and 0.5.7.

2. Runtime probe (the actual materialization). A throwaway Java test driving a real embedded Pemja interpreter materialized the values and inspected their Java types:

• b"hello" → Java [B (byte[], len 5) — safe
• bytearray(...) → pemja.core.object.PyObject — unsafe wrapper
• str → java.lang.String — known-good baseline

This is the Python-object → Java-object conversion that FlinkMemoryObject.set()'s j_memory_object.set(path, value) bridge invokes, so it reflects the real memory.set path.

Why that settles restore safety. byte[] is a first-class Flink-serializable primitive array, so once bytes materializes as byte[] it joins the already-proven byte[] checkpoint path. A literal checkpoint-restart round-trip still can't run on the MiniCluster (in-place recovery doesn't recreate the JVM, so the Pemja conversion path isn't crossed) — a permanent end-to-end assertion is deferred to the recovery harness in #836 (noted there).

Key narrowing — exact type only. The safe Pemja branch is gated on PyBytes_CheckExact, so only exact bytes is safe; bytearray and bytes subclasses wrap as PyObject. This lines up exactly with the validator's existing exact-type check (type(value) in _CHECKPOINT_STABLE_SCALARS), so admitting bytes is a one-line addition: exact bytes is accepted, bytearray and bytes subclasses stay rejected for free.

PR: #846 (validator + accept/reject tests pinning the exact-type boundary + contract docs).