[Discussion] Memory Observation Capability Design

[rosemarYuan]

Memory Observability Design

Introduction

This doc introduces Memory Observability for Flink Agents — the ability to observe how an agent's memory changes as the agent runs.

The framework already provides event-based observability (EventLogger / EventListener), but memory reads and writes do not produce events today, so memory operations are invisible to it.

This doc proposes to close that gap by reusing the existing event-based observability: the framework records each memory operation as it happens, and at the action execution boundary converts the records into events, so that all existing tooling (logging, listening, routing, action triggering) works for memory out of the box.

The sections below will cover:

• The motivation and design goals.
• The API design: configuration options and the memory event format.
• How events are generated internally, and how the value payload is encoded.
• Rebuilding the full Short-Term / Sensory Memory state from the event log.

1. Motivation

Background

Flink Agents provides an event-based observability stack: users can record and export framework events with EventLogger / EventListener. Meanwhile, the framework defines three types of memory with the following operations:

• Sensory Memory / Short-Term Memory (STM): Read / Write
• Long-Term Memory (LTM): Add / Delete / Get / Search

Problem

Users can observe an agent's run results through events. But for agents that use memory, there is currently no way to observe how memory evolves inside the agent.

Goal

Use the existing event-based observability and turn memory operation into a memory-event. Concretely: record each memory operation synchronously as it happens, then converting those records into events at the action boundary. Existing observability tooling then applies to memory unchanged.

• Keep the existing event observation and routing logic untouched.
• Optionally support per-memory-type, per-operation observation.

// Goal: N memory operations inside one action → a set of subscribable memory-events

user action :  memory operations ──▶ action boundary
            [recorded synchronously]       │
                                           ▼
                        [converted into N types of memory events]
                                           │
              ┌────────────────────────────┼────────────────────────────┐
              ▼                            ▼                            ▼
        EventLogger                 EventListener            routing + action triggering

Target usage:

// Enable memory observability
AgentConfiguration config = new AgentConfiguration();
config.set(MemoryEventOptions.MEMORY_GENERATE_EVENT, true);

// From now on, the memory operations of every action are emitted as
// _<scope>_<operation>_event at the action boundary. For example, the user
// can subscribe to short-term-memory writes of the user.id field:
@Action("type == '_short_term_write_event' && 'user.id' in value")
public void onIdChanged(Event event, RunnerContext ctx) { ... }

2. API Design

2.1 Configuration Options

By default, memory observability generates a set of commonly useful events. Without any configuration, users can observe the high-value memory operations: short-term writes, sensory writes, long-term updates, long-term gets, and long-term searches.

Users who need to adjust the observation scope can control whether each type of memory operation generates events via AgentConfiguration. The framework provides a master switch and per-operation switches:

• memory.generate-event: the master switch. It has no default value; when a per-operation switch is not configured, the resolution falls back to the master switch's configured value.
• memory.generate-event.<scope>-<operation>: per-operation switches, each controlling whether one specific kind of memory operation generates events.

Option key	Default behavior (when neither the per-operation switch nor the master switch is configured)	Description
memory.generate-event	— (no default value)	Master switch for memory events. Acts as the fallback when a per-operation switch is not configured; if the master switch is not configured either, the operation's own default behavior applies
memory.generate-event.short-term-write	record	Whether to record short-term memory writes
memory.generate-event.short-term-read	do not record	Whether to record short-term memory reads
memory.generate-event.sensory-write	record	Whether to record sensory memory writes
memory.generate-event.sensory-read	do not record	Whether to record sensory memory reads
memory.generate-event.long-term-update	record	Whether to record long-term memory updates (both writes and deletes)
memory.generate-event.long-term-get	record	Whether to record long-term memory gets
memory.generate-event.long-term-search	record	Whether to record long-term memory searches

Per-operation switches take precedence over the master switch. The effective behavior is resolved in the following order:

1. If the per-operation switch is configured, use its configured value.
2. Otherwise, if the master switch memory.generate-event is explicitly configured, use the master switch's configured value.
3. Otherwise (neither is configured), use the operation's own default behavior (see the table above).

Usage examples:

// Default behavior (nothing configured): write operations and LTM operations
// are recorded, read operations are not
AgentConfiguration config = new AgentConfiguration();

