diff --git a/.changeset/py-durable-executor-seam.md b/.changeset/py-durable-executor-seam.md new file mode 100644 index 00000000..95cab6aa --- /dev/null +++ b/.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. diff --git a/python/server/src/smooth_operator_server/turn_runner.py b/python/server/src/smooth_operator_server/turn_runner.py index 658b9c57..8d624e08 100644 --- a/python/server/src/smooth_operator_server/turn_runner.py +++ b/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 diff --git a/python/server/tests/test_durable_executor_seam.py b/python/server/tests/test_durable_executor_seam.py new file mode 100644 index 00000000..756bf38f --- /dev/null +++ b/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"