weiqingy commented on code in PR #938:
URL: https://github.com/apache/flink-agents/pull/938#discussion_r3837524310
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -172,6 +172,7 @@ void open(
}
if (containPythonAction || mem0Configured) {
initPythonActionExecutor(agentPlan, jobIdentifier);
+ resourceCache.setPythonActionExecutor(pythonActionExecutor);
Review Comment:
The interpreter starts whenever a plan has Python-owned resources, but
`pythonActionExecutor` only gets wired when there is also a Python action or
Mem0 (`:173`). `ResourceCache.eagerMaterialize` needs that executor for any
Python-owned provider though (`ResourceCache.java:196-200`), and
`ActionExecutionOperator.open()` always reaches it through
`registerSubagentSetups()` (`:230`, then `:757`).
So a plan with a Python-owned AGENT resource but no Python action and no
Mem0 would start the interpreter and then fail at operator open. It looks
constructible from Java today: `AgentPlan.java:334` turns any descriptor
carrying a `pythonClazz` argument into a `PythonResourceProvider`, and
`addResource(name, AGENT, descriptor)` goes through that branch at `:597-602`.
That said, this is also the cross-language case the description says is
deferred, so I am not sure it is meant to work yet. Is it?
One part seems worth tidying either way: the `checkState` message says the
Python runtime is not running, but it is. The interpreter, runner context and
adapter are all up, only the executor was skipped, so the message could send
someone debugging in the wrong direction.
Would widening the guard to `containPythonAction || containPythonResource ||
mem0Configured` do it? I did notice the "is this provider Python-owned?" test
is spelled out in three places now (`:133-142`, `ResourceCache.java:186-187`,
`flink_runner_context.py:364-366`), and the first two already disagree about
`PythonSerializableResourceProvider`, so a shared helper might be the better
home for it. Curious which way you would go.
##########
python/flink_agents/e2e_tests/e2e_tests_integration/subagent_external_integration_agent.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.
+################################################################################
+"""Agents exercising the Python external sub-agent modes.
+
+An async (durable pub/sub) setup and a deferred setup, each driven by a Python
+action running on the Java runtime over pemja. The backend is an in-memory run
+store held on the setup instance (pemja's Python runs in the MiniCluster JVM,
+not the test process), so the test needs no external service while still
+exercising the full submit / poll / fetch machinery of each mode.
+"""
+
+from typing import Any
+
+from pydantic import PrivateAttr
+from typing_extensions import override
+
+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.resource import ResourceType
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.async_subagent import BaseAsyncSubagentSetup,
RunStatus
+from flink_agents.runtime.deferred_subagent import (
+ DeferredSubagentSetup,
+ PreparedTriple,
+)
+
+
+def _outcome(result: SubagentResult) -> str:
+ """Render a sub-agent result as the string emitted downstream."""
+ return result.result if result.success else f"ERR:{result.error_message}"
+
+
+class InMemoryAsyncSubagentSetup(BaseAsyncSubagentSetup):
+ """External async setup backed by an in-memory run store.
+
+ A prompt containing ``fail`` produces a failed run; any other prompt
+ completes and echoes back, tagged with the injected sub-agent name.
+ """
+
+ _runs: dict = PrivateAttr(default_factory=dict)
+
+ @override
+ def call_submit_request(self, session_id: str, call_id: str, prompt: Any)
-> None:
+ """Record the run under its (session_id, call_id) identity."""
+ self._runs[(session_id, call_id)] = prompt
+
+ @override
+ def call_query_status(self, session_id: str, call_id: str) -> RunStatus:
Review Comment:
nit: `call_query_status` reports terminal on the very first probe, so
`_await_until_terminal` (`async_subagent.py:272-281`) runs a single iteration
and never reaches the sleep. To be fair the loop is properly covered at unit
level through `_MockAsyncSetup`'s probe counter
(`test_async_subagent.py:124-137`), so this is not really an untested path. It
is more that the module docstring at `:24` reaches a bit further than the tests
do when it says they exercise "the full submit / poll / fetch machinery of each
mode".
Should that counter come across into `InMemoryAsyncSubagentSetup`, or is
trimming the docstring claim to what the e2e actually walks the simpler call?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetup.java:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.runtime.subagent;
+
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentFuture;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Production base for sub-agents whose protocol is an asynchronous job, run
in durable pub/sub
+ * mode: {@code submit} publishes the run through one durable POST, the
returned handle subscribes
+ * to it.
+ */
+public abstract class BaseAsyncSubagentSetup extends BaseSubagentSetup {
+
+ /**
+ * Delay between status probes while waiting for the run to reach a
terminal state. Defaults to
+ * {@code 500}. The descriptor-based constructor reads the optional {@code
+ * statusPollIntervalMillis} argument over it, and subclasses may override
it directly.
+ */
+ protected long statusPollIntervalMillis = 500;
+
+ protected BaseAsyncSubagentSetup() {}
+
+ /**
+ * Descriptor-based construction, as used by YAML-declared {@code
subagents:} entries: reads the
+ * optional {@code statusPollIntervalMillis} argument, falling back to the
default of {@code
+ * 500} when absent.
+ */
+ protected BaseAsyncSubagentSetup(
+ ResourceDescriptor descriptor, ResourceContext resourceContext) {
+ Number statusPollInterval =
descriptor.getArgument("statusPollIntervalMillis");
Review Comment:
nit: the descriptor argument here and the pydantic alias on the Python side
are both new, and I could not find a test that goes through either one. Every
test I found sets the field directly (`MockAsyncSubagentSetup.java:73`,
`external/ExternalAsyncSubagentSetup.java:57`, `test_async_subagent.py:114`),
so nothing builds a setup from a `ResourceDescriptor` carrying the argument,
and nothing feeds the camelCase spelling to the aliases at
`async_subagent.py:200-202`.
Those two literal strings are what keeps the Java and Python spellings in
step, so a typo in either would quietly fall back to 500 rather than fail. Is a
line per side worth adding, a descriptor carrying the argument on the Java side
and a `statusPollIntervalMillis=...` construction on the Python side?
##########
python/flink_agents/runtime/base_subagent.py:
##########
@@ -0,0 +1,278 @@
+################################################################################
+# 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.
+################################################################################
+"""The framework-level runtime base shared by every sub-agent execution
mode."""
+
+import hashlib
+import json
+import uuid
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, replace
+from typing import Any
+
+from pydantic import PrivateAttr
+
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import SubagentFuture, SubagentSetup
+from flink_agents.runtime.subagent_handles import PendingSubagentCallRegistry
+from flink_agents.runtime.task_lifecycle_listener import TaskLifecycleListener
+
+
+def _event_attributes(event: Any) -> dict[str, Any]:
+ """Normalize an event's attributes into a plain dict.
+
+ Accepts both a Java ``Event`` reference passed across the bridge and a
+ plain Python mapping, copying Java maps entry by entry.
+ """
+ attributes = event.getAttributes()
+ if attributes is None:
+ return {}
+ if isinstance(attributes, dict):
+ return dict(attributes)
+ try:
+ return {str(k): v for k, v in attributes.entrySet()}
+ except AttributeError:
+ return {str(k): v for k, v in dict(attributes).items()}
+
+
+@dataclass(frozen=True)
+class Namespace:
+ """The caller-side identity of one action task execution.
+
+ Provides the task identity keying the runtime bookkeeping and the
+ namespace digest seeding the deterministic ids of the sub-agent
+ calls the task issues.
+
+ Key, sequence number, action name, and the event's type and
+ attributes are facts of the execution itself, identical for every
+ sub-agent called from it. The subagent name distinguishes the
+ sub-agents called from one action, so it alone keeps their id
+ ranges apart.
+ """
+
+ key: str
+ sequence_number: int
+ action_name: str
+ event_type: str
+ event_attributes: dict[str, Any]
+ event_id: str
+ subagent_name: str = ""
+
+ @staticmethod
+ def from_task(task: Any) -> "Namespace":
+ """Extract the facts from an ``ActionTask`` reference or fake."""
+ return Namespace(
+ key=str(task.getKey()),
+ sequence_number=int(task.getSequenceNumber()),
+ action_name=str(task.getAction().getName()),
+ event_type=str(task.getEvent().getType()),
+ event_attributes=_event_attributes(task.getEvent()),
+ event_id=str(task.getEvent().getId()),
+ )
+
+ @property
+ def task_identity(self) -> str:
+ """A key unique among live task executions and stable across the
+ steps of one task.
+ """
+ return
f"{self.key}#{self.sequence_number}#{self.action_name}#{self.event_id}"
+
+ def namespace_digest(self) -> str:
+ """Digest the id-bearing facts into a name-based UUID string.
+
+ The ids are reproducible across a failover replay. The event id
+ stays out of the digest: it keys the runtime bookkeeping only.
+ """
+ fields = {
+ "actionName": self.action_name,
+ "eventAttributes": self.event_attributes,
+ "eventType": self.event_type,
+ "key": self.key,
+ "sequenceNumber": self.sequence_number,
+ "subagentName": self.subagent_name,
+ }
+ payload = json.dumps(
+ fields, sort_keys=True, separators=(",", ":"), default=str
+ ).encode("utf-8")
+ # MD5 with the version/variant bits, as in Java's
+ # UUID.nameUUIDFromBytes (a version 3 UUID).
+ digest = bytearray(hashlib.md5(payload).digest())
+ digest[6] = (digest[6] & 0x0F) | 0x30
+ digest[8] = (digest[8] & 0x3F) | 0x80
+ return str(uuid.UUID(bytes=bytes(digest)))
+
+
+class SubagentIdAllocator:
+ """Deterministic ``(session_id, call_id)`` source for one task execution.
+
+ The namespace digest fixes the counting range, so a failover replay
+ of the same task hands out the same ids in the same call order.
+ """
+
+ def __init__(self, namespace: Namespace) -> None:
+ """Create an allocator over one task's namespace."""
+ self._namespace = namespace
+ self._session_ordinal = 0
+ self._per_session_call_ordinals: dict[str, int] = {}
+
+ def next_session_id(self) -> str:
+ """Create a session id scoped to this task's namespace."""
+ ordinal = self._session_ordinal
+ self._session_ordinal += 1
+ return f"{self._namespace.namespace_digest()}-{ordinal}"
+
+ def next_call_id(self, session_id: str) -> str:
+ """Create a call id by appending the per-session ordinal."""
+ ordinal = self._per_session_call_ordinals.get(session_id, 0) + 1
+ self._per_session_call_ordinals[session_id] = ordinal
+ return f"{session_id}-{ordinal}"
+
+
+class BaseSubagentSetup(SubagentSetup, TaskLifecycleListener, ABC):
+ """Runtime base for sub-agent setups, holding the per-task id allocators
+ and pending-call registries keyed to the currently executing action task.
+ How an invocation is issued stays an execution mode owned by the concrete
+ subclass.
+ """
+
+ _per_task_allocators: dict[str, SubagentIdAllocator] = PrivateAttr(
+ default_factory=dict
+ )
+ _per_task_registries: dict[str, PendingSubagentCallRegistry] = PrivateAttr(
+ default_factory=dict
+ )
+ _current_namespace: Namespace | None = PrivateAttr(default=None)
+ _subagent_name: str | None = PrivateAttr(default=None)
+
+ #
--------------------------------------------------------------------------------
+ # Task lifecycle hooks (keyword-invoked by the runtime bridge)
+ #
--------------------------------------------------------------------------------
+
+ def on_task_prepared(self, task: Any) -> None:
+ """Record the task whose execution is currently issuing calls."""
+ namespace = Namespace.from_task(task)
+ self._current_namespace = replace(
+ namespace, subagent_name=self._subagent_name or ""
+ )
+
+ def on_task_transferred(self, from_task: Any, to_task: Any) -> None:
+ """Move the finishing task's bookkeeping onto the generated task."""
+ from_identity = Namespace.from_task(from_task).task_identity
+ to_identity = Namespace.from_task(to_task).task_identity
+ allocator = self._per_task_allocators.pop(from_identity, None)
+ if allocator is not None:
+ self._per_task_allocators[to_identity] = allocator
+ registry = self._per_task_registries.pop(from_identity, None)
+ if registry is not None:
+ registry.set_action_name(
+ Namespace.from_task(to_task).action_name
+ )
+ self._per_task_registries[to_identity] = registry
+
+ def on_task_finished(self, task: Any) -> None:
+ """Drop the task's bookkeeping and enforce resolved handles."""
+ self._current_namespace = None
+ identity = Namespace.from_task(task).task_identity
+ self._per_task_allocators.pop(identity, None)
+ registry = self._per_task_registries.pop(identity, None)
+ if registry is not None:
+ registry.check_empty()
+
+ #
--------------------------------------------------------------------------------
+ # Identity injected by the framework
+ #
--------------------------------------------------------------------------------
+
+ def set_subagent_name(self, subagent_name: str) -> None:
+ """Record the resource name the framework injects as the subagent
name."""
+ self._subagent_name = subagent_name
+
+ @property
+ def subagent_name(self) -> str | None:
+ """The injected subagent name, or None outside the framework."""
+ return self._subagent_name
+
+ #
--------------------------------------------------------------------------------
+ # Submit dispatch: complete missing ids, then delegate to the mode
+ #
--------------------------------------------------------------------------------
+
+ def submit(
+ self,
+ ctx: RunnerContext,
+ prompt: Any,
+ session_id: str | None = None,
+ call_id: str | None = None,
+ ) -> SubagentFuture:
+ """Issue an invocation, assigning the missing ids deterministically."""
+ if session_id is None or call_id is None:
+ allocator = self._current_allocator()
+ if session_id is None:
+ session_id = allocator.next_session_id()
+ if call_id is None:
+ call_id = allocator.next_call_id(session_id)
+ return self.submit_with_identity(ctx, prompt, session_id, call_id)
Review Comment:
`submit` is annotated `-> SubagentFuture` (`:218`, and at
`api/subagent.py:169`), but the `submit_with_identity` it delegates to is
`async def` in async mode (`async_subagent.py:205`) and a plain `def` in
deferred mode (`deferred_subagent.py:137`). So the same method with the same
annotation hands back a coroutine in one mode and a handle in the other.
The e2e agent comments at `subagent_external_integration_agent.py:113` and
`:129` show that is intentional, so I am not questioning the design. What
caught my eye is that the split is not visible anywhere a user would look: not
in the annotation, not in the `SubagentSetup.submit` docstring. Java has no
such split (`BaseAsyncSubagentSetup.java:70` and
`BaseDeferredSubagentSetup.java:36` both return the handle), and `AGENTS.md:18`
asks for Java and Python to stay semantically aligned.
The direction that worries me is the quiet one. If someone writes the
deferred shape (`future = reviewer.submit(...)` then `result = await future`)
against an async setup, it still runs, but awaiting the coroutine gives back
the `AsyncSubagentFuture`, so `result` ends up a handle instead of a
`SubagentResult` and flows downstream with nothing raised. The other direction
at least fails loudly with a `TypeError`.
Is it worth making `BaseSubagentSetup.submit` itself `async def` and
awaiting the deferred return, so both modes share one protocol? Or if the split
is here to stay, where would you want the shape written down so a caller sees
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
