[api][runtime][python] Add memory events for agent memory observability

[rosemarYuan]

Linked issue: #886

Purpose of change

Follow-up to #876, #886 .This change makes agent memory activity observable through the existing event infrastructure.

Emission model.

• Memory events are emitted at the action finish boundary, together with the action's user events.
• The framework emits at most one event per memory scope and operation kind, for example, one short-term write event per action.
• Each event's value is a dot-key flat map describing the operation's net effect. Within one action, the last read/write for the same path wins.
• Failed or unfinished actions emit no memory events.
• Actions triggered by memory events do not emit memory events themselves, preventing recursive subscriptions.

Recovery semantics. Generation remains consistent with the framework's existing delivery and durability guarantees:

• At-least-once on replay. After a failure, a replayed completed action may re-log its persisted memory events, and a replayed input may re-emit its run-begin event with restored short-term memory. Event Log consumers should tolerate duplicates.
• Durable execution. A long-term operation wrapped in durable_execute is not recorded again when its result is replayed from state. Generated events describe operations that actually executed in the current attempt.

Tests

Coverage targets the design-critical paths:

• Cross-language wire-format conformance — MemoryEventWireFormatTest in Java and test_memory_event_wire_format.py in Python pin the same attribute JSON, preventing SDK drift.
• Flush semantics — TestMemoryObservationFlush covers emit-once-per-finished-action behavior, no emission on unfinished/failed drains, config gating, suppression awareness, and drain-then-discard semantics. A suppressed or disabled action still drains the long-term buffer, so records cannot leak into the next same-key action.
• Config fallback — TestMemoryEventSettings covers all three resolution levels, including the case where the master switch is off but an operation-specific sub-key is explicitly on.
• Dot-key folding — TestMemoryEventBuilder covers empty paths, single paths, nested paths, last-wins behavior, and long-term add/delete/get/search value shapes.
• Polymorphic serde — ActionStateSerdeTest verifies that memory events round-trip through the @class mixin, so they can be persisted and replayed with the action's output events.
• Agent-run begin — AgentRunBeginEventEmissionTest, AgentRunBeginEventTest, and test_run_event cover snapshot content, per-key isolation, and the guarantee that the snapshot is taken before the run's own writes.
• Long-term recording in Python — test_mem0_op_recording and test_mem0_recording_hook cover thread-safe buffering and drain-by-key behavior, including re-buffering of non-matching keys.
• TTL default — ShortTermMemoryTTLIntegrationTest pins the new ON_CREATE_AND_WRITE default.
• End-to-end logging — memory_event_logging_test.py exercises the full emit → Event Log path.

API

Java:

• org.apache.flink.agents.api.event.MemoryEvent
• org.apache.flink.agents.api.event.ShortTermWriteEvent
• org.apache.flink.agents.api.event.ShortTermReadEvent
• org.apache.flink.agents.api.event.SensoryWriteEvent
• org.apache.flink.agents.api.event.SensoryReadEvent
• org.apache.flink.agents.api.event.LongTermUpdateEvent
• org.apache.flink.agents.api.event.LongTermGetEvent
• org.apache.flink.agents.api.event.LongTermSearchEvent
• org.apache.flink.agents.api.event.AgentRunBeginEvent
• The corresponding EventType.* constants
• org.apache.flink.agents.api.configuration.MemoryEventOptions

Python:

• flink_agents.api.events.memory_event
• flink_agents.api.events.run_event
• flink_agents.api.core_options.MemoryEventOptions

The new memory-event surface — event classes, EventType.* constants, and MemoryEventOptions — does not depend on any long-term-memory backend.

The observation recording seam is a runtime-side default-method extension point: InternalBaseLongTermMemory.drainObservationRecordsJson, whose default implementation returns []. A future pure-Java backend can therefore implement its own recording without another API change.

Configuration. Memory event generation is controlled per operation kind. Configuration is resolved in the following order:

1. Operation-specific sub-key.
2. memory.generate-event master switch.
3. Built-in per-operation default.

The built-in defaults are:

• Writes: on.
• Long-term operations: on.
• Short-term/sensory reads: off.

The master switch and operation-specific switches intentionally use null defaults, so the built-in defaults remain reachable.
agent-run.begin-event has its own switch, defaults to on, and is independent of the memory.generate-event master switch.

Notable behavior change.

• The repo-wide default of short-term-memory.state-ttl.update-type changes from ON_READ_AND_WRITE to ON_CREATE_AND_WRITE.
• The run-begin event is enabled by default and scans the key's full short-term memory on every input. Under ON_READ_AND_WRITE, that scan would refresh every entry's TTL and prevent expiry. Users who require read-refresh TTL semantics can set the option back explicitly, or disable agent-run.begin-event. See the documentation's Semantics & Caveats section for details.

Documentation

