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

api/src/main/java/org/apache/flink/agents/api/EventType.java

@@ -40,5 +40,23 @@ public final class EventType {
     public static final String ContextRetrievalResponseEvent =
             org.apache.flink.agents.api.event.ContextRetrievalResponseEvent.EVENT_TYPE;
 
+    public static final String ShortTermWriteEvent =
+            org.apache.flink.agents.api.event.ShortTermWriteEvent.EVENT_TYPE;
+    public static final String ShortTermReadEvent =
+            org.apache.flink.agents.api.event.ShortTermReadEvent.EVENT_TYPE;
+    public static final String SensoryWriteEvent =
+            org.apache.flink.agents.api.event.SensoryWriteEvent.EVENT_TYPE;
+    public static final String SensoryReadEvent =
+            org.apache.flink.agents.api.event.SensoryReadEvent.EVENT_TYPE;
+    public static final String LongTermUpdateEvent =
+            org.apache.flink.agents.api.event.LongTermUpdateEvent.EVENT_TYPE;
+    public static final String LongTermGetEvent =
+            org.apache.flink.agents.api.event.LongTermGetEvent.EVENT_TYPE;
+    public static final String LongTermSearchEvent =
+            org.apache.flink.agents.api.event.LongTermSearchEvent.EVENT_TYPE;
+
+    public static final String AgentRunBeginEvent =
+            org.apache.flink.agents.api.event.AgentRunBeginEvent.EVENT_TYPE;
+
     private EventType() {}
 }

api/src/main/java/org/apache/flink/agents/api/agents/AgentExecutionOptions.java

