th-137b91: Python server durable-executor selection seam (Q parity)

.changeset/py-durable-executor-seam.md

@@ -0,0 +1,16 @@
+---
+"@smooai/smooth-operator": patch
+---
+
+feat(python-server): env-gated durable-executor selection seam (th-137b91, Q parity)
+
+The Python server ran every turn by calling `SmoothAgent.run_stream` directly, so there was no single
+place a durable backend (ADR-030) could be selected — unlike the Rust server's `turn_executor`
+(`runner.rs`). `TurnRunner` now routes each turn through the engine's `AgentExecutor` seam, chosen
+once in `select_turn_executor`: a durable backend is dependency-injected as an opaque `AgentExecutor`
+(so the server keeps no hard dependency on the Temporal package), and it is used only when
+`SMOOTH_AGENT_DURABLE_EXECUTOR` opts in (`1/true/on/yes`). With nothing injected — the default — the
+turn runs on `InProcessExecutor`, a verbatim delegation to `run_stream`, so behavior is unchanged.
+Asking for durable mode with nothing injected warns and falls back rather than silently pretending a
+turn is durable. `durable_requested` is split out for a testable parse. Tests cover the parse table,
+the selection logic, and two real turns driven through a fake injected executor.

python/server/src/smooth_operator_server/turn_runner.py