// Turn off memory observability entirely
config.setBool("memory.generate-event", false);

// Enable only selected operations: turn the master switch off, then observe
// only long-term updates and short-term writes
config.setBool("memory.generate-event", false);
config.setBool("memory.generate-event.long-term-update", true);
config.setBool("memory.generate-event.short-term-write", true);

2.2 Memory Events

A memory event is a regular event that can be routed and can trigger actions like any other event. It contains the following fields:

• id: the UUID of the event.
• type: the kind of memory operation, one of:
  • _short_term_write_event: a write to short-term memory
  • _short_term_read_event: a read from short-term memory
  • _sensory_write_event: a write to sensory memory
  • _sensory_read_event: a read from sensory memory
  • _long_term_update_event: an update to long-term memory, covering both add and delete
  • _long_term_get_event: a get from long-term memory by id
  • _long_term_search_event: a semantic search over long-term memory
• key: the Flink keyed-state key this operation belongs to, identifying which key the operation happened under.
• value: the content of the memory operation. Concretely:
  • for write events, the data that was written;
  • for read events, the data that was returned.

3. Design

3.1 Event Generation

At runtime the framework turns every memory operation into an event in two steps: it records the operation synchronously when it happens, then converts all records into events at the action boundary.

Recording — synchronous, at the operation site. Recording is done inside the framework's memory operation implementations. The records are written into the action state and persisted with checkpoints. Users do not notice it, and no API changes.
Emission — at the action boundary. When an action finishes executing, all memory records produced by that execution are converted into multiple observation events, grouped by "scope-operation":

{
  "id":    "<UUID>",                    // UUID of the event
  "type":  "_short_term_write_event",   // memory operation type
  "key":   "key1",                      // the keyed-state key of this operation
  "value": { ... }                      // content of the memory operation
}
{
  "id":    "<UUID>",
  "type":  "_long_term_search_event",
  "key":   "key1",
  "value": { ... }
}

When an action performs multiple operations of the same kind, the records are merged by the following rules:

• Operations on different fields: coexist in one event, keyed by field: {k1: v1, k2: v2, ...}.
• Operations on the same field: only the last value is kept.

In addition, memory operations performed inside an action that was itself triggered by a memory event do not generate new observation events. This prevents infinite feedback loops.

3.2 The value Field

A MemoryEvent carries the fields **id**, **type**, **key**, and **value**:

{
  "id":    "<UUID>",                       // UUID of the event
  "type":  "_<scope>_<operation>_event",   // memory operation type
  "key":   "<keyed-state key>",            // the Flink keyed-state key of this operation
  "value": { ... }                         // updates / reads
}

For **value**, three internal encodings were considered. Take recording an STM write of user.tier = "gold" and user.address.city = "SF" as an example:

Candidate	Shape	How a downstream action subscribes to a field	Pros	Cons
Leaf names	{"tier":"gold","city":"SF"}	has(tier)	Simple and intuitive	❌ Leaf names are not globally unique — ambiguity cannot be resolved (e.g. user.id vs address.id)
Dotted keys	{"user.tier":"gold",``"user.address.city":"SF"}	"user.tier" in value	Unambiguous semantics	Some redundancy: entries sharing a common prefix are not merged
Nested map	{"user":{ "tier":"gold", "address":``"city":"SF"}}}	has(value.user.tier)	1. Structured, more readable. 2. Shared prefixes are merged, saving space	❌ Cannot tell which key was operated on — the semantics are ambiguous.E.g. for {a: {b: 1, c: 2}}, the following two operation sequences are indistinguishable:1. set(a.b, 1); set(a.c, 2)2. set(a, {b: 1, c: 2})

Since the nested-map encoding cannot distinguish writing multiple leaf values field by field from writing a whole map at once, we choose the dotted-key encoding:

// ①  STM write — writes the two leaves user.tier and user.address.city
{ "type": "_short_term_write_event", "id": "<uuid>", "key": "user-42",
  "value": { "user.tier": "gold",           // add value
             "user.address.city": {
                  "Shanghai": 1201,
                  "Beijing" : 2021 }        // add map
                  } }

