refactor: drop ModelRunner/AgentRunner compat from managed layer

ManagedModel and ManagedAgent now require a Runner. The compat shims
(_invoke_runner, isinstance(result, RunnerResult) branches, Union
type annotations) are removed; result handling is direct on
RunnerResult fields.

The deprecated ManagedModel.invoke() is preserved for backwards compat
but now delegates to run() and adapts the ManagedResult into the legacy
ModelResponse shape.

ModelRunner and AgentRunner protocol definitions remain in place so
downstream provider packages that import them continue to work.

Author: jsonbailey

packages/sdk/server-ai/src/ldai/client.py

@@ -443,7 +443,7 @@ async def create_model(
         if not runner:
             return None
 
-        return ManagedModel(config, runner)
+        return ManagedModel(config, runner)  # type: ignore[arg-type]
 
     async def create_chat(
         self,
@@ -517,7 +517,7 @@ async def create_agent(
         if not runner:
             return None
 
-        return ManagedAgent(config, runner)
+        return ManagedAgent(config, runner)  # type: ignore[arg-type]
 
     def agent_config(
         self,

packages/sdk/server-ai/src/ldai/managed_agent.py

@@ -1,26 +1,22 @@
 """ManagedAgent — LaunchDarkly managed wrapper for agent invocations."""
 
-from typing import Union
-
 from ldai.models import AIAgentConfig
-from ldai.providers import AgentResult, AgentRunner
 from ldai.providers.runner import Runner
-from ldai.providers.types import ManagedResult, RunnerResult
+from ldai.providers.types import ManagedResult
 
 
 class ManagedAgent:
     """
     LaunchDarkly managed wrapper for AI agent invocations.
 
-    Holds an AgentRunner or Runner. Handles tracking automatically via
-    ``create_tracker()``.
+    Holds a Runner. Handles tracking automatically via ``create_tracker()``.
     Obtain an instance via ``LDAIClient.create_agent()``.
     """
 
     def __init__(
         self,
         ai_config: AIAgentConfig,
-        agent_runner: Union[Runner, AgentRunner],
+        agent_runner: Runner,
     ):
         self._ai_config = ai_config
         self._agent_runner = agent_runner
@@ -33,23 +29,21 @@ async def run(self, input: str) -> ManagedResult:
         :return: ManagedResult containing the agent's output and metric summary
         """
         tracker = self._ai_config.create_tracker()
-        result: Union[RunnerResult, AgentResult] = await tracker.track_metrics_of_async(
+        result = await tracker.track_metrics_of_async(
             lambda r: r.metrics,
             lambda: self._agent_runner.run(input),
         )
-        # Support both RunnerResult (content) and legacy AgentResult (output)
-        content = result.content if isinstance(result, RunnerResult) else result.output  # type: ignore[union-attr]
         return ManagedResult(
-            content=content,
+            content=result.content,
             metrics=tracker.get_summary(),
             raw=result.raw,
         )
 
-    def get_agent_runner(self) -> Union[Runner, AgentRunner]:
+    def get_agent_runner(self) -> Runner:
         """
         Return the underlying runner for advanced use.
 
-        :return: The Runner or AgentRunner instance.
+        :return: The Runner instance.
         """
         return self._agent_runner
 

packages/sdk/server-ai/src/ldai/managed_model.py

@@ -1,28 +1,27 @@
 import asyncio
 import warnings
-from typing import List, Union
+from typing import List
 
 from ldai import log
 from ldai.models import AICompletionConfig, LDMessage
-from ldai.providers.model_runner import ModelRunner
 from ldai.providers.runner import Runner
-from ldai.providers.types import JudgeResult, ManagedResult, ModelResponse, RunnerResult
+from ldai.providers.types import JudgeResult, LDAIMetrics, ManagedResult, ModelResponse
 from ldai.tracker import LDAIConfigTracker
 
 
 class ManagedModel:
     """
     LaunchDarkly managed wrapper for AI model invocations.
 
-    Holds a Runner (or legacy ModelRunner). Handles conversation management,
-    judge evaluation dispatch, and tracking automatically via ``create_tracker()``.
+    Holds a Runner. Handles conversation management, judge evaluation
+    dispatch, and tracking automatically via ``create_tracker()``.
     Obtain an instance via ``LDAIClient.create_model()``.
     """
 
     def __init__(
         self,
         ai_config: AICompletionConfig,
-        model_runner: Union[Runner, ModelRunner],
+        model_runner: Runner,
     ):
         self._ai_config = ai_config
         self._model_runner = model_runner
@@ -48,50 +47,27 @@ async def run(self, prompt: str) -> ManagedResult:
         config_messages = self._ai_config.messages or []
         all_messages = config_messages + self._messages
 
-        result: Union[RunnerResult, ModelResponse] = await tracker.track_metrics_of_async(
+        result = await tracker.track_metrics_of_async(
             lambda r: r.metrics,
-            lambda: self._invoke_runner(all_messages),
+            lambda: self._model_runner.run(all_messages),
         )
 
-        # Support both new RunnerResult and legacy ModelResponse
-        if isinstance(result, RunnerResult):
-            content = result.content
-            raw = result.raw
-            parsed = result.parsed
-            assistant_message = LDMessage(role='assistant', content=content)
-        else:
-            content = result.message.content
-            raw = getattr(result, 'raw', None)
-            parsed = getattr(result, 'parsed', None)
-            assistant_message = result.message
+        assistant_message = LDMessage(role='assistant', content=result.content)
 
         input_text = '\r\n'.join(m.content for m in self._messages) if self._messages else ''
 
-        evaluations_task = self._track_judge_results(tracker, input_text, content)
+        evaluations_task = self._track_judge_results(tracker, input_text, result.content)
 
         self._messages.append(assistant_message)
 
         return ManagedResult(
-            content=content,
+            content=result.content,
             metrics=tracker.get_summary(),
-            raw=raw,
-            parsed=parsed,
+            raw=result.raw,
+            parsed=result.parsed,
             evaluations=evaluations_task,
         )
 
-    async def _invoke_runner(
-        self, all_messages: List[LDMessage]
-    ) -> Union[RunnerResult, ModelResponse]:
-        """
-        Delegate to the runner.  Supports both the new ``Runner`` protocol
-        (``run(messages) → RunnerResult``) and the legacy ``ModelRunner``
-        (``invoke_model(messages) → ModelResponse``).
-        """
-        if isinstance(self._model_runner, Runner):
-            return await self._model_runner.run(all_messages)
-        # Legacy ModelRunner path
-        return await self._model_runner.invoke_model(all_messages)  # type: ignore[union-attr]
-
     async def invoke(self, prompt: str) -> ModelResponse:
         """
         Invoke the model with a prompt string.
@@ -108,26 +84,16 @@ async def invoke(self, prompt: str) -> ModelResponse:
             DeprecationWarning,
             stacklevel=2,
         )
-        tracker = self._ai_config.create_tracker()
-
-        user_message = LDMessage(role='user', content=prompt)
-        self._messages.append(user_message)
-
-        config_messages = self._ai_config.messages or []
-        all_messages = config_messages + self._messages
-
-        response: ModelResponse = await tracker.track_metrics_of_async(
-            lambda result: result.metrics,
-            lambda: self._model_runner.invoke_model(all_messages),  # type: ignore[union-attr]
+        result = await self.run(prompt)
+        return ModelResponse(
+            message=LDMessage(role='assistant', content=result.content),
+            metrics=LDAIMetrics(
+                success=bool(result.metrics.success),
+                usage=result.metrics.usage,
+            ),
+            evaluations=result.evaluations,
         )
 
-        input_text = '\r\n'.join(m.content for m in self._messages) if self._messages else ''
-        output_text = response.message.content
-        response.evaluations = self._track_judge_results(tracker, input_text, output_text)
-
-        self._messages.append(response.message)
-        return response
-
     def _track_judge_results(
         self,
         tracker: LDAIConfigTracker,
@@ -169,11 +135,11 @@ def append_messages(self, messages: List[LDMessage]) -> None:
         """
         self._messages.extend(messages)
 
-    def get_model_runner(self) -> Union[Runner, ModelRunner]:
+    def get_model_runner(self) -> Runner:
         """
         Return the underlying runner for advanced use.
 
-        :return: The Runner or legacy ModelRunner instance.
+        :return: The Runner instance.
         """
         return self._model_runner
 

packages/sdk/server-ai/tests/test_managed_model.py

@@ -9,7 +9,7 @@
 from ldai.evaluator import Evaluator
 from ldai.managed_model import ManagedModel
 from ldai.models import AICompletionConfig, LDMessage, ModelConfig, ProviderConfig
-from ldai.providers.types import JudgeResult, LDAIMetrics, ManagedResult, ModelResponse, RunnerResult
+from ldai.providers.types import JudgeResult, LDAIMetrics, ManagedResult, RunnerResult
 from ldai.tracker import LDAIConfigTracker, LDAIMetricSummary
 
 
@@ -21,13 +21,6 @@ def _make_runner_result(content: str = 'response text') -> RunnerResult:
     )
 
 
-def _make_model_response(content: str = 'response text') -> ModelResponse:
-    return ModelResponse(
-        message=LDMessage(role='assistant', content=content),
-        metrics=LDAIMetrics(success=True, usage=None),
-    )
-
-
 def _make_summary() -> LDAIMetricSummary:
     summary = LDAIMetricSummary()
     summary._success = True
@@ -237,11 +230,9 @@ async def test_invoke_emits_deprecation_warning(self):
         """invoke() should emit a DeprecationWarning."""
         evaluator = Evaluator.noop()
         mock_runner = MagicMock()
-        mock_runner.invoke_model = AsyncMock(return_value=_make_model_response())
+        mock_runner.run = AsyncMock(return_value=_make_runner_result())
 
-        config, mock_tracker = _make_config_with_tracker(evaluator)
-        # invoke() expects a ModelResponse from the tracker, not a RunnerResult.
-        mock_tracker.track_metrics_of_async = AsyncMock(return_value=_make_model_response())
+        config, _mock_tracker = _make_config_with_tracker(evaluator)
         model = ManagedModel(config, mock_runner)
 
         with pytest.warns(DeprecationWarning, match=r"ManagedModel\.invoke\(\) is deprecated"):