diff --git a/README.md b/README.md index 74b1a68..03b0efc 100644 --- a/README.md +++ b/README.md @@ -319,6 +319,21 @@ See the [examples/](./examples) directory for runnable scripts: MCP examples are provided for both clouds and explicitly send `ark-beta-mcp: true`. Other built-in-tool examples are CN-only and show their required beta headers. +## Managed Agents multiagent updates + +`MultiagentType` is a string enum. Its wire value remains `coordinator` or +`multiagent_20261001`, while `str(config.type)` includes the enum class name. +Use `.value` when an application needs the wire string. + +`MultiagentConfig` represents Create and Response values, while +`MultiagentPatch` represents Update fields: `coordinator` uses `agents` and +`multiagent_20261001` uses `subagents`. The patch model separately represents +omitted and explicit-null values. + +When updating an Agent, omit `multiagent` to leave the topology unchanged, pass +`multiagent=None` to remove it, or pass `agents=[]` to clear a coordinator's +agent roster. + ## Self-hosted MCP tools The self-hosted worker core exposes a protocol-independent `MCPClient` @@ -336,6 +351,15 @@ declaration and `mcp_tools()` for the worker registry. Keep both alive for the entire worker lifetime. See the runnable [`self_hosted_mcp_worker`](./examples/self_hosted_mcp_worker) example. +`EnvironmentWorkerOptions` exposes `prefer_stream` and +`stream_timeout_seconds`; the defaults remain stream mode. The timeout is applied +to httpx connect/read phases, so a stream that receives no bytes for 30 seconds +reconnects rather than expiring after 30 seconds in total. Worker logs record +stream connection attempts, the first business event, session status +transitions, reconnects, and a warning when a running session has made no +business progress for 60 seconds. Periodic `span.*` events do not count as +business progress. These logs do not add an automatic progress timeout. + The adapter wraps an already connected MCP `ClientSession`, so applications may use stdio or another transport supported by their MCP client. One client session is reused across all Managed Agents Sessions handled by the worker; diff --git a/src/arkruntime/resources/agents/agents.py b/src/arkruntime/resources/agents/agents.py index 51887e6..c777b15 100644 --- a/src/arkruntime/resources/agents/agents.py +++ b/src/arkruntime/resources/agents/agents.py @@ -15,6 +15,7 @@ from ...types.agent.mcp_server import MCPServer from ...types.agent.model_config import ModelConfig from ...types.agent.multiagent_config import MultiagentConfig +from ...types.agent.multiagent_patch import MultiagentPatch from ...types.agent.tag import Tag from ...types.agent.tool_item import ToolItem @@ -42,6 +43,14 @@ def _list_query( return q +def _validate_create_multiagent( + multiagent: MultiagentConfig | NotGiven | None, +) -> MultiagentConfig | NotGiven | None: + if multiagent is NOT_GIVEN or multiagent is None: + return multiagent + return MultiagentConfig.model_validate(multiagent) + + class Agents(SyncAPIResource): def create( self, @@ -61,6 +70,7 @@ def create( extra_body: Body | None = None, timeout: float | httpx.Timeout | None = None, ) -> Agent: + multiagent = _validate_create_multiagent(multiagent) return self._post( _PREFIX, body=dump_body( @@ -151,7 +161,7 @@ def update( mcp_servers: Optional[List[MCPServer]] | NotGiven = NOT_GIVEN, tools: Optional[List[ToolItem]] | NotGiven = NOT_GIVEN, skills: Optional[List[AgentSkillRef]] | NotGiven = NOT_GIVEN, - multiagent: Optional[MultiagentConfig] | NotGiven = NOT_GIVEN, + multiagent: Optional[MultiagentPatch] | NotGiven = NOT_GIVEN, metadata: Optional[dict] | NotGiven = NOT_GIVEN, extra_headers=None, extra_query=None, @@ -241,6 +251,7 @@ async def create( extra_body=None, timeout=None, ) -> Agent: + multiagent = _validate_create_multiagent(multiagent) return await self._post( _PREFIX, body=dump_body( diff --git a/src/arkruntime/resources/sessions/session_events.py b/src/arkruntime/resources/sessions/session_events.py index d796a5a..b46aea7 100644 --- a/src/arkruntime/resources/sessions/session_events.py +++ b/src/arkruntime/resources/sessions/session_events.py @@ -212,7 +212,10 @@ def stream( client = self._client._client # httpx.Client base = str(self._client._base_url).rstrip("/") url = f"{base}/sessions/{session_id}/events/stream" - headers = {"Accept": "text/event-stream", **(self._client.auth_headers or {})} + headers = { + **self._client.default_headers(), + "Accept": "text/event-stream", + } with client.stream("GET", url, headers=headers, timeout=timeout) as resp: _raise_for_stream_status(self, resp) event_name = "" @@ -312,7 +315,10 @@ async def stream( client = self._client._client # httpx.AsyncClient base = str(self._client._base_url).rstrip("/") url = f"{base}/sessions/{session_id}/events/stream" - headers = {"Accept": "text/event-stream", **(self._client.auth_headers or {})} + headers = { + **self._client.default_headers(), + "Accept": "text/event-stream", + } async with client.stream("GET", url, headers=headers, timeout=timeout) as resp: await _araise_for_stream_status(self, resp) event_name = "" diff --git a/src/arkruntime/resources/sessions/session_threads.py b/src/arkruntime/resources/sessions/session_threads.py index e37389c..38b0253 100644 --- a/src/arkruntime/resources/sessions/session_threads.py +++ b/src/arkruntime/resources/sessions/session_threads.py @@ -134,7 +134,10 @@ def stream( client = self._client._client base = str(self._client._base_url).rstrip("/") url = f"{base}/sessions/{session_id}/threads/{thread_id}/stream" - headers = {"Accept": "text/event-stream", **(self._client.auth_headers or {})} + headers = { + **self._client.default_headers(), + "Accept": "text/event-stream", + } with client.stream("GET", url, headers=headers, timeout=timeout) as resp: _raise_for_stream_status(self, resp) event_name = "" @@ -251,7 +254,10 @@ async def stream( client = self._client._client base = str(self._client._base_url).rstrip("/") url = f"{base}/sessions/{session_id}/threads/{thread_id}/stream" - headers = {"Accept": "text/event-stream", **(self._client.auth_headers or {})} + headers = { + **self._client.default_headers(), + "Accept": "text/event-stream", + } async with client.stream("GET", url, headers=headers, timeout=timeout) as resp: await _araise_for_stream_status(self, resp) event_name = "" diff --git a/src/arkruntime/selfhosted/session_tool_runner.py b/src/arkruntime/selfhosted/session_tool_runner.py index aa64dc2..0fe9bec 100644 --- a/src/arkruntime/selfhosted/session_tool_runner.py +++ b/src/arkruntime/selfhosted/session_tool_runner.py @@ -52,6 +52,7 @@ STREAM_BACKOFF_START = 0.5 STREAM_BACKOFF_CAP = 10.0 STREAM_HEALTHY_AFTER = 30.0 +STREAM_PROGRESS_WARN_SECONDS = 60.0 SEND_RETRIES = 3 STREAM_QUEUE_SIZE = 256 @@ -98,6 +99,7 @@ class SessionToolRunnerOptions: stop_event: Any = None logger: logging.Logger = logging.getLogger("arkruntime.selfhosted.session_tool_runner") on_tool_error: Any = None + initial_session_status: str = "" @dataclass @@ -149,6 +151,15 @@ def _is_stopped(self) -> bool: return self._stop.is_set() or bool(self.options.stop_event and self.options.stop_event.is_set()) def run(self) -> List[ToolCallResult]: + mode = "stream" if self.options.prefer_stream and hasattr(self.api, "stream_events") else "list" + self.options.logger.info( + "session runner started session_id=%s work_id=%s mode=%s max_idle_seconds=%s stream_timeout_seconds=%.3fs", + self.session_id, + self.options.work_id, + mode, + self.options.max_idle_seconds, + self.options.stream_timeout_seconds, + ) try: if self.options.result_store is not None: pending, processed = self.options.result_store.recover() @@ -161,7 +172,11 @@ def run(self) -> List[ToolCallResult]: self._consume_stream_loop() return self._results except EventStreamUnsupported: - pass + mode = "list" + self.options.logger.info( + "session stream unsupported session_id=%s falling_back=list", + self.session_id, + ) except (IdleTimeout, SessionTerminated): raise except Exception: @@ -172,19 +187,38 @@ def run(self) -> List[ToolCallResult]: return self._results finally: self._state.cancel_active_execution() + self.options.logger.info( + "session runner stopped session_id=%s work_id=%s mode=%s events=%d stream_attempts=%d", + self.session_id, + self.options.work_id, + mode, + self._state.event_count, + self._state.stream_attempts, + ) def _consume_stream_loop(self) -> None: backoff = STREAM_BACKOFF_START while not self._is_stopped(): + self._state.stream_attempts += 1 + attempt = self._state.stream_attempts + self.options.logger.info( + "session stream connecting session_id=%s work_id=%s attempt=%d timeout=%.3fs", + self.session_id, + self.options.work_id, + attempt, + self.options.stream_timeout_seconds, + ) event_queue: "queue.Queue[object]" = queue.Queue(maxsize=STREAM_QUEUE_SIZE) opened_at = time.monotonic() pump = threading.Thread(target=self._pump_stream, args=(event_queue,), daemon=True) pump.start() self._state.reconcile() + reconnect_error: Optional[BaseException] = None while not self._is_stopped() and (pump.is_alive() or not event_queue.empty()): self._state.drain_execution_done() self._state.flush_results() self._raise_if_idle_expired() + self._state.maybe_log_no_progress() try: item = event_queue.get(timeout=self._state.next_wait_seconds(0.5)) except queue.Empty: @@ -196,9 +230,30 @@ def _consume_stream_loop(self) -> None: raise item if is_fatal_4xx(item): raise item + reconnect_error = item break self._state.handle_stream_event(item) # type: ignore[arg-type] - self._sleep_or_idle(_jitter(backoff)) + if self._is_stopped(): + return + reconnect_in = _jitter(backoff) + if reconnect_error is None: + self.options.logger.info( + "session stream ended session_id=%s work_id=%s attempt=%d reconnect_in=%.3fs", + self.session_id, + self.options.work_id, + attempt, + reconnect_in, + ) + else: + self.options.logger.warning( + "session stream disconnected session_id=%s work_id=%s attempt=%d reconnect_in=%.3fs err=%s", + self.session_id, + self.options.work_id, + attempt, + reconnect_in, + reconnect_error, + ) + self._sleep_or_idle(reconnect_in) backoff = min(backoff * 2, STREAM_BACKOFF_CAP) def _pump_stream(self, event_queue: "queue.Queue[object]") -> None: @@ -233,6 +288,7 @@ def _consume_list(self) -> None: self._state.reconcile(reconcile=False) self._state.flush_results() self._raise_if_idle_expired() + self._state.maybe_log_no_progress() self._sleep_or_idle(self.options.event_poll_interval_seconds) def _sleep_or_idle(self, seconds: float) -> None: @@ -240,6 +296,7 @@ def _sleep_or_idle(self, seconds: float) -> None: while not self._is_stopped(): self._state.drain_execution_done() self._raise_if_idle_expired() + self._state.maybe_log_no_progress() remaining = deadline - time.monotonic() if remaining <= 0: return @@ -258,6 +315,13 @@ class _RunnerState: def __init__(self, runner: SessionToolRunner) -> None: self.runner = runner self.page = runner.options.event_page + self.event_count = 0 + self.business_event_count = 0 + self.stream_attempts = 0 + self.session_status = runner.options.initial_session_status.strip().lower() + self.last_event_at = time.monotonic() + self.last_business_event_at = time.monotonic() + self.last_progress_warning_at = 0.0 self.processed: Dict[str, bool] = {} self.seen: Dict[str, bool] = {} self.answered: Dict[str, bool] = {} @@ -266,7 +330,9 @@ def __init__(self, runner: SessionToolRunner) -> None: self.pending_ask: Dict[str, Event] = {} self.confirmations: Dict[str, Event] = {} self.external_tools: Dict[str, Event] = {} + self.unposted_tool_use_results: Set[str] = set() self.tool_use_events: Dict[str, Event] = {} + self.tool_use_call_id_by_event_id: Dict[str, str] = {} self.scheduled: Set[str] = set() self.execution_queue: Deque[_PendingToolEvent] = deque() self.active_execution: Optional[_ActiveToolExecution] = None @@ -274,6 +340,8 @@ def __init__(self, runner: SessionToolRunner) -> None: self.session_tool_uses: Set[str] = set() self.tool_uses_since_status: Set[str] = set() self.blocking_event_ids: Set[str] = set() + self.pending_blocking_event_ids: Set[str] = set() + self.requires_action_fallback: Set[str] = set() self.blocking_events_known = False self.idle_armed_at = 0.0 self.idle_arm_pending = False @@ -344,6 +412,7 @@ def process_listed_events(self, events: Iterable[Event], reconcile: bool = False pending_ids[call_id] = True elif event.type in (EVENT_TYPE_SESSION_STATUS_TERMINATED, EVENT_TYPE_SESSION_DELETED): raise SessionTerminated("session terminated") + self.reconcile_unmatched_requires_action() self.reconcile_recovered_results() if touched_idle: self.disarm_idle() @@ -394,11 +463,44 @@ def mark_event_seen(self, event: Event) -> bool: if self.seen.get(key): return False self.seen[key] = True + self.record_event(event) return True + def record_event(self, event: Event) -> None: + self.event_count += 1 + now = time.monotonic() + self.last_event_at = now + is_business_event = not event.type.startswith("span.") + if is_business_event: + self.business_event_count += 1 + self.last_business_event_at = now + self.last_progress_warning_at = 0.0 + if self.business_event_count == 1: + self.runner.options.logger.info( + "session first business event session_id=%s work_id=%s event_type=%s event_id=%s", + self.runner.session_id, + self.runner.options.work_id, + event.type, + event.id, + ) + else: + self.runner.options.logger.debug( + "session event received session_id=%s work_id=%s event_type=%s event_id=%s", + self.runner.session_id, + self.runner.options.work_id, + event.type, + event.id, + ) + def mark_answered(self, call_id: str) -> None: if not call_id: return + event = self.tool_use_events.get(call_id) + if event is not None and event.id: + self.pending_blocking_event_ids.discard(event.id) + self.tool_use_call_id_by_event_id.pop(event.id, None) + self.requires_action_fallback.discard(call_id) + self.pending_blocking_event_ids.discard(call_id) self.answered[call_id] = True self.processed[call_id] = True self.pending_results.pop(call_id, None) @@ -435,12 +537,15 @@ def handle_tool_use(self, event: Event, custom: bool) -> None: if not self.owns_tool(event, custom): self.external_tools[call_id] = event self.maybe_arm_pending_idle() - self.runner._results.append(ToolCallResult(call_id, event.name, custom, posted=False, event=event)) + self.record_unposted_tool_use_result_once(call_id, event, custom) return confirmation, allowed = self.permission_allows(event, custom, call_id) if not allowed: - self.runner._results.append( - ToolCallResult(call_id, event.name, custom, confirmation=confirmation, posted=False, event=event) + self.record_unposted_tool_use_result_once( + call_id, + event, + custom, + confirmation=confirmation, ) return if self.runner.options.result_store is not None: @@ -456,6 +561,23 @@ def handle_tool_use(self, event: Event, custom: bool) -> None: self.execution_queue.append(_PendingToolEvent(event, custom, confirmation)) self.start_next_tool_execution() + def record_unposted_tool_use_result_once( + self, call_id: str, event: Event, custom: bool, confirmation: str = "" + ) -> None: + if call_id in self.unposted_tool_use_results: + return + self.unposted_tool_use_results.add(call_id) + self.runner._results.append( + ToolCallResult( + tool_use_id=call_id, + name=event.name, + custom=custom, + confirmation=confirmation, + posted=False, + event=event, + ) + ) + def start_next_tool_execution(self) -> None: if self.active_execution is not None: return @@ -658,6 +780,7 @@ def send_result(self, call_id: str, event: Event, custom: bool, confirmation: st result=out, ) ) + self.maybe_arm_pending_idle() def retry_send_event(self, event: Event, call_id: str) -> bool: last_exc: Optional[BaseException] = None @@ -700,30 +823,149 @@ def observe_session_state(self, event: Event) -> None: self.session_tool_uses.add(call_id) if not self.is_answered(call_id): self.tool_use_events[call_id] = event + if event.id: + self.tool_use_call_id_by_event_id[event.id] = call_id self.tool_uses_since_status.add(call_id) return if event.type == EVENT_TYPE_SESSION_STATUS_IDLE: + self.set_session_status("idle", event) self.blocking_events_known = True self.blocking_event_ids = set() if event.stop_reason_type() == SESSION_STOP_REASON_REQUIRES_ACTION: - self.blocking_event_ids.update(event.stop_reason_event_ids()) + event_ids = set(event.stop_reason_event_ids()) + self.blocking_event_ids.update(event_ids) + self.pending_blocking_event_ids.update(event_ids) + for event_id in event_ids: + call_id = self.tool_use_call_id_by_event_id.get(event_id) + if call_id: + self.blocking_event_ids.add(call_id) + self.pending_blocking_event_ids.add(call_id) + else: + self.clear_unmatched_pending_blockers() self.tool_uses_since_status.clear() return if event.type in (EVENT_TYPE_SESSION_STATUS_RUNNING, EVENT_TYPE_SESSION_STATUS_RESCHEDULED): + self.set_session_status("running", event) self.blocking_events_known = True - self.blocking_event_ids.clear() - self.tool_uses_since_status.clear() + self.clear_resolved_pending_blockers() + elif event.type in (EVENT_TYPE_SESSION_STATUS_TERMINATED, EVENT_TYPE_SESSION_DELETED): + self.set_session_status("terminated", event) + self.clear_pending_blockers() + + def clear_resolved_pending_blockers(self) -> None: + for call_id, tool_event in self.tool_use_events.items(): + if call_id not in self.pending_blocking_event_ids and ( + not tool_event.id or tool_event.id not in self.pending_blocking_event_ids + ): + continue + self.pending_blocking_event_ids.discard(call_id) + if tool_event.id: + self.pending_blocking_event_ids.discard(tool_event.id) + self.tool_use_call_id_by_event_id.pop(tool_event.id, None) + self.blocking_event_ids.clear() + self.tool_uses_since_status.clear() + + def clear_unmatched_pending_blockers(self) -> None: + matched = set() + for call_id, tool_event in self.tool_use_events.items(): + if call_id in self.pending_blocking_event_ids or ( + tool_event.id and tool_event.id in self.pending_blocking_event_ids + ): + matched.add(call_id) + if tool_event.id: + matched.add(tool_event.id) + self.pending_blocking_event_ids.intersection_update(matched) + + def clear_pending_blockers(self) -> None: + self.blocking_event_ids.clear() + self.pending_blocking_event_ids.clear() + self.tool_use_call_id_by_event_id.clear() + self.tool_uses_since_status.clear() + + def set_session_status(self, status: str, event: Event) -> None: + if status == self.session_status: + return + previous = self.session_status or "unknown" + self.session_status = status + if status == "running": + self.last_business_event_at = time.monotonic() + self.last_progress_warning_at = 0.0 + self.runner.options.logger.info( + "session status changed session_id=%s work_id=%s previous=%s current=%s event_type=%s stop_reason=%s", + self.runner.session_id, + self.runner.options.work_id, + previous, + status, + event.type, + event.stop_reason_type() if status == "idle" else "", + ) + + def maybe_log_no_progress(self) -> None: + if self.session_status != "running": + return + now = time.monotonic() + no_progress_seconds = now - self.last_business_event_at + if no_progress_seconds < STREAM_PROGRESS_WARN_SECONDS: + return + if self.last_progress_warning_at and now - self.last_progress_warning_at < STREAM_PROGRESS_WARN_SECONDS: + return + self.last_progress_warning_at = now + self.runner.options.logger.warning( + "session has no business progress session_id=%s work_id=%s status=%s " + "no_progress_seconds=%.1f events=%d stream_attempts=%d", + self.runner.session_id, + self.runner.options.work_id, + self.session_status, + no_progress_seconds, + self.event_count, + self.stream_attempts, + ) def should_handle_tool_use(self, call_id: str) -> bool: if not self.blocking_events_known: return True - return call_id in self.blocking_event_ids or call_id in self.tool_uses_since_status + return self.is_blocking_tool_use(call_id) or call_id in self.tool_uses_since_status + + def is_blocking_tool_use(self, call_id: str) -> bool: + if call_id in self.requires_action_fallback: + return True + # StopReason.EventIDs 里是 agent.tool_use 的事件 id，而 runner 内部以 + # tool_use_id 作为 call id，两者不能直接互相查表。 + if call_id in self.blocking_event_ids or call_id in self.pending_blocking_event_ids: + return True + event = self.tool_use_events.get(call_id) + return event is not None and ( + event.id in self.blocking_event_ids or event.id in self.pending_blocking_event_ids + ) + + def reconcile_unmatched_requires_action(self) -> None: + if not self.blocking_events_known or not self.pending_blocking_event_ids: + return + for call_id, event in self.tool_use_events.items(): + if call_id in self.pending_blocking_event_ids or (event.id and event.id in self.pending_blocking_event_ids): + return + candidates = [] + for call_id, event in self.tool_use_events.items(): + if self.is_answered(call_id) or call_id in self.scheduled: + continue + if not self.owns_tool(event, event.type == EVENT_TYPE_AGENT_CUSTOM_TOOL_USE): + continue + candidates.append(call_id) + if len(candidates) != 1: + return + call_id = candidates[0] + if call_id not in self.requires_action_fallback: + self.runner.options.logger.warning( + "using sole unresolved tool use for unmatched requires_action tool_use_id=%s", + call_id, + ) + self.requires_action_fallback.add(call_id) def reconcile_recovered_results(self) -> None: if not self.blocking_events_known: return for call_id in list(self.recovered_results): - if call_id in self.blocking_event_ids and call_id in self.session_tool_uses: + if self.is_blocking_tool_use(call_id) and call_id in self.session_tool_uses: self.recovered_results.discard(call_id) continue if call_id in self.tool_uses_since_status: @@ -763,7 +1005,7 @@ def maybe_arm_pending_idle(self) -> None: self.idle_armed_at = time.monotonic() def has_idle_blockers(self) -> bool: - return bool(self.pending_ask or self.pending_results or self.external_tools or self.scheduled) + return bool(self.pending_ask or self.external_tools or self.scheduled) def idle_expired(self) -> bool: max_idle = self.max_idle_seconds() diff --git a/src/arkruntime/selfhosted/worker.py b/src/arkruntime/selfhosted/worker.py index 827938c..a67011a 100644 --- a/src/arkruntime/selfhosted/worker.py +++ b/src/arkruntime/selfhosted/worker.py @@ -212,6 +212,8 @@ class EnvironmentWorkerOptions: custom_tools: Dict[str, Tool] = field(default_factory=dict) logger: logging.Logger = logging.getLogger("arkruntime.selfhosted.environment_worker") tool_timeout_seconds: Optional[float] = None + prefer_stream: bool = True + stream_timeout_seconds: float = 30.0 class EnvironmentWorker: @@ -309,8 +311,11 @@ def _handle_item(self, work: _ClaimedWork) -> None: result_store=store, max_idle_seconds=self.options.max_idle_seconds, tool_timeout_seconds=self.options.tool_timeout_seconds, + prefer_stream=self.options.prefer_stream, + stream_timeout_seconds=self.options.stream_timeout_seconds, stop_event=work_stop, logger=self.options.logger, + initial_session_status=str(session.raw.get("status") or ""), ), ) runner.run() diff --git a/src/arkruntime/types/agent/__init__.py b/src/arkruntime/types/agent/__init__.py index 8a62c58..b394548 100644 --- a/src/arkruntime/types/agent/__init__.py +++ b/src/arkruntime/types/agent/__init__.py @@ -19,6 +19,10 @@ from .model_config import ModelConfig from .model_speed import ModelSpeed from .multiagent_config import MultiagentConfig +from .multiagent_feature_type import MultiagentFeatureType +from .multiagent_inline_agents_config import MultiagentInlineAgentsConfig +from .multiagent_patch import MultiagentPatch +from .multiagent_subagents_config import MultiagentSubagentsConfig from .multiagent_type import MultiagentType from .permission_policy import PermissionPolicy from .permission_policy_type import PermissionPolicyType @@ -44,6 +48,10 @@ "ModelConfig", "ModelSpeed", "MultiagentConfig", + "MultiagentFeatureType", + "MultiagentInlineAgentsConfig", + "MultiagentPatch", + "MultiagentSubagentsConfig", "MultiagentType", "PermissionPolicy", "PermissionPolicyType", diff --git a/src/arkruntime/types/agent/create_agent_request.py b/src/arkruntime/types/agent/create_agent_request.py index 8d71918..bfbf4c0 100644 --- a/src/arkruntime/types/agent/create_agent_request.py +++ b/src/arkruntime/types/agent/create_agent_request.py @@ -21,6 +21,8 @@ class CreateAgentRequest(BaseModel): """ 创建 Agent 的请求体。 + + multiagent_20261001 必须携带非 null 的 subagents；该条件由服务端校验。 """ name: str @@ -57,7 +59,7 @@ class CreateAgentRequest(BaseModel): """ multiagent: Optional[MultiagentConfig] = None """ - 多 Agent 编排拓扑。 + 多 Agent 编排拓扑。multiagent_20261001 时 subagents 必填且不能为 null。 """ metadata: Optional[Dict[str, str]] = None """ diff --git a/src/arkruntime/types/agent/multiagent_config.py b/src/arkruntime/types/agent/multiagent_config.py index 78a581c..9cecea4 100644 --- a/src/arkruntime/types/agent/multiagent_config.py +++ b/src/arkruntime/types/agent/multiagent_config.py @@ -6,24 +6,53 @@ from __future__ import annotations -from typing import List +from typing import List, Optional + +from pydantic import model_validator from arkruntime._models import BaseModel from .agent_ref import AgentRef +from .multiagent_subagents_config import MultiagentSubagentsConfig from .multiagent_type import MultiagentType class MultiagentConfig(BaseModel): """ - 多 Agent 协调拓扑。 + 多 Agent 协调拓扑。`type` 当前支持 `coordinator` 与 + `multiagent_20261001`。 + + - `coordinator` 使用 `agents`；`subagents` 会被服务端忽略。 + - `multiagent_20261001` 使用 `subagents`；非空 `agents` 返回 400。 + - 拓扑迁移只允许 `coordinator` 或单 Agent 升级到 + `multiagent_20261001`；反之返回 400。 + - `subagents` 表示拓扑配置值；Update 的 presence / null patch 语义由 + `MultiagentPatch` 单独表达。 """ type: MultiagentType """ 拓扑类型。 """ - agents: List[AgentRef] + agents: Optional[List[AgentRef]] = None + """ + coordinator 的下属 Agent 数组，0–20 项；深度限制为 1。 + `multiagent_20261001` 不支持该字段。 """ - 下属 Agent 数组，1–20 项；深度限制为 1（被引用的 Agent 自身不能再有 `multiagent`）。 + subagents: Optional[MultiagentSubagentsConfig] = None """ + `multiagent_20261001` 的子代理配置。 + + Create 时必须提供。该属性不接受显式 `null`。 + """ + + # ark-apis generated validation: dynamic multiagent subagents + @model_validator(mode="after") + def _validate_dynamic_subagents(self) -> "MultiagentConfig": + type_value = getattr(self.type, "value", self.type) + if type_value == "multiagent_20261001" and self.subagents is None: + raise ValueError( + 'multiagent.subagents is required when multiagent.type is ' + '"multiagent_20261001"' + ) + return self diff --git a/src/arkruntime/types/agent/multiagent_feature_type.py b/src/arkruntime/types/agent/multiagent_feature_type.py new file mode 100644 index 0000000..1e41bae --- /dev/null +++ b/src/arkruntime/types/agent/multiagent_feature_type.py @@ -0,0 +1,18 @@ +# Generated by datamodel-code-generator from ark-apis typespec/openapi. +# DO NOT EDIT — regenerate with `make gen-py-` in ark-apis. +# +# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from enum import Enum + + +class MultiagentFeatureType(str, Enum): + """ + 多 Agent 子代理能力开关。 + """ + + enabled = "enabled" + disabled = "disabled" diff --git a/src/arkruntime/types/agent/multiagent_inline_agents_config.py b/src/arkruntime/types/agent/multiagent_inline_agents_config.py new file mode 100644 index 0000000..0168161 --- /dev/null +++ b/src/arkruntime/types/agent/multiagent_inline_agents_config.py @@ -0,0 +1,22 @@ +# Generated by datamodel-code-generator from ark-apis typespec/openapi. +# DO NOT EDIT — regenerate with `make gen-py-` in ark-apis. +# +# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from arkruntime._models import BaseModel + +from .multiagent_feature_type import MultiagentFeatureType + + +class MultiagentInlineAgentsConfig(BaseModel): + """ + `multiagent_20261001` 的 inline 子代理能力开关。 + """ + + type: MultiagentFeatureType + """ + inline 子代理能力开关。 + """ diff --git a/src/arkruntime/types/agent/multiagent_patch.py b/src/arkruntime/types/agent/multiagent_patch.py new file mode 100644 index 0000000..2a38c0c --- /dev/null +++ b/src/arkruntime/types/agent/multiagent_patch.py @@ -0,0 +1,43 @@ +# Generated by datamodel-code-generator from ark-apis typespec/openapi. +# DO NOT EDIT — regenerate with `make gen-py-` in ark-apis. +# +# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import List, Optional + +from arkruntime._models import BaseModel + +from .agent_ref import AgentRef +from .multiagent_subagents_config import MultiagentSubagentsConfig +from .multiagent_type import MultiagentType + + +class MultiagentPatch(BaseModel): + """ + UpdateAgent 的多 Agent 编排 patch。 + + 顶层省略 = 不改，显式 `null` = 移除。同为 multiagent_20261001 时， + subagents 省略会保留当前配置；subagents 显式 `null` 会重置为默认配置。 + subagents 内层的 inline_agents / predefined_agents 按字段是否显式出现合并， + 省略保留，显式 `null` 按各自重置语义处理，predefined_agents 显式 `[]` + 清空名册。coordinator 的 agents 传值时整体替换名册，省略等价于空名册。 + """ + + type: MultiagentType + """ + 拓扑类型。 + """ + agents: Optional[List[AgentRef]] = None + """ + coordinator 的下属 Agent 数组，0–20 项；深度限制为 1。 + `multiagent_20261001` 不支持该字段。省略或显式 `[]` 都表示清空名册， + 非空数组整体替换名册。 + """ + subagents: Optional[MultiagentSubagentsConfig] = None + """ + `multiagent_20261001` 的子代理 patch。省略保留当前配置，显式 `null` + 重置为 `enabled` + inline `enabled` + 空名册。 + """ diff --git a/src/arkruntime/types/agent/multiagent_subagents_config.py b/src/arkruntime/types/agent/multiagent_subagents_config.py new file mode 100644 index 0000000..42321f3 --- /dev/null +++ b/src/arkruntime/types/agent/multiagent_subagents_config.py @@ -0,0 +1,50 @@ +# Generated by datamodel-code-generator from ark-apis typespec/openapi. +# DO NOT EDIT — regenerate with `make gen-py-` in ark-apis. +# +# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import List, Optional + +from arkruntime._models import BaseModel + +from .agent_ref import AgentRef +from .multiagent_feature_type import MultiagentFeatureType +from .multiagent_inline_agents_config import MultiagentInlineAgentsConfig + + +class MultiagentSubagentsConfig(BaseModel): + """ + `multiagent_20261001` 的子代理配置。 + + 在 UpdateAgent patch 中，`inline_agents` / `predefined_agents` 的显式 + `null` 与省略具有不同语义。Java typed model 将 `null` 序列化为省略； + 需要显式 `null` 的 Java 调用方必须使用 raw request body。 + """ + + type: MultiagentFeatureType + """ + 子代理能力开关。 + """ + inline_agents: Optional[MultiagentInlineAgentsConfig] = None + """ + inline 子代理能力开关。`subagents.type=enabled` 的 Create 中，省略或 + 显式 `null` 按 `enabled` 处理。UpdateAgent patch 中省略表示保留当前值， + 显式 `null` 重置为 `enabled`。 + 服务端响应在 `subagents.type` 为 `disabled` 时可能返回 `null`；客户端 + 必须接受该 wire，并将它视为未配置 inline 能力。 + """ + predefined_agents: Optional[List[AgentRef]] = None + """ + 预定义 Agent 引用数组，最多 20 项；显式传入时整体替换。 + + 服务端查询被引用 Agent 时会按 `id` 去重校验，但不会改写成员名册； + 重复引用、自引用和循环引用按请求中的原顺序保存、冻结和展开，并受 + Session 的最大展开深度限制。 + `inline_agents` 省略时按 `enabled` 处理；此时 Create 省略本字段等价于 + 空名册。UpdateAgent patch 中省略表示保留当前值，显式 `null` 等价于 + `[]`，显式数组整体替换。`inline_agents.type` 显式为 `disabled` 时必须 + 提供本字段（可为 `[]`）。服务端响应可能返回 `null`；客户端须按空名册处理。 + """ diff --git a/src/arkruntime/types/agent/multiagent_type.py b/src/arkruntime/types/agent/multiagent_type.py index a115213..68111ef 100644 --- a/src/arkruntime/types/agent/multiagent_type.py +++ b/src/arkruntime/types/agent/multiagent_type.py @@ -6,11 +6,13 @@ from __future__ import annotations -from typing import Literal +from enum import Enum -from typing_extensions import TypeAliasType -MultiagentType = TypeAliasType("MultiagentType", Literal["coordinator"]) -""" -多 Agent 拓扑类型。 -""" +class MultiagentType(str, Enum): + """ + 多 Agent 拓扑类型。 + """ + + coordinator = "coordinator" + multiagent_20261001 = "multiagent_20261001" diff --git a/src/arkruntime/types/agent/update_agent_request.py b/src/arkruntime/types/agent/update_agent_request.py index 08b7940..9b566d1 100644 --- a/src/arkruntime/types/agent/update_agent_request.py +++ b/src/arkruntime/types/agent/update_agent_request.py @@ -13,7 +13,7 @@ from .agent_skill_ref import AgentSkillRef from .mcp_server import MCPServer from .model_config import ModelConfig -from .multiagent_config import MultiagentConfig +from .multiagent_patch import MultiagentPatch from .tag import Tag from .tool_item import ToolItem @@ -65,9 +65,17 @@ class UpdateAgentRequest(BaseModel): """ Skill 引用列表（整体替换）。 """ - multiagent: Optional[MultiagentConfig] = None + multiagent: Optional[MultiagentPatch] = None """ - 多 Agent 编排拓扑。 + 多 Agent 编排拓扑（patch）：顶层省略 = 不改，显式 `null` = 移除。 + 同为 multiagent_20261001 时，subagents 省略会保留当前配置；subagents + 显式 `null` 会重置为默认配置。subagents 内层的 inline_agents / + predefined_agents 按字段是否显式出现合并，省略保留，显式 `null` 按各自 + 重置语义处理，predefined_agents 显式 `[]` 清空名册。coordinator 的 agents + 传值时整体替换名册，省略等价于空名册。 + + 生成器无法让所有语言在强类型模型中统一区分省略和显式 `null`；不能表达时 + 使用 raw HTTP/request-body escape hatch。 """ metadata: Optional[Dict[str, str]] = None """ diff --git a/src/arkruntime/types/session/_init_extras_shim.py b/src/arkruntime/types/session/_init_extras_shim.py index f7e02ab..67a4ea0 100644 --- a/src/arkruntime/types/session/_init_extras_shim.py +++ b/src/arkruntime/types/session/_init_extras_shim.py @@ -13,6 +13,7 @@ from .session_stream_shim import ( ListSessionEventsResponse, + ManagedAgentsAgentCustomToolUseEvent, ManagedAgentsAgentMCPToolResultEvent, ManagedAgentsAgentMCPToolUseEvent, ManagedAgentsAgentMessageEvent, @@ -51,6 +52,7 @@ __all__ = [ "ListSessionEventsResponse", + "ManagedAgentsAgentCustomToolUseEvent", "ManagedAgentsAgentMCPToolResultEvent", "ManagedAgentsAgentMCPToolUseEvent", "ManagedAgentsAgentMessageEvent", diff --git a/src/arkruntime/types/session/session_stream_shim.py b/src/arkruntime/types/session/session_stream_shim.py index 9dde9d8..85898ad 100644 --- a/src/arkruntime/types/session/session_stream_shim.py +++ b/src/arkruntime/types/session/session_stream_shim.py @@ -258,6 +258,17 @@ class ManagedAgentsAgentToolUseEvent(_SessionEventBase): input: Any = None +class ManagedAgentsAgentCustomToolUseEvent(_SessionEventBase): + """Agent invoked a custom tool owned by the caller. The runtime waits + for a matching ``user.custom_tool_result`` before continuing. + """ + + session_thread_id: str = "" + custom_tool_use_id: str = "" + name: str = "" + input: Any = None + + class ManagedAgentsAgentToolResultEvent(_SessionEventBase): """Tool's result (post ``user.tool_result`` if the tool required client-side execution, or runtime-owned if it was a builtin/MCP tool). @@ -405,6 +416,7 @@ class ManagedAgentsUnknownSessionEvent(_SessionEventBase): "agent.message": ManagedAgentsAgentMessageEvent, "agent.thinking": ManagedAgentsAgentThinkingEvent, "agent.tool_use": ManagedAgentsAgentToolUseEvent, + "agent.custom_tool_use": ManagedAgentsAgentCustomToolUseEvent, "agent.tool_result": ManagedAgentsAgentToolResultEvent, "agent.mcp_tool_use": ManagedAgentsAgentMCPToolUseEvent, "agent.mcp_tool_result": ManagedAgentsAgentMCPToolResultEvent, diff --git a/tests/selfhosted/test_session_tool_runner.py b/tests/selfhosted/test_session_tool_runner.py index e62ea04..4380486 100644 --- a/tests/selfhosted/test_session_tool_runner.py +++ b/tests/selfhosted/test_session_tool_runner.py @@ -8,7 +8,14 @@ import pytest -from arkruntime.selfhosted import Event, ListEventsResponse, SessionToolRunner, SessionToolRunnerOptions +import arkruntime.selfhosted.session_tool_runner as session_tool_runner_module +from arkruntime.selfhosted import ( + Event, + ListEventsResponse, + SessionTerminated, + SessionToolRunner, + SessionToolRunnerOptions, +) from arkruntime.selfhosted.tool_result_store import ToolCallStoreDecision from arkruntime.selfhosted.tools import FunctionTool, ToolContext, ToolSet, text_result @@ -158,6 +165,179 @@ def test_duplicate_stream_idle_event_does_not_reset_idle_deadline(tmp_path) -> N assert runner._state.idle_armed_at == armed_at +def test_stream_runner_logs_lifecycle_and_timeout(tmp_path, caplog) -> None: + class StreamAPI: + timeout = None + + def list_events(self, _session_id, **_kwargs): + return ListEventsResponse() + + def stream_events(self, _session_id, *, timeout): + self.timeout = timeout + yield Event(id="terminated", type="session.status_terminated") + + api = StreamAPI() + runner = SessionToolRunner( + api, + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + stream_timeout_seconds=12.5, + initial_session_status="running", + ), + ) + + with caplog.at_level("INFO"), pytest.raises(SessionTerminated): + runner.run() + + assert api.timeout == 12.5 + assert "session runner started" in caplog.text + assert "session stream connecting" in caplog.text + assert "attempt=1 timeout=12.500s" in caplog.text + assert "session first business event" in caplog.text + assert "session status changed" in caplog.text + assert "session runner stopped" in caplog.text + + +def test_stream_runner_logs_reconnect_after_transport_error(tmp_path, caplog, monkeypatch) -> None: + class StreamAPI: + attempts = 0 + + def list_events(self, _session_id, **_kwargs): + return ListEventsResponse() + + def stream_events(self, _session_id, *, timeout): + self.attempts += 1 + if self.attempts == 1: + raise RuntimeError("temporary stream failure") + yield Event(id="terminated", type="session.status_terminated") + + monkeypatch.setattr(session_tool_runner_module, "_jitter", lambda _seconds: 0.0) + api = StreamAPI() + runner = SessionToolRunner( + api, + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + ), + ) + + with caplog.at_level("INFO"), pytest.raises(SessionTerminated): + runner.run() + + assert api.attempts == 2 + assert "session stream disconnected" in caplog.text + assert "temporary stream failure" in caplog.text + assert "attempt=2" in caplog.text + + +def test_running_session_logs_no_progress_warning(tmp_path, caplog, monkeypatch) -> None: + monkeypatch.setattr(session_tool_runner_module, "STREAM_PROGRESS_WARN_SECONDS", 0.0) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + initial_session_status="running", + ), + ) + + with caplog.at_level("WARNING"): + runner._state.maybe_log_no_progress() + + assert "session has no business progress" in caplog.text + assert "events=0" in caplog.text + + +def test_entering_running_resets_no_progress_baseline(tmp_path, caplog, monkeypatch) -> None: + monkeypatch.setattr(session_tool_runner_module, "STREAM_PROGRESS_WARN_SECONDS", 60.0) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + initial_session_status="idle", + ), + ) + runner._state.last_business_event_at -= 61 + + with caplog.at_level("WARNING"): + runner._state.handle_stream_event(Event(type="session.status_running")) + runner._state.maybe_log_no_progress() + + assert "session has no business progress" not in caplog.text + + +def test_span_progress_frame_does_not_hide_business_no_progress_warning(tmp_path, caplog, monkeypatch) -> None: + monkeypatch.setattr(session_tool_runner_module, "STREAM_PROGRESS_WARN_SECONDS", 60.0) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + initial_session_status="running", + ), + ) + runner._state.last_business_event_at -= 61 + runner._state.last_progress_warning_at = 0.0 + + with caplog.at_level("WARNING"): + runner._state.handle_stream_event(Event(id="span-1", type="span.outcome_evaluation_ongoing")) + runner._state.maybe_log_no_progress() + + assert "session has no business progress" in caplog.text + + +def test_keyless_event_does_not_refresh_progress_or_event_count(tmp_path, caplog, monkeypatch) -> None: + monkeypatch.setattr(session_tool_runner_module, "STREAM_PROGRESS_WARN_SECONDS", 0.0) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + initial_session_status="running", + ), + ) + + with caplog.at_level("WARNING"): + runner._state.handle_stream_event(Event(type="agent.message")) + runner._state.maybe_log_no_progress() + + assert runner._state.event_count == 0 + assert "session has no business progress" in caplog.text + + +def test_idle_session_does_not_log_no_progress_warning(tmp_path, caplog, monkeypatch) -> None: + monkeypatch.setattr(session_tool_runner_module, "STREAM_PROGRESS_WARN_SECONDS", 0.0) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + work_id="work-1", + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + initial_session_status="idle", + ), + ) + + with caplog.at_level("WARNING"): + runner._state.maybe_log_no_progress() + + assert "session has no business progress" not in caplog.text + + def test_reconcile_does_not_reset_idle_deadline_for_seen_history(tmp_path) -> None: runner = SessionToolRunner( object(), @@ -642,6 +822,448 @@ def test_current_blocker_reuses_recovered_result_without_reexecution(tmp_path) - assert not store.discarded +def test_earlier_requires_action_survives_later_idle(tmp_path) -> None: + executions = [] + tool = FunctionTool( + "custom", + lambda value, _context: executions.append(value) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + max_idle_seconds=60, + ), + ) + runner._state.retry_send_event = lambda _event, _call_id: True + + runner._state.process_listed_events( + [ + Event( + id="use-1", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-1", + input={"value": "one"}, + ), + Event( + id="idle-1", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["use-1"]}, + ), + Event(id="user-2", type="user.message", content=[]), + Event( + id="use-2", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-2", + input={"value": "two"}, + ), + Event( + id="idle-2", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["use-2"]}, + ), + ], + reconcile=False, + ) + + _finish_execution(runner) + _finish_execution(runner) + + assert executions == [{"value": "one"}, {"value": "two"}] + + +def test_known_remote_blocker_does_not_trigger_sole_tool_fallback(tmp_path) -> None: + executions = [] + tool = FunctionTool( + "local", + lambda _input, _context: executions.append(True) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"local": tool}, + max_idle_seconds=60, + ), + ) + runner._state.retry_send_event = lambda _event, _call_id: True + + runner._state.process_listed_events( + [ + Event( + id="local-use", + type="agent.custom_tool_use", + name="local", + custom_tool_use_id="local-call", + input={}, + ), + Event(id="idle-end", type="session.status_idle", stop_reason={"type": "end_turn"}), + Event(id="running", type="session.status_running"), + Event( + id="remote-use", + type="agent.custom_tool_use", + name="remote", + custom_tool_use_id="remote-call", + session_thread_id="child", + input={}, + ), + Event( + id="idle-requires-action", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["remote-use"]}, + ), + ], + reconcile=True, + ) + + assert not executions + assert [result.tool_use_id for result in runner.results] == ["remote-call"] + assert runner.results[0].posted is False + + +def test_running_status_clears_abandoned_pending_blockers(tmp_path) -> None: + executions = [] + tool = FunctionTool( + "custom", + lambda _input, _context: executions.append(True) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + max_idle_seconds=60, + ), + ) + runner._state.retry_send_event = lambda _event, _call_id: True + + runner._state.process_listed_events( + [ + Event( + id="tool-use", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-1", + input={}, + ), + Event( + id="idle-requires-action", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["tool-use"]}, + ), + Event(id="user-message", type="user.message", content=[]), + Event(id="running", type="session.status_running"), + Event(id="idle-end-turn", type="session.status_idle", stop_reason={"type": "end_turn"}), + ], + reconcile=True, + ) + + assert not executions + assert not runner._state.pending_blocking_event_ids + assert not runner._state.tool_use_call_id_by_event_id + + +def test_later_idle_drops_unmatched_pending_blockers(tmp_path) -> None: + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions(tools=ToolSet(), tool_context=ToolContext(workdir=str(tmp_path))), + ) + + runner._state.process_listed_events( + [ + Event( + id="idle-requires-action", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["foreign-use"]}, + ), + Event(id="running", type="session.status_running"), + Event(id="idle-end-turn", type="session.status_idle", stop_reason={"type": "end_turn"}), + ], + reconcile=True, + ) + + assert not runner._state.pending_blocking_event_ids + + +def test_reconcile_records_external_tool_use_only_once(tmp_path) -> None: + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions(tools=ToolSet(), tool_context=ToolContext(workdir=str(tmp_path))), + ) + events = [ + Event( + id="tool-use", + type="agent.custom_tool_use", + name="remote", + custom_tool_use_id="call-1", + input={}, + ), + Event( + id="idle", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["tool-use"]}, + ), + ] + + for _ in range(4): + runner._state.process_listed_events(events, reconcile=True) + + assert [result.tool_use_id for result in runner.results] == ["call-1"] + + +def test_reconcile_records_pending_confirmation_tool_use_only_once(tmp_path) -> None: + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet([FunctionTool("approval", lambda _input, _context: text_result("ok"))]), + tool_context=ToolContext(workdir=str(tmp_path)), + ), + ) + events = [ + Event( + id="tool-use", + type="agent.tool_use", + name="approval", + tool_use_id="call-1", + evaluated_permission="ask", + input={}, + ), + Event( + id="idle", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["tool-use"]}, + ), + ] + + for _ in range(4): + runner._state.process_listed_events(events, reconcile=True) + + assert [result.tool_use_id for result in runner.results] == ["call-1"] + assert runner.results[0].posted is False + + +def test_failed_result_delivery_still_arms_idle_deadline(tmp_path) -> None: + executions = [] + store = _RecordingStore() + tool = FunctionTool( + "custom", + lambda _input, _context: executions.append(True) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + result_store=store, + max_idle_seconds=0.01, + ), + ) + runner._state.retry_send_event = lambda _event, _call_id: False + + runner._state.process_listed_events( + [ + Event( + id="tool-use", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-1", + input={}, + ), + Event( + id="idle-requires-action", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["tool-use"]}, + ), + Event(id="idle-end-turn", type="session.status_idle", stop_reason={"type": "end_turn"}), + ], + reconcile=True, + ) + _finish_execution(runner) + + assert executions == [True] + assert "call-1" in runner._state.pending_results + assert runner._state.idle_armed_at + time.sleep(0.02) + assert runner._state.idle_expired() + + +def test_empty_ledger_replays_unresolved_tool_use(tmp_path) -> None: + executions = [] + store = _RecordingStore() + tool = FunctionTool( + "custom", + lambda _input, _context: executions.append(True) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + result_store=store, + ), + ) + sent = [] + runner._state.retry_send_event = lambda event, _call_id: sent.append(event) or True + + runner._state.process_listed_events( + [ + Event(id="running", type="session.status_running"), + Event( + id="sevt-use-1", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-1", + input={}, + ), + Event( + id="idle", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["sevt-use-1"]}, + ), + ], + reconcile=True, + ) + _finish_execution(runner) + + assert executions == [True] + assert [event.custom_tool_use_id for event in sent] == ["call-1"] + assert store.marked == ["call-1"] + + +def test_unmatched_requires_action_replays_sole_owned_tool(tmp_path) -> None: + executions = [] + tool = FunctionTool( + "custom", + lambda _input, _context: executions.append(True) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + ), + ) + sent = [] + runner._state.retry_send_event = lambda event, _call_id: sent.append(event) or True + + runner._state.process_listed_events( + [ + Event( + id="idle", type="session.status_idle", stop_reason={"type": "requires_action", "event_ids": ["opaque"]} + ), + Event( + id="sevt-use-1", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-1", + input={}, + ), + Event(id="running", type="session.status_running"), + ], + reconcile=True, + ) + _finish_execution(runner) + + assert executions == [True] + assert [event.custom_tool_use_id for event in sent] == ["call-1"] + assert not runner._state.requires_action_fallback + + +def test_unmatched_requires_action_does_not_guess_among_multiple_tools(tmp_path) -> None: + executions = [] + tool = FunctionTool( + "custom", + lambda _input, _context: executions.append(True) or text_result("ok"), + ) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + ), + ) + runner._state.retry_send_event = lambda _event, _call_id: True + + runner._state.process_listed_events( + [ + Event(id="running", type="session.status_running"), + Event(id="sevt-use-1", type="agent.custom_tool_use", name="custom", custom_tool_use_id="call-1", input={}), + Event(id="sevt-use-2", type="agent.custom_tool_use", name="custom", custom_tool_use_id="call-2", input={}), + Event( + id="idle", type="session.status_idle", stop_reason={"type": "requires_action", "event_ids": ["opaque"]} + ), + ], + reconcile=True, + ) + + assert not executions + assert not runner._state.requires_action_fallback + + +def test_current_blocker_matches_event_id_to_tool_use_id(tmp_path) -> None: + executions = [] + store = _RecordingStore() + tool = FunctionTool("custom", lambda _input, _context: executions.append(True) or text_result("ok")) + runner = SessionToolRunner( + object(), + "session-1", + SessionToolRunnerOptions( + tools=ToolSet(), + tool_context=ToolContext(workdir=str(tmp_path)), + custom_tools={"custom": tool}, + result_store=store, + ), + ) + sent = [] + result = Event(type="user.custom_tool_result", custom_tool_use_id="call-1") + runner._state.pending_results["call-1"] = result + runner._state.recovered_results.add("call-1") + runner._state.retry_send_event = lambda event, _call_id: sent.append(event) or True + + runner._state.process_listed_events( + [ + Event( + id="sevt-use-1", + type="agent.custom_tool_use", + name="custom", + custom_tool_use_id="call-1", + input={}, + ), + Event( + id="idle", + type="session.status_idle", + stop_reason={"type": "requires_action", "event_ids": ["sevt-use-1"]}, + ), + ], + reconcile=True, + ) + + assert not executions + assert sent == [result] + assert store.marked == ["call-1"] + assert not store.discarded + + def test_recovered_result_waits_for_authoritative_status(tmp_path) -> None: executions = [] tool = FunctionTool("custom", lambda _input, _context: executions.append(True) or text_result("ok")) diff --git a/tests/selfhosted/test_worker.py b/tests/selfhosted/test_worker.py index 6fcd3ef..590c85a 100644 --- a/tests/selfhosted/test_worker.py +++ b/tests/selfhosted/test_worker.py @@ -254,6 +254,39 @@ def test_worker_tool_timeout_overrides_tool_context(tmp_path) -> None: assert context.tool_timeout_seconds == 0.02 +def test_worker_passes_stream_configuration_and_initial_status(tmp_path, monkeypatch) -> None: + captured = {} + + class Runner: + def __init__(self, _api, _session_id, options): + captured["options"] = options + + def run(self): + return None + + class API(_ActiveHeartbeatAPI): + def get_session(self, session_id): + return Session(id=session_id, raw={"status": "running"}) + + monkeypatch.setattr("arkruntime.selfhosted.worker.SessionToolRunner", Runner) + worker = EnvironmentWorker( + API(), + EnvironmentWorkerOptions( + environment_id="env-1", + workdir=str(tmp_path), + prefer_stream=False, + stream_timeout_seconds=12.5, + ), + ) + + worker.handle_item(HandleItemOptions(work_id="work-1", environment_id="env-1", session_id="session-1")) + + options = captured["options"] + assert options.prefer_stream is False + assert options.stream_timeout_seconds == 12.5 + assert options.initial_session_status == "running" + + @pytest.mark.parametrize("timeout", [None, 0, -1]) def test_worker_nonpositive_tool_timeout_preserves_tool_context(tmp_path, timeout) -> None: worker = EnvironmentWorker( @@ -279,6 +312,8 @@ def test_worker_options_preserve_legacy_positional_order() -> None: assert options.custom_tools is custom_tools assert options.logger is logger assert options.tool_timeout_seconds is None + assert options.prefer_stream is True + assert options.stream_timeout_seconds == 30.0 def test_worker_uses_configured_workdir(tmp_path) -> None: diff --git a/tests/test_managed_agents_serialize.py b/tests/test_managed_agents_serialize.py new file mode 100644 index 0000000..ad4726f --- /dev/null +++ b/tests/test_managed_agents_serialize.py @@ -0,0 +1,93 @@ +import json + +import httpx +import pytest +from pydantic import ValidationError + +from arkruntime import Ark +from arkruntime._managed_agents_serialize import dump_body +from arkruntime.types.agent import ( + ModelConfig, + MultiagentPatch, + MultiagentSubagentsConfig, +) + + +def test_multiagent_update_preserves_explicit_null_and_empty_array() -> None: + body = dump_body( + { + "version": 1, + "multiagent": MultiagentPatch( + type="multiagent_20261001", + subagents=MultiagentSubagentsConfig( + type="enabled", + inline_agents=None, + predefined_agents=[], + ), + ), + } + ) + + assert json.loads(json.dumps(body)) == { + "version": 1, + "multiagent": { + "type": "multiagent_20261001", + "subagents": { + "type": "enabled", + "inline_agents": None, + "predefined_agents": [], + }, + }, + } + + +def test_multiagent_update_explicit_null_is_sent() -> None: + assert dump_body({"version": 1, "multiagent": None}) == { + "version": 1, + "multiagent": None, + } + + +def test_agent_update_wrapper_sends_explicit_multiagent_null() -> None: + bodies = [] + + def handle(request: httpx.Request) -> httpx.Response: + bodies.append(json.loads(request.content)) + return httpx.Response( + 200, + json={ + "id": "agent-1", + "type": "agent", + "name": "agent", + "version": 2, + "created_at": "2026-10-28T00:00:00Z", + "updated_at": "2026-10-28T00:00:00Z", + }, + ) + + with Ark( + api_key="placeholder", + http_client=httpx.Client(transport=httpx.MockTransport(handle)), + ) as client: + client.agents.update("agent-1", version=1, multiagent=None) + + assert bodies == [{"version": 1, "multiagent": None}] + + +@pytest.mark.parametrize( + "multiagent", + [ + {"type": "multiagent_20261001"}, + {"type": "multiagent_20261001", "subagents": None}, + ], +) +def test_agent_create_rejects_dynamic_multiagent_without_subagents( + multiagent: dict, +) -> None: + with Ark(api_key="placeholder") as client: + with pytest.raises(ValidationError, match="multiagent.subagents"): + client.agents.create( + name="agent", + model=ModelConfig(id="model-id"), + multiagent=multiagent, + ) diff --git a/tests/test_multiagent_gate_e2e.py b/tests/test_multiagent_gate_e2e.py new file mode 100644 index 0000000..512eb93 --- /dev/null +++ b/tests/test_multiagent_gate_e2e.py @@ -0,0 +1,78 @@ +import os +import time + +import pytest + +from arkruntime import Ark +from arkruntime.types.agent import ( + AgentRef, + ModelConfig, + MultiagentConfig, + MultiagentPatch, + MultiagentSubagentsConfig, +) + + +@pytest.mark.skipif( + not os.environ.get("ARK_STG_TOKEN") or not os.environ.get("ARK_E2E_BASE_URL"), + reason="ARK_STG_TOKEN and ARK_E2E_BASE_URL are required for Managed Agents E2E", +) +def test_multiagent_client_gate_against_stg() -> None: + client = Ark( + api_key=os.environ["ARK_STG_TOKEN"], + base_url=os.environ["ARK_E2E_BASE_URL"], + max_retries=0, + ) + name = f"sdk-gate-py-{int(time.time() * 1000)}" + created = None + try: + created = client.agents.create( + name=name, + model=ModelConfig(id=os.environ.get("ARK_E2E_MODEL_ID", "doubao-seed-1-6-250615")), + multiagent=MultiagentConfig( + type="multiagent_20261001", + subagents=MultiagentSubagentsConfig( + type="enabled", + predefined_agents=[AgentRef(type="self")], + ), + ), + ) + assert created.multiagent is not None + assert created.multiagent.subagents is not None + + fetched = client.agents.retrieve(created.id) + assert fetched.id == created.id + + updated = client.agents.update( + created.id, + version=created.version, + multiagent=MultiagentPatch( + type="multiagent_20261001", + subagents=MultiagentSubagentsConfig( + type="enabled", + predefined_agents=[], + ), + ), + ) + assert updated.version > created.version + + listed = client.agents.list(limit=100) + assert any(item.id == created.id for item in listed.data) + + legacy_headers = {"User-Agent": "ark-runtime-python/0.11.0"} + with pytest.raises(Exception, match="MULTIAGENT_UNSUPPORTED_BY_CLIENT"): + client.agents.retrieve(created.id, extra_headers=legacy_headers) + with pytest.raises(Exception, match="MULTIAGENT_UNSUPPORTED_BY_CLIENT"): + client.agents.update( + created.id, + version=updated.version, + description="must be rejected", + extra_headers=legacy_headers, + ) + + legacy_list = client.agents.list(limit=100, extra_headers=legacy_headers) + legacy_item = next(item for item in legacy_list.data if item.id == created.id) + assert legacy_item.multiagent is None + finally: + if created is not None: + client.agents.delete(created.id) diff --git a/tests/test_session_stream_shim.py b/tests/test_session_stream_shim.py new file mode 100644 index 0000000..04553a3 --- /dev/null +++ b/tests/test_session_stream_shim.py @@ -0,0 +1,79 @@ +import json +from importlib.metadata import version + +import httpx +import pytest + +from arkruntime import Ark, AsyncArk +from arkruntime._constants import CLIENT_REQUEST_HEADER +from arkruntime.resources.sessions.session_events import _decode_stream_frame +from arkruntime.types.session import ManagedAgentsAgentCustomToolUseEvent + + +def _assert_stream_headers(request: httpx.Request) -> None: + assert request.headers["Accept"] == "text/event-stream" + assert request.headers["Authorization"] == "Bearer placeholder" + assert request.headers["X-Ark-Env"] == "lane-test" + assert request.headers["X-Maas-Env"] == "multiagent" + assert request.headers["User-Agent"] == "ark-runtime-python/" + version("arkruntime") + assert request.headers[CLIENT_REQUEST_HEADER] + + +def test_session_stream_propagates_sdk_default_headers() -> None: + requests = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, text="") + + with Ark( + api_key="placeholder", + http_client=httpx.Client(transport=httpx.MockTransport(handle)), + ) as client: + client._custom_headers.update({"X-Ark-Env": "lane-test", "X-Maas-Env": "multiagent"}) + assert list(client.sessions.events.stream("session-1")) == [] + + assert len(requests) == 1 + _assert_stream_headers(requests[0]) + + +@pytest.mark.asyncio +async def test_async_session_stream_propagates_sdk_default_headers() -> None: + requests = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, text="") + + async with AsyncArk( + api_key="placeholder", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handle)), + ) as client: + client._custom_headers.update({"X-Ark-Env": "lane-test", "X-Maas-Env": "multiagent"}) + assert [frame async for frame in client.sessions.events.stream("session-1")] == [] + + assert len(requests) == 1 + _assert_stream_headers(requests[0]) + + +def test_session_stream_decodes_child_custom_tool_use_event() -> None: + payload = { + "type": "agent.custom_tool_use", + "id": "event-1", + "processed_at": "2026-10-11T00:00:00Z", + "session_thread_id": "thread-child", + "custom_tool_use_id": "call-1", + "name": "submit", + "input": {"text": "hello"}, + } + + frame = _decode_stream_frame("", json.dumps(payload)) + event = frame.data + + assert frame.type == "agent.custom_tool_use" + assert isinstance(event, ManagedAgentsAgentCustomToolUseEvent) + assert event.id == "event-1" + assert event.custom_tool_use_id == "call-1" + assert event.session_thread_id == "thread-child" + assert event.name == "submit" + assert event.input == {"text": "hello"} diff --git a/tests/test_session_thread_stream_headers.py b/tests/test_session_thread_stream_headers.py new file mode 100644 index 0000000..76bc257 --- /dev/null +++ b/tests/test_session_thread_stream_headers.py @@ -0,0 +1,100 @@ +from uuid import uuid4 + +import httpx +import pytest + +from arkruntime.resources.sessions.session_threads import ( + AsyncSessionThreads, + SessionThreads, +) + + +class _SyncClient: + def __init__(self, transport) -> None: + self._client = httpx.Client(transport=transport) + self._base_url = "https://api.example.com" + + def default_headers(self) -> dict: + return { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "ark-runtime-test", + "X-Client-Request-Id": "request-id", + "X-Ark-Env": "lane-test", + "X-Maas-Env": "multiagent", + } + + def post(self, *args, **kwargs): + raise AssertionError("stream tests do not use JSON requests") + + get = post + delete = post + post_without_retry = post + get_api_list = post + + +class _AsyncClient: + def __init__(self, transport) -> None: + self._client = httpx.AsyncClient(transport=transport) + self._base_url = "https://api.example.com" + + def default_headers(self) -> dict: + return { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "ark-runtime-test", + "X-Client-Request-Id": "request-id", + "X-Ark-Env": "lane-test", + "X-Maas-Env": "multiagent", + } + + def post(self, *args, **kwargs): + raise AssertionError("stream tests do not use JSON requests") + + get = post + delete = post + post_without_retry = post + get_api_list = post + + +def _assert_stream_headers(request: httpx.Request) -> None: + assert request.headers["Accept"] == "text/event-stream" + assert request.headers["User-Agent"] == "ark-runtime-test" + assert request.headers["X-Client-Request-Id"] == "request-id" + assert request.headers["X-Ark-Env"] == "lane-test" + assert request.headers["X-Maas-Env"] == "multiagent" + + +def test_session_thread_stream_propagates_default_headers() -> None: + session_id = str(uuid4()) + thread_id = str(uuid4()) + requests = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, text="") + + client = _SyncClient(httpx.MockTransport(handle)) + assert list(SessionThreads(client).stream(session_id, thread_id)) == [] + client._client.close() + + assert len(requests) == 1 + _assert_stream_headers(requests[0]) + + +@pytest.mark.asyncio +async def test_async_session_thread_stream_propagates_default_headers() -> None: + session_id = str(uuid4()) + thread_id = str(uuid4()) + requests = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, text="") + + client = _AsyncClient(httpx.MockTransport(handle)) + assert [frame async for frame in AsyncSessionThreads(client).stream(session_id, thread_id)] == [] + await client._client.aclose() + + assert len(requests) == 1 + _assert_stream_headers(requests[0])