[runtime][python][docs] Fix multi-agent examples and clarify sub-agent usage

docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md

@@ -159,7 +159,9 @@ The Java table declares only the `id` and `review` columns.
 
 ### Analyze Each Row with the Review Agent
 
-A Table row reaches the Action as a `dict` in Python and a `Row` in Java. Build Your First Agent
+A Table row reaches the Action as a `pyflink.common.Row` in Python and an
+`org.apache.flink.types.Row` in Java. In Python, access fields with `row["id"]` or call
+`row.as_dict()` when dictionary methods such as `.get()` are needed. Build Your First Agent
 gave its Agent a `ProductReview` object in Python and a JSON string in Java, so this example
 defines a second Agent class, `TableReviewAnalysisAgent`, for the row input. It redeclares the
 prompt, tool, and chat model setup that `ReviewAnalysisAgent` declares, and `ReviewAnalysisAgent`
@@ -171,7 +173,7 @@ itself is unchanged. The key selector that reads `id` out of a row travels with
 {{< tab "Python" >}}
 ```python
 class TableKeySelector(KeySelector):
-    """KeySelector for extracting key from dictionary input (Table data)."""
+    """KeySelector for extracting the product id from a Table Row."""
 
     def get_key(self, value: Any) -> str:
         if isinstance(value, dict):
@@ -186,14 +188,14 @@ class TableReviewAnalysisAgent(Agent):
     @action(EventType.InputEvent)
     @staticmethod
     def process_input(event: Event, ctx: RunnerContext) -> None:
-        """Process input event from Table data (dictionary format).
+        """Process input event from Table data (Row format).
 
-        When using from_table(), the input is a dictionary with keys matching
-        the table column names.
+        When using from_table(), the input is a Row with fields matching
+        the table column names. Use row.as_dict() if a dictionary is needed.
         """
-        input_dict = cast("InputEvent", event).input
-        product_id = str(input_dict["id"])
-        review_text = str(input_dict["review"])
+        input_row = cast("InputEvent", event).input
+        product_id = str(input_row["id"])
+        review_text = str(input_row["review"])
         # Rest of the body is the one Build Your First Agent shows. It keeps the id in
         # short-term memory and sends ChatRequestEvent.
         ...
@@ -274,7 +276,8 @@ One review is too little to suggest an improvement, so the job groups the per-re
 product id in a one-minute tumbling window and reduces each group to a `ProductReviewSummary`. That
 type is declared in the shared types file and carries the product id, the score distribution, and
 the collected unsatisfied reasons. Python returns the summary object, and Java emits it as JSON
-text, because the next stage reads it back with Jackson.
+text, because the next stage reads it back with Jackson. Scores outside 1–5 are excluded from
+the distribution; if none are valid, all five percentages are `0.0%`. The reasons are still retained.
 
 {{< tabs "Aggregate in a Window" >}}
 
@@ -298,7 +301,9 @@ class AggregateScoreDistributionAndDislikeReasons(ProcessWindowFunction):
                 rating_counts[rating - 1] += 1
             reason_list = reason_list + element.reasons
         total = sum(rating_counts)
-        percentages = [round((x / total) * 100, 1) for x in rating_counts]
+        percentages = [
+            round((x / total) * 100, 1) if total else 0.0 for x in rating_counts
+        ]
         formatted_percentages = [f"{p}%" for p in percentages]
         return [
             ProductReviewSummary(

python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py

@@ -39,6 +39,7 @@
 from pathlib import Path
 from typing import Any
 
+import pytest
 from pyflink.common import Encoder
 from pyflink.common.typeinfo import Types
 from pyflink.datastream import KeySelector, StreamExecutionEnvironment
@@ -61,6 +62,7 @@
 from flink_agents.api.decorators import action
 from flink_agents.api.events.event import Event, InputEvent, OutputEvent
 from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+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.subagent import InternalSubagentResourceDescriptorBuilder
@@ -172,28 +174,28 @@ def chat(
         )
 
 
-def _child_react_agent() -> ReActAgent:
-    """A ReActAgent with neither a prompt nor an output schema.
-
-    The connection is registered on the child itself, keeping the child plan
-    self-contained: a child scope resolves resources against the child plan
-    only, so the connection must live there.
-    """
+def _child_react_agent(*, local_connection: bool = True) -> ReActAgent:
+    """A bare ReActAgent using either its own or the root's connection."""
     child = ReActAgent(
         chat_model=ResourceDescriptor(
             clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}",
             connection="child_connection",
             model="mock-model",
         ),
     )
-    child.add_resource(
+    if local_connection:
+        _register_child_connection(child)
+    return child
+
+
+def _register_child_connection(agent: Agent) -> None:
+    agent.add_resource(
         "child_connection",
         ResourceType.CHAT_MODEL_CONNECTION,
         ResourceDescriptor(
             clazz=f"{ChildScriptedConnection.__module__}.{ChildScriptedConnection.__name__}",
         ),
     )
-    return child
 
 
 def _tool_calling_child_react_agent() -> ReActAgent:
@@ -234,8 +236,8 @@ async def orchestrate(event: Event, ctx: RunnerContext) -> None:
             ctx.send_event(OutputEvent(output=msg))
 
 
-def _delegating_parent() -> ReActAgent:
-    """A bare ReActAgent parent that declares the child to its model.
+def _delegating_parent(*, shared_connection: bool = False) -> ReActAgent:
+    """A prompted ReActAgent parent that declares the child to its model.
 
     Sub-classing ReActAgent would trip the plan compiler, which rejects
     inherited ``@action`` methods on a concrete agent, so the parent is a
@@ -248,7 +250,10 @@ def _delegating_parent() -> ReActAgent:
             model="mock-model",
             subagents=[CHILD_SCOPE],
         ),
+        prompt=Prompt.from_text("PARENT ONLY: {input}"),
     )
+    if shared_connection:
+        _register_child_connection(parent)
     parent.add_resource(
         "parent_connection",
         ResourceType.CHAT_MODEL_CONNECTION,
@@ -261,7 +266,7 @@ def _delegating_parent() -> ReActAgent:
         CHILD_SCOPE,
         ResourceType.AGENT,
         InternalSubagentResourceDescriptorBuilder()
-        .agent(_child_react_agent())
+        .agent(_child_react_agent(local_connection=not shared_connection))
         .description("Reviews a code diff.")
         .input_schema(
             '{"type":"object","properties":{"input":{"type":"string"}},'
@@ -297,10 +302,28 @@ def _run_and_collect(root: Agent, tmp_path: Path) -> str:
     return "".join(p.read_text() for p in result_dir.rglob("*") if p.is_file())
 
 
-def test_caller_driven_react_subagent(tmp_path: Path) -> None:
+@pytest.mark.parametrize("connection_scope", ["child", "root", "nested_root", "override"])
+def test_caller_driven_react_subagent(tmp_path: Path, connection_scope: str) -> None:
     """A submitted string runs the child's ReAct loop and returns its answer."""
     root = CallerRootAgent()
-    root.add_resource(CHILD_SCOPE, ResourceType.AGENT, _child_react_agent())
+    child = _child_react_agent(local_connection=connection_scope in {"child", "override"})
+    if connection_scope in {"root", "nested_root"}:
+        _register_child_connection(root)
+    if connection_scope == "override":
+        # This connection would ask for an unavailable tool if the child failed
+        # to prefer its own connection registered under the same name and type.
+        root.add_resource(
+            "child_connection",
+            ResourceType.CHAT_MODEL_CONNECTION,
+            ResourceDescriptor(
+                clazz=f"{ToolCallingChildConnection.__module__}.{ToolCallingChildConnection.__name__}",
+            ),
+        )
+    if connection_scope == "nested_root":
+        middle = CallerRootAgent()
+        middle.add_resource(CHILD_SCOPE, ResourceType.AGENT, child)
+        child = middle
+    root.add_resource(CHILD_SCOPE, ResourceType.AGENT, child)
 
     contents = _run_and_collect(root, tmp_path)
 
@@ -311,9 +334,14 @@ def test_caller_driven_react_subagent(tmp_path: Path) -> None:
     )
 
 
-def test_model_driven_react_subagent_delegation(tmp_path: Path) -> None:
+@pytest.mark.parametrize("shared_connection", [False, True])
+def test_model_driven_react_subagent_delegation(
+    tmp_path: Path, shared_connection: bool
+) -> None:
     """A model-issued delegation reaches the declared child and closes the loop."""
-    contents = _run_and_collect(_delegating_parent(), tmp_path)
+    contents = _run_and_collect(
+        _delegating_parent(shared_connection=shared_connection), tmp_path
+    )
 
     # The final answer echoes the tool observation, proving the whole loop: the
     # model's {"input": ...} arguments became the child's user message through
@@ -322,6 +350,7 @@ def test_model_driven_react_subagent_delegation(tmp_path: Path) -> None:
     # this parent answer does -- so both fragments here mean the round-trip
     # completed; the assertion stays split because the observation carries the
     # result in the framework's string form, not verbatim.
+    assert "PARENT ONLY" not in contents
     assert "observed:" in contents, f"no observation echoed; collected: {contents!r}"
     assert "child saw: review the diff" in contents, (
         f"child output missing from the loop; collected: {contents!r}"

python/flink_agents/examples/quickstart/agents/table_review_analysis_agent.py

@@ -47,15 +47,15 @@
 
 
 class TableKeySelector(KeySelector):
-    """KeySelector for extracting key from dictionary input (Table data)."""
+    """KeySelector for extracting the product id from a Table Row."""
 
     def get_key(self, value: Any) -> str:
-        """Extract key from dictionary.
+        """Extract the product id from a Table Row or mapping.
 
         Parameters
         ----------
         value : Any
-            The input value, expected to be a dictionary with 'id' key.
+            The input value, with an 'id' field.
 
         Returns:
         -------
@@ -72,11 +72,11 @@ class TableReviewAnalysisAgent(Agent):
     """An agent that analyzes product reviews from Flink Table input.
 
     This agent is designed to work with Flink Table API. It receives input as
-    dictionary (when using from_table()) and produces analysis results including
-    satisfaction score and reasons for dissatisfaction.
+    a pyflink.common.Row (when using from_table()) and produces analysis results
+    including satisfaction score and reasons for dissatisfaction.
 
     The main difference from ReviewAnalysisAgent is that this agent handles
-    dictionary input instead of Pydantic objects.
+    Row input instead of Pydantic objects.
     """
 
     @prompt
@@ -116,14 +116,14 @@ def review_analysis_model() -> ResourceDescriptor:
     @action(EventType.InputEvent)
     @staticmethod
     def process_input(event: Event, ctx: RunnerContext) -> None:
-        """Process input event from Table data (dictionary format).
+        """Process input event from Table data (Row format).
 
-        When using from_table(), the input is a dictionary with keys matching
-        the table column names.
+        When using from_table(), the input is a Row with fields matching
+        the table column names. Use row.as_dict() if a dictionary is needed.
         """
-        input_dict = cast("InputEvent", event).input
-        product_id = str(input_dict["id"])
-        review_text = str(input_dict["review"])
+        input_row = cast("InputEvent", event).input
+        product_id = str(input_row["id"])
+        review_text = str(input_row["review"])
 
         ctx.short_term_memory.set("id", product_id)
 

python/flink_agents/examples/quickstart/tests/test_multiple_agent_example.py

@@ -0,0 +1,47 @@
+################################################################################
+#  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.
+#################################################################################
+"""Regression tests for score aggregation in the multi-agent example."""
+
+import pytest
+
+from flink_agents.examples.quickstart.agents.custom_types_and_resources import (
+    ProductReviewAnalysisRes,
+)
+from flink_agents.examples.quickstart.workflow_multiple_agent_example import (
+    AggregateScoreDistributionAndDislikeReasons,
+)
+
+
+@pytest.mark.parametrize(
+    ("scores", "expected"),
+    [
+        ([0, 6], ["0.0%"] * 5),
+        ([1, 5, 0, 6], ["50.0%", "0.0%", "0.0%", "0.0%", "50.0%"]),
+        ([3, 3], ["0.0%", "0.0%", "100.0%", "0.0%", "0.0%"]),
+    ],
+)
+def test_score_distribution_ignores_out_of_range_scores(scores, expected):
+    reviews = [
+        ProductReviewAnalysisRes(id="product", score=score, reasons=[str(score)])
+        for score in scores
+    ]
+    summary, = AggregateScoreDistributionAndDislikeReasons().process(
+        "product", None, reviews
+    )
+    assert summary.score_hist == expected
+    assert summary.unsatisfied_reasons == [str(score) for score in scores]

python/flink_agents/examples/quickstart/tests/test_prompt_driven_actions.py

@@ -20,6 +20,7 @@
 from unittest.mock import MagicMock
 
 import pytest
+from pyflink.common import Row
 
 from flink_agents.api.chat_message import MessageRole
 from flink_agents.api.events.event import InputEvent
@@ -38,7 +39,7 @@
     ("agent", "payload"),
     [
         (ReviewAnalysisAgent, {"id": "product", "review": "damaged"}),
-        (TableReviewAnalysisAgent, {"id": "product", "review": "damaged"}),
+        (TableReviewAnalysisAgent, Row(id="product", review="damaged")),
         (
             ProductSuggestionAgent,
             {"id": "product", "score_hist": ["1"], "unsatisfied_reasons": ["damaged"]},

python/flink_agents/examples/quickstart/workflow_multiple_agent_example.py

@@ -66,7 +66,9 @@ def process(
                 rating_counts[rating - 1] += 1
             reason_list = reason_list + element.reasons
         total = sum(rating_counts)
-        percentages = [round((x / total) * 100, 1) for x in rating_counts]
+        percentages = [
+            round((x / total) * 100, 1) if total else 0.0 for x in rating_counts
+        ]
         formatted_percentages = [f"{p}%" for p in percentages]
         return [
             ProductReviewSummary(

python/flink_agents/runtime/flink_runner_context.py

@@ -749,7 +749,13 @@ def __scoped_plan_and_cache(self, plan_json: str) -> tuple:
             from flink_agents.plan.agent_plan import AgentPlan
 
             scoped_plan = AgentPlan.model_validate_json(plan_json)
-            cache = ResourceCache(scoped_plan.resource_providers, scoped_plan.config)
+            # Match Java's child scopes: local resources override resources
+            # shared by the root, including for nested sub-agents.
+            cache = ResourceCache(
+                scoped_plan.resource_providers,
+                scoped_plan.config,
+                parent=self.__resource_cache,
+            )
             cache.set_java_resource_adapter(self.__j_resource_adapter)
             entry = (scoped_plan, cache)
             self.__scoped_resource_caches[plan_json] = entry

python/flink_agents/runtime/remote_execution_environment.py

@@ -99,10 +99,10 @@ def __init__(
 
     @property
     def t_env(self) -> StreamTableEnvironment:
-        """Get or crate table environment."""
+        """Get or create the table environment for the input stream."""
         if self.__t_env is None:
             self.__t_env = StreamTableEnvironment.create(
-                stream_execution_environment=self.__env
+                stream_execution_environment=self.__input.get_execution_environment()
             )
         return self.__t_env
 

python/flink_agents/runtime/resource_cache.py

@@ -41,6 +41,7 @@ def __init__(
         self,
         resource_providers: Dict[ResourceType, Dict[str, ResourceProvider]],
         config: AgentConfiguration | None = None,
+        parent: "ResourceCache | None" = None,
     ) -> None:
         """Create a ResourceCache from the given resource providers and config.
 
@@ -50,9 +51,13 @@ def __init__(
             Two-level mapping of resource type to resource name to provider.
         config : AgentConfiguration | None
             Agent configuration passed to providers during resource creation.
+        parent : ResourceCache | None
+            Fallback cache for resources not declared in this scope. Resources
+            resolved there remain owned by that cache, including their lifecycle.
         """
         self._resource_providers = resource_providers or {}
         self._config = config
+        self._parent = parent
         self._cache: Dict[ResourceType, Dict[str, Resource]] = {}
         self._j_resource_adapter: Any = None
         self._resource_context = ResourceContextImpl(self)
@@ -86,6 +91,8 @@ def get_resource(self, name: str, type: ResourceType) -> Resource:
             return cached
         providers = self._resource_providers.get(type)
         if providers is None or name not in providers:
+            if self._parent is not None:
+                return self._parent.get_resource(name, type)
             msg = f"Resource not found: '{name}' of type {type}"
             raise KeyError(msg)
         resource_provider = providers[name]