Isolate the OpenAI Agents SDK, simplify compactor wiring

- Engine builds one AsyncOpenAI and installs the SDK default with
  use_for_tracing=False at run start; closed in stream_engine_async's
  outer finally so connection-pool lifecycle is deterministic.
- Compaction is a pure async def compact(*, client, compaction_model,
  item) -> str. build_compactor / CompactorFactory / Compactor type
  alias are gone; AgentContext.compact_old_items takes the client
  directly and calls compact() with self.compaction_model.
- OpenAiAgentRunner takes the client (not a compactor); the runner_protocol
  seam is deleted from production code (tests monkeypatch via the SDK seam).
- SynthesisTool requires its injected client; lazy construction is gone.
- Tests: shared probe_kit helpers updated; integration tests cover client
  lifecycle (normal exit + early consumer break); compaction tests
  monkeypatch agent_context.compact instead of build_compactor; example
  probes switched to duck-typed fakes and post-state assertions.

Closes INF-2933 items 8 and 9.

Author: declanatinference

engine/agents/agent_context.py

@@ -1,18 +1,18 @@
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable
-from typing import TYPE_CHECKING, TypeAlias
+from typing import TYPE_CHECKING
+
+from openai import AsyncOpenAI
 
 from engine.agents.agent_context_items import AgentContextItem
+from engine.agents.compactor import compact
 from engine.agents.prompt_templates import render_root_system_prompt
 from engine.model_config import ModelConfig
 from engine.models.messages import AgentMessage
 
 if TYPE_CHECKING:
     from engine.engine_config import EngineConfig
 