// ②  STM read — reads user.address.city
{ "type": "_short_term_read_event", "id": "<uuid>", "key": "user-42",
  "value": { "user.address.city": "SF" } }

// ③  LTM update — key = "memorySet.itemId", value = content; null = delete
{ "type": "_long_term_update_event", "id": "<uuid>", "key": "user-42",
  "value": {
    "user_prefs.m_8f3": "user prefers email notifications",   // add
    "profile.m_a01":    null                                  // delete
  } }

// ④  LTM get (by id)
{ "type": "_long_term_get_event", "id": "<uuid>", "key": "user-42",
  "value": {
    "user_prefs.m_8f3": "user prefers email notifications",
    "user_prefs.m_a01": "user is located in SF"
  } }

// ⑤  LTM search — not dotted-key encoded; keeps the query → ordered result list form
{ "type": "_long_term_search_event", "id": "<uuid>", "key": "user-42",
  "value": {
    "refund policy": [
      { "id": "p_01", "value": "7-day no-questions-asked refund", "score": 0.92 },
      { "id": "p_02", "value": "customized items are non-refundable", "score": 0.81 }
    ]
  } }

4. Rebuilding Short-Term / Sensory Memory State from the Event Log

Besides observing memory at runtime and triggering new actions, there is a second class of demand: during log-based troubleshooting, reconstructing the memory state at a given moment from the event log.

The basic idea: since the framework records every memory update from job start, applying the updates in order on top of a known initial state restores the state at any given moment. Feasibility differs per memory type, though:

• LTM: user-written data goes through LLM processing (compaction, merging, deletion), so the state cannot be rebuilt by applying updates to an initial state.
• Sensory Memory: automatically cleared at the end of every agent run, so applying the write records of the current run is sufficient.
• STM: its content keeps evolving for the lifetime of the job. Replaying from job start makes the rebuild cost grow linearly with job age, so periodic full snapshots are needed as rebuild starting points.

The rest of this section discusses when to record the full snapshots and how the rebuild works.

4.1 When to Record Full Memory Snapshots

The rebuild needs periodic full STM snapshots as starting points, i.e. recording the complete STM state per key every so often.

Chosen: record at agent-run begin. Whenever an input for some key arrives and an agent run is about to start, record a full STM snapshot for that key. Concretely, when an InputEvent arrives, the framework emits an AgentRunBeginEvent marking the start of the agent run. The event carries the full STM as of the input's arrival, to be used for state reconstruction from the event log. The memory payload format is the same as the value structure in §3.2.
Rejected alternative: record at checkpoint time. On checkpoint completion (notifyCheckpointComplete), iterate over all keys held by the subtask via applyToAllKeys and record a full STM snapshot per key. This was rejected because the checkpointing path already carries a lot of responsibilities.

Usage:

// Enable AgentRunBeginEvent
config.set(MemoryEventOptions.AGENT_RUN_BEGIN_EVENT, true);

Configuration:

// AgentRunBeginEvent is a run-lifecycle event. It is not governed by the
// memory event master switch, and is recorded by default.
public static final ConfigOption<Boolean> AGENT_RUN_BEGIN_EVENT =
        new ConfigOption<>("agent-run.begin-event", Boolean.class, true);

4.2 How the Rebuild Works

Since every agent run starts with a full STM snapshot, the rebuild needs neither a standalone command-line tool nor replaying long changelogs merged across runs. To rebuild the state of a key at time t:

1. Locate, in the event log, the latest full snapshot of that key before t (i.e. the AgentRunBeginEvent of the run containing t).
2. Starting from that snapshot, apply the _short_term_write_events of that run before t in order — this yields the state at t. If t is exactly a run boundary, the snapshot itself is the answer.

The rebuild cost is proportional to the number of operations within a single run and does not grow with job age.

Likewise, since Sensory Memory always starts empty, applying the write records of the current run reconstructs its state at any moment.

Future Works

• A safety valve for large value payloads (e.g. size limits / truncation).
• Optionally recording the previous value alongside the new value for richer diffs.
• Recording full snapshots every N runs (counter-based) instead of every run, to reduce snapshot overhead.
• An AgentRunEndEvent complementing AgentRunBeginEvent.