@@ -48,6 +48,10 @@ public class AgentExecutionOptions {
     public static final ConfigOption<Boolean> RAG_ASYNC =
             new ConfigOption<>("rag.async", Boolean.class, true);
 
+    /** Opt-in lifecycle event emitted at the beginning of each agent run. */
+    public static final ConfigOption<Boolean> AGENT_RUN_BEGIN_EVENT =
+            new ConfigOption<>("agent-run.begin-event", Boolean.class, false);
+
     /** Set to a positive value in milliseconds to enable short-term memory TTL; 0 disables it. */
     public static final ConfigOption<Long> SHORT_TERM_MEMORY_STATE_TTL_MS =
             new ConfigOption<>("short-term-memory.state-ttl.ms", Long.class, 0L);

api/src/main/java/org/apache/flink/agents/api/configuration/MemoryEventOptions.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.api.configuration;
+
+/**
+ * Config options controlling memory observation events.
+ *
+ * <p>Resolution order per operation: the operation's own sub-key if explicitly configured, else the
+ * {@code memory.generate-event} master switch if explicitly configured, else the operation's
+ * built-in default (writes and all long-term ops: on; short-term/sensory reads: off).
+ */
+public class MemoryEventOptions {
+
+    /** Master switch. Fallback for unset sub-switches; when unset itself, per-op defaults apply. */
+    public static final ConfigOption<Boolean> MEMORY_GENERATE_EVENT =
+            new ConfigOption<>("memory.generate-event", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> SHORT_TERM_WRITE =
+            new ConfigOption<>("memory.generate-event.short-term-write", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> SHORT_TERM_READ =
+            new ConfigOption<>("memory.generate-event.short-term-read", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> SENSORY_WRITE =
+            new ConfigOption<>("memory.generate-event.sensory-write", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> SENSORY_READ =
+            new ConfigOption<>("memory.generate-event.sensory-read", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> LONG_TERM_UPDATE =
+            new ConfigOption<>("memory.generate-event.long-term-update", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> LONG_TERM_GET =
+            new ConfigOption<>("memory.generate-event.long-term-get", Boolean.class, null);
+
+    public static final ConfigOption<Boolean> LONG_TERM_SEARCH =
+            new ConfigOption<>("memory.generate-event.long-term-search", Boolean.class, null);
+
+    private MemoryEventOptions() {}
+}

api/src/main/java/org/apache/flink/agents/api/event/AgentRunBeginEvent.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.event;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.agents.api.Event;
+
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Lifecycle event marking the begin of one agent run. When enabled, it is emitted after an
+ * InputEvent arrives for a String key and before any action of that run executes. Carries the key's
+ * short-term VALUE nodes as a dot-key flat map. Object nodes and empty-object structure are
+ * intentionally outside this event contract.
+ *
+ * <p>This event is opt-in through {@code agent-run.begin-event} (default false), independent of the
+ * {@code memory.generate-event} master switch. On the wire: {@code {id, type, attributes: {key,
+ * value}}}.
+ */
+public class AgentRunBeginEvent extends Event {
+
+    public static final String EVENT_TYPE = "_agent_run_begin_event";
+
+    public AgentRunBeginEvent(String key, Map<String, Object> value) {
+        super(EVENT_TYPE, MemoryEvent.normalizeAttributes(key, value));
+    }
+
+    /** Deserialization constructor ({@code @JsonCreator}: see {@code MemoryEvent}'s note). */
+    @JsonCreator
+    public AgentRunBeginEvent(
+            @JsonProperty("id") UUID id,
+            @JsonProperty("attributes") Map<String, Object> attributes) {
+        super(id, EVENT_TYPE, MemoryEvent.normalizeAttributes(attributes));
+    }
+
+    /** Converts a generic {@link Event} of this type into the typed view. */
+    public static AgentRunBeginEvent fromEvent(Event event) {
+        AgentRunBeginEvent result = new AgentRunBeginEvent(event.getId(), event.getAttributes());
+        if (event.hasSourceTimestamp()) {
+            result.setSourceTimestamp(event.getSourceTimestamp());
+        }
+        return result;
+    }
+
+    @JsonIgnore
+    public String getKey() {
+        return (String) getAttr("key");
+    }
+
+    @SuppressWarnings("unchecked")
+    @JsonIgnore
+    public Map<String, Object> getValue() {
+        return (Map<String, Object>) getAttr("value");
+    }
+}

api/src/main/java/org/apache/flink/agents/api/event/LongTermGetEvent.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.event;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Memory observation event: a long-term memory read by id. See {@link MemoryEvent} for the
+ * attribute contract.
+ */
+public class LongTermGetEvent extends MemoryEvent {
+
+    public static final String EVENT_TYPE = "_long_term_get_event";
+
+    public LongTermGetEvent(String key, Map<String, Object> value) {
+        super(EVENT_TYPE, key, value);
+    }
+
+    /** Deserialization constructor ({@code @JsonCreator}: see {@link MemoryEvent}'s note). */
+    @JsonCreator
+    public LongTermGetEvent(
+            @JsonProperty("id") UUID id,
+            @JsonProperty("attributes") Map<String, Object> attributes) {
+        super(id, EVENT_TYPE, attributes);
+    }
+}

api/src/main/java/org/apache/flink/agents/api/event/LongTermSearchEvent.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.event;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Memory observation event: a semantic search over long-term memory. Unlike the other memory
+ * events, {@code value} maps each memory set to a map from query string to ordered hit list; {@link
+ * #getResults()} exposes that shape.
+ */
+public class LongTermSearchEvent extends MemoryEvent {
+
+    public static final String EVENT_TYPE = "_long_term_search_event";
+
+    public LongTermSearchEvent(String key, Map<String, Object> value) {
+        super(EVENT_TYPE, key, value);
+    }
+
+    /** Deserialization constructor ({@code @JsonCreator}: see {@link MemoryEvent}'s note). */
+    @JsonCreator
+    public LongTermSearchEvent(
+            @JsonProperty("id") UUID id,
+            @JsonProperty("attributes") Map<String, Object> attributes) {
+        super(id, EVENT_TYPE, attributes);
+    }
+
+    /** Typed view of {@link #getValue()}: memory set → query string → ordered hit list. */
+    @SuppressWarnings("unchecked")
+    @JsonIgnore
+    public Map<String, Map<String, List<Map<String, Object>>>> getResults() {
+        return (Map<String, Map<String, List<Map<String, Object>>>>) (Map<?, ?>) getValue();
+    }
+}

api/src/main/java/org/apache/flink/agents/api/event/LongTermUpdateEvent.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api.event;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Memory observation event: a long-term memory update (adds and deletes). Item changes are grouped
+ * by memory set in {@code value}; {@code cleared_sets} identifies whole-set deletes. See {@link
+ * MemoryEvent} for the common attribute contract.
+ */
+public class LongTermUpdateEvent extends MemoryEvent {
+
+    public static final String EVENT_TYPE = "_long_term_update_event";
+
+    public LongTermUpdateEvent(String key, Map<String, Object> value) {
+        this(key, value, Collections.emptyList());
+    }
+
+    public LongTermUpdateEvent(String key, Map<String, Object> value, List<String> clearedSets) {
+        super(EVENT_TYPE, attributes(key, value, clearedSets));
+    }
+
+    /** Deserialization constructor ({@code @JsonCreator}: see {@link MemoryEvent}'s note). */
+    @JsonCreator
+    public LongTermUpdateEvent(
+            @JsonProperty("id") UUID id,
+            @JsonProperty("attributes") Map<String, Object> attributes) {
+        super(id, EVENT_TYPE, normalizeUpdateAttributes(attributes));
+    }
+
+    private static Map<String, Object> attributes(
+            String key, Map<String, Object> value, List<String> clearedSets) {
+        Map<String, Object> attributes = new LinkedHashMap<>();
+        attributes.put("key", key);
+        attributes.put("value", value);
+        attributes.put(
+                "cleared_sets",
+                clearedSets == null ? Collections.emptyList() : List.copyOf(clearedSets));
+        return attributes;
+    }
+
+    private static Map<String, Object> normalizeUpdateAttributes(Map<String, Object> attributes) {
+        Map<String, Object> normalized = new LinkedHashMap<>(attributes);
+        Object clearedSets = normalized.get("cleared_sets");
+        if (clearedSets == null) {
+            normalized.put("cleared_sets", Collections.emptyList());
+        } else if (clearedSets instanceof List
+                && ((List<?>) clearedSets).stream().allMatch(String.class::isInstance)) {
+            normalized.put("cleared_sets", List.copyOf((List<?>) clearedSets));
+        } else {
+            throw new IllegalArgumentException(
+                    "Long-term update attribute 'cleared_sets' must be a list of strings.");
+        }
+        return normalized;
+    }
+
+    @SuppressWarnings("unchecked")
+    @JsonIgnore
+    public List<String> getClearedSets() {
+        Object clearedSets = getAttr("cleared_sets");
+        return clearedSets instanceof List ? (List<String>) clearedSets : Collections.emptyList();
+    }
+}