-Compactor: TypeAlias = Callable[[AgentContextItem], Awaitable[str]]
-
 
 class AgentContext:
     """One agent's conversation memory, with compaction-aware rendering to AgentMessage.
@@ -101,7 +101,7 @@ def to_messages_array(self) -> list[AgentMessage]:
         """Render stored items into provider-compatible messages, swapping in summaries for compacted items."""
         return [_render_item(item) for item in self.items]
 
-    async def compact_old_items(self, compactor: "Compactor") -> None:
+    async def compact_old_items(self, client: AsyncOpenAI) -> None:
         """Compact eligible older items in place using two independent keep-last thresholds.
 
         Text messages and tool turns are tracked separately; tool turns (assistant
@@ -128,7 +128,9 @@ async def compact_old_items(self, compactor: "Compactor") -> None:
 
         for idx in sorted(set(eligible)):
             item = self.items[idx]
-            summary = await compactor(item)
+            summary = await compact(
+                client=client, compaction_model=self.compaction_model, item=item
+            )
             self.items[idx] = item.model_copy(
                 update={"is_compacted": True, "compaction_summary": summary}
             )

engine/agents/compactor.py

@@ -1,65 +1,33 @@
 from __future__ import annotations
 
-from collections.abc import Callable
-
 from openai import AsyncOpenAI, omit
 
-from engine.agents.agent_context import Compactor
 from engine.agents.agent_context_items import AgentContextItem
-from engine.agents.agent_execution import AgentExecution
 from engine.agents.prompt_templates import COMPACTION_SYSTEM_PROMPT
-from engine.engine_config import EngineConfig
-
-CompactorFactory = Callable[[AgentExecution], Compactor]
-
-
-def build_compactor_factory(
-    engine_config: EngineConfig,
-    client: AsyncOpenAI | None = None,
-) -> CompactorFactory:
-    """Returns a factory that produces a Compactor bound to an OpenAI-compatible client.
-
-    The factory takes an AgentExecution (currently unused but reserved for
-    future per-agent compaction policies) and returns a callable that the
-    AgentContext can invoke per item it wants compacted. When ``client`` is
-    not supplied, it is constructed from ``engine_config.model_provider`` so
-    compaction routes through whichever OpenAI-compatible endpoint the run is
-    configured for.
-    """
-    openai_client = client
-
-    def factory(_execution: AgentExecution) -> Compactor:
-        async def compact(item: AgentContextItem) -> str:
-            nonlocal openai_client
-            if openai_client is None:
-                openai_client = AsyncOpenAI(
-                    base_url=engine_config.model_provider.base_url,
-                    api_key=engine_config.model_provider.api_key,
-                    default_headers=engine_config.model_provider.default_headers,
-                )
-
-            user_text = _item_as_prompt(item)
-            # Frontier models (gpt-5.x, claude-opus-4-7+, …) reject
-            # ``temperature`` as deprecated; only forward it when
-            # explicitly set on the compaction model.
-            temperature = (
-                engine_config.compaction_model.temperature
-                if engine_config.compaction_model.temperature is not None
-                else omit
-            )
-            response = await openai_client.chat.completions.create(
-                model=engine_config.compaction_model.name,
-                messages=[
-                    {"role": "system", "content": COMPACTION_SYSTEM_PROMPT},
-                    {"role": "user", "content": user_text},
-                ],
-                temperature=temperature,
-            )
-            return (response.choices[0].message.content or "").strip()
-
-        return compact
-
-    return factory
+from engine.model_config import ModelConfig
+
+
+async def compact(
+    *,
+    client: AsyncOpenAI,
+    compaction_model: ModelConfig,
+    item: AgentContextItem,
+) -> str:
+    """Summarize one ``AgentContextItem`` via ``client`` using ``compaction_model``."""
+    user_text = _item_as_prompt(item)
+    # Frontier models (gpt-5.x, claude-opus-4-7+, …) reject ``temperature``
+    # as deprecated; only forward it when explicitly set on the compaction
+    # model.
+    temperature = compaction_model.temperature if compaction_model.temperature is not None else omit
+    response = await client.chat.completions.create(
+        model=compaction_model.name,
+        messages=[
+            {"role": "system", "content": COMPACTION_SYSTEM_PROMPT},
+            {"role": "user", "content": user_text},
+        ],
+        temperature=temperature,
+    )
+    return (response.choices[0].message.content or "").strip()
 
 
 def _item_as_prompt(item: AgentContextItem) -> str:

engine/agents/engine_run_state.py

@@ -2,11 +2,10 @@
 
 from dataclasses import dataclass, field
 
-from agents import Runner
+from openai import AsyncOpenAI
 
 from engine.agents.agent_execution import AgentExecution
 from engine.agents.engine_output_bus import EngineOutputBus
-from engine.agents.runner_protocol import RunnerProtocol
 from engine.engine_config import EngineConfig
 from engine.sandbox.sandbox import Sandbox
 from engine.traces.trace_store import TraceStore
@@ -17,9 +16,7 @@ class EngineRunState:
     """Shared mutable state for one Engine run.
 
     Holds the singleton TraceStore, output bus, and config, plus lookup tables for
-    AgentExecutions by ``agent_id`` and by the ``tool_call_id`` that spawned them. The
-    ``runner`` field is a test seam: production uses ``agents.Runner``, probes inject
-    a fake (see ``RunnerProtocol``).
+    AgentExecutions by ``agent_id`` and by the ``tool_call_id`` that spawned them.
 
     ``sandbox`` is resolved once at run start. ``None`` means the host could not
     provide a working sandbox (e.g. Deno not installed or Pyodide wheels could
@@ -31,9 +28,9 @@ class EngineRunState:
     output_bus: EngineOutputBus
     config: EngineConfig
     sandbox: Sandbox | None
+    openai_client: AsyncOpenAI
     executions_by_agent_id: dict[str, AgentExecution] = field(default_factory=dict)
     executions_by_tool_call_id: dict[str, AgentExecution] = field(default_factory=dict)
-    runner: RunnerProtocol = field(default_factory=lambda: Runner)
 
     def register(self, execution: AgentExecution) -> None:
         """Index a newly-created AgentExecution by agent_id, and by tool_call_id when subagent."""

engine/agents/openai_agent_runner.py

@@ -4,7 +4,6 @@
 from collections.abc import Awaitable, Callable
 from typing import Any
 
-from agents import set_default_openai_client
 from openai import (
     APIConnectionError,
     APIStatusError,
@@ -13,38 +12,11 @@
     RateLimitError,
 )
 
-from engine.agents.agent_context import AgentContext, Compactor
+from engine.agents.agent_context import AgentContext
 from engine.agents.agent_execution import AgentExecution
 from engine.agents.engine_output_bus import EngineOutputBus
 from engine.agents.openai_event_mapper import OpenAiEventMapper
 from engine.errors import EngineAgentExhaustedError, EngineAgentRefusedError
-from engine.model_provider_config import ModelProviderConfig
-
-
-def configure_default_sdk_client(provider: ModelProviderConfig) -> None:
-    """Bind the OpenAI Agents SDK's default client to the configured endpoint.
-
-    The SDK uses a process-global client, so this is best-effort for callers
-    running multiple engines in one process. We only override when at least
-    one of ``base_url`` / ``api_key`` / ``default_headers`` is set; otherwise
-    the SDK keeps using its env-driven default.
-
-    ``use_for_tracing=False`` keeps the SDK's tracing exporter on its
-    default OpenAI path. Without this, redirecting model calls to a non-
-    OpenAI provider (vLLM, Ollama, OpenRouter, etc.) also redirects
-    tracing POSTs there — those endpoints don't speak the tracing API,
-    causing spurious errors or silent trace loss.
-    """
-    if provider.base_url is None and provider.api_key is None and provider.default_headers is None:
-        return
-    set_default_openai_client(
-        AsyncOpenAI(
-            base_url=provider.base_url,
-            api_key=provider.api_key,
-            default_headers=provider.default_headers,
-        ),
-        use_for_tracing=False,
-    )
 
 
 def _is_retriable_llm_error(exc: BaseException) -> bool:
@@ -59,7 +31,6 @@ def _is_retriable_llm_error(exc: BaseException) -> bool:
 MAX_CONSECUTIVE_LLM_FAILURES = 10
 
 RunStreamedCallable = Callable[..., Awaitable[Any]]
-CompactorFactory = Callable[[AgentExecution], Compactor]
 logger = logging.getLogger(__name__)
 
 
@@ -76,15 +47,15 @@ class OpenAiAgentRunner:
     def __init__(
         self,
         run_streamed: RunStreamedCallable,
-        compactor_factory: CompactorFactory,
+        client: AsyncOpenAI,
         event_mapper: OpenAiEventMapper | None = None,
         refusal_retries: int = 0,
     ) -> None:
         """``run_streamed`` is injected so root and subagent paths can supply their own
-        max_turns and starting agent. ``compactor_factory`` produces a per-execution
-        compactor bound to whatever model EngineConfig pins for compaction."""
+        max_turns and starting agent. ``client`` is the per-run AsyncOpenAI used for
+        compaction calls."""
         self._run_streamed = run_streamed
-        self._compactor_factory = compactor_factory
+        self._client = client
         self._mapper = event_mapper or OpenAiEventMapper()
         self._refusal_retries = refusal_retries
 
@@ -188,7 +159,7 @@ async def run(
                 )
 
             agent_execution.record_llm_success()
-            await agent_context.compact_old_items(self._compactor_factory(agent_execution))
+            await agent_context.compact_old_items(self._client)
             return
 
         raise EngineAgentExhaustedError(