This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 2f56f97f [python] Backfill remote test coverage ahead of local
execution path removal (#881)
2f56f97f is described below
commit 2f56f97ff951e455e56e6c9d1b623636c3ad4ce5
Author: Weiqing Yang <[email protected]>
AuthorDate: Fri Jul 10 01:40:24 2026 -0700
[python] Backfill remote test coverage ahead of local execution path
removal (#881)
---
.../chat_model_multi_provider_remote_test.py | 147 ++++++++++
.../e2e_tests_mcp/mcp_test.py | 114 ++++++--
.../e2e_tests_integration/execute_test.py | 106 +++++++
.../e2e_tests_integration/execute_test_agent.py | 34 +++
.../e2e_tests_integration/mock_chat_model_agent.py | 303 +++++++++++++++++++++
.../e2e_tests_integration/mock_chat_model_test.py | 164 +++++++++++
.../workflow_memory_remote_agent.py | 83 ++++++
.../workflow_memory_remote_test.py | 103 +++++++
.../chat_model_multi_provider_input/input_data.txt | 2 +
.../ground_truth/test_built_in_action_content.txt | 2 +
.../ground_truth/test_chat_model_get_resource.txt | 2 +
.../ground_truth/test_execute_sync_exception.txt | 3 +
.../ground_truth/test_execute_with_kwargs.txt | 3 +
.../ground_truth/test_workflow_memory.txt | 4 +
.../e2e_tests/resources/mcp_input/input_data.txt | 2 +
.../mock_chat_model_built_in_input/input_data.txt | 2 +
.../input_data.txt | 2 +
.../resources/workflow_memory_input/input_data.txt | 4 +
.../tests/test_remote_execution_environment.py | 22 ++
19 files changed, 1078 insertions(+), 24 deletions(-)
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/chat_model_multi_provider_remote_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/chat_model_multi_provider_remote_test.py
new file mode 100644
index 00000000..af243984
--- /dev/null
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/chat_model_multi_provider_remote_test.py
@@ -0,0 +1,147 @@
+################################################################################
+# 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.
+#################################################################################
+import os
+import sysconfig
+from pathlib import Path
+
+import pytest
+from pyflink.common import Encoder, WatermarkStrategy
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import (
+ RuntimeExecutionMode,
+ StreamExecutionEnvironment,
+)
+from pyflink.datastream.connectors.file_system import (
+ FileSource,
+ StreamFormat,
+ StreamingFileSink,
+)
+
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from flink_agents.e2e_tests.e2e_tests_integration.chat_model_integration_agent
import (
+ ChatModelTestAgent,
+)
+
+current_dir = Path(__file__).parent
+
+TONGYI_MODEL = os.environ.get("TONGYI_CHAT_MODEL", "qwen-plus")
+OPENAI_MODEL = os.environ.get("OPENAI_CHAT_MODEL", "gpt-3.5-turbo")
+AZURE_OPENAI_MODEL = os.environ.get("AZURE_OPENAI_CHAT_MODEL", "gpt-5")
+AZURE_OPENAI_API_VERSION = os.environ.get(
+ "AZURE_OPENAI_API_VERSION", "2025-04-01-preview"
+)
+
+DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY")
+OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
+AZURE_OPENAI_API_KEY = os.environ.get("AZURE_OPENAI_API_KEY")
+
+os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
+
+
[email protected](
+ "model_provider",
+ [
+ pytest.param(
+ "Tongyi",
+ marks=pytest.mark.skipif(
+ not DASHSCOPE_API_KEY, reason="Tongyi api key is not set."
+ ),
+ ),
+ pytest.param(
+ "OpenAI",
+ marks=pytest.mark.skipif(
+ not OPENAI_API_KEY, reason="OpenAI api key is not set."
+ ),
+ ),
+ pytest.param(
+ "AzureOpenAI",
+ marks=pytest.mark.skipif(
+ not AZURE_OPENAI_API_KEY, reason="Azure OpenAI api key is not
set."
+ ),
+ ),
+ ],
+)
+def test_chat_model_integration_remote(
+ model_provider: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ """Non-Ollama providers answer math and creative prompts on the remote
path.
+
+ Mirrors the from_list ``chat_model_integration_test`` for the three
+ credential-gated providers (Tongyi/OpenAI/AzureOpenAI), but drives the
agent
+ through a real StreamExecutionEnvironment with a FileSource and file sink.
+ Each parameter skips cleanly when its credential is absent.
+ """
+ monkeypatch.setenv("TONGYI_CHAT_MODEL", TONGYI_MODEL)
+ monkeypatch.setenv("OPENAI_CHAT_MODEL", OPENAI_MODEL)
+ monkeypatch.setenv("AZURE_OPENAI_CHAT_MODEL", AZURE_OPENAI_MODEL)
+ monkeypatch.setenv("AZURE_OPENAI_API_VERSION", AZURE_OPENAI_API_VERSION)
+ monkeypatch.setenv("MODEL_PROVIDER", model_provider)
+
+ env = StreamExecutionEnvironment.get_execution_environment()
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ # currently, bounded source is not supported due to runtime
implementation, so
+ # we use continuous file source here.
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+
f"file:///{current_dir}/../resources/chat_model_multi_provider_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="chat_model_multi_provider_source",
+ )
+
+ deserialize_datastream = input_datastream.map(lambda x: str(x))
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream, key_selector=lambda x: x
+ )
+ .apply(ChatModelTestAgent())
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+
+ output_datastream.map(
+ lambda x: str(x).replace("\n", "").replace("\r", ""), Types.STRING()
+ ).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+
+ agents_env.execute()
+
+ actual_result = []
+ for file in result_dir.iterdir():
+ if file.is_dir():
+ for child in file.iterdir():
+ with child.open() as f:
+ actual_result.extend(f.readlines())
+ if file.is_file():
+ with file.open() as f:
+ actual_result.extend(f.readlines())
+
+ joined = "\n".join(actual_result).lower()
+ assert "3" in joined, f"sum answer missing '3': {actual_result!r}"
+ assert "cat" in joined, f"creative answer missing 'cat': {actual_result!r}"
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/e2e_tests_mcp/mcp_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/e2e_tests_mcp/mcp_test.py
index 2c994f1f..19ab5aed 100644
---
a/python/flink_agents/e2e_tests/e2e_tests_integration/e2e_tests_mcp/mcp_test.py
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/e2e_tests_mcp/mcp_test.py
@@ -30,11 +30,20 @@ Prerequisites:
import multiprocessing
import os
import runpy
+import sysconfig
import time
from pathlib import Path
import pytest
from pydantic import BaseModel
+from pyflink.common import Configuration, Encoder, WatermarkStrategy
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import RuntimeExecutionMode, StreamExecutionEnvironment
+from pyflink.datastream.connectors.file_system import (
+ FileSource,
+ StreamFormat,
+ StreamingFileSink,
+)
from flink_agents.api.agents.agent import Agent
from flink_agents.api.chat_message import ChatMessage, MessageRole
@@ -55,6 +64,8 @@ from flink_agents.api.resource import (
from flink_agents.api.runner_context import RunnerContext
from flink_agents.e2e_tests.test_utils import pull_model
+os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
+
OLLAMA_MODEL = os.environ.get("MCP_OLLAMA_CHAT_MODEL", "qwen3:1.7b")
MCP_SERVER_ENDPOINT = "http://127.0.0.1:8000/mcp";
MCP_SERVER_ENDPOINT_WITHOUT_PROMPTS = "http://127.0.0.1:8001/mcp";
@@ -170,42 +181,97 @@ client = pull_model(OLLAMA_MODEL)
@pytest.mark.skipif(
client is None, reason="Ollama client is not available or test model is
missing"
)
-def test_mcp(mcp_server_mode: str, server_file: str, server_endpoint: str) ->
None:
- """Test MCP integration with different server modes.
+def test_mcp(
+ mcp_server_mode: str,
+ server_file: str,
+ server_endpoint: str,
+ tmp_path: Path,
+) -> None:
+ """Test MCP integration with different server modes on a MiniCluster.
+
+ Drives ``MyMCPAgent`` (which binds the MCP ``add`` tool and ``ask_sum``
+ prompt into an Ollama chat model setup) through a real
+ ``StreamExecutionEnvironment`` with a FileSource and file sink. LLM output
+ is non-deterministic, so the assertion counts output records rather than
+ matching content.
Args:
mcp_server_mode: "with_prompts" or "without_prompts"
server_file: Name of the MCP server file to run
server_endpoint: Endpoint URL of the MCP server
+ tmp_path: pytest fixture providing the sink output directory
"""
# Start MCP server in background
print(f"Starting MCP server: {server_file}...")
server_process = multiprocessing.Process(target=run_mcp_server,
args=(server_file,))
server_process.start()
- time.sleep(5)
-
- # Set environment variable to control agent behavior
- os.environ["MCP_SERVER_MODE"] = mcp_server_mode
-
- print(f"\nRunning MyMCPAgent with Ollama model: {OLLAMA_MODEL}")
- print(f"MCP server mode: {mcp_server_mode}")
- print(f"MCP server endpoint: {server_endpoint}\n")
+ try:
+ time.sleep(5)
+
+ # Set environment variable to control agent behavior
+ os.environ["MCP_SERVER_MODE"] = mcp_server_mode
+
+ print(f"\nRunning MyMCPAgent with Ollama model: {OLLAMA_MODEL}")
+ print(f"MCP server mode: {mcp_server_mode}")
+ print(f"MCP server endpoint: {server_endpoint}\n")
+
+ config = Configuration()
+ config.set_string("state.backend.type", "rocksdb")
+ config.set_string("checkpointing.interval", "1s")
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ # currently, bounded source is not supported due to runtime
+ # implementation, so we use a continuous file source here.
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+ f"file:///{current_dir}/../../resources/mcp_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="mcp_source",
+ )
- env = AgentsExecutionEnvironment.get_execution_environment()
- input_list = []
- agent = MyMCPAgent()
+ deserialize_datastream = input_datastream.map(
+ lambda x: CalculationInput.model_validate_json(x)
+ )
- output_list = env.from_list(input_list).apply(agent).to_list()
+ agents_env =
AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream, key_selector=lambda x: str(x.a)
+ )
+ .apply(MyMCPAgent())
+ .to_datastream()
+ )
- # Add test inputs
- input_list.append({"key": "calc1", "value": CalculationInput(a=1, b=2)})
- input_list.append({"key": "calc2", "value": CalculationInput(a=12, b=34)})
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
- env.execute()
+ output_datastream.map(
+ lambda x: str(x).replace("\n", "").replace("\r", ""),
Types.STRING()
+ ).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
- print("Results:")
- for output in output_list:
- for key, value in output.items():
- print(f"{key}: {value}")
- assert len(output_list) == 2
- server_process.kill()
+ agents_env.execute()
+
+ actual_result = []
+ for file in result_dir.iterdir():
+ if file.is_dir():
+ for child in file.iterdir():
+ with child.open() as f:
+ actual_result.extend(f.readlines())
+ if file.is_file():
+ with file.open() as f:
+ actual_result.extend(f.readlines())
+
+ records = [line for line in actual_result if line.strip()]
+ assert len(records) == 2, f"expected 2 output records, got {records!r}"
+ finally:
+ server_process.kill()
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test.py
index eca25ced..edd2d4b9 100644
--- a/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test.py
+++ b/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test.py
@@ -42,6 +42,8 @@ from
flink_agents.e2e_tests.e2e_tests_integration.execute_test_agent import (
ExecuteTestKeySelector,
ExecuteWithAsyncExceptionTestAgent,
ExecuteWithAsyncTestAgent,
+ ExecuteWithKwargsTestAgent,
+ ExecuteWithSyncExceptionTestAgent,
)
from flink_agents.e2e_tests.test_utils import check_result
@@ -256,3 +258,107 @@ def test_durable_execute_async_exception_flink(tmp_path:
Path) -> None:
f"{current_dir}/../resources/ground_truth/test_execute_async_exception.txt"
),
)
+
+
+def test_durable_execute_sync_exception_flink(tmp_path: Path) -> None:
+ """Test synchronous durable_execute() exception handling in Flink
environment."""
+ config = Configuration()
+ config.set_string("state.backend.type", "rocksdb")
+ config.set_string("checkpointing.interval", "1s")
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+ f"file:///{current_dir}/../resources/execute_test_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="execute_test_source",
+ )
+
+ deserialize_datastream = input_datastream.map(
+ lambda x: ExecuteTestData.model_validate_json(x)
+ )
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream, key_selector=ExecuteTestKeySelector()
+ )
+ .apply(ExecuteWithSyncExceptionTestAgent())
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+
+ output_datastream.map(lambda x: json.dumps(x), Types.STRING()).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+
+ agents_env.execute()
+
+ check_result(
+ result_dir=result_dir,
+ ground_truth_dir=Path(
+
f"{current_dir}/../resources/ground_truth/test_execute_sync_exception.txt"
+ ),
+ )
+
+
+def test_durable_execute_with_kwargs_flink(tmp_path: Path) -> None:
+ """Test durable_execute() with keyword arguments in Flink environment."""
+ config = Configuration()
+ config.set_string("state.backend.type", "rocksdb")
+ config.set_string("checkpointing.interval", "1s")
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+ f"file:///{current_dir}/../resources/execute_test_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="execute_test_source",
+ )
+
+ deserialize_datastream = input_datastream.map(
+ lambda x: ExecuteTestData.model_validate_json(x)
+ )
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream, key_selector=ExecuteTestKeySelector()
+ )
+ .apply(ExecuteWithKwargsTestAgent())
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+
+ output_datastream.map(lambda x: json.dumps(x), Types.STRING()).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+
+ agents_env.execute()
+
+ check_result(
+ result_dir=result_dir,
+ ground_truth_dir=Path(
+
f"{current_dir}/../resources/ground_truth/test_execute_with_kwargs.txt"
+ ),
+ )
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test_agent.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test_agent.py
index 6bd662f4..171e8068 100644
--- a/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test_agent.py
+++ b/python/flink_agents/e2e_tests/e2e_tests_integration/execute_test_agent.py
@@ -160,3 +160,37 @@ class ExecuteWithAsyncExceptionTestAgent(Agent):
output=ExecuteTestErrorOutput(id=input_data.id,
error=str(exc))
)
)
+
+
+class ExecuteWithSyncExceptionTestAgent(Agent):
+ """Agent that tests exception handling in synchronous durable_execute()."""
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def process(event: Event, ctx: RunnerContext) -> None:
+ """Process an event and capture synchronous durable_execute()
exceptions."""
+ input_data =
ExecuteTestData.model_validate(InputEvent.from_event(event).input)
+ try:
+ ctx.durable_execute(raise_exception, f"Test error:
{input_data.value}")
+ except ValueError as exc:
+ ctx.send_event(
+ OutputEvent(
+ output=ExecuteTestErrorOutput(
+ id=input_data.id, error=f"Caught: {exc}"
+ )
+ )
+ )
+
+
+class ExecuteWithKwargsTestAgent(Agent):
+ """Agent that uses durable_execute() with keyword arguments."""
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def process(event: Event, ctx: RunnerContext) -> None:
+ """Process an event using durable_execute() with keyword arguments."""
+ input_data =
ExecuteTestData.model_validate(InputEvent.from_event(event).input)
+ result = ctx.durable_execute(compute_value, x=input_data.value, y=20)
+ ctx.send_event(
+ OutputEvent(output=ExecuteTestOutput(id=input_data.id,
result=result))
+ )
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/mock_chat_model_agent.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/mock_chat_model_agent.py
new file mode 100644
index 00000000..625bb69f
--- /dev/null
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/mock_chat_model_agent.py
@@ -0,0 +1,303 @@
+################################################################################
+# 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.
+#################################################################################
+"""Agent definitions exercising a deterministic mock chat model in Flink.
+
+These agents run the built-in ``chat_model_action``/``tool_call_action`` flow
and
+in-action ``get_resource`` resolution against a mock chat model, so the
resulting
+content is stable without any external LLM service.
+"""
+
+import uuid
+from typing import Any, Dict, List, Sequence
+
+from pydantic import BaseModel
+from pyflink.datastream import KeySelector
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import (
+ BaseChatModelConnection,
+ BaseChatModelSetup,
+)
+from flink_agents.api.decorators import (
+ action,
+ chat_model_connection,
+ chat_model_setup,
+ prompt,
+ tool,
+)
+from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.events.event_type import EventType
+from flink_agents.api.prompts.prompt import Prompt
+from flink_agents.api.resource import ResourceDescriptor, ResourceType
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.tools.tool import ToolType
+
+
+class MockChatModelInput(BaseModel):
+ """Input record for the mock chat model agents.
+
+ Attributes:
+ ----------
+ id : int
+ Unique identifier used as the partition key.
+ content : str
+ The user message content fed to the agent.
+ """
+
+ id: int
+ content: str
+
+
+class MockChatModelOutput(BaseModel):
+ """Output record capturing the deterministic chat model content.
+
+ Attributes:
+ ----------
+ id : int
+ Unique identifier echoed from the input record.
+ result : str
+ The chat model content produced for the record.
+ """
+
+ id: int
+ result: str
+
+
+class MockChatModelKeySelector(KeySelector):
+ """KeySelector extracting the partition key from a MockChatModelInput."""
+
+ def get_key(self, value: MockChatModelInput) -> int:
+ """Extract key from MockChatModelInput."""
+ return value.id
+
+
+class MockChatModelConnection(BaseChatModelConnection):
+ """Mock chat model connection integrating prompt and tool."""
+
+ def chat(
+ self,
+ messages: Sequence[ChatMessage],
+ tools: List | None = None,
+ **kwargs: Any,
+ ) -> ChatMessage:
+ """Generate a tool call or a response according to input."""
+ if "sum" in messages[-1].content:
+ input = messages[-1].content
+ # Validate the tool was bound before the model was invoked.
+ assert tools[0].name == "add"
+ function = {"name": "add", "arguments": {"a": 1, "b": 2}}
+ tool_call = {
+ "id": uuid.uuid4(),
+ "type": ToolType.FUNCTION,
+ "function": function,
+ }
+ return ChatMessage(
+ role=MessageRole.ASSISTANT, content=input,
tool_calls=[tool_call]
+ )
+ content = "\n".join([message.content for message in messages])
+ return ChatMessage(role=MessageRole.ASSISTANT, content=content)
+
+
+class MockChatModel(BaseChatModelSetup):
+ """Mock chat model setup integrating prompt and tool."""
+
+ def open(self) -> None:
+ """Do nothing."""
+
+ @property
+ def model_kwargs(self) -> Dict[str, Any]:
+ """Return model kwargs."""
+ return {}
+
+ def chat(
+ self,
+ messages: Sequence[ChatMessage],
+ prompt_args: Dict[str, Any] | None = None,
+ **kwargs: Any,
+ ) -> ChatMessage:
+ """Execute chat conversation."""
+ server = self.resource_context.get_resource(
+ self.connection, ResourceType.CHAT_MODEL_CONNECTION
+ )
+
+ if self.prompt is not None:
+ if isinstance(self.prompt, str):
+ prompt = self.resource_context.get_resource(
+ self.prompt, ResourceType.PROMPT
+ )
+ else:
+ prompt = self.prompt
+
+ if "sum" in messages[-1].content:
+ str_prompt_args = (
+ {k: str(v) for k, v in prompt_args.items()} if prompt_args
else {}
+ )
+ messages = prompt.format_messages(**str_prompt_args)
+
+ tools = None
+ if self.tools is not None:
+ tools = [
+ self.resource_context.get_resource(tool_name,
ResourceType.TOOL)
+ for tool_name in self.tools
+ ]
+
+ return server.chat(messages, tools=tools, **kwargs)
+
+
+class BuiltInActionAgent(Agent):
+ """Agent driving the built-in chat/tool actions with a mock chat model."""
+
+ @prompt
+ @staticmethod
+ def prompt() -> Prompt:
+ """Prompt used by the mock chat model."""
+ return Prompt.from_text(
+ text="Please call the appropriate tool to do the following task:
{task}",
+ )
+
+ @chat_model_connection
+ @staticmethod
+ def mock_connection() -> ResourceDescriptor:
+ """Chat model connection used by the mock chat model."""
+ return ResourceDescriptor(
+ clazz=f"{MockChatModelConnection.__module__}."
+ f"{MockChatModelConnection.__name__}"
+ )
+
+ @chat_model_setup
+ @staticmethod
+ def mock_chat_model() -> ResourceDescriptor:
+ """Chat model referenced by the ChatRequestEvent."""
+ return ResourceDescriptor(
+ clazz=f"{MockChatModel.__module__}.{MockChatModel.__name__}",
+ connection="mock_connection",
+ model="mock-model",
+ prompt="prompt",
+ tools=["add"],
+ )
+
+ @tool
+ @staticmethod
+ def add(a: int, b: int) -> int:
+ """Calculate the sum of a and b.
+
+ Parameters
+ ----------
+ a : int
+ The first operand
+ b : int
+ The second operand
+
+ Returns:
+ -------
+ int:
+ The sum of a and b
+ """
+ return a + b
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def process_input(event: Event, ctx: RunnerContext) -> None:
+ """Send a ChatRequestEvent to trigger the built-in actions."""
+ input_data = MockChatModelInput.model_validate(
+ InputEvent.from_event(event).input
+ )
+ # Carry the record id across the chat round-trip via per-key memory,
+ # since ChatResponseEvent does not echo the original input.
+ ctx.short_term_memory.set("input_id", input_data.id)
+ ctx.send_event(
+ ChatRequestEvent(
+ model="mock_chat_model",
+ messages=[
+ ChatMessage(role=MessageRole.USER,
content=input_data.content)
+ ],
+ prompt_args={"task": input_data.content},
+ )
+ )
+
+ @action(EventType.ChatResponseEvent)
+ @staticmethod
+ def process_chat_response(event: Event, ctx: RunnerContext) -> None:
+ """Emit the chat model content, keyed by the original record id."""
+ response = ChatResponseEvent.from_event(event).response
+ input_id = ctx.short_term_memory.get("input_id")
+ ctx.send_event(
+ OutputEvent(
+ output=MockChatModelOutput(id=input_id,
result=response.content)
+ )
+ )
+
+
+class GetResourceChatModel(BaseChatModelSetup):
+ """Mock chat model setup carrying custom descriptor fields."""
+
+ host: str
+ desc: str
+
+ def open(self) -> None:
+ """Do nothing."""
+
+ @property
+ def model_kwargs(self) -> Dict[str, Any]:
+ """Return model kwargs."""
+ return {}
+
+ def chat(self, messages: Sequence[ChatMessage], **kwargs: Any) ->
ChatMessage:
+ """Echo the input alongside the custom descriptor fields."""
+ return ChatMessage(
+ role=MessageRole.ASSISTANT,
+ content=f"{messages[0].content} {self.host} {self.desc}",
+ )
+
+
+class GetResourceAgent(Agent):
+ """Agent resolving a chat model via get_resource inside an action."""
+
+ @chat_model_setup
+ @staticmethod
+ def mock_chat_model() -> ResourceDescriptor:
+ """Chat model carrying custom descriptor fields."""
+ return ResourceDescriptor(
+ clazz=f"{GetResourceChatModel.__module__}."
+ f"{GetResourceChatModel.__name__}",
+ host="8.8.8.8",
+ desc="mock chat model just for testing.",
+ connection="mock",
+ model="mock-model",
+ )
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def mock_action(event: Event, ctx: RunnerContext) -> None:
+ """Resolve the chat model and emit its content, keyed by record id."""
+ input_data = MockChatModelInput.model_validate(
+ InputEvent.from_event(event).input
+ )
+ mock_chat_model = ctx.get_resource(
+ name="mock_chat_model", type=ResourceType.CHAT_MODEL
+ )
+ content = mock_chat_model.chat(
+ messages=[ChatMessage(role=MessageRole.USER,
content=input_data.content)]
+ ).content
+ ctx.send_event(
+ OutputEvent(
+ output=MockChatModelOutput(id=input_data.id, result=content)
+ )
+ )
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/mock_chat_model_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/mock_chat_model_test.py
new file mode 100644
index 00000000..dd7bc3c7
--- /dev/null
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/mock_chat_model_test.py
@@ -0,0 +1,164 @@
+################################################################################
+# 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.
+#################################################################################
+"""E2E tests running a deterministic mock chat model through the Flink
runner."""
+
+import json
+import os
+import sysconfig
+from pathlib import Path
+
+from pyflink.common import Configuration, Encoder, WatermarkStrategy
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import (
+ RuntimeExecutionMode,
+ StreamExecutionEnvironment,
+)
+from pyflink.datastream.connectors.file_system import (
+ FileSource,
+ StreamFormat,
+ StreamingFileSink,
+)
+
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from flink_agents.e2e_tests.e2e_tests_integration.mock_chat_model_agent import
(
+ BuiltInActionAgent,
+ GetResourceAgent,
+ MockChatModelInput,
+ MockChatModelKeySelector,
+)
+from flink_agents.e2e_tests.test_utils import check_result
+
+current_dir = Path(__file__).parent
+
+os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
+
+
+def test_built_in_chat_tool_action_content(tmp_path: Path) -> None:
+ """Built-in chat/tool actions produce deterministic content with a mock
model.
+
+ Exercises the built-in ``chat_model_action`` and ``tool_call_action`` flow
end
+ to end on a real MiniCluster: prompt-template formatting, the ``add`` tool
+ binding, and the final assembled content are all verified via the file
sink.
+ """
+ config = Configuration()
+ config.set_string("state.backend.type", "rocksdb")
+ config.set_string("checkpointing.interval", "1s")
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+
f"file:///{current_dir}/../resources/mock_chat_model_built_in_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="mock_chat_model_built_in_source",
+ )
+
+ deserialize_datastream = input_datastream.map(
+ lambda x: MockChatModelInput.model_validate_json(x)
+ )
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream,
key_selector=MockChatModelKeySelector()
+ )
+ .apply(BuiltInActionAgent())
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+
+ output_datastream.map(lambda x: json.dumps(x), Types.STRING()).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+
+ agents_env.execute()
+
+ check_result(
+ result_dir=result_dir,
+ ground_truth_dir=Path(
+ f"{current_dir}/../resources/ground_truth/"
+ f"test_built_in_action_content.txt"
+ ),
+ )
+
+
+def test_chat_model_get_resource_in_action(tmp_path: Path) -> None:
+ """get_resource(CHAT_MODEL) in an action resolves custom descriptor fields.
+
+ Verifies that a chat model resolved via ``ctx.get_resource`` inside an
action
+ carries the custom ``host``/``desc`` fields from its ResourceDescriptor
through
+ to the instance, and that those fields appear in the produced content.
+ """
+ config = Configuration()
+ config.set_string("state.backend.type", "rocksdb")
+ config.set_string("checkpointing.interval", "1s")
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+
f"file:///{current_dir}/../resources/mock_chat_model_get_resource_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="mock_chat_model_get_resource_source",
+ )
+
+ deserialize_datastream = input_datastream.map(
+ lambda x: MockChatModelInput.model_validate_json(x)
+ )
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream,
key_selector=MockChatModelKeySelector()
+ )
+ .apply(GetResourceAgent())
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+
+ output_datastream.map(lambda x: json.dumps(x), Types.STRING()).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+
+ agents_env.execute()
+
+ check_result(
+ result_dir=result_dir,
+ ground_truth_dir=Path(
+ f"{current_dir}/../resources/ground_truth/"
+ f"test_chat_model_get_resource.txt"
+ ),
+ )
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/workflow_memory_remote_agent.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/workflow_memory_remote_agent.py
new file mode 100644
index 00000000..840ec131
--- /dev/null
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/workflow_memory_remote_agent.py
@@ -0,0 +1,83 @@
+################################################################################
+# 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.
+#################################################################################
+import uuid
+
+from pydantic import BaseModel
+from pyflink.datastream import KeySelector
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.decorators import action
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.events.event_type import EventType
+from flink_agents.api.runner_context import RunnerContext
+
+
+class WorkflowData(BaseModel):
+ """Input record for the short-term-memory workflow.
+
+ Attributes:
+ ----------
+ value : str
+ Payload echoed back in the output.
+ key : str | None
+ Routing key. A record that arrives without a key is assigned an
+ auto-generated one during deserialization, mirroring the key
+ auto-generation the from_list runner performs for keyless records.
+ """
+
+ value: str
+ key: str | None = None
+
+
+def deserialize_with_key(line: str) -> WorkflowData:
+ """Parse a JSON line, assigning a unique key when the record has none."""
+ data = WorkflowData.model_validate_json(line)
+ if data.key is None:
+ data.key = str(uuid.uuid4())
+ return data
+
+
+class WorkflowKeySelector(KeySelector):
+ """KeySelector routing each record by its (possibly auto-generated) key."""
+
+ def get_key(self, value: WorkflowData) -> str:
+ """Extract the routing key from a WorkflowData."""
+ return value.key
+
+
+class VisitCountAgent(Agent):
+ """Agent that counts how many records it has seen per key.
+
+ Each key owns an isolated short-term memory, so ``visit_count`` accumulates
+ independently per key. A keyless input record lands under its own
+ auto-generated key and therefore counts from one.
+ """
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def count_visits(event: Event, ctx: RunnerContext) -> None:
+ """Increment and emit the per-key visit count."""
+ data = WorkflowData.model_validate(InputEvent.from_event(event).input)
+ stm = ctx.short_term_memory
+
+ count = (stm.get("visit_count") or 0) + 1
+ stm.set("visit_count", count)
+
+ ctx.send_event(
+ OutputEvent(output={"content": data.value, "visit_count": count})
+ )
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/workflow_memory_remote_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/workflow_memory_remote_test.py
new file mode 100644
index 00000000..93f5acbf
--- /dev/null
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/workflow_memory_remote_test.py
@@ -0,0 +1,103 @@
+################################################################################
+# 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.
+#################################################################################
+import json
+import os
+import sysconfig
+from pathlib import Path
+
+from pyflink.common import Configuration, Encoder, WatermarkStrategy
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import (
+ RuntimeExecutionMode,
+ StreamExecutionEnvironment,
+)
+from pyflink.datastream.connectors.file_system import (
+ FileSource,
+ StreamFormat,
+ StreamingFileSink,
+)
+
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from flink_agents.e2e_tests.e2e_tests_integration.workflow_memory_remote_agent
import (
+ VisitCountAgent,
+ WorkflowKeySelector,
+ deserialize_with_key,
+)
+from flink_agents.e2e_tests.test_utils import check_result
+
+current_dir = Path(__file__).parent
+
+os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
+
+
+def test_short_term_memory_same_key_accumulation(tmp_path: Path) -> None:
+ """Same-key records accumulate visit_count; a keyless record gets its own
key.
+
+ Three records share key ``alpha`` and drive its short-term-memory
+ ``visit_count`` to 1, 2, 3 in arrival order. A fourth record has no key, is
+ assigned an auto-generated one at deserialization, and lands in an isolated
+ memory that counts from one.
+ """
+ config = Configuration()
+ config.set_string("state.backend.type", "rocksdb")
+ config.set_string("checkpointing.interval", "1s")
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+
+ # currently, bounded source is not supported due to runtime
implementation, so
+ # we use continuous file source here.
+ input_datastream = env.from_source(
+ source=FileSource.for_record_stream_format(
+ StreamFormat.text_line_format(),
+ f"file:///{current_dir}/../resources/workflow_memory_input",
+ ).build(),
+ watermark_strategy=WatermarkStrategy.no_watermarks(),
+ source_name="workflow_memory_source",
+ )
+
+ deserialize_datastream = input_datastream.map(deserialize_with_key)
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(
+ input=deserialize_datastream, key_selector=WorkflowKeySelector()
+ )
+ .apply(VisitCountAgent())
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+
+ output_datastream.map(lambda x: json.dumps(x), Types.STRING()).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+
+ agents_env.execute()
+
+ check_result(
+ result_dir=result_dir,
+ ground_truth_dir=Path(
+ f"{current_dir}/../resources/ground_truth/test_workflow_memory.txt"
+ ),
+ )
diff --git
a/python/flink_agents/e2e_tests/resources/chat_model_multi_provider_input/input_data.txt
b/python/flink_agents/e2e_tests/resources/chat_model_multi_provider_input/input_data.txt
new file mode 100644
index 00000000..059df8ce
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/chat_model_multi_provider_input/input_data.txt
@@ -0,0 +1,2 @@
+calculate the sum of 1 and 2.
+Tell me a joke about cats.
diff --git
a/python/flink_agents/e2e_tests/resources/ground_truth/test_built_in_action_content.txt
b/python/flink_agents/e2e_tests/resources/ground_truth/test_built_in_action_content.txt
new file mode 100644
index 00000000..2b7cd0aa
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/ground_truth/test_built_in_action_content.txt
@@ -0,0 +1,2 @@
+{"id": 1, "result": "calculate the sum of 1 and 2.\nPlease call the
appropriate tool to do the following task: calculate the sum of 1 and 2.\n3"}
+{"id": 2, "result": "calculate the sum of 1 and 2.\nPlease call the
appropriate tool to do the following task: calculate the sum of 1 and 2.\n3"}
diff --git
a/python/flink_agents/e2e_tests/resources/ground_truth/test_chat_model_get_resource.txt
b/python/flink_agents/e2e_tests/resources/ground_truth/test_chat_model_get_resource.txt
new file mode 100644
index 00000000..403b1d4d
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/ground_truth/test_chat_model_get_resource.txt
@@ -0,0 +1,2 @@
+{"id": 1, "result": "the first message. 8.8.8.8 mock chat model just for
testing."}
+{"id": 2, "result": "the second message. 8.8.8.8 mock chat model just for
testing."}
diff --git
a/python/flink_agents/e2e_tests/resources/ground_truth/test_execute_sync_exception.txt
b/python/flink_agents/e2e_tests/resources/ground_truth/test_execute_sync_exception.txt
new file mode 100644
index 00000000..bea0dc6d
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/ground_truth/test_execute_sync_exception.txt
@@ -0,0 +1,3 @@
+{"id":1,"error":"Caught: Test error: 5"}
+{"id":2,"error":"Caught: Test error: 15"}
+{"id":3,"error":"Caught: Test error: 10"}
diff --git
a/python/flink_agents/e2e_tests/resources/ground_truth/test_execute_with_kwargs.txt
b/python/flink_agents/e2e_tests/resources/ground_truth/test_execute_with_kwargs.txt
new file mode 100644
index 00000000..9010c3b4
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/ground_truth/test_execute_with_kwargs.txt
@@ -0,0 +1,3 @@
+{"id":1,"result":25}
+{"id":2,"result":35}
+{"id":3,"result":30}
diff --git
a/python/flink_agents/e2e_tests/resources/ground_truth/test_workflow_memory.txt
b/python/flink_agents/e2e_tests/resources/ground_truth/test_workflow_memory.txt
new file mode 100644
index 00000000..760273a4
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/ground_truth/test_workflow_memory.txt
@@ -0,0 +1,4 @@
+{"content": "a1", "visit_count": 1}
+{"content": "a2", "visit_count": 2}
+{"content": "a3", "visit_count": 3}
+{"content": "orphan", "visit_count": 1}
diff --git a/python/flink_agents/e2e_tests/resources/mcp_input/input_data.txt
b/python/flink_agents/e2e_tests/resources/mcp_input/input_data.txt
new file mode 100644
index 00000000..e3bf070c
--- /dev/null
+++ b/python/flink_agents/e2e_tests/resources/mcp_input/input_data.txt
@@ -0,0 +1,2 @@
+{"a": 1, "b": 2}
+{"a": 12, "b": 34}
diff --git
a/python/flink_agents/e2e_tests/resources/mock_chat_model_built_in_input/input_data.txt
b/python/flink_agents/e2e_tests/resources/mock_chat_model_built_in_input/input_data.txt
new file mode 100644
index 00000000..930c983b
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/mock_chat_model_built_in_input/input_data.txt
@@ -0,0 +1,2 @@
+{"id": 1, "content": "calculate the sum of 1 and 2."}
+{"id": 2, "content": "calculate the sum of 1 and 2."}
diff --git
a/python/flink_agents/e2e_tests/resources/mock_chat_model_get_resource_input/input_data.txt
b/python/flink_agents/e2e_tests/resources/mock_chat_model_get_resource_input/input_data.txt
new file mode 100644
index 00000000..fa755068
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/mock_chat_model_get_resource_input/input_data.txt
@@ -0,0 +1,2 @@
+{"id": 1, "content": "the first message."}
+{"id": 2, "content": "the second message."}
diff --git
a/python/flink_agents/e2e_tests/resources/workflow_memory_input/input_data.txt
b/python/flink_agents/e2e_tests/resources/workflow_memory_input/input_data.txt
new file mode 100644
index 00000000..0e7ba98a
--- /dev/null
+++
b/python/flink_agents/e2e_tests/resources/workflow_memory_input/input_data.txt
@@ -0,0 +1,4 @@
+{"key": "alpha", "value": "a1"}
+{"key": "alpha", "value": "a2"}
+{"key": "alpha", "value": "a3"}
+{"value": "orphan"}
diff --git
a/python/flink_agents/runtime/tests/test_remote_execution_environment.py
b/python/flink_agents/runtime/tests/test_remote_execution_environment.py
index 51482f04..28eca85e 100644
--- a/python/flink_agents/runtime/tests/test_remote_execution_environment.py
+++ b/python/flink_agents/runtime/tests/test_remote_execution_environment.py
@@ -20,6 +20,7 @@ import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
+import pytest
import yaml
from flink_agents.plan.configuration import AgentConfiguration
@@ -199,6 +200,27 @@ def test_execute_without_job_name() -> None:
mock_stream_env.execute.assert_called_once_with(job_name=None)
+def test_apply_by_unknown_name_errors() -> None:
+ """Applying an unregistered agent name raises ValueError before execution.
+
+ The guard fires at apply() time on the remote builder, so no cluster is
+ started and no job is submitted. The Flink environment and input datastream
+ are never exercised to reach the guard, so both are mocked. Mirrors the
same
+ contract on the from_list builder.
+ """
+ with patch(
+
"flink_agents.runtime.remote_execution_environment.StreamExecutionEnvironment"
+ ):
+ remote_env = RemoteExecutionEnvironment(env=MagicMock())
+
+ builder = remote_env.from_datastream(
+ input=MagicMock(), key_selector=lambda record: record["key"]
+ )
+
+ with pytest.raises(ValueError, match="ghost"):
+ builder.apply("ghost")
+
+
def _verify_config(config: AgentConfiguration) -> None:
assert config.get_str("database.host") == "localhost"
assert config.get_int("database.port") == 5432