@@ -21,10 +21,12 @@
 from typing import Any, Callable
 
 from smooth_operator_core import (
+    AgentExecutor,
     AgentOptions,
     DoneEvent,
     HumanApprovalRequest,
     HumanApprovalResponse,
+    InProcessExecutor,
     Knowledge,
     SmoothAgent,
     SmoothAgentThread,
@@ -118,6 +120,50 @@
 logger = logging.getLogger(__name__)
 
 
+#: Env var a deployment sets to run turns on a durable backend (ADR-030) instead of
+#: in-process. Unset — the default — is the in-process executor, a verbatim
+#: delegation to ``SmoothAgent.run_stream``, so a deployment that never sets this
+#: behaves exactly as it did before the seam existed. Mirrors the Rust server's
+#: ``DURABLE_EXECUTOR_ENV`` (``runner.rs``).
+DURABLE_EXECUTOR_ENV = "SMOOTH_AGENT_DURABLE_EXECUTOR"
+
+
+def durable_requested(value: str | None) -> bool:
+    """Whether ``value`` opts into durable execution.
+
+    Off unless explicitly asked — an unset, empty, or unrecognized value stays off.
+    Separated from the env read so the parse is testable without mutating
+    process-global state (mirrors the Rust ``durable_requested``)."""
+    return (value or "").strip().lower() in {"1", "true", "on", "yes"}
+
+
+def select_turn_executor(injected: AgentExecutor | None, env_value: str | None = None) -> AgentExecutor:
+    """The executor a turn runs on — the one place a durable backend plugs in.
+
+    A durable backend is dependency-**injected** (``TurnRunner(executor=...)``), so
+    this module needs no hard dependency on the Temporal package: the seam works with
+    the injected executor as an opaque ``AgentExecutor``. Selection is env-gated to
+    match the parity spec — the injected durable backend is used only when
+    ``SMOOTH_AGENT_DURABLE_EXECUTOR`` opts in; otherwise (and whenever nothing is
+    injected) this resolves to the in-process executor, a verbatim delegation to
+    ``SmoothAgent.run_stream``. Asking for durable mode without supplying an executor
+    warns and falls back, rather than silently pretending a turn is durable.
+
+    (This differs deliberately from the Rust ``turn_executor``, where an injected
+    executor wins unconditionally: here the env var is the explicit opt-in, so an
+    injected backend can never silently take over a deployment that didn't ask.)"""
+    if env_value is None:
+        env_value = os.environ.get(DURABLE_EXECUTOR_ENV)
+    if durable_requested(env_value):
+        if injected is not None:
+            return injected
+        logger.warning(
+            "%s requested but no durable executor was supplied on the turn; running the turn in-process",
+            DURABLE_EXECUTOR_ENV,
+        )
+    return InProcessExecutor()
+
+
 def preamble_model() -> str | None:
     """The fast model id for the parallel preamble, or ``None`` when the feature is
     off. Unset / empty / whitespace ⇒ off (no extra LLM call, behavior unchanged) —
@@ -201,9 +247,16 @@ def __init__(
         judge_model: str | None = None,
         tool_hooks: list[Any] | None = None,
         org_id: str | None = None,
+        executor: AgentExecutor | None = None,
     ) -> None:
         self._chat_client = chat_client
         self._store = store
+        #: The executor the turn's engine runs on. Selected once here from the
+        #: injected backend + ``SMOOTH_AGENT_DURABLE_EXECUTOR`` (see
+        #: :func:`select_turn_executor`): a durable backend arrives by injection, and
+        #: with nothing injected (the default) this is the in-process executor, a
+        #: verbatim delegation to ``SmoothAgent.run_stream`` — behavior unchanged.
+        self._executor = select_turn_executor(executor)
         self._knowledge = knowledge
         self._system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT
         #: Resolved per-agent config (instructions / workflow / persona). ``None`` →
@@ -423,7 +476,10 @@ async def _gate(req: HumanApprovalRequest) -> HumanApprovalResponse:
                 # 3. Persist the inbound user message.
                 await self._store.append_message(conversation_id, MessageDirection.INBOUND, user_message)
 
-                async for event in agent.run_stream(model_message, thread=thread):
+                # Run the turn through the executor seam (default in-process, a durable
+                # backend when injected + opted in), not `agent.run_stream` directly —
+                # the one place a durable backend plugs in (mirrors the Rust runner).
+                async for event in self._executor.execute_streaming(agent, model_message, thread=thread):
                     if isinstance(event, TextEvent):
                         if event.text:
                             # Close the preamble window BEFORE the first answer token

python/server/tests/test_durable_executor_seam.py

@@ -0,0 +1,114 @@
+"""The server's durable-executor selection seam (ADR-030, item Q).
+
+``TurnRunner`` runs each turn through an ``AgentExecutor`` rather than calling
+``SmoothAgent.run_stream`` directly, so a durable backend can be selected in one
+place — mirroring the Rust server's ``turn_executor`` (``runner.rs``). The seam is
+dependency-**injected**: the durable backend is passed in as an opaque
+``AgentExecutor``, so the server needs no hard dependency on the Temporal package.
+
+These tests cover the selection logic (``durable_requested`` parse +
+``select_turn_executor``) and prove end to end that an injected executor is used
+when — and only when — ``SMOOTH_AGENT_DURABLE_EXECUTOR`` opts in.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from smooth_operator_core import InProcessExecutor, MockLlmProvider
+
+from smooth_operator_server.session_store import InMemorySessionStore
+from smooth_operator_server.turn_runner import (
+    DURABLE_EXECUTOR_ENV,
+    TurnRunner,
+    durable_requested,
+    select_turn_executor,
+)
+
+
+class SpyExecutor:
+    """A fake durable ``AgentExecutor`` that records it was used, then delegates to
+    the real agent so the turn still completes."""
+
+    def __init__(self) -> None:
+        self.used = False
+
+    async def execute(self, agent: Any, message: str, history: Any = None, thread: Any = None) -> Any:
+        self.used = True
+        return await agent.run(message, history, thread)
+
+    def execute_streaming(self, agent: Any, message: str, history: Any = None, thread: Any = None) -> Any:
+        self.used = True
+        return agent.run_stream(message, history, thread)
+
+
+# ── durable_requested: opt-in parse (mirrors the Rust table) ─────────────────
+
+
+@pytest.mark.parametrize("value", ["1", "true", "TRUE", " on ", "yes"])
+def test_durable_requested_opts_in(value: str):
+    assert durable_requested(value) is True
+
+
+@pytest.mark.parametrize("value", ["", " ", "0", "false", "off", "no", "maybe", None])
+def test_durable_requested_stays_off(value: str | None):
+    assert durable_requested(value) is False
+
+
+# ── select_turn_executor: injection + env gating ─────────────────────────────
+
+
+def test_injected_executor_used_when_env_opts_in():
+    spy = SpyExecutor()
+    assert select_turn_executor(spy, env_value="1") is spy
+
+
+def test_injected_executor_ignored_when_env_off():
+    """Env off ⇒ in-process, even with a durable executor injected — an injected
+    backend never silently takes over a deployment that didn't ask."""
+    spy = SpyExecutor()
+    selected = select_turn_executor(spy, env_value="0")
+    assert selected is not spy
+    assert isinstance(selected, InProcessExecutor)
+
+
+def test_env_on_without_injection_falls_back_to_in_process():
+    selected = select_turn_executor(None, env_value="true")
+    assert isinstance(selected, InProcessExecutor)
+
+
+def test_each_fallback_builds_its_own_in_process_executor():
+    a = select_turn_executor(None, env_value=None)
+    b = select_turn_executor(None, env_value=None)
+    assert isinstance(a, InProcessExecutor)
+    assert a is not b
+
+
+# ── TurnRunner wires the seam, and the turn actually routes through it ────────
+
+
+async def test_turn_runs_through_injected_executor_when_env_opts_in(monkeypatch):
+    monkeypatch.setenv(DURABLE_EXECUTOR_ENV, "1")
+    spy = SpyExecutor()
+    mock = MockLlmProvider().push_text("durable reply")
+
+    runner = TurnRunner(chat_client=mock, store=InMemorySessionStore(), executor=spy)
+    result = await runner.run(conversation_id="c1", request_id="r1", user_message="go", sink=lambda _e: None)
+
+    assert spy.used is True
+    assert result.reply == "durable reply"
+
+
+async def test_turn_runs_in_process_when_env_off(monkeypatch):
+    monkeypatch.delenv(DURABLE_EXECUTOR_ENV, raising=False)
+    spy = SpyExecutor()
+    mock = MockLlmProvider().push_text("in-process reply")
+
+    runner = TurnRunner(chat_client=mock, store=InMemorySessionStore(), executor=spy)
+    result = await runner.run(conversation_id="c2", request_id="r2", user_message="go", sink=lambda _e: None)
+
+    # The injected executor was NOT used — the in-process path ran instead.
+    assert spy.used is False
+    assert isinstance(runner._executor, InProcessExecutor)
+    assert result.reply == "in-process reply"