Adopt felix 0.3.0 native async providers

scripts/run_benchmark.py

@@ -175,6 +175,7 @@ def _make_mock_provider() -> MagicMock:
         content=_MOCK_ACTION_PLAN, model="mock-model",
         usage={"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150},
     )
+    provider.acomplete.return_value = provider.complete.return_value
     return provider
 
 

src/rle/agents/base_role.py

@@ -2,14 +2,17 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
+import time
 from abc import abstractmethod
 from typing import Any, ClassVar, Tuple
 
 from felix_agent_sdk import LLMAgent, LLMResult, LLMTask
 from felix_agent_sdk.communication import Spoke
 from felix_agent_sdk.core import HelixGeometry
+from felix_agent_sdk.events.types import EventType
 from felix_agent_sdk.providers.base import BaseProvider
 from felix_agent_sdk.providers.types import ChatMessage, CompletionResult, MessageRole
 from felix_agent_sdk.tokens.budget import TokenBudget
@@ -227,14 +230,8 @@ def _get_spoke_context(self) -> list[dict[str, Any]]:
             })
         return context
 
-    def _call_provider(
-        self,
-        system_prompt: str,
-        user_prompt: str,
-        temperature: float,
-        max_tokens: int,
-    ) -> CompletionResult:
-        """Override to pass extra provider kwargs and no-think prefix."""
+    def _build_messages(self, system_prompt: str, user_prompt: str) -> list[ChatMessage]:
+        """Build the message list, including the no-think prefill when enabled."""
         messages = [
             ChatMessage(role=MessageRole.SYSTEM, content=system_prompt),
             ChatMessage(role=MessageRole.USER, content=user_prompt),
@@ -246,6 +243,22 @@ def _call_provider(
             messages.append(
                 ChatMessage(role=MessageRole.ASSISTANT, content="</think>"),
             )
+        return messages
+
+    def _record_completion(self, result: CompletionResult) -> CompletionResult:
+        self._last_raw_output = result.content
+        self._last_usage = result.usage if hasattr(result, "usage") else None
+        return result
+
+    def _call_provider(
+        self,
+        system_prompt: str,
+        user_prompt: str,
+        temperature: float,
+        max_tokens: int,
+    ) -> CompletionResult:
+        """Override to pass extra provider kwargs and no-think prefix."""
+        messages = self._build_messages(system_prompt, user_prompt)
 
         def _do_complete() -> CompletionResult:
             return self.provider.complete(
@@ -261,9 +274,40 @@ def _do_complete() -> CompletionResult:
         else:
             result = _do_complete()
 
-        self._last_raw_output = result.content
-        self._last_usage = result.usage if hasattr(result, "usage") else None
-        return result
+        return self._record_completion(result)
+
+    async def _acall_provider(
+        self,
+        system_prompt: str,
+        user_prompt: str,
+        temperature: float,
+        max_tokens: int,
+    ) -> CompletionResult:
+        """Async mirror of ``_call_provider`` using the provider's native
+        ``acomplete()`` (felix 0.3.0). Providers without async support fall
+        back to running the sync path in a worker thread.
+        """
+        messages = self._build_messages(system_prompt, user_prompt)
+
+        async def _do_complete() -> CompletionResult:
+            return await self.provider.acomplete(
+                messages,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                **self._provider_kwargs,
+            )
+
+        try:
+            if self._weave_op is not None:
+                result: CompletionResult = await self._weave_op(_do_complete)()
+            else:
+                result = await _do_complete()
+        except NotImplementedError:
+            return await asyncio.to_thread(
+                self._call_provider, system_prompt, user_prompt, temperature, max_tokens,
+            )
+
+        return self._record_completion(result)
 
     # ------------------------------------------------------------------
     # Abstract methods — subclasses must implement
@@ -603,20 +647,118 @@ def deliberate(
         self._last_action_plan = plan
         return plan
 
-    def _retry_with_correction(self, bad_output: str, error: str) -> LLMResult:
-        """Retry with a short correction prompt asking the LLM to fix its JSON."""
-        system_prompt = (
-            "You previously produced invalid JSON. Fix the error and return ONLY "
-            "valid JSON matching the required schema. No explanation, no markdown."
+    async def adeliberate(
+        self,
+        state: GameState,
+        current_time: float,
+        context_history: list[dict[str, Any]] | None = None,
+    ) -> ActionPlan:
+        """Async mirror of ``deliberate()`` — the LLM call runs on the event
+        loop via the provider's native ``acomplete()`` (felix 0.3.0), so a
+        timeout cancellation actually aborts the request instead of leaving
+        an orphaned worker thread.
+        """
+        if not hasattr(self, "_state") or self._state.value == "waiting":
+            self.spawn(current_time)
+        self.update_position(current_time)
+
+        spoke_context = self._get_spoke_context()
+        effective_context = spoke_context if spoke_context else (context_history or [])
+        task = self.build_task(state, effective_context)
+        result = await self.aprocess_task(task)
+
+        try:
+            plan = self.parse_action_plan(result, state.colony.tick)
+        except ActionPlanParseError as first_error:
+            logger.warning(
+                "%s: parse failed, retrying with correction prompt: %s",
+                self.ROLE_NAME, first_error.reason,
+            )
+            try:
+                retry_result = await self._aretry_with_correction(
+                    result.content, first_error.reason,
+                )
+                plan = self.parse_action_plan(retry_result, state.colony.tick)
+            except (ActionPlanParseError, Exception) as retry_error:
+                raise ActionPlanParseError(
+                    result.content, f"Retry also failed: {retry_error}",
+                ) from retry_error
+
+        self._last_action_plan = plan
+        return plan
+
+    async def aprocess_task(self, task: LLMTask) -> LLMResult:
+        """Async mirror of ``LLMAgent.process_task()`` using ``_acall_provider``.
+
+        Replicates the SDK's sync pipeline (events, adaptive temperature,
+        confidence, accounting) — the SDK exposes async only at the provider
+        layer, not on LLMAgent.
+        """
+        start = time.perf_counter()
+
+        self.emit_event(
+            EventType.TASK_STARTED,
+            {"task_id": task.task_id, "agent_type": self.agent_type},
+            source=f"agent:{self.agent_id}",
         )
-        user_prompt = (
+
+        temperature = self.get_adaptive_temperature()
+        system_prompt, user_prompt = self.create_position_aware_prompt(task)
+
+        completion = await self._acall_provider(
+            system_prompt, user_prompt, temperature, self.max_tokens,
+        )
+
+        confidence = self.calculate_confidence(completion.content)
+        self.record_confidence(confidence)
+
+        elapsed = time.perf_counter() - start
+        self.total_tokens_used += completion.total_tokens
+        self.total_processing_time += elapsed
+
+        result = LLMResult(
+            agent_id=self.agent_id,
+            task_id=task.task_id,
+            content=completion.content,
+            position_info=self.get_position_info(),
+            completion_result=completion,
+            processing_time=elapsed,
+            confidence=confidence,
+            temperature_used=temperature,
+            token_budget_used=completion.total_tokens,
+        )
+        self.processing_results.append(result)
+
+        self.emit_event(
+            EventType.TASK_COMPLETED,
+            {
+                "task_id": task.task_id,
+                "agent_type": self.agent_type,
+                "confidence": round(confidence, 4),
+                "temperature": round(temperature, 4),
+                "tokens": completion.total_tokens,
+                "processing_time": round(elapsed, 4),
+                "phase": result.position_info.get("phase", ""),
+            },
+            source=f"agent:{self.agent_id}",
+        )
+
+        return result
+
+    _CORRECTION_SYSTEM_PROMPT = (
+        "You previously produced invalid JSON. Fix the error and return ONLY "
+        "valid JSON matching the required schema. No explanation, no markdown."
+    )
+
+    @staticmethod
+    def _correction_user_prompt(bad_output: str, error: str) -> str:
+        return (
             f"Your previous output had this error: {error}\n\n"
             f"Original output (fix this):\n{bad_output[:2000]}\n\n"
             "Return ONLY the corrected JSON object."
         )
-        completion = self._call_provider(
-            system_prompt, user_prompt, temperature=0.1, max_tokens=self.max_tokens,
-        )
+
+    def _make_retry_result(self, completion: CompletionResult) -> LLMResult:
         self.total_tokens_used += completion.total_tokens
         return LLMResult(
             agent_id=self.agent_id,
@@ -629,3 +771,21 @@ def _retry_with_correction(self, bad_output: str, error: str) -> LLMResult:
             temperature_used=0.1,
             token_budget_used=completion.total_tokens,
         )
+
+    def _retry_with_correction(self, bad_output: str, error: str) -> LLMResult:
+        """Retry with a short correction prompt asking the LLM to fix its JSON."""
+        completion = self._call_provider(
+            self._CORRECTION_SYSTEM_PROMPT,
+            self._correction_user_prompt(bad_output, error),
+            temperature=0.1, max_tokens=self.max_tokens,
+        )
+        return self._make_retry_result(completion)
+
+    async def _aretry_with_correction(self, bad_output: str, error: str) -> LLMResult:
+        """Async mirror of ``_retry_with_correction``."""
+        completion = await self._acall_provider(
+            self._CORRECTION_SYSTEM_PROMPT,
+            self._correction_user_prompt(bad_output, error),
+            temperature=0.1, max_tokens=self.max_tokens,
+        )
+        return self._make_retry_result(completion)

src/rle/orchestration/game_loop.py

@@ -176,18 +176,17 @@ async def _deliberate_agent_with_timeout(
         self, agent: RimWorldRoleAgent, state: object,
         current_time: float, tick_num: int,
     ) -> tuple[RimWorldRoleAgent, ActionPlan | None]:
-        """Run one agent's deliberation in a worker thread with a hard timeout.
+        """Run one agent's deliberation with a hard timeout.
 
-        On timeout: emits a deliberation_timeout ERROR event, records a
-        provider_error entry in _deliberation_log, and returns (agent, None)
-        so the tick can continue. The hung thread keeps running until the
-        provider's own timeout cleans it up (Python threads can't be killed).
+        Deliberation is natively async (felix 0.3.0 ``acomplete()``), so a
+        timeout cancels the in-flight provider request instead of leaving an
+        orphaned worker thread. On timeout: emits a deliberation_timeout
+        ERROR event, records it in _deliberation_log, and returns
+        (agent, None) so the tick can continue.
         """
         try:
             return await asyncio.wait_for(
-                asyncio.to_thread(
-                    self._deliberate_agent, agent, state, current_time, tick_num,
-                ),
+                self._deliberate_agent(agent, state, current_time, tick_num),
                 timeout=self._role_timeout_s,
             )
         except asyncio.TimeoutError:
@@ -212,7 +211,7 @@ async def _deliberate_agent_with_timeout(
     async def _deliberate_parallel(
         self, state: object, current_time: float, tick_num: int,
     ) -> list[tuple[RimWorldRoleAgent, ActionPlan | None]]:
-        """Run role agents concurrently in worker threads with per-task timeout."""
+        """Run role agents concurrently on the event loop with per-task timeout."""
         return list(await asyncio.gather(*[
             self._deliberate_agent_with_timeout(a, state, current_time, tick_num)
             for a in self._role_agents
@@ -230,17 +229,17 @@ async def _deliberate_sequential(
             results.append((agent_result, plan))
         return results
 
-    def _deliberate_agent(
+    async def _deliberate_agent(
         self, agent: RimWorldRoleAgent, state: object,
         current_time: float, tick_num: int,
     ) -> tuple[RimWorldRoleAgent, ActionPlan | None]:
-        """Run one agent's deliberation. Thread-safe for parallel execution.
+        """Run one agent's deliberation.
 
         Agents read inter-agent context from their CentralPost spoke internally.
         """
         t0 = _time.monotonic()
         try:
-            plan = agent.deliberate(state, current_time)  # type: ignore[arg-type]
+            plan = await agent.adeliberate(state, current_time)  # type: ignore[arg-type]
         except ActionPlanParseError as e:
             latency_ms = round((_time.monotonic() - t0) * 1000, 1)
             logger.warning(