You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "addu390 (via GitHub)" <gi...@apache.org> on 2026/05/27 19:50:14 UTC
[PR] [Feature] Cross Language Actions [flink-agents]
addu390 opened a new pull request, #709:
URL: https://github.com/apache/flink-agents/pull/709
Linked issue: #622
### Purpose of change
Add symmetrical cross-language action support so a Java agent can declare a Python-bodied action via `@Action(target=@PythonFunction(...))` and a Python agent can declare a Java-bodied action via `@action(target=JavaFunction(...))`. Aligns built-in event SerDe (`Event`, `ChatMessage`, `MessageRole`, `ToolResponse`, `Document`) so events round-trip across the Java<->Python wire.
### Tests
- Plan-compile and JSON-shape tests on both sides (`AgentPlanCrossLanguageTest`, `test_agent_plan_cross_language`).
- Event snapshot round-trip tests against shared fixtures under `e2e-test/cross-language-event-snapshots/{java,python}/`.
- Runtime dispatch tests (`CrossLanguageActionRuntimeTest`, `test_local_runner_cross_language`).
- Flink mini-cluster e2e: `JavaAgentWithPythonActionTest` and `python_agent_with_java_action_test`.
### API
Yes:
- Java: `@Action.target()` field, new `@PythonFunction` annotation.
- Python: `@action(target=...)` kwarg.
### Documentation
- [x] `doc-not-needed`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong merged PR #709:
URL: https://github.com/apache/flink-agents/pull/709
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on PR #709:
URL: https://github.com/apache/flink-agents/pull/709#issuecomment-4559284765
Hi @wzhero1, please review when you get a chance. Noticed that cross-language actions is planned to be part of `0.3` with code-freeze on May 31st. Although close, I think it's possible to target.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3318031018
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapter.java:
##########
@@ -180,6 +180,18 @@ public Object invokeJavaTool(
return response.getResult();
}
+ /** Invoke a Java static action method with positional arguments from Python. */
+ public Object invokeJavaAction(
+ String className,
+ String methodName,
+ List<String> parameterTypes,
+ List<Object> arguments)
+ throws Exception {
+ Method method = resolveMethod(className, methodName, parameterTypes);
+ Object[] args = arguments == null ? new Object[0] : arguments.toArray();
+ return method.invoke(null, args);
Review Comment:
`invoke(null, args)` assumes the resolved target is `static`, but `Class.getMethod(...)` doesn't enforce that. A `JavaFunction` whose `method_name` points at an instance method resolves fine here and then throws `NullPointerException` from `Method.invoke` with a message that says nothing about the static requirement — hard to diagnose from the user side.
Could we check `Modifier.isStatic(method.getModifiers())` at resolution time and throw a message naming the method (mirroring `FunctionTool`'s static check), so the misconfiguration surfaces where the user can act on it?
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assertTrue(
+ Files.exists(committed),
+ "Java snapshot "
+ + fileName
+ + " missing from "
+ + committed
+ + ". If you added a new event, regenerate with -Dregenerate.snapshots=true and commit alongside the test.");
+ JsonNode expected = MAPPER.readTree(Files.readString(committed));
+
+ assertEquals(
+ expected,
+ actual,
+ "Java serialization of "
+ + fileName
+ + " drifted from committed snapshot; if intentional, regenerate.");
+ }
+
+ private static Event readPythonSnapshot(String fileName) throws Exception {
+ Path pythonSnapshot = snapshotDir.resolve("python/" + fileName);
+ assertTrue(
+ Files.exists(pythonSnapshot),
+ "Python snapshot "
+ + fileName
+ + " missing from "
+ + pythonSnapshot
+ + ". Regenerate the Python side with REGENERATE_SNAPSHOTS=1 and commit alongside this test.");
+ return Event.fromJson(Files.readString(pythonSnapshot));
+ }
+
+ // ── InputEvent ─────────────────────────────────────────────────────────
+
+ private static InputEvent buildInputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("input", "hello");
+ return new InputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateInputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void inputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeInputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("input_event.json");
+ InputEvent typed = InputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python InputEvent.");
+ assertEquals(InputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("hello", typed.getInput(), "InputEvent.input mismatch.");
+ }
+
+ // ── OutputEvent ────────────────────────────────────────────────────────
+
+ private static OutputEvent buildOutputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("output", "world");
+ return new OutputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateOutputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void outputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeOutputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("output_event.json");
+ OutputEvent typed = OutputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python OutputEvent.");
+ assertEquals(OutputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("world", typed.getOutput(), "OutputEvent.output mismatch.");
+ }
+
+ // ── ChatRequestEvent ───────────────────────────────────────────────────
+
+ private static ChatRequestEvent buildChatRequestEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("messages", List.of(new ChatMessage(MessageRole.USER, "hello world")));
+ return new ChatRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void chatRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_request_event.json");
+ ChatRequestEvent typed = ChatRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ assertNotNull(typed.getMessages());
+ assertEquals(1, typed.getMessages().size(), "Expected one message.");
+ ChatMessage msg = typed.getMessages().get(0);
+ assertEquals(MessageRole.USER, msg.getRole(), "Role mismatch on Python-produced message.");
+ assertEquals("hello world", msg.getContent());
+ }
+
+ @Test
+ void chatRequestOutputSchemaWireFormatIsJavaShaped() throws Exception {
+ OutputSchema schema =
+ new OutputSchema(
+ new RowTypeInfo(
+ new TypeInformation[] {BasicTypeInfo.STRING_TYPE_INFO},
+ new String[] {"name"}));
+ ChatRequestEvent event =
+ new ChatRequestEvent(
+ "test-model", List.of(new ChatMessage(MessageRole.USER, "hi")), schema);
+ String json = MAPPER.writeValueAsString(event);
+
+ assertTrue(json.contains("\"fieldNames\""), "Java wire format uses `fieldNames`.");
+ assertFalse(json.contains("\"names\""), "Java wire format does not use Python's `names`.");
+ }
+
+ // ── ChatResponseEvent ──────────────────────────────────────────────────
+
+ private static ChatResponseEvent buildChatResponseEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("request_id", FIXED_REQUEST_ID);
+ attrs.put("response", new ChatMessage(MessageRole.ASSISTANT, "hi there"));
+ attrs.put("retry_count", 0);
+ attrs.put("total_retry_wait_sec", 0);
+ return new ChatResponseEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatResponseEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_response_event.json", buildChatResponseEvent());
+ }
+
+ @Test
+ void chatResponseEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_response_event.json", buildChatResponseEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatResponseEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_response_event.json");
+ ChatResponseEvent typed = ChatResponseEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatResponseEvent.EVENT_TYPE, typed.getType());
+ assertEquals(FIXED_REQUEST_ID, typed.getRequestId(), "request_id mismatch.");
+ ChatMessage response = typed.getResponse();
+ assertNotNull(response, "response field is null.");
+ assertEquals(MessageRole.ASSISTANT, response.getRole(), "Role mismatch on response.");
+ assertEquals("hi there", response.getContent());
+ }
+
+ // ── ToolRequestEvent ───────────────────────────────────────────────────
+
+ private static ToolRequestEvent buildToolRequestEvent() {
+ Map<String, Object> toolCall = new LinkedHashMap<>();
+ toolCall.put("id", FIXED_TOOL_CALL_ID);
+ toolCall.put("name", "echo");
+ toolCall.put("arguments", Map.of("value", "ping"));
+
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("tool_calls", List.of(toolCall));
+ return new ToolRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateToolRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("tool_request_event.json", buildToolRequestEvent());
+ }
+
+ @Test
+ void toolRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("tool_request_event.json", buildToolRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeToolRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("tool_request_event.json");
+ ToolRequestEvent typed = ToolRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ToolRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ List<Map<String, Object>> toolCalls = typed.getToolCalls();
+ assertNotNull(toolCalls);
+ assertEquals(1, toolCalls.size());
+ assertEquals(FIXED_TOOL_CALL_ID, toolCalls.get(0).get("id"));
+ }
+
+ // ── ToolResponseEvent ──────────────────────────────────────────────────
+
+ private static ToolResponseEvent buildToolResponseEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("request_id", FIXED_REQUEST_ID);
+ attrs.put("responses", Map.of(FIXED_TOOL_CALL_ID, ToolResponse.success("pong")));
+ attrs.put("success", Map.of(FIXED_TOOL_CALL_ID, true));
+ attrs.put("error", new HashMap<String, String>());
+ attrs.put("external_ids", new HashMap<String, String>());
+ attrs.put("timestamp", FIXED_TIMESTAMP);
+ return new ToolResponseEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateToolResponseEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("tool_response_event.json", buildToolResponseEvent());
+ }
+
+ @Test
+ void toolResponseEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("tool_response_event.json", buildToolResponseEvent());
+ }
+
+ @Test
+ void pythonToolResponseEventLosesDataWhenConsumedByJava() throws Exception {
Review Comment:
This test pins the data loss as the expected contract: `ToolResponseEvent.fromEvent` only handles `ToolResponse`/`Map` values (no `String` branch), so a Python action emitting `responses={"call_id": "pong"}` (string-valued, which Python does emit) yields an empty `responses` map on the Java side with no exception — the core Python-action → Java-action path. Because this asserts the loss, the suite stays green while the data is gone and a fix has to break this test first.
Could we add a `String` branch in `fromEvent` so string responses survive, or — if they're genuinely out of scope for 0.3 — rename this to make the gap explicit (e.g. `..._knownGap_NotYetSupported`) and note it in the PR description rather than encoding silent loss as the expected behavior?
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assertTrue(
+ Files.exists(committed),
+ "Java snapshot "
+ + fileName
+ + " missing from "
+ + committed
+ + ". If you added a new event, regenerate with -Dregenerate.snapshots=true and commit alongside the test.");
+ JsonNode expected = MAPPER.readTree(Files.readString(committed));
+
+ assertEquals(
+ expected,
+ actual,
+ "Java serialization of "
+ + fileName
+ + " drifted from committed snapshot; if intentional, regenerate.");
+ }
+
+ private static Event readPythonSnapshot(String fileName) throws Exception {
+ Path pythonSnapshot = snapshotDir.resolve("python/" + fileName);
+ assertTrue(
+ Files.exists(pythonSnapshot),
+ "Python snapshot "
+ + fileName
+ + " missing from "
+ + pythonSnapshot
+ + ". Regenerate the Python side with REGENERATE_SNAPSHOTS=1 and commit alongside this test.");
+ return Event.fromJson(Files.readString(pythonSnapshot));
+ }
+
+ // ── InputEvent ─────────────────────────────────────────────────────────
+
+ private static InputEvent buildInputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("input", "hello");
+ return new InputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateInputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void inputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeInputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("input_event.json");
+ InputEvent typed = InputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python InputEvent.");
+ assertEquals(InputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("hello", typed.getInput(), "InputEvent.input mismatch.");
+ }
+
+ // ── OutputEvent ────────────────────────────────────────────────────────
+
+ private static OutputEvent buildOutputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("output", "world");
+ return new OutputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateOutputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void outputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeOutputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("output_event.json");
+ OutputEvent typed = OutputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python OutputEvent.");
+ assertEquals(OutputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("world", typed.getOutput(), "OutputEvent.output mismatch.");
+ }
+
+ // ── ChatRequestEvent ───────────────────────────────────────────────────
+
+ private static ChatRequestEvent buildChatRequestEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("messages", List.of(new ChatMessage(MessageRole.USER, "hello world")));
+ return new ChatRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void chatRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_request_event.json");
+ ChatRequestEvent typed = ChatRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ assertNotNull(typed.getMessages());
+ assertEquals(1, typed.getMessages().size(), "Expected one message.");
+ ChatMessage msg = typed.getMessages().get(0);
+ assertEquals(MessageRole.USER, msg.getRole(), "Role mismatch on Python-produced message.");
+ assertEquals("hello world", msg.getContent());
+ }
+
+ @Test
+ void chatRequestOutputSchemaWireFormatIsJavaShaped() throws Exception {
Review Comment:
This (and the Python mirror `test_chat_request_output_schema_wire_format_is_python_shaped`) pins a wire incompatibility rather than reconciling it: for a `RowTypeInfo`-typed `output_schema`, Java emits `{"fieldNames": [...], "types": [<Class>]}` while Python emits `{"names": [...], "types": [<BasicType int>]}` — both the keys and the type encodings differ, so a `ChatRequestEvent` carrying a non-null `RowTypeInfo` output_schema can't be deserialized on the other side. (The `BaseModel` `module`/`class` path does look symmetric; this is specifically the `RowTypeInfo` branch.)
Is structured `RowTypeInfo` output across the boundary in scope for 0.3? If yes the two sides need a shared key/format; if no, same ask — make it an explicit documented limitation instead of a green test asserting the mismatch.
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -228,20 +228,39 @@ private void extractActionsFromAgent(Agent agent) throws Exception {
// Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
for (Method method : agentClass.getDeclaredMethods()) {
- if (method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
- org.apache.flink.agents.api.annotation.Action actionAnnotation =
- method.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
-
- String[] listenEventTypeStrings =
- Objects.requireNonNull(actionAnnotation).listenEventTypes();
-
- org.apache.flink.agents.plan.JavaFunction javaFunction =
+ if (!method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
Review Comment:
`getDeclaredMethods()` returns only methods declared on the concrete class, so `@Action` methods inherited from a parent agent class aren't registered. Every agent today is a single-level `extends Agent` with `@Action` on the final class, so nothing triggers this — but factoring shared actions into an intermediate base class (a natural refactor) would make them silently stop registering.
Note `getMethods()` only half-fixes it (catches `public` `@Action` by convention but misses `protected`/package-private); walking the superclass chain calling `getDeclaredMethods()` is the visibility-insensitive fix. Low urgency — worth a guard or a documented constraint.
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
Review Comment:
Forcing a fixed id here masks a real cross-language divergence: Java assigns event ids with `UUID.randomUUID()` (`Event.java`) while Python derives a deterministic content-based id (`event.py` `_generate_content_based_id`). Because the two sides use different identity models, the committed snapshots only line up if the id is forced — so they don't reflect what either side naturally emits, and the same logical event gets a different id depending on its producer, which breaks id-based dedup/replay/correlation across languages.
Is cross-language event-id stability a requirement? If so the two sides should agree on one scheme (content-derived on both, most likely); if not, worth documenting that ids are language-local and must not be used for cross-boundary correlation.
##########
python/flink_agents/plan/agent_plan.py:
##########
@@ -242,21 +242,31 @@ def _get_actions(agent: Agent) -> List[Action]:
actions = []
for name, value in agent.__class__.__dict__.items():
if isinstance(value, staticmethod) and hasattr(value, "_listen_events"):
+ exec_ = (
Review Comment:
`hasattr(value, "_listen_events")` is checked on the `staticmethod` wrapper, which makes this order-sensitive. With `@action` outermost over `@staticmethod` (the order the examples use), the attribute lands on the staticmethod object and this branch hits. With the conventional `@staticmethod` outermost over `@action`, the attribute is on `__func__`, `hasattr(wrapper, ...)` is False (3.9), and the `elif callable(value)` below also misses (staticmethod isn't callable pre-3.10) — so the action is silently dropped with no error.
Unwrapping first (`inner = value.__func__ if isinstance(value, staticmethod) else value`, then check `inner`) would make both orders work. No example hits the broken order today, so it's preventive — but it's an easy silent-drop trap.
##########
python/flink_agents/plan/function.py:
##########
@@ -272,17 +272,23 @@ def set_java_resource_adapter(self, adapter: Any) -> None:
def __call__(self, *args: Tuple[Any, ...], **kwargs: Dict[str, Any]) -> Any:
"""Invoke the Java method via the JVM resource adapter.
- LLM tool calls always arrive as keyword arguments — positional
- ``*args`` are ignored because the Java side reorders parameters
- by name via reflection.
+ Positional args route to ``invokeJavaAction`` (action dispatch);
+ keyword args route to ``invokeJavaTool`` (LLM tool dispatch).
"""
if self._j_resource_adapter is None:
msg = (
"JavaFunction requires the JVM resource adapter; not set "
- "on this descriptor. The runtime should inject it via "
+ "on this descriptor. The runtime injects it via "
"set_java_resource_adapter before invocation."
)
raise RuntimeError(msg)
+ if args:
Review Comment:
Routing on `if args:` (positional → `invokeJavaAction`, else → `invokeJavaTool`) leaves the "positional = action, keyword = tool" contract enforced only by the docstring. A mixed call `func(event, x=1)` would go to `invokeJavaAction` and silently drop `kwargs`; a no-arg call falls to the tool path. Both real call sites are clean today (the action path passes exactly `(event, context)` positionally, the tool path passes pure kwargs), so this isn't a live bug — but it's an unenforced convention that a future caller or refactor could misroute silently.
A cheap assertion (reject simultaneous `args` and `kwargs`, or dispatch on an explicit flag) would make the contract enforced rather than assumed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315600549
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/EventRouter.java:
##########
@@ -186,31 +186,33 @@ Event wrapToInputEvent(IN input, PythonActionExecutor pythonActionExecutor) thro
/**
* Extracts the downstream output payload from an output {@link Event}.
*
- * <p>For a Java {@link OutputEvent}, returns the payload directly. For Python-side output
- * events (cross-language unified {@link Event} with output type), the event is JSON-serialized
- * and handed to {@link PythonActionExecutor#getOutputFromOutputEvent(String)} for extraction.
+ * <p>Dispatch is by pipeline wire format, not action language:
+ *
+ * <ul>
+ * <li>Java pipelines ({@code inputIsJava}) emit the raw payload directly.
+ * <li>Python pipelines re-encode through {@link
+ * PythonActionExecutor#getOutputFromOutputEvent(String)} so the downstream Python sink
+ * receives cloudpickle bytes.
+ * </ul>
*
* @param event the output event (must satisfy {@link EventUtil#isOutputEvent(Event)}).
- * @param pythonActionExecutor the Python action executor (used only for Python output events).
+ * @param pythonActionExecutor used only on Python pipelines.
* @return the typed output payload.
*/
@SuppressWarnings("unchecked")
OUT getOutputFromOutputEvent(Event event, PythonActionExecutor pythonActionExecutor) {
checkState(EventUtil.isOutputEvent(event));
- if (event instanceof OutputEvent) {
- return (OUT) ((OutputEvent) event).getOutput();
- } else {
- // Python output events arrive as unified Event with type "_output_event".
- // Pass the JSON representation to Python for extraction.
- try {
- String eventJson = new ObjectMapper().writeValueAsString(event);
- Object outputFromOutputEvent =
- pythonActionExecutor.getOutputFromOutputEvent(eventJson);
- return (OUT) outputFromOutputEvent;
- } catch (Exception e) {
- throw new IllegalStateException(
- "Failed to extract output from event: " + event.getType(), e);
- }
+ OutputEvent typedEvent =
+ (event instanceof OutputEvent) ? (OutputEvent) event : OutputEvent.fromEvent(event);
+ if (inputIsJava) {
+ return (OUT) typedEvent.getOutput();
Review Comment:
Missed the tests, added 3 new tests in `EventRouterTest` and also added a `call-outs` section in the PR description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3321916734
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -227,21 +227,56 @@ private void extractActionsFromAgent(Agent agent) throws Exception {
// Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
+ // getDeclaredMethods() skips inherited @Action methods; reject loudly.
+ for (Class<?> parent = agentClass.getSuperclass();
+ parent != null && parent != Agent.class;
+ parent = parent.getSuperclass()) {
+ for (Method inherited : parent.getDeclaredMethods()) {
+ if (inherited.isAnnotationPresent(
+ org.apache.flink.agents.api.annotation.Action.class)) {
+ throw new IllegalStateException(
+ "Inherited @Action '"
+ + parent.getName()
+ + "#"
+ + inherited.getName()
+ + "' is not supported; declare on the concrete agent.");
+ }
+ }
+ }
Review Comment:
Ideally, inherited actions should also be included. Given that this issue is not introduced by this PR, I'm okay with not fixing it in this PR. We'd better create an issue to track and fix this in future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315600549
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/EventRouter.java:
##########
@@ -186,31 +186,33 @@ Event wrapToInputEvent(IN input, PythonActionExecutor pythonActionExecutor) thro
/**
* Extracts the downstream output payload from an output {@link Event}.
*
- * <p>For a Java {@link OutputEvent}, returns the payload directly. For Python-side output
- * events (cross-language unified {@link Event} with output type), the event is JSON-serialized
- * and handed to {@link PythonActionExecutor#getOutputFromOutputEvent(String)} for extraction.
+ * <p>Dispatch is by pipeline wire format, not action language:
+ *
+ * <ul>
+ * <li>Java pipelines ({@code inputIsJava}) emit the raw payload directly.
+ * <li>Python pipelines re-encode through {@link
+ * PythonActionExecutor#getOutputFromOutputEvent(String)} so the downstream Python sink
+ * receives cloudpickle bytes.
+ * </ul>
*
* @param event the output event (must satisfy {@link EventUtil#isOutputEvent(Event)}).
- * @param pythonActionExecutor the Python action executor (used only for Python output events).
+ * @param pythonActionExecutor used only on Python pipelines.
* @return the typed output payload.
*/
@SuppressWarnings("unchecked")
OUT getOutputFromOutputEvent(Event event, PythonActionExecutor pythonActionExecutor) {
checkState(EventUtil.isOutputEvent(event));
- if (event instanceof OutputEvent) {
- return (OUT) ((OutputEvent) event).getOutput();
- } else {
- // Python output events arrive as unified Event with type "_output_event".
- // Pass the JSON representation to Python for extraction.
- try {
- String eventJson = new ObjectMapper().writeValueAsString(event);
- Object outputFromOutputEvent =
- pythonActionExecutor.getOutputFromOutputEvent(eventJson);
- return (OUT) outputFromOutputEvent;
- } catch (Exception e) {
- throw new IllegalStateException(
- "Failed to extract output from event: " + event.getType(), e);
- }
+ OutputEvent typedEvent =
+ (event instanceof OutputEvent) ? (OutputEvent) event : OutputEvent.fromEvent(event);
+ if (inputIsJava) {
+ return (OUT) typedEvent.getOutput();
Review Comment:
Missed the tests, added 3 new unit tests in `EventRouterTest`, `JavaAgentWithPythonActionTest` would be the Pemja regression guard and also added a `call-outs` section in the PR description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on PR #709:
URL: https://github.com/apache/flink-agents/pull/709#issuecomment-4570043357
Thanks for the update! LGTM. @xintongsong could you take a final look before merge?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347863
##########
plan/src/main/java/org/apache/flink/agents/plan/serializer/ActionJsonDeserializer.java:
##########
@@ -172,8 +172,16 @@ private Object deserializePythonConfig(JsonNode node) {
List<Object> list = new ArrayList<>();
node.forEach(element -> list.add(deserializePythonConfig(element)));
return list;
- } else if (node.isValueNode()) {
- return node.asText();
+ } else if (node.isNull()) {
Review Comment:
Nice — splitting `node.asText()` into typed branches (`booleanValue`/`intValue`/`longValue`/`doubleValue`/`textValue`) restores the type information that Python-side config carries through the JSON. Pre-change, a Python config of `{"max_iters": 10}` would have come back as the string `"10"` on the Java side. Worth a callout in the PR description as an incidental correctness fix shipped along with the cross-language work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3318598273
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assertTrue(
+ Files.exists(committed),
+ "Java snapshot "
+ + fileName
+ + " missing from "
+ + committed
+ + ". If you added a new event, regenerate with -Dregenerate.snapshots=true and commit alongside the test.");
+ JsonNode expected = MAPPER.readTree(Files.readString(committed));
+
+ assertEquals(
+ expected,
+ actual,
+ "Java serialization of "
+ + fileName
+ + " drifted from committed snapshot; if intentional, regenerate.");
+ }
+
+ private static Event readPythonSnapshot(String fileName) throws Exception {
+ Path pythonSnapshot = snapshotDir.resolve("python/" + fileName);
+ assertTrue(
+ Files.exists(pythonSnapshot),
+ "Python snapshot "
+ + fileName
+ + " missing from "
+ + pythonSnapshot
+ + ". Regenerate the Python side with REGENERATE_SNAPSHOTS=1 and commit alongside this test.");
+ return Event.fromJson(Files.readString(pythonSnapshot));
+ }
+
+ // ── InputEvent ─────────────────────────────────────────────────────────
+
+ private static InputEvent buildInputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("input", "hello");
+ return new InputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateInputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void inputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeInputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("input_event.json");
+ InputEvent typed = InputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python InputEvent.");
+ assertEquals(InputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("hello", typed.getInput(), "InputEvent.input mismatch.");
+ }
+
+ // ── OutputEvent ────────────────────────────────────────────────────────
+
+ private static OutputEvent buildOutputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("output", "world");
+ return new OutputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateOutputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void outputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeOutputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("output_event.json");
+ OutputEvent typed = OutputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python OutputEvent.");
+ assertEquals(OutputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("world", typed.getOutput(), "OutputEvent.output mismatch.");
+ }
+
+ // ── ChatRequestEvent ───────────────────────────────────────────────────
+
+ private static ChatRequestEvent buildChatRequestEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("messages", List.of(new ChatMessage(MessageRole.USER, "hello world")));
+ return new ChatRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void chatRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_request_event.json");
+ ChatRequestEvent typed = ChatRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ assertNotNull(typed.getMessages());
+ assertEquals(1, typed.getMessages().size(), "Expected one message.");
+ ChatMessage msg = typed.getMessages().get(0);
+ assertEquals(MessageRole.USER, msg.getRole(), "Role mismatch on Python-produced message.");
+ assertEquals("hello world", msg.getContent());
+ }
+
+ @Test
+ void chatRequestOutputSchemaWireFormatIsJavaShaped() throws Exception {
+ OutputSchema schema =
+ new OutputSchema(
+ new RowTypeInfo(
+ new TypeInformation[] {BasicTypeInfo.STRING_TYPE_INFO},
+ new String[] {"name"}));
+ ChatRequestEvent event =
+ new ChatRequestEvent(
+ "test-model", List.of(new ChatMessage(MessageRole.USER, "hi")), schema);
+ String json = MAPPER.writeValueAsString(event);
+
+ assertTrue(json.contains("\"fieldNames\""), "Java wire format uses `fieldNames`.");
+ assertFalse(json.contains("\"names\""), "Java wire format does not use Python's `names`.");
+ }
+
+ // ── ChatResponseEvent ──────────────────────────────────────────────────
+
+ private static ChatResponseEvent buildChatResponseEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("request_id", FIXED_REQUEST_ID);
+ attrs.put("response", new ChatMessage(MessageRole.ASSISTANT, "hi there"));
+ attrs.put("retry_count", 0);
+ attrs.put("total_retry_wait_sec", 0);
+ return new ChatResponseEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatResponseEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_response_event.json", buildChatResponseEvent());
+ }
+
+ @Test
+ void chatResponseEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_response_event.json", buildChatResponseEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatResponseEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_response_event.json");
+ ChatResponseEvent typed = ChatResponseEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatResponseEvent.EVENT_TYPE, typed.getType());
+ assertEquals(FIXED_REQUEST_ID, typed.getRequestId(), "request_id mismatch.");
+ ChatMessage response = typed.getResponse();
+ assertNotNull(response, "response field is null.");
+ assertEquals(MessageRole.ASSISTANT, response.getRole(), "Role mismatch on response.");
+ assertEquals("hi there", response.getContent());
+ }
+
+ // ── ToolRequestEvent ───────────────────────────────────────────────────
+
+ private static ToolRequestEvent buildToolRequestEvent() {
+ Map<String, Object> toolCall = new LinkedHashMap<>();
+ toolCall.put("id", FIXED_TOOL_CALL_ID);
+ toolCall.put("name", "echo");
+ toolCall.put("arguments", Map.of("value", "ping"));
+
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("tool_calls", List.of(toolCall));
+ return new ToolRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateToolRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("tool_request_event.json", buildToolRequestEvent());
+ }
+
+ @Test
+ void toolRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("tool_request_event.json", buildToolRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeToolRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("tool_request_event.json");
+ ToolRequestEvent typed = ToolRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ToolRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ List<Map<String, Object>> toolCalls = typed.getToolCalls();
+ assertNotNull(toolCalls);
+ assertEquals(1, toolCalls.size());
+ assertEquals(FIXED_TOOL_CALL_ID, toolCalls.get(0).get("id"));
+ }
+
+ // ── ToolResponseEvent ──────────────────────────────────────────────────
+
+ private static ToolResponseEvent buildToolResponseEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("request_id", FIXED_REQUEST_ID);
+ attrs.put("responses", Map.of(FIXED_TOOL_CALL_ID, ToolResponse.success("pong")));
+ attrs.put("success", Map.of(FIXED_TOOL_CALL_ID, true));
+ attrs.put("error", new HashMap<String, String>());
+ attrs.put("external_ids", new HashMap<String, String>());
+ attrs.put("timestamp", FIXED_TIMESTAMP);
+ return new ToolResponseEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateToolResponseEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("tool_response_event.json", buildToolResponseEvent());
+ }
+
+ @Test
+ void toolResponseEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("tool_response_event.json", buildToolResponseEvent());
+ }
+
+ @Test
+ void pythonToolResponseEventLosesDataWhenConsumedByJava() throws Exception {
Review Comment:
Fixed. `ToolResponseEvent.fromEvent` now wraps non-`ToolResponse`/non-Map values via `ToolResponse.success(v)`, so covers primitives `string/number/bool..` round-trip. Renamed the test to infer that and extended the Python snapshot with numeric and boolean entries to pin all three.
Root cause: `responses` is `Dict[UUID, Any]` on Python but `Map<String, ToolResponse>` on Java, so raw scalars satisfied Python's schema with nowhere to land on Java. Tightening Python's type to `Dict[UUID, ToolResponse]` would be the cleaner long-term fix/follow-up.
One caveat: when Python encodes an error case as a `string` in responses (e.g. `tool_call_action.py:48`), Java now wraps it as `success(error_string)` because the Python wire shape doesn't distinguish success from error. The cleaner fix would be tightening Python's responses in a follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on PR #709:
URL: https://github.com/apache/flink-agents/pull/709#issuecomment-4565952039
@weiqingy @wzhero1 Thanks for the review, replied/Addressed the comments. Ready for an other review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347855
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/EventRouter.java:
##########
@@ -186,31 +186,33 @@ Event wrapToInputEvent(IN input, PythonActionExecutor pythonActionExecutor) thro
/**
* Extracts the downstream output payload from an output {@link Event}.
*
- * <p>For a Java {@link OutputEvent}, returns the payload directly. For Python-side output
- * events (cross-language unified {@link Event} with output type), the event is JSON-serialized
- * and handed to {@link PythonActionExecutor#getOutputFromOutputEvent(String)} for extraction.
+ * <p>Dispatch is by pipeline wire format, not action language:
+ *
+ * <ul>
+ * <li>Java pipelines ({@code inputIsJava}) emit the raw payload directly.
+ * <li>Python pipelines re-encode through {@link
+ * PythonActionExecutor#getOutputFromOutputEvent(String)} so the downstream Python sink
+ * receives cloudpickle bytes.
+ * </ul>
*
* @param event the output event (must satisfy {@link EventUtil#isOutputEvent(Event)}).
- * @param pythonActionExecutor the Python action executor (used only for Python output events).
+ * @param pythonActionExecutor used only on Python pipelines.
* @return the typed output payload.
*/
@SuppressWarnings("unchecked")
OUT getOutputFromOutputEvent(Event event, PythonActionExecutor pythonActionExecutor) {
checkState(EventUtil.isOutputEvent(event));
- if (event instanceof OutputEvent) {
- return (OUT) ((OutputEvent) event).getOutput();
- } else {
- // Python output events arrive as unified Event with type "_output_event".
- // Pass the JSON representation to Python for extraction.
- try {
- String eventJson = new ObjectMapper().writeValueAsString(event);
- Object outputFromOutputEvent =
- pythonActionExecutor.getOutputFromOutputEvent(eventJson);
- return (OUT) outputFromOutputEvent;
- } catch (Exception e) {
- throw new IllegalStateException(
- "Failed to extract output from event: " + event.getType(), e);
- }
+ OutputEvent typedEvent =
+ (event instanceof OutputEvent) ? (OutputEvent) event : OutputEvent.fromEvent(event);
+ if (inputIsJava) {
+ return (OUT) typedEvent.getOutput();
Review Comment:
For the `inputIsJava == true && event is unified Event` case (e.g. a Python action emitting `_output_event` in a Java pipeline), this now calls `OutputEvent.fromEvent(event).getOutput()` and bypasses `PythonActionExecutor` entirely. The old branch routed everything that wasn't `instanceof OutputEvent` through Python — so this is a deliberate behavior change to a hot path, not just a refactor, and it affects every output event in every pipeline, not just cross-language ones.
Two asks:
- Could this be called out in the PR description so reviewers (and the 0.3 release notes) don't miss it?
- Is there a test under `runtime/` that specifically pins the Java-pipeline + Python-action output path so a future Pemja change doesn't silently regress payload reconstruction? `CrossLanguageActionRuntimeTest` exercises dispatch but I didn't spot one that asserts on the recovered output type/value here.
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -228,20 +228,30 @@ private void extractActionsFromAgent(Agent agent) throws Exception {
// Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
for (Method method : agentClass.getDeclaredMethods()) {
- if (method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
- org.apache.flink.agents.api.annotation.Action actionAnnotation =
- method.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
-
- String[] listenEventTypeStrings =
- Objects.requireNonNull(actionAnnotation).listenEventTypes();
-
- org.apache.flink.agents.plan.JavaFunction javaFunction =
+ if (!method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
+ continue;
+ }
+ org.apache.flink.agents.api.annotation.Action actionAnnotation =
+ Objects.requireNonNull(
+ method.getAnnotation(
+ org.apache.flink.agents.api.annotation.Action.class));
+ String[] listenEventTypeStrings = actionAnnotation.listenEventTypes();
+ org.apache.flink.agents.api.annotation.PythonFunction target =
+ actionAnnotation.target();
+
+ org.apache.flink.agents.plan.Function execFunction;
+ if (target.module().isEmpty()) {
Review Comment:
The `module().isEmpty()` branch picks Java vs Python, but `qualname()` is read unconditionally on the Python path without a non-empty check. `@Action(target = @PythonFunction(module = "pkg"))` (no qualname) would build a `PythonFunction("pkg", "")` and fail much later inside the interpreter with an obscure error.
A small validation here — something like: both empty → Java; exactly one set → reject with a message naming the action; both set → Python — would fail at the layer where the user can actually correct it. (Same concern applies in mirror on the Python side: does `@action(target=...)` reject a `JavaFunction` with an empty `method_name` or `parameter_types`?)
##########
plan/src/main/java/org/apache/flink/agents/plan/serializer/ActionJsonDeserializer.java:
##########
@@ -172,8 +172,16 @@ private Object deserializePythonConfig(JsonNode node) {
List<Object> list = new ArrayList<>();
node.forEach(element -> list.add(deserializePythonConfig(element)));
return list;
- } else if (node.isValueNode()) {
- return node.asText();
+ } else if (node.isNull()) {
Review Comment:
Nice — splitting `node.asText()` into typed branches (`booleanValue`/`intValue`/`longValue`/`doubleValue`/`textValue`) restores the type information that Python-side config carries through the JSON. Pre-change, a Python config of `{"max_iters": 10}` would have come back as the string `"10"` on the Java side. Worth a callout in the PR description as an incidental correctness fix shipped along with the cross-language work.
##########
python/flink_agents/api/tests/test_cross_language_event_snapshots.py:
##########
@@ -0,0 +1,455 @@
+################################################################################
+# 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.
+#################################################################################
+"""Cross-language event SerDe snapshot tests."""
+
+import json
+import os
+from pathlib import Path
+from typing import ClassVar
+from uuid import UUID
+
+import pytest
+
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
+from flink_agents.api.events.context_retrieval_event import (
+ ContextRetrievalRequestEvent,
+ ContextRetrievalResponseEvent,
+)
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.events.tool_event import ToolRequestEvent, ToolResponseEvent
+from flink_agents.api.vector_stores.vector_store import Document
+
+_REPO_ROOT = Path(__file__).resolve().parents[4]
+_SNAPSHOT_DIR = _REPO_ROOT / "e2e-test" / "cross-language-event-snapshots"
+
+_FIXED_EVENT_ID = UUID("00000000-0000-0000-0000-000000000001")
+_FIXED_REQUEST_ID = UUID("00000000-0000-0000-0000-000000000002")
+_FIXED_TOOL_CALL_ID = "call_aaaa"
+
+
+def _regenerate_enabled() -> bool:
+ return os.environ.get("REGENERATE_SNAPSHOTS", "").lower() in {"1", "true", "yes"}
+
+
+def _force_id(event: Event, fixed_id: UUID) -> Event:
+ object.__setattr__(event, "id", fixed_id)
+ return event
+
+
+def _write_python_snapshot(name: str, event: Event) -> None:
+ target = _SNAPSHOT_DIR / "python" / name
+ target.parent.mkdir(parents=True, exist_ok=True)
+ target.write_text(event.model_dump_json(indent=2) + "\n")
+
+
+def _assert_python_snapshot_stable(name: str, event: Event) -> None:
+ actual = json.loads(event.model_dump_json())
+ committed_path = _SNAPSHOT_DIR / "python" / name
+ if not committed_path.exists():
Review Comment:
Mirror of the same concern on the Java side: `pytest.skip` when the committed snapshot is missing turns a real wire-format regression into a silent pass-as-skipped. The `REGENERATE_SNAPSHOTS=1` skips on the regenerate tests are legitimate (the env var is the gate), but the stability / round-trip helpers (lines 64 and 76) would be stronger as `pytest.fail(...)` or a plain `assert committed_path.exists()` — the snapshot suite is the only thing pinning the cross-language wire format, so a missing file should be loud, not quiet.
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,483 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assumeTrue(
Review Comment:
`assumeTrue` here means the test reports as "skipped, not failed" if `chat_request_event.json` (or any other committed snapshot) gets renamed or deleted. The whole point of the snapshot suite is to catch silent wire-format drift, so a silently-skipped "stable" test undercuts that guarantee.
The `regenerate*` tests genuinely need conditional skipping (the system property is the gate), but the `*JavaSnapshotIsStable` and `javaCanDeserialize*FromPythonSnapshot` cases would be stronger as hard assertions — committed snapshot missing means the test must fail. Worth doing the same on `readPythonSnapshot` at line 111.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347859
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,483 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assumeTrue(
Review Comment:
`assumeTrue` here means the test reports as "skipped, not failed" if `chat_request_event.json` (or any other committed snapshot) gets renamed or deleted. The whole point of the snapshot suite is to catch silent wire-format drift, so silently-skipped "stable" tests undercut that guarantee.
The `regenerate*` tests genuinely need conditional skipping (the system property is the gate). For `*JavaSnapshotIsStable` and `javaCanDeserialize*FromPythonSnapshot`, was there a reason for the symmetric `assumeTrue` — any case where a missing snapshot should soft-pass? If not, would a hard `assertTrue(Files.exists(...))` be better? Same question for `readPythonSnapshot:111`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3318621593
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assertTrue(
+ Files.exists(committed),
+ "Java snapshot "
+ + fileName
+ + " missing from "
+ + committed
+ + ". If you added a new event, regenerate with -Dregenerate.snapshots=true and commit alongside the test.");
+ JsonNode expected = MAPPER.readTree(Files.readString(committed));
+
+ assertEquals(
+ expected,
+ actual,
+ "Java serialization of "
+ + fileName
+ + " drifted from committed snapshot; if intentional, regenerate.");
+ }
+
+ private static Event readPythonSnapshot(String fileName) throws Exception {
+ Path pythonSnapshot = snapshotDir.resolve("python/" + fileName);
+ assertTrue(
+ Files.exists(pythonSnapshot),
+ "Python snapshot "
+ + fileName
+ + " missing from "
+ + pythonSnapshot
+ + ". Regenerate the Python side with REGENERATE_SNAPSHOTS=1 and commit alongside this test.");
+ return Event.fromJson(Files.readString(pythonSnapshot));
+ }
+
+ // ── InputEvent ─────────────────────────────────────────────────────────
+
+ private static InputEvent buildInputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("input", "hello");
+ return new InputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateInputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void inputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeInputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("input_event.json");
+ InputEvent typed = InputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python InputEvent.");
+ assertEquals(InputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("hello", typed.getInput(), "InputEvent.input mismatch.");
+ }
+
+ // ── OutputEvent ────────────────────────────────────────────────────────
+
+ private static OutputEvent buildOutputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("output", "world");
+ return new OutputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateOutputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void outputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeOutputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("output_event.json");
+ OutputEvent typed = OutputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python OutputEvent.");
+ assertEquals(OutputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("world", typed.getOutput(), "OutputEvent.output mismatch.");
+ }
+
+ // ── ChatRequestEvent ───────────────────────────────────────────────────
+
+ private static ChatRequestEvent buildChatRequestEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("messages", List.of(new ChatMessage(MessageRole.USER, "hello world")));
+ return new ChatRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void chatRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_request_event.json");
+ ChatRequestEvent typed = ChatRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ assertNotNull(typed.getMessages());
+ assertEquals(1, typed.getMessages().size(), "Expected one message.");
+ ChatMessage msg = typed.getMessages().get(0);
+ assertEquals(MessageRole.USER, msg.getRole(), "Role mismatch on Python-produced message.");
+ assertEquals("hello world", msg.getContent());
+ }
+
+ @Test
+ void chatRequestOutputSchemaWireFormatIsJavaShaped() throws Exception {
Review Comment:
Renamed the test (and Python mirror) to `...RowTypeInfoOutputSchemaIsNotPortableAcrossLanguages_knownGap` with Javadoc spelling out the divergence: Java's `{"fieldNames", types:[<Class>]}` vs Python's `{"names", types:[<BasicTypeInfo ordinal>]}`. The BaseModel branch is symmetric and works; only `RowTypeInfo` is affected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347856
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -228,20 +228,30 @@ private void extractActionsFromAgent(Agent agent) throws Exception {
// Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
for (Method method : agentClass.getDeclaredMethods()) {
- if (method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
- org.apache.flink.agents.api.annotation.Action actionAnnotation =
- method.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
-
- String[] listenEventTypeStrings =
- Objects.requireNonNull(actionAnnotation).listenEventTypes();
-
- org.apache.flink.agents.plan.JavaFunction javaFunction =
+ if (!method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
+ continue;
+ }
+ org.apache.flink.agents.api.annotation.Action actionAnnotation =
+ Objects.requireNonNull(
+ method.getAnnotation(
+ org.apache.flink.agents.api.annotation.Action.class));
+ String[] listenEventTypeStrings = actionAnnotation.listenEventTypes();
+ org.apache.flink.agents.api.annotation.PythonFunction target =
+ actionAnnotation.target();
+
+ org.apache.flink.agents.plan.Function execFunction;
+ if (target.module().isEmpty()) {
Review Comment:
The `module().isEmpty()` branch picks Java vs Python, but `qualname()` is read unconditionally on the Python path without a non-empty check. `@Action(target = @PythonFunction(module = "pkg"))` (no qualname) would build a `PythonFunction("pkg", "")` and fail much later inside the interpreter with an obscure error.
Would it make sense to validate here — e.g. both empty → Java; exactly one set → reject with a message naming the action; both set → Python? Catching it at extraction time would surface the typo at the layer where the user can actually correct it. (Same question on the Python side — does @action(target=...) reject a JavaFunction with an empty method_name or parameter_types?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315600549
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/EventRouter.java:
##########
@@ -186,31 +186,33 @@ Event wrapToInputEvent(IN input, PythonActionExecutor pythonActionExecutor) thro
/**
* Extracts the downstream output payload from an output {@link Event}.
*
- * <p>For a Java {@link OutputEvent}, returns the payload directly. For Python-side output
- * events (cross-language unified {@link Event} with output type), the event is JSON-serialized
- * and handed to {@link PythonActionExecutor#getOutputFromOutputEvent(String)} for extraction.
+ * <p>Dispatch is by pipeline wire format, not action language:
+ *
+ * <ul>
+ * <li>Java pipelines ({@code inputIsJava}) emit the raw payload directly.
+ * <li>Python pipelines re-encode through {@link
+ * PythonActionExecutor#getOutputFromOutputEvent(String)} so the downstream Python sink
+ * receives cloudpickle bytes.
+ * </ul>
*
* @param event the output event (must satisfy {@link EventUtil#isOutputEvent(Event)}).
- * @param pythonActionExecutor the Python action executor (used only for Python output events).
+ * @param pythonActionExecutor used only on Python pipelines.
* @return the typed output payload.
*/
@SuppressWarnings("unchecked")
OUT getOutputFromOutputEvent(Event event, PythonActionExecutor pythonActionExecutor) {
checkState(EventUtil.isOutputEvent(event));
- if (event instanceof OutputEvent) {
- return (OUT) ((OutputEvent) event).getOutput();
- } else {
- // Python output events arrive as unified Event with type "_output_event".
- // Pass the JSON representation to Python for extraction.
- try {
- String eventJson = new ObjectMapper().writeValueAsString(event);
- Object outputFromOutputEvent =
- pythonActionExecutor.getOutputFromOutputEvent(eventJson);
- return (OUT) outputFromOutputEvent;
- } catch (Exception e) {
- throw new IllegalStateException(
- "Failed to extract output from event: " + event.getType(), e);
- }
+ OutputEvent typedEvent =
+ (event instanceof OutputEvent) ? (OutputEvent) event : OutputEvent.fromEvent(event);
+ if (inputIsJava) {
+ return (OUT) typedEvent.getOutput();
Review Comment:
Missed a few tests, added 3 new unit tests in `EventRouterTest`, `JavaAgentWithPythonActionTest` would be the Pemja regression guard and also added a `call-outs` section in the PR description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347860
##########
python/flink_agents/api/tests/test_cross_language_event_snapshots.py:
##########
@@ -0,0 +1,455 @@
+################################################################################
+# 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.
+#################################################################################
+"""Cross-language event SerDe snapshot tests."""
+
+import json
+import os
+from pathlib import Path
+from typing import ClassVar
+from uuid import UUID
+
+import pytest
+
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
+from flink_agents.api.events.context_retrieval_event import (
+ ContextRetrievalRequestEvent,
+ ContextRetrievalResponseEvent,
+)
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.events.tool_event import ToolRequestEvent, ToolResponseEvent
+from flink_agents.api.vector_stores.vector_store import Document
+
+_REPO_ROOT = Path(__file__).resolve().parents[4]
+_SNAPSHOT_DIR = _REPO_ROOT / "e2e-test" / "cross-language-event-snapshots"
+
+_FIXED_EVENT_ID = UUID("00000000-0000-0000-0000-000000000001")
+_FIXED_REQUEST_ID = UUID("00000000-0000-0000-0000-000000000002")
+_FIXED_TOOL_CALL_ID = "call_aaaa"
+
+
+def _regenerate_enabled() -> bool:
+ return os.environ.get("REGENERATE_SNAPSHOTS", "").lower() in {"1", "true", "yes"}
+
+
+def _force_id(event: Event, fixed_id: UUID) -> Event:
+ object.__setattr__(event, "id", fixed_id)
+ return event
+
+
+def _write_python_snapshot(name: str, event: Event) -> None:
+ target = _SNAPSHOT_DIR / "python" / name
+ target.parent.mkdir(parents=True, exist_ok=True)
+ target.write_text(event.model_dump_json(indent=2) + "\n")
+
+
+def _assert_python_snapshot_stable(name: str, event: Event) -> None:
+ actual = json.loads(event.model_dump_json())
+ committed_path = _SNAPSHOT_DIR / "python" / name
+ if not committed_path.exists():
Review Comment:
Mirror of the same concern on the Java side: `pytest.skip` when the committed snapshot is missing turns a real wire-format regression into a silent pass-as-skipped. The `REGENERATE_SNAPSHOTS=1` skips on the regenerate tests are legitimate (the env var is the gate).
For the stability / round-trip helpers at lines 64 and 76, was the symmetric `pytest.skip` intentional — any case where a missing snapshot should soft-pass? If not, would `pytest.fail(...)` or a plain `assert committed_path.exists()` be a tighter fit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347856
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -228,20 +228,30 @@ private void extractActionsFromAgent(Agent agent) throws Exception {
// Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
for (Method method : agentClass.getDeclaredMethods()) {
- if (method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
- org.apache.flink.agents.api.annotation.Action actionAnnotation =
- method.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
-
- String[] listenEventTypeStrings =
- Objects.requireNonNull(actionAnnotation).listenEventTypes();
-
- org.apache.flink.agents.plan.JavaFunction javaFunction =
+ if (!method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
+ continue;
+ }
+ org.apache.flink.agents.api.annotation.Action actionAnnotation =
+ Objects.requireNonNull(
+ method.getAnnotation(
+ org.apache.flink.agents.api.annotation.Action.class));
+ String[] listenEventTypeStrings = actionAnnotation.listenEventTypes();
+ org.apache.flink.agents.api.annotation.PythonFunction target =
+ actionAnnotation.target();
+
+ org.apache.flink.agents.plan.Function execFunction;
+ if (target.module().isEmpty()) {
Review Comment:
The `module().isEmpty()` branch picks Java vs Python, but `qualname()` is read unconditionally on the Python path without a non-empty check. `@Action(target = @PythonFunction(module = "pkg"))` (no qualname) would build a `PythonFunction("pkg", "")` and fail much later inside the interpreter with an obscure error.
Would it make sense to validate here — e.g. both empty → Java; exactly one set → reject with a message naming the action; both set → Python? Catching it at extraction time would surface the typo at the layer where the user can actually correct it. (Same question on the Python side — does `@action(target=...)` reject a `JavaFunction` with an empty `method_name` or `parameter_types`?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315593576
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -228,20 +228,30 @@ private void extractActionsFromAgent(Agent agent) throws Exception {
// Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
for (Method method : agentClass.getDeclaredMethods()) {
- if (method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
- org.apache.flink.agents.api.annotation.Action actionAnnotation =
- method.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
-
- String[] listenEventTypeStrings =
- Objects.requireNonNull(actionAnnotation).listenEventTypes();
-
- org.apache.flink.agents.plan.JavaFunction javaFunction =
+ if (!method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class)) {
+ continue;
+ }
+ org.apache.flink.agents.api.annotation.Action actionAnnotation =
+ Objects.requireNonNull(
+ method.getAnnotation(
+ org.apache.flink.agents.api.annotation.Action.class));
+ String[] listenEventTypeStrings = actionAnnotation.listenEventTypes();
+ org.apache.flink.agents.api.annotation.PythonFunction target =
+ actionAnnotation.target();
+
+ org.apache.flink.agents.plan.Function execFunction;
+ if (target.module().isEmpty()) {
Review Comment:
Agreed. Validation in `AgentPlan` now checks both fields: both empty is Java, both set is Python, anything in between throws with the action name. Same check on the Python decorator for both `PythonFunction` and `JavaFunction` targets. Also added tests for the reject cases on both sides.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3315347855
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/EventRouter.java:
##########
@@ -186,31 +186,33 @@ Event wrapToInputEvent(IN input, PythonActionExecutor pythonActionExecutor) thro
/**
* Extracts the downstream output payload from an output {@link Event}.
*
- * <p>For a Java {@link OutputEvent}, returns the payload directly. For Python-side output
- * events (cross-language unified {@link Event} with output type), the event is JSON-serialized
- * and handed to {@link PythonActionExecutor#getOutputFromOutputEvent(String)} for extraction.
+ * <p>Dispatch is by pipeline wire format, not action language:
+ *
+ * <ul>
+ * <li>Java pipelines ({@code inputIsJava}) emit the raw payload directly.
+ * <li>Python pipelines re-encode through {@link
+ * PythonActionExecutor#getOutputFromOutputEvent(String)} so the downstream Python sink
+ * receives cloudpickle bytes.
+ * </ul>
*
* @param event the output event (must satisfy {@link EventUtil#isOutputEvent(Event)}).
- * @param pythonActionExecutor the Python action executor (used only for Python output events).
+ * @param pythonActionExecutor used only on Python pipelines.
* @return the typed output payload.
*/
@SuppressWarnings("unchecked")
OUT getOutputFromOutputEvent(Event event, PythonActionExecutor pythonActionExecutor) {
checkState(EventUtil.isOutputEvent(event));
- if (event instanceof OutputEvent) {
- return (OUT) ((OutputEvent) event).getOutput();
- } else {
- // Python output events arrive as unified Event with type "_output_event".
- // Pass the JSON representation to Python for extraction.
- try {
- String eventJson = new ObjectMapper().writeValueAsString(event);
- Object outputFromOutputEvent =
- pythonActionExecutor.getOutputFromOutputEvent(eventJson);
- return (OUT) outputFromOutputEvent;
- } catch (Exception e) {
- throw new IllegalStateException(
- "Failed to extract output from event: " + event.getType(), e);
- }
+ OutputEvent typedEvent =
+ (event instanceof OutputEvent) ? (OutputEvent) event : OutputEvent.fromEvent(event);
+ if (inputIsJava) {
+ return (OUT) typedEvent.getOutput();
Review Comment:
For the `inputIsJava == true && event is unified Event` case (e.g. a Python action emitting `_output_event` in a Java pipeline), this now calls `OutputEvent.fromEvent(event).getOutput()` and bypasses `PythonActionExecutor` entirely. The old branch routed everything that wasn't `instanceof OutputEvent` through Python — so this is a deliberate behavior change to a hot path, not just a refactor, and it affects every output event in every pipeline, not just cross-language ones.
Two asks:
- Could this be called out in the PR description so reviewers (and the 0.3 release notes) don't miss it?
- Is there a test under `runtime/` that specifically pins the Java-pipeline + Python-action output path so a future Pemja change doesn't silently regress payload reconstruction? `CrossLanguageActionRuntimeTest` exercises dispatch — point me at the one covering the recovered output type/value if I missed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3318387054
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
Review Comment:
1. **ID survives the language hop via `from_event`.** There's a fix in this PR. `InputEvent.from_event` and `OutputEvent.from_event` in `event.py` now does `result.id = event.id` instead of letting `Event.__setattr__` regenerate a content-based ID.
The `assertEquals(FIXED_EVENT_ID, typed.getId(), ...)` lines in the round-trip tests would fail without that fix, so `FIXED_EVENT_ID` isn't masking. It's just the fixture that makes the snapshot JSON byte-stable so the preservation assertion can exist.
2. **Independent Java and Python producers don't agree on the ID for logically equivalent events.** Java uses `UUID.randomUUID()` so two identical-payload events get different IDs anyway. Python uses content-derived UUIDv3 so two identical-payload events get the same ID, plus the ID regenerates on every `__setattr__` write (maybe a bug?). So cross-producer dedup/replay across the boundary doesn't work today. Aligning that is a real design call, out of scope for 0.3, unless you'd rather tackle it here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Cross Language Actions [flink-agents]
Posted by "addu390 (via GitHub)" <gi...@apache.org>.
addu390 commented on code in PR #709:
URL: https://github.com/apache/flink-agents/pull/709#discussion_r3318598273
##########
api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java:
##########
@@ -0,0 +1,489 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.vectorstores.Document;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Cross-language event SerDe snapshot tests. */
+class CrossLanguageEventSnapshotTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final UUID FIXED_EVENT_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000001");
+ private static final UUID FIXED_REQUEST_ID =
+ UUID.fromString("00000000-0000-0000-0000-000000000002");
+ private static final String FIXED_TOOL_CALL_ID = "call_aaaa";
+ private static final long FIXED_TIMESTAMP = 1_700_000_000_000L;
+
+ private static Path snapshotDir;
+
+ @BeforeAll
+ static void resolveSnapshotDir() {
+ Path repoRoot = Paths.get(System.getProperty("user.dir")).getParent();
+ snapshotDir = repoRoot.resolve("e2e-test/cross-language-event-snapshots");
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────
+
+ private static boolean regenerateRequested() {
+ return Boolean.parseBoolean(System.getProperty("regenerate.snapshots", "false"));
+ }
+
+ private static void writeJavaSnapshot(String fileName, Event event) throws Exception {
+ String json = MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(event);
+ Path target = snapshotDir.resolve("java/" + fileName);
+ Files.createDirectories(target.getParent());
+ Files.writeString(target, json + "\n");
+ }
+
+ private static void assertJavaSnapshotStable(String fileName, Event event) throws Exception {
+ String actualJson = MAPPER.writeValueAsString(event);
+ JsonNode actual = MAPPER.readTree(actualJson);
+
+ Path committed = snapshotDir.resolve("java/" + fileName);
+ assertTrue(
+ Files.exists(committed),
+ "Java snapshot "
+ + fileName
+ + " missing from "
+ + committed
+ + ". If you added a new event, regenerate with -Dregenerate.snapshots=true and commit alongside the test.");
+ JsonNode expected = MAPPER.readTree(Files.readString(committed));
+
+ assertEquals(
+ expected,
+ actual,
+ "Java serialization of "
+ + fileName
+ + " drifted from committed snapshot; if intentional, regenerate.");
+ }
+
+ private static Event readPythonSnapshot(String fileName) throws Exception {
+ Path pythonSnapshot = snapshotDir.resolve("python/" + fileName);
+ assertTrue(
+ Files.exists(pythonSnapshot),
+ "Python snapshot "
+ + fileName
+ + " missing from "
+ + pythonSnapshot
+ + ". Regenerate the Python side with REGENERATE_SNAPSHOTS=1 and commit alongside this test.");
+ return Event.fromJson(Files.readString(pythonSnapshot));
+ }
+
+ // ── InputEvent ─────────────────────────────────────────────────────────
+
+ private static InputEvent buildInputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("input", "hello");
+ return new InputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateInputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void inputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("input_event.json", buildInputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeInputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("input_event.json");
+ InputEvent typed = InputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python InputEvent.");
+ assertEquals(InputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("hello", typed.getInput(), "InputEvent.input mismatch.");
+ }
+
+ // ── OutputEvent ────────────────────────────────────────────────────────
+
+ private static OutputEvent buildOutputEvent() {
+ Map<String, Object> attrs = new HashMap<>();
+ attrs.put("output", "world");
+ return new OutputEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateOutputEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void outputEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("output_event.json", buildOutputEvent());
+ }
+
+ @Test
+ void javaCanDeserializeOutputEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("output_event.json");
+ OutputEvent typed = OutputEvent.fromEvent(base);
+
+ assertEquals(
+ FIXED_EVENT_ID, typed.getId(), "ID lost when deserializing Python OutputEvent.");
+ assertEquals(OutputEvent.EVENT_TYPE, typed.getType());
+ assertEquals("world", typed.getOutput(), "OutputEvent.output mismatch.");
+ }
+
+ // ── ChatRequestEvent ───────────────────────────────────────────────────
+
+ private static ChatRequestEvent buildChatRequestEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("messages", List.of(new ChatMessage(MessageRole.USER, "hello world")));
+ return new ChatRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void chatRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_request_event.json", buildChatRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_request_event.json");
+ ChatRequestEvent typed = ChatRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ assertNotNull(typed.getMessages());
+ assertEquals(1, typed.getMessages().size(), "Expected one message.");
+ ChatMessage msg = typed.getMessages().get(0);
+ assertEquals(MessageRole.USER, msg.getRole(), "Role mismatch on Python-produced message.");
+ assertEquals("hello world", msg.getContent());
+ }
+
+ @Test
+ void chatRequestOutputSchemaWireFormatIsJavaShaped() throws Exception {
+ OutputSchema schema =
+ new OutputSchema(
+ new RowTypeInfo(
+ new TypeInformation[] {BasicTypeInfo.STRING_TYPE_INFO},
+ new String[] {"name"}));
+ ChatRequestEvent event =
+ new ChatRequestEvent(
+ "test-model", List.of(new ChatMessage(MessageRole.USER, "hi")), schema);
+ String json = MAPPER.writeValueAsString(event);
+
+ assertTrue(json.contains("\"fieldNames\""), "Java wire format uses `fieldNames`.");
+ assertFalse(json.contains("\"names\""), "Java wire format does not use Python's `names`.");
+ }
+
+ // ── ChatResponseEvent ──────────────────────────────────────────────────
+
+ private static ChatResponseEvent buildChatResponseEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("request_id", FIXED_REQUEST_ID);
+ attrs.put("response", new ChatMessage(MessageRole.ASSISTANT, "hi there"));
+ attrs.put("retry_count", 0);
+ attrs.put("total_retry_wait_sec", 0);
+ return new ChatResponseEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateChatResponseEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("chat_response_event.json", buildChatResponseEvent());
+ }
+
+ @Test
+ void chatResponseEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("chat_response_event.json", buildChatResponseEvent());
+ }
+
+ @Test
+ void javaCanDeserializeChatResponseEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("chat_response_event.json");
+ ChatResponseEvent typed = ChatResponseEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ChatResponseEvent.EVENT_TYPE, typed.getType());
+ assertEquals(FIXED_REQUEST_ID, typed.getRequestId(), "request_id mismatch.");
+ ChatMessage response = typed.getResponse();
+ assertNotNull(response, "response field is null.");
+ assertEquals(MessageRole.ASSISTANT, response.getRole(), "Role mismatch on response.");
+ assertEquals("hi there", response.getContent());
+ }
+
+ // ── ToolRequestEvent ───────────────────────────────────────────────────
+
+ private static ToolRequestEvent buildToolRequestEvent() {
+ Map<String, Object> toolCall = new LinkedHashMap<>();
+ toolCall.put("id", FIXED_TOOL_CALL_ID);
+ toolCall.put("name", "echo");
+ toolCall.put("arguments", Map.of("value", "ping"));
+
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put("tool_calls", List.of(toolCall));
+ return new ToolRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateToolRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("tool_request_event.json", buildToolRequestEvent());
+ }
+
+ @Test
+ void toolRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("tool_request_event.json", buildToolRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeToolRequestEventFromPythonSnapshot() throws Exception {
+ Event base = readPythonSnapshot("tool_request_event.json");
+ ToolRequestEvent typed = ToolRequestEvent.fromEvent(base);
+
+ assertEquals(FIXED_EVENT_ID, typed.getId());
+ assertEquals(ToolRequestEvent.EVENT_TYPE, typed.getType());
+ assertEquals("test-model", typed.getModel());
+ List<Map<String, Object>> toolCalls = typed.getToolCalls();
+ assertNotNull(toolCalls);
+ assertEquals(1, toolCalls.size());
+ assertEquals(FIXED_TOOL_CALL_ID, toolCalls.get(0).get("id"));
+ }
+
+ // ── ToolResponseEvent ──────────────────────────────────────────────────
+
+ private static ToolResponseEvent buildToolResponseEvent() {
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("request_id", FIXED_REQUEST_ID);
+ attrs.put("responses", Map.of(FIXED_TOOL_CALL_ID, ToolResponse.success("pong")));
+ attrs.put("success", Map.of(FIXED_TOOL_CALL_ID, true));
+ attrs.put("error", new HashMap<String, String>());
+ attrs.put("external_ids", new HashMap<String, String>());
+ attrs.put("timestamp", FIXED_TIMESTAMP);
+ return new ToolResponseEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateToolResponseEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to refresh.");
+ writeJavaSnapshot("tool_response_event.json", buildToolResponseEvent());
+ }
+
+ @Test
+ void toolResponseEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable("tool_response_event.json", buildToolResponseEvent());
+ }
+
+ @Test
+ void pythonToolResponseEventLosesDataWhenConsumedByJava() throws Exception {
Review Comment:
Fixed. `ToolResponseEvent.fromEvent` now wraps non-`ToolResponse`/non-Map values via `ToolResponse.success(v)`, so covers primitives `string/number/bool..` round-trip. Renamed the test to infer that and extended the Python snapshot with numeric and boolean entries to pin all three.
Root cause: `responses` is `Dict[UUID, Any]` on Python but `Map<String, ToolResponse>` on Java, so raw scalars satisfied Python's schema with nowhere to land on Java. Tightening Python's type to `Dict[UUID, ToolResponse]` would be the cleaner long-term fix/follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org