[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]

This puts the raw memory value object straight into the event's value map (update.getValue(), no normalization). Those events get drained into pendingEvents, persisted into ActionState.outputEvents, and re-emitted on replay via the completed-action path. The Kryo {serde,version,payload} envelope that preserves an untyped memory value's concrete type across durable recovery is bound class-scoped to MemoryUpdate.value (addMixIn(MemoryUpdate.class, ...) in ActionStateSerde) — it doesn't cover Event.attributes, which serializes through stock Jackson. So a byte[] short-term value (a supported type — ActionStateSerdeTest.testByteArrayMemoryValuePreserved) comes back as a base64 String after recovery, and a POJO as a LinkedHashMap. An action subscribing to _short_term_write_event would then see byte[] on the live run and String on the recovered run.

It only bites under a narrow combination — durable execution enabled, a non-JSON-native STM value, and an action that actually depends on the concrete type — so I don't think it blocks merge. And it may well be intentional: if memory events are audit-only, a base64 string is arguably fine. Is that the intent, or should the event path reuse the same envelope? Either way, testMemoryEventsSurviveActionStateRoundTrip currently uses only "gold" and 1 (both JSON-native, so type is trivially preserved) — a byte[] assertion there would pin whichever answer you land on.

[weiqingy]

There's an enforced parity guard that reflects a Java options class against its Python mirror and asserts the field names, keys, types, and defaults all match — check_java_python_config_options_parity.py, run in CI via test_java_config_in_python.sh. Its main() registers AgentConfigOptions and AgentExecutionOptions, but this new MemoryEventOptions family isn't wired in on either side. test_memory_event.py:104 looks like it fills that gap, but it checks the Python constants against hardcoded literals — it can't see Java, so by construction it can't catch Java↔Python drift. Given the 9 new keys and their tri-state null defaults are exactly the shape that's easiest to break asymmetrically, would registering the class in that checker's main() be worth it? normalize_java_default already maps a Java null default to Python None, so it should compare cleanly with no checker changes.

[weiqingy]

The "ALWAYS drain" invariant here holds for suppressed/disabled actions — they finish normally and hit this drain before the early returns. But flushMemoryObservation only runs from drainEvents(_, true), which the action tasks call only on normal finish (JavaActionTask:85-86, and the Python task). When an action throws after doing LTM ops, invoke() rethrows, drainEvents(_, true) is never reached, and the Python-side buffer for this key isn't drained — so the guarantee is actually narrower than "a suppressed or disabled action still drains the buffer" reads. In practice it looks masked: a thrown action fails the mailbox task → operator restart → the interpreter's buffer is recreated fresh. I traced that path and restart appears to be the only exit, but I couldn't prove it exhaustively — is operator restart guaranteed to be the only way out here? A small test driving an action that throws and asserting no memory events leak into the next same-key action would settle both the invariant and my uncertainty.

[weiqingy]

If the Pemja drain_ltm_observation_records call throws, this swallows it at DEBUG and moves on. Two things go quiet: this action's LTM observation is lost, and if the Python side threw before clearing its buffer, those records survive and get attributed to the next same-key action's flush — the exact cross-action leak the drain exists to prevent. A persistent drain failure would misattribute LTM records with no visible signal. Would a throttled WARN fit better here, so the condition is at least observable?

[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.