feat: make agent runner config schema driven

src/langbot/pkg/agent/runner/orchestrator.py

@@ -3,9 +3,12 @@ from __future__ import annotations
 
 import typing
 import traceback
+import asyncio
+import time
 
 from langbot_plugin.api.entities.builtin.provider import message as provider_message
 from langbot_plugin.api.entities.builtin.pipeline import query as pipeline_query
+from langbot_plugin.entities.io.errors import ActionCallTimeoutError
 
 from ...core import app
 from .descriptor import AgentRunnerDescriptor
@@ -155,14 +158,32 @@ class AgentRunOrchestrator:
             )
 
         try:
-            async for result_dict in self.ap.plugin_connector.run_agent(
+            gen = self.ap.plugin_connector.run_agent(
                 plugin_author=descriptor.plugin_author,
                 plugin_name=descriptor.plugin_name,
                 runner_name=descriptor.runner_name,
                 context=context,
-            ):
+            )
+
+            while True:
+                try:
+                    result_dict = await self._next_with_deadline(gen, descriptor, context)
+                except StopAsyncIteration:
+                    break
                 yield result_dict
 
+        except asyncio.TimeoutError as e:
+            raise RunnerExecutionError(
+                descriptor.id,
+                'Runner timed out (code: runner.timeout)',
+                retryable=True,
+            ) from e
+        except ActionCallTimeoutError as e:
+            raise RunnerExecutionError(
+                descriptor.id,
+                f'{e} (code: runner.timeout)',
+                retryable=True,
+            ) from e
         except RunnerExecutionError:
             raise
         except Exception as e:
@@ -176,6 +197,57 @@ class AgentRunOrchestrator:
                 retryable=False,
             )
 
+    async def _next_with_deadline(
+        self,
+        gen: typing.AsyncGenerator[dict[str, typing.Any], None],
+        descriptor: AgentRunnerDescriptor,
+        context: AgentRunContextPayload,
+    ) -> dict[str, typing.Any]:
+        """Read the next runner result while enforcing the run deadline."""
+        remaining = self._remaining_deadline_seconds(context)
+        if remaining is not None and remaining <= 0:
+            await self._close_generator(gen, descriptor)
+            raise asyncio.TimeoutError
+
+        try:
+            if remaining is None:
+                return await anext(gen)
+            return await asyncio.wait_for(anext(gen), timeout=remaining)
+        except StopAsyncIteration:
+            if self._is_deadline_exhausted(context):
+                raise asyncio.TimeoutError
+            raise
+        except asyncio.TimeoutError:
+            await self._close_generator(gen, descriptor)
+            raise
+
+    def _remaining_deadline_seconds(
+        self,
+        context: AgentRunContextPayload,
+    ) -> float | None:
+        runtime = context.get('runtime') or {}
+        deadline_at = runtime.get('deadline_at')
+        if deadline_at is None:
+            return None
+        try:
+            return float(deadline_at) - time.time()
+        except (TypeError, ValueError):
+            return None
+
+    def _is_deadline_exhausted(self, context: AgentRunContextPayload) -> bool:
+        remaining = self._remaining_deadline_seconds(context)
+        return remaining is not None and remaining <= 0
+
+    async def _close_generator(
+        self,
+        gen: typing.AsyncGenerator[dict[str, typing.Any], None],
+        descriptor: AgentRunnerDescriptor,
+    ) -> None:
+        try:
+            await gen.aclose()
+        except Exception as e:
+            self.ap.logger.warning(f'Failed to close timed-out runner {descriptor.id}: {e}')
+
     def resolve_runner_id_for_telemetry(self, query: pipeline_query.Query) -> str | None:
         """Resolve runner ID for telemetry/logging without full execution.