• doc-needed
• doc-not-needed
• doc-included Documentation is included in `docs/content/docs/development/memory/memory_events.md
  This change also adds notes to the sensory/short-term memory page and the configuration reference.

[zxs1633079383]

I took a light design pass, mainly around whether the new memory events pollute user payloads or collide with the event-log lineage direction.

The current shape looks good to me:

• Memory observations are emitted as separate typed events (_short_term_write_event, _long_term_search_event, etc.) rather than mutating the triggering user event or putting framework fields into user-owned event attributes.
• The memory event payload is scoped under attributes: { key, value }, so the observation payload is explicit and reconstructable without changing the user event schema.
• The action-finish flush boundary is a good place to fold per-action memory records into net-effect events, and the suppression for memory-triggered actions avoids recursive observation noise.
• AgentRunBeginEvent is kept as a lifecycle/reconstruction anchor rather than overloading memory events themselves.

One thing I would keep intentionally separate from this PR: lineage/run identity fields from #841. This PR should probably continue to avoid adding runId, sourceEventId, emittingAction, etc. directly into the memory event value payload. If/when the event-log lineage schema lands, those fields should sit in event context/envelope metadata, not inside the user-facing memory observation value map. The docs already point in that direction by describing key/value as the memory event attributes, so this is more of a boundary note than a blocker.

[rosemarYuan]

Thanks for the design review @zxs1633079383 — agreed, and that is exactly the boundary I want to keep.

For this PR, the memory event payload under attributes: { key, value } stays observation-only: it only describes the memory content/effect being observed. No #841 lineage or run-identity fields such as runId, sourceEventId, or emittingAction should be added inside the memory event value/attributes map.

On the lineage direction, we are aligned. If/when the event-log lineage schema lands, those fields should be carried as event-log metadata/envelope-level information and apply uniformly to logged events, memory events included. The exact carrier shape, for example whether it is a separate lineage object on the log record or another context/envelope mechanism, belongs in the lineage design thread rather than this PR.

Same principle you described: memory events expose memory observation data; lineage stays outside the user-facing event payload.

[weiqingy]

Thanks for taking this on — the ordering in flushMemoryObservation is a nice call: draining the LTM buffer unconditionally before the suppressed/disabled early returns (RunnerContextImpl.java:226-243) is what stops a suppressed action from leaking records into the next same-key action. A few questions inline, plus one design-level one up top.

1. agent-run.begin-event defaulting to true (MemoryEventOptions.java:58) carries two coupled costs that might be worth weighing together, since they share that one root. First, every input record now pays a full MapState.entries() scan of the key's short-term memory in maybeEmitAgentRunBeginEvent (ActionExecutionOperator.java:299-317) plus an AgentRunBeginEvent allocation carrying the whole STM snapshot — even when the EventLog is disabled and nothing subscribes to consume it. On a RocksDB backend that's a state-backend iterator per record, not a cheap in-memory read. Second, because that scan reads every entry on every input and STM state is TTL-configured for read-refresh, the PR flips the repo-wide default of short-term-memory.state-ttl.update-type from ON_READ_AND_WRITE to ON_CREATE_AND_WRITE (AgentExecutionOptions.java:63-67) so expiry can still fire. That flip is global and unconditional — disabling agent-run.begin-event doesn't restore the old read-refresh semantics, so a user who enables STM TTL and never touches memory events still sees changed expiry behavior.

   Was defaulting the begin-event to false — or gating emission on the EventLog being configured — considered? Either would drop the per-record scan for everyone not reconstructing STM from the log, and would make the global TTL default flip unnecessary, while still letting the feature be turned on when someone wants log-based reconstruction. I can also see the counter-argument: if the snapshot only starts when you enable it, there's no reconstruction anchor for runs that happened before you flipped it on, which is a real reason to keep it on by default. So this is genuinely a question about which default serves the common case better, not a claim that on-by-default is wrong — curious how you weighed the two.

2. docs/content/docs/operations/configuration.md:128 introduces its table as "the list of all built-in core configuration options", but the family this PR adds isn't in it — the 8 memory.generate-event[.*] keys and agent-run.begin-event are absent (grepping the page for begin-event/agent-run returns nothing). They're documented in the new memory_events.md, so this is a completeness gap on the page that advertises itself as exhaustive, not a docs hole. Of the omitted keys, agent-run.begin-event is the one a reader is most likely to come to this page looking for — its default is exactly what makes the TTL default flip in item (1) necessary. Adding the 9 keys to the table (or softening the "all" claim) would close it.

[weiqingy]

DELETE_SET adds set -> null but doesn't purge the earlier set + "." + id entries folded in above, so ADD s.m1=v then DELETE_SET s within one action yields {"s.m1": v, "s": null}. A log consumer reconstructing state from this event would see s.m1=v as a live member even though the same event's s -> null says the set — and that member with it — was deleted. It's observability-only, but the emitted value misreports the member. Purging the set + "." prefix entries when folding in a DELETE_SET would keep the reconstruction accurate.

[rosemarYuan]

Hi @weiqingy, sorry for the delayed response. I was working on the Action Condition changes last week, so I didn’t get a chance to address this PR sooner. I’ve now addressed all the review comments and added the corresponding tests. Thanks again for the thorough review.

For begin-event, I hadn’t fully considered the cost you pointed out: enabling it by default makes every input scan the full STM state, even when log-based reconstruction is not needed. Given that overhead, I changed begin-event to be opt-in (false by default) and restored the original ON_READ_AND_WRITE TTL default. Users who need an STM reconstruction anchor can still enable the event explicitly.

[weiqingy]

Thanks for working through all of these, they all look resolved to me. Left one optional nit on the drain-logging thread, otherwise nothing further from me.

[wenjin272]

Thanks for taking this on @rosemarYuan. I left some comments.

[wenjin272]

With this wiring, could we remove the existing Python-side context switch in PythonActionExecutor.executePythonFunction()?

ActionTaskContextManager.createAndSetRunnerContext() calls RunnerContextImpl.switchActionContext() before every action execution or coroutine resume. Now that PythonRunnerContextImpl holds this Mem0LongTermMemory wrapper, that call
already switches the same Python Mem0 instance using the same key and also forwards the observation flags.

The following invocation in executePythonFunction() therefore appears redundant and switches the LTM context a second time for the initial Python action:

interpreter.invoke(
        FLINK_RUNNER_CONTEXT_SWITCH_ACTION_CONTEXT,
        pythonRunnerContext,
        hashOfKey);

Could we remove this invocation, the now-unused hashOfKey parameter, and the corresponding Python helper so that LTM context switching has a single lifecycle path?

[wenjin272]

Could we avoid restricting framework memory events to String-keyed streams?

Flink guarantees that the current key is non-null, so for Java inputs we can derive the event key text directly with:

String.valueOf(key)

For PyFlink inputs, getCurrentKey() returns a single-field Row produced by KeyByKeySelector. We should first unwrap that field:

• If an explicit key_type was configured, the field is already the corresponding Java value and can use String.valueOf(logicalKey).
• If no key_type was configured, PyFlink uses PICKLED_BYTE_ARRAY; the byte[] should be decoded to the original Python key and then converted with str().

The current implementation accepts only logical String keys, so common keys such as Long or Integer silently disable all framework memory events and AgentRunBeginEvent. Could we instead support the string representation of
any logical Flink key and document that the event's key attribute is an observational representation rather than a routing key?

[wenjin272]

Could we avoid resolving eventKeyText inside processActionTaskForKey()?

The event key is invariant for the entire run, but this method is invoked once for every action task and coroutine/continuation resume. For PyFlink keys, resolveEventKeyText() may need to unwrap the key Row, decode pickle data,
and invoke the Python interpreter, so repeatedly resolving the same key is unnecessary.

Could we resolve it once in processEvent() when an InputEvent actually begins processing, then pass it to both tryEmitAgentRunBeginEvent() and the action-task processing chain?

tryProcessActionTaskForKey(key, eventKeyText);
processActionTaskForKey(key, eventKeyText);

When additional action tasks or continuations are scheduled, they can forward the same value. For checkpoint recovery, tryResumeProcessActionTasks() can resolve it once per restored processing key before submitting the first
mailbox task.

This would avoid repeated cross-language key decoding without introducing an additional cache or checkpointed state.

[wenjin272]

Could we use a single resolved key here instead of passing both ltmPartitionKey and eventKeyText?

Both values are derived from the same Flink key and identify the same run context. If the logical key is resolved and converted to text once, that value could be used for both the LTM context and framework events:

switchActionContext(
        actionName,
        memoryContext,
        contextKey,
        observationSuppressed);

This would simplify the contract and avoid using key.hashCode() as a separate LTM identity, which introduces an additional collision risk.

[wenjin272]

The signature here is inconsistent with InternalBaseLongTermMemory.switch_context(), which only declares the key parameter. Could we keep the parent and subclass signatures aligned?

[wenjin272]

Is this check necessary? Mem0LongTermMemory is created with the FlinkRunnerContext constructed by create_flink_runner_context(), where _j_runner_context is always initialized with the Java runner context. Therefore, it cannot be None on this path.

[wenjin272]

The per-operation observation settings are resolved from the AgentPlan and remain constant for the lifetime of the job. With the logical key always available, the only action-specific state here is observationSuppressed.

Could we configure the update/get/search settings once when initializing the LTM, and make switchActionContext() only switch the key and suppression state? This would separate job-level configuration from action-level context switching.