diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index afdefc3..1e29fdf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -85,11 +85,42 @@ jobs: mv submilli-server-x86_64-unknown-linux-musl submilli-server chmod +x submilli-server - run: uv sync --all-packages --all-groups --locked - - run: uv run pytest ${{ matrix.tests }} + - run: uv run pytest ${{ matrix.tests }} -m 'not durable' working-directory: packages/${{ matrix.package }} env: SUBMILLI_BIN: ${{ github.workspace }}/.submilli-bin + durable-integration: + runs-on: ubuntu-latest + timeout-minutes: 45 + steps: + - uses: actions/checkout@v7 + - uses: astral-sh/setup-uv@v10.3.0 + - uses: actions/checkout@v7 + with: + repository: submilli/submilli-runtime + # Durable callbacks landed after v0.4.3. Pin the tested upstream + # contract until a release containing them is available. + ref: 5f67954152a8fa812f64ffc9cb1d17cd4ccd82a8 + path: runtime + - uses: dtolnay/rust-toolchain@1.98.1 + - name: Build upstream submilli-server with durable callbacks + run: cargo build --release -p submilli-server + working-directory: runtime + env: + CARGO_BUILD_JOBS: "2" + - run: uv sync --all-packages --all-groups --locked + - name: SDK durable execution and crash recovery + run: uv run pytest tests/integration/test_durable.py + working-directory: packages/submilli + env: + SUBMILLI_BIN: ${{ github.workspace }}/runtime/target/release + - name: LangChain durable tools and approvals + run: uv run pytest tests/integration_tests/test_durable.py + working-directory: packages/langchain-submilli + env: + SUBMILLI_BIN: ${{ github.workspace }}/runtime/target/release + build: runs-on: ubuntu-latest steps: diff --git a/README.md b/README.md index fbdc74d..391ae81 100644 --- a/README.md +++ b/README.md @@ -150,10 +150,16 @@ Integration tests start submilli-server from `$SUBMILLI_BIN` or `PATH`. A releas live in the ignored `.submilli-bin/` directory: ```sh -SUBMILLI_BIN=$PWD/../../.submilli-bin uv run pytest tests/integration # submilli -SUBMILLI_BIN=$PWD/../../.submilli-bin uv run pytest tests/integration_tests # langchain-submilli +SUBMILLI_BIN=$PWD/../../.submilli-bin uv run pytest tests/integration -m 'not durable' # submilli +SUBMILLI_BIN=$PWD/../../.submilli-bin uv run pytest tests/integration_tests -m 'not durable' # langchain-submilli ``` +Durable integration tests require an upstream runtime build containing durable callbacks +(tested at `5f67954152a8fa812f64ffc9cb1d17cd4ccd82a8`, after v0.4.3). Set `SUBMILLI_BIN` +to its binary directory and run the same commands with `-m durable`. CI builds that pinned +revision separately and exercises queued/running HTTP responses, crash recovery, duplicate +resumes, concurrent tools, and approvals after restart. Each test starts an isolated server. + Some `langchain-submilli` integration tests also need the `research` Blueprint. `packages/langchain-submilli/scripts/test-server.sh` starts a suitable server. diff --git a/packages/langchain-submilli/CHANGELOG.md b/packages/langchain-submilli/CHANGELOG.md index 4590744..c040e05 100644 --- a/packages/langchain-submilli/CHANGELOG.md +++ b/packages/langchain-submilli/CHANGELOG.md @@ -6,6 +6,17 @@ All notable changes to this project are recorded here. The format follows ## [Unreleased] +### Fixed + +- Expired idempotency responses reach the model as errors that advise checking + effects before repeating the program. +- On a server with durable execution, a run that stops without a result + (`stuck`, `cancelled`, `aborted`, `indeterminate`, `expired`, or a state the + SDK does not know) reaches the model as an error result with the run's status + and reason, as a lost run does, instead of an empty result. The artifact's + `run_status` names the state. A queued run is waited for. Needs the next + `submilli` release, which adds `RunStateError`. + ## [0.1.1] - 2026-10-15 ### Changed diff --git a/packages/langchain-submilli/NOTES.md b/packages/langchain-submilli/NOTES.md index fa667a5..3a5c7ca 100644 --- a/packages/langchain-submilli/NOTES.md +++ b/packages/langchain-submilli/NOTES.md @@ -174,6 +174,12 @@ each call suspends the run in memory, and the harness answers with `POST /v1/run `idempotency_incomplete`. `409 run_not_suspended` (a call timed out) and `400 unknown_call` make the middleware read the run with `GET /v1/runs/{id}` and carry on. A `running` state is polled with backoff. +- **Durable runs.** On a server with `durable_execution`, a `queued` run is polled like a + `running` one. A run with no result to return (`submilli.RunStateError`: `stuck`, `cancelled`, + `aborted`, `indeterminate`, `expired`, or an unknown state) becomes an error result for the + model, as a lost run does, with the status and reason. It does not raise: the model needs to + know the run's effects may be partial, and a stuck run holds back later programs in the + session. - **Discovery.** With exposed tools, `packages__search` and `packages__docs` use `/v1/sessions/{id}/packages/...`, so `@tools` is included, and those calls open the session (in `aafter_model` too). Without exposed tools nothing changes. diff --git a/packages/langchain-submilli/README.md b/packages/langchain-submilli/README.md index 9591b82..f655e49 100644 --- a/packages/langchain-submilli/README.md +++ b/packages/langchain-submilli/README.md @@ -82,6 +82,9 @@ instructions for using them from the Submilli server. in-progress, or unavailable replays without silently rerunning the program. - When a session has ended, the middleware opens a replacement. If a restarted server asks for the session's secrets, it rebinds them once and retries. +- On a server with durable execution, the middleware waits for a queued run. A run that stops + without a result, such as one that is stuck or cancelled, comes back to the model as an error + with the run's `status` and `reason`, and the artifact's `run_status` names it. - The execute tool message contains `result`, `console`, `error`, and `denied`. Its artifact contains `execution_id`, `session_id`, and any application-tool calls. diff --git a/packages/langchain-submilli/pyproject.toml b/packages/langchain-submilli/pyproject.toml index b991a2f..3e085bd 100644 --- a/packages/langchain-submilli/pyproject.toml +++ b/packages/langchain-submilli/pyproject.toml @@ -62,7 +62,10 @@ include = ["src", "tests", "README.md", "NOTES.md", "CHANGELOG.md", "LICENSE"] [tool.pytest.ini_options] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" -markers = ["integration: needs a running submilli-server with the research Blueprint"] +markers = [ + "integration: needs a running submilli-server with the research Blueprint", + "durable: requires a runtime build with durable execution and tool callbacks", +] testpaths = ["tests/unit_tests"] pythonpath = ["."] diff --git a/packages/langchain-submilli/src/langchain_submilli/_middleware.py b/packages/langchain-submilli/src/langchain_submilli/_middleware.py index 5913afe..d29502a 100644 --- a/packages/langchain-submilli/src/langchain_submilli/_middleware.py +++ b/packages/langchain-submilli/src/langchain_submilli/_middleware.py @@ -29,6 +29,7 @@ IdempotencyError, PendingCall, RunState, + RunStateError, SessionRequiresSecretsError, SubmilliError, ToolResult, @@ -289,6 +290,9 @@ async def _execute(self, code: str, runtime: ToolRuntime) -> ToolMessage | Comma except IdempotencyError as error: message = t.idempotency_message(error, tool_call_id=tool_call_id, session_id=session["id"]) return _with_session(message, session, stored) + except RunStateError as error: + message = t.run_state_message(error, tool_call_id=tool_call_id, session_id=session["id"], calls=[]) + return _with_session(message, session, stored) calls: list[dict[str, Any]] = [] updates: list[tuple[str, Any]] = [] if not isinstance(result, ExecuteResult): @@ -299,6 +303,9 @@ async def _execute(self, code: str, runtime: ToolRuntime) -> ToolMessage | Comma tool_call_id=tool_call_id, execution_id=result.execution_id, session_id=session["id"], calls=calls ) return _with_session(message, session, stored, updates) + except RunStateError as error: + message = t.run_state_message(error, tool_call_id=tool_call_id, session_id=session["id"], calls=calls) + return _with_session(message, session, stored, updates) message = t.execute_message(result, tool_call_id=tool_call_id, notice=notice, tool_calls=calls) return _with_session(message, session, stored, updates) diff --git a/packages/langchain-submilli/src/langchain_submilli/_tools.py b/packages/langchain-submilli/src/langchain_submilli/_tools.py index 84ad8d7..f2ccb95 100644 --- a/packages/langchain-submilli/src/langchain_submilli/_tools.py +++ b/packages/langchain-submilli/src/langchain_submilli/_tools.py @@ -17,7 +17,7 @@ from langchain_core.tools import BaseTool, InjectedToolArg, StructuredTool from pydantic import BaseModel, Field -from submilli import BlueprintPrompt, ExecuteResult, IdempotencyError +from submilli import BlueprintPrompt, ExecuteResult, IdempotencyError, RunStateError # The server's own names, as its MCP endpoint lists them. EXECUTE = "submilli__typescript__execute" @@ -154,6 +154,10 @@ def unsupported(runtime: ToolRuntime, **_: Any) -> Any: "the program rather than risk running it twice. Trying again later may work." ), "idempotency_key_invalid": "Nothing ran. The server refused this call's idempotency key.", + "idempotency_response_expired": ( + "Nothing new ran. The earlier program finished, but the server no longer keeps its result. " + "Check its effects before running anything that repeats it; changing the key would run it again." + ), } @@ -223,6 +227,71 @@ def lost_run_message( ) +_UNKNOWN_EFFECTS = ( + "Check what it would have changed before running anything that repeats it, and tell the user if you cannot tell." +) +_NO_RESULT = ( + "run_result_unavailable", + "The run finished, but the server no longer keeps its result. Check its effects, or the last run, " + "before running anything that repeats it.", +) + +# By run status: the error kind the model sees, and what it is told. +RUN_STATE_MESSAGES = { + "stuck": ( + "run_stuck", + "The run stopped before it finished, and the server holds it until an operator recovers it. Its " + "effects up to that point are unknown, and programs run after it in this session wait until it is " + "recovered. Tell the user before running anything else.", + ), + "cancelled": ( + "run_cancelled", + "The run was cancelled before it finished, because its session was closed. Its effects up to that " + "point are unknown. " + _UNKNOWN_EFFECTS, + ), + "aborted": ("run_aborted", "The run was abandoned before the program started, so nothing ran."), + "indeterminate": ( + "run_indeterminate", + "The run was interrupted, and the server cannot tell how far it got: it may have done some, all, " + "or none of its work. " + _UNKNOWN_EFFECTS, + ), + "expired": _NO_RESULT, + "completed": _NO_RESULT, +} + + +def run_state_message( + error: RunStateError, *, tool_call_id: str, session_id: str, calls: list[dict[str, Any]] +) -> ToolMessage: + """A run the server reports with no result to return, such as a durable run that is `stuck`.""" + kind, message = RUN_STATE_MESSAGES.get( + error.status, + ( + "run_state_unknown", + f"The server reported the run as {error.status!r}, which this version of langchain-submilli " + "does not know, so its outcome is unknown. " + _UNKNOWN_EFFECTS, + ), + ) + reported: dict[str, Any] = {"kind": kind, "message": message, "status": error.status} + if error.reason: + reported["reason"] = error.reason + content: dict[str, Any] = {"result": None, "console": [], "error": reported, "denied": False} + artifact: dict[str, Any] = { + "execution_id": error.execution_id, + "session_id": session_id, + "run_status": error.status, + } + if calls: + artifact["tool_calls"] = calls + return ToolMessage( + content=json.dumps(content, ensure_ascii=False), + tool_call_id=tool_call_id, + name=EXECUTE, + artifact=artifact, + status="error", + ) + + def idempotency_message(error: IdempotencyError, *, tool_call_id: str, session_id: str) -> ToolMessage: """An idempotency refusal, explained to the model. Nothing ran.""" message = IDEMPOTENCY_MESSAGES.get(error.code or "", f"Nothing ran. The server refused the call: {error.detail}") diff --git a/packages/langchain-submilli/tests/integration_tests/conftest.py b/packages/langchain-submilli/tests/integration_tests/conftest.py index 34b977a..fa069ea 100644 --- a/packages/langchain-submilli/tests/integration_tests/conftest.py +++ b/packages/langchain-submilli/tests/integration_tests/conftest.py @@ -61,7 +61,7 @@ def _free_port() -> int: class ManagedServer: """A submilli-server in a temporary home, which a test may restart.""" - def __init__(self, home: Path) -> None: + def __init__(self, home: Path, *, durable: bool = False) -> None: server_bin = _binary("submilli-server") if not server_bin: pytest.skip("submilli-server not found (set SUBMILLI_BIN)") @@ -72,7 +72,10 @@ def __init__(self, home: Path) -> None: self.token = secrets.token_hex(32) self.process: subprocess.Popen[bytes] | None = None (home / "store.key").write_text(__import__("base64").b64encode(os.urandom(32)).decode()) - (home / "server.yaml").write_text(f"secret_store:\n key_file: {home / 'store.key'}\n") + config = f"secret_store:\n key_file: {home / 'store.key'}\n" + if durable: + config += "durable_execution: true\nmax_durable_runs: 2\n" + (home / "server.yaml").write_text(config) self.env = { **os.environ, "SUBMILLI_HOME": str(home), @@ -120,8 +123,8 @@ def apply(self, blueprint: str) -> None: @pytest.fixture -def managed(tmp_path: Path) -> Iterator[ManagedServer]: - server = ManagedServer(tmp_path) +def managed(tmp_path: Path, request: pytest.FixtureRequest) -> Iterator[ManagedServer]: + server = ManagedServer(tmp_path, durable=getattr(request, "param", False)) server.start() try: yield server diff --git a/packages/langchain-submilli/tests/integration_tests/test_app_tools.py b/packages/langchain-submilli/tests/integration_tests/test_app_tools.py index 9f2912a..2ddb43d 100644 --- a/packages/langchain-submilli/tests/integration_tests/test_app_tools.py +++ b/packages/langchain-submilli/tests/integration_tests/test_app_tools.py @@ -24,6 +24,7 @@ BLUEPRINT = """kind: blueprint name: shop +vfs: none variables: userId: required: true diff --git a/packages/langchain-submilli/tests/integration_tests/test_durable.py b/packages/langchain-submilli/tests/integration_tests/test_durable.py new file mode 100644 index 0000000..1d4d982 --- /dev/null +++ b/packages/langchain-submilli/tests/integration_tests/test_durable.py @@ -0,0 +1,74 @@ +"""Durable tool callbacks through an agent, including approvals across restarts.""" + +from __future__ import annotations + +import asyncio +from typing import Any, cast + +import pytest + +from tests.integration_tests.conftest import ManagedServer +from tests.integration_tests.test_app_tools import THREAD, agent, body, decide, first_tool, sent, start +from tests.scripted import execute + +pytestmark = [pytest.mark.integration, pytest.mark.durable, pytest.mark.parametrize("managed", [True], indirect=True)] + +PROGRAM = """import tools from "@tools"; +function main(): string { + const first = tools.send_email({to: "first@x", body: "one"}); + const second = tools.send_email({to: "second@x", body: "two"}); + return first + ";" + second; +} +""" + + +@pytest.fixture(autouse=True) +def clear_sent() -> None: + sent.clear() + + +@pytest.mark.parametrize("decision", ["approve", "reject"]) +async def test_approval_survives_restart_without_repeating_an_earlier_tool( + managed: ManagedServer, decision: str +) -> None: + graph = agent(managed, [execute(PROGRAM)], "done", interrupt_on={"send_email": True}) + paused = await start(graph) + assert paused["__interrupt__"] and sent == [] + paused = await decide(graph, {"type": "approve"}) + assert paused["__interrupt__"] + assert sent == [{"to": "first@x", "body": "one"}] + session = (await graph.aget_state(THREAD)).values["submilli_session"] + managed.restart() + result = await decide(graph, {"type": decision}) + message = first_tool(result) + if decision == "approve": + assert body(message)["result"] == "sent to first@x;sent to second@x" + assert sent == [{"to": "first@x", "body": "one"}, {"to": "second@x", "body": "two"}] + else: + assert body(message)["denied"] is True + assert body(message)["error"]["source"] == "harness" + assert sent == [{"to": "first@x", "body": "one"}] + assert (await graph.aget_state(THREAD)).values["submilli_session"] == session + assert len(message.artifact["tool_calls"]) == 2 + + +async def test_concurrent_agents_keep_durable_tool_calls_separate(managed: ManagedServer) -> None: + source = 'import tools from "@tools"; function main(): string { return tools.send_email({to: "TO", body: "hi"}); }' + + # Each agent gets its own scripted model iterator and middleware instance. + # This also checks several clients driving the same server simultaneously. + async def independent(number: int) -> dict[str, Any]: + independent_graph = agent(managed, [execute(source.replace("TO", f"user-{number}@x"))], "done") + return cast( + "dict[str, Any]", + await independent_graph.ainvoke( + {"messages": [{"role": "user", "content": "go"}]}, + {"configurable": {"thread_id": f"t-{number}"}}, + context={"user_id": f"u-{number}"}, + ), + ) + + results = await asyncio.wait_for(asyncio.gather(*(independent(i) for i in range(12))), timeout=60) + assert sorted(item["to"] for item in sent) == sorted(f"user-{i}@x" for i in range(12)) + assert len({first_tool(result).artifact["execution_id"] for result in results}) == 12 + assert all(body(first_tool(result))["result"].startswith("sent to user-") for result in results) diff --git a/packages/langchain-submilli/tests/unit_tests/fake_server.py b/packages/langchain-submilli/tests/unit_tests/fake_server.py index 071ab23..2c5e73a 100644 --- a/packages/langchain-submilli/tests/unit_tests/fake_server.py +++ b/packages/langchain-submilli/tests/unit_tests/fake_server.py @@ -72,6 +72,10 @@ class FakeServer: programs: dict[str, list[tuple[str, dict[str, Any]]]] = field(default_factory=dict) suspended: dict[str, Run] = field(default_factory=dict) open_warnings: list[str] = field(default_factory=list) + # As a server with `durable_execution` answers: the next executes answer + # these run views, and `GET /v1/runs/{id}` answers `views[id]`, in turn. + durable_next: list[dict[str, Any]] = field(default_factory=list) + views: dict[str, list[dict[str, Any]]] = field(default_factory=dict) def install(self, router: respx.MockRouter) -> None: router.route(host="submilli.test").mock(side_effect=self.handle) @@ -93,6 +97,8 @@ def handle(self, request: httpx.Request) -> httpx.Response: if match := re.fullmatch(r"/v1/runs/([^/]+)/resume", path): return self.resume(match[1], body) if match := re.fullmatch(r"/v1/runs/([^/]+)", path): + if self.views.get(match[1]): + return _json(200, self.views[match[1]].pop(0)) run = self.suspended.get(match[1]) if run is None: return _json(404, {"error": "unknown_run", "execution_id": match[1], "message": "no such run"}) @@ -138,6 +144,8 @@ def execute(self, session_id: str, body: dict[str, Any], key: str | None) -> htt if self.refuse_next: status, refusal = self.refuse_next.pop(0) return _json(status, {**refusal, "session_id": session_id}) + if self.durable_next: + return _json(200, {**self.durable_next.pop(0), "session_id": session_id}) code = body["code"] if key is not None and key in session.ledger: recorded_code, recorded = session.ledger[key] diff --git a/packages/langchain-submilli/tests/unit_tests/test_app_tools.py b/packages/langchain-submilli/tests/unit_tests/test_app_tools.py index 3d9975f..10f94ff 100644 --- a/packages/langchain-submilli/tests/unit_tests/test_app_tools.py +++ b/packages/langchain-submilli/tests/unit_tests/test_app_tools.py @@ -7,6 +7,7 @@ from collections.abc import Iterator from typing import Annotated, Any, TypedDict, cast +import httpx import pytest import respx from langchain.agents import AgentState, create_agent @@ -210,6 +211,57 @@ async def test_a_program_call_runs_the_tool_with_the_runs_context_and_returns_it ] +@pytest.mark.parametrize("approval", [False, True]) +async def test_durable_queues_recovery_and_tool_approvals_wait_for_the_final_result( + server: FakeServer, monkeypatch: pytest.MonkeyPatch, approval: bool +) -> None: + server.programs["mail"] = [("send_email", {"to": "a@x", "body": "hi"})] + original_execute = server.execute + original_resume = server.resume + started = False + interrupted = False + + def queued_execute(session_id: str, body: dict[str, Any], key: str | None) -> httpx.Response: + nonlocal started + response = original_execute(session_id, body, key) + if not started: + started = True + pending = response.json() + run_id = pending["execution_id"] + queued = {"status": "queued", "execution_id": run_id, "session_id": session_id} + server.views[run_id] = [queued, {**queued, "status": "running"}, pending] + return httpx.Response(200, json=queued) + return response + + def recovering_resume(run_id: str, body: dict[str, Any]) -> httpx.Response: + nonlocal interrupted + if not interrupted: + interrupted = True + # The answer has not committed; recovery exposes the same call. + return httpx.Response(409, json={"error": "run_not_suspended"}) + done = original_resume(run_id, body).json() + working = {"status": "running", "execution_id": run_id, "session_id": done["session_id"]} + server.views[run_id] = [working, {**working, "status": "completed", "response": done}] + return httpx.Response(200, json=working) + + monkeypatch.setattr(server, "execute", queued_execute) + monkeypatch.setattr(server, "resume", recovering_resume) + graph = agent( + calls_then_done(execute("mail")), + middleware(ptc=[send_email], interrupt_on={"send_email": True} if approval else None), + ) + result = await start(graph) + if approval: + assert result["__interrupt__"] and ran == [] + result = await decide(graph, {"type": "approve"}) + assert ran == [("send_email", {"to": "a@x", "body": "hi"})] + message = executed(result) + assert message.status == "success" + assert json.loads(message.text)["result"] == json.dumps(["sent to a@x"]) + assert len(message.artifact["tool_calls"]) == 1 + assert server.total_runs() == 1 + + async def test_a_tool_that_raises_fails_the_call_in_the_program(server: FakeServer) -> None: server.programs["boom"] = [("broken", {})] result = await start(agent(calls_then_done(execute("boom")), middleware(ptc=[broken]))) @@ -251,6 +303,26 @@ def restart_then_resume(run_id: str, body: dict[str, Any]) -> Any: assert message.artifact["refusal"] == "unknown_run" +async def test_a_run_that_stops_while_waiting_on_a_tool_is_reported_without_raising(server: FakeServer) -> None: + server.programs["orders"] = [("query_orders", {"customer": "c_42"})] + + def close_session(run_id: str, body: dict[str, Any]) -> httpx.Response: + # A durable server answers a resume of a cancelled run with its view. + view = {"execution_id": run_id, "session_id": "s", "status": "cancelled", "reason": "session_closed"} + return httpx.Response(200, json={**view, "durable": True, "revision": 4}) + + server.resume = close_session # type: ignore[method-assign] + result = await start(agent(calls_then_done(execute("orders")), middleware(ptc=[query_orders]))) + + message = executed(result) + error = json.loads(message.text)["error"] + assert (error["kind"], error["reason"]) == ("run_cancelled", "session_closed") + assert message.status == "error" + assert message.artifact["run_status"] == "cancelled" + [record] = message.artifact["tool_calls"] + assert record["tool"] == "query_orders" + + async def test_a_changed_tool_catalog_opens_a_new_session(server: FakeServer) -> None: graph = agent(calls_then_done(execute("a")), middleware(ptc=[query_orders])) await start(graph) diff --git a/packages/langchain-submilli/tests/unit_tests/test_middleware.py b/packages/langchain-submilli/tests/unit_tests/test_middleware.py index 4d08173..7dd0ba5 100644 --- a/packages/langchain-submilli/tests/unit_tests/test_middleware.py +++ b/packages/langchain-submilli/tests/unit_tests/test_middleware.py @@ -227,6 +227,7 @@ async def test_lost_secrets_are_sent_again_with_rebind(server: FakeServer) -> No [ (409, "idempotency_conflict", "different code"), (409, "idempotency_incomplete", "side effects are unknown"), + (409, "idempotency_response_expired", "no longer keeps its result"), (503, "idempotency_in_progress", "still running"), (503, "idempotency_unavailable", "could not read or write"), ], @@ -249,6 +250,63 @@ async def test_incomplete_tells_the_model_to_check_before_retrying(server: FakeS assert "Check what it would have changed" in body["error"]["message"] +def lifecycle(status: str, reason: str | None = None, **extra: Any) -> dict[str, Any]: + """A durable run's lifecycle view, as submilli-server answers it.""" + return { + "execution_id": "run-1", + "status": status, + "reason": reason, + "durable": True, + "revision": 2, + "attempt": 1, + "admission_order": 1, + "created_at_unix_ms": 1_761_000_000_000, + "completed_at_unix_ms": None, + **extra, + } + + +@pytest.mark.parametrize( + ("status", "reason", "kind", "phrase"), + [ + ("stuck", "credentials_unavailable", "run_stuck", "until an operator recovers it"), + ("cancelled", "session_closed", "run_cancelled", "its session was closed"), + ("aborted", "never_dispatched", "run_aborted", "nothing ran"), + ("indeterminate", "interrupted_without_history", "run_indeterminate", "some, all, or none"), + ("expired", None, "run_result_unavailable", "no longer keeps its result"), + ("completed", None, "run_result_unavailable", "no longer keeps its result"), + ("paused", "operator", "run_state_unknown", "does not know"), + ], +) +async def test_a_durable_run_without_a_result_is_reported_not_raised( + server: FakeServer, status: str, reason: str | None, kind: str, phrase: str +) -> None: + server.durable_next.append(lifecycle(status, reason)) + result = await run(agent(calls_then_done(execute("code")), middleware())) + + message = tool_message(result) + assert message.status == "error" + error = content(message)["error"] + assert (error["kind"], error["status"], error.get("reason")) == (kind, status, reason) + assert phrase in error["message"] + assert content(message)["result"] is None + assert message.artifact["execution_id"] == "run-1" + assert message.artifact["run_status"] == status + assert result["messages"][-1].content == "done" + + +async def test_a_queued_durable_run_is_waited_for(server: FakeServer) -> None: + server.durable_next.append(lifecycle("queued")) + retained = {"status": "finished", "execution_id": "run-1", "session_id": "s", "result": "42", "console": []} + server.views["run-1"] = [lifecycle("running"), lifecycle("completed", response={**retained, "error": None})] + result = await run(agent(calls_then_done(execute("code")), middleware())) + + message = tool_message(result) + assert content(message) == {"result": "42", "console": [], "error": None, "denied": False} + assert message.status == "success" + assert len(server.calls("GET", "/v1/runs/run-1")) == 2 + + async def test_the_same_tool_call_replays(server: FakeServer) -> None: submilli = middleware() graph = agent( diff --git a/packages/submilli/CHANGELOG.md b/packages/submilli/CHANGELOG.md index c15440d..1271c7e 100644 --- a/packages/submilli/CHANGELOG.md +++ b/packages/submilli/CHANGELOG.md @@ -6,6 +6,31 @@ All notable changes to this project are recorded here. The format follows ## [Unreleased] +### Added + +- `RunStateError`, raised for a run with no result to return. Servers with + `durable_execution` report runs that are `stuck`, `cancelled`, `aborted`, + `indeterminate`, `expired`, or `completed` without a kept result. A status + the SDK does not know raises it too. It carries the run's `execution_id`, + `session_id`, `status`, `reason`, and the `body` the server sent. + +### Changed + +- `RunningRun.status` is `"queued"` or `"running"`. + +### Fixed + +- Recovery exposing the same uncommitted tool call during `runs.complete` + resends its cached answer instead of executing the Python tool again. +- An expired idempotency response raises `IdempotencyError` with code + `idempotency_response_expired`. +- Runs on a server with `durable_execution` no longer read as finished runs + with no result. A `queued` run is a `RunningRun`, which `runs.complete` and + `sessions.run` poll as they poll a `running` one. A `completed` run from + `GET /v1/runs/{id}` returns the `ExecuteResult` kept in its `response`. +- `GET /v1/runs/{id}` answers an unknown run with a plain-text `404`, which now + raises `UnknownRunError` as the JSON `unknown_run` does. + ## [0.1.1] - 2026-10-15 ### Changed diff --git a/packages/submilli/README.md b/packages/submilli/README.md index c9d6ed4..4c21732 100644 --- a/packages/submilli/README.md +++ b/packages/submilli/README.md @@ -111,6 +111,13 @@ else: print(result.result, result.console) ``` +**Durable runs.** On a server with `durable_execution`, an execute still queued or working after +the server's wait answers `RunningRun`, which `sessions.run` and `runs.complete` poll. A run that +stops without a result, such as one that is `stuck` or `cancelled`, raises `RunStateError`. +Polling continues until a result or a stopped state arrives; cancel the async task if your +application needs an overall deadline. `sessions.execute`, `runs.retrieve`, and `runs.resume` +return the current state; use `sessions.run` or `runs.complete` to wait for completion. + **Idempotency.** Pass an `idempotency_key` with every execute that may be retried. When the same request reaches the server twice, the program runs once and the retry gets the recorded result. `idempotency_key(call_id)` turns any id, such as a model's tool call id, into a key the @@ -163,8 +170,14 @@ function main(): string { as a Blueprint that grants no `tools.call`, or a tool whose result is typed `unknown`. The server doesn't resolve `$ref`, so pass Pydantic schemas through `submilli.schema.inline_refs`. - **Lower level.** `runs.resume`, `runs.retrieve`, and `runs.complete` drive a run yourself. -- **Restarts.** A suspended run is lost if the server restarts, and resuming it raises - `UnknownRunError`. +- **Restarts.** With durable execution enabled, the server recovers suspended runs and + committed tool answers after a restart, preserving call IDs. Without durability, a + suspended run is lost and resuming it raises `UnknownRunError`. +- **Tool retries.** During one `runs.complete` invocation, a repeated call ID reuses the + answer already produced, including when recovery exposes an uncommitted call again. + This cache does not survive a Python process restart. Tools with external side effects + should persistently deduplicate by `PendingCall.call_id`; server durability cannot + make an external effect and its answer commit atomically. ## Discovery @@ -193,6 +206,7 @@ Every error is a `SubmilliError`: | `SessionRequiresSecretsError` | The server asks for the session's harness secrets again. `rebind` them; `.required` names them | | `IdempotencyError` | The server refused the key and ran nothing. `.code` says why | | `UnknownRunError` | The server no longer holds a suspended run | +| `RunStateError` | The run has no result to return: it is `stuck`, `cancelled`, `aborted` or `indeterminate`, its result expired, or the SDK doesn't know its status. `.status` and `.reason` say which | | `RunNotSuspendedError`, `UnknownCallError` | A resume that no longer applies. `runs.complete` handles these | | `UnknownBlueprintError` | No Blueprint of that name | | `BadRequestError`, `ForbiddenError`, `NotFoundError`, `ConflictError`, `ServiceUnavailableError`, `InternalServerError` | Other refusals, by status | diff --git a/packages/submilli/pyproject.toml b/packages/submilli/pyproject.toml index 2eac292..eb57ab8 100644 --- a/packages/submilli/pyproject.toml +++ b/packages/submilli/pyproject.toml @@ -51,7 +51,10 @@ include = ["src", "tests", "README.md", "CHANGELOG.md", "LICENSE"] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" testpaths = ["tests/unit"] -markers = ["integration: starts a submilli-server from $SUBMILLI_BIN or PATH"] +markers = [ + "integration: starts a submilli-server from $SUBMILLI_BIN or PATH", + "durable: requires a runtime build with durable execution and tool callbacks", +] filterwarnings = ["error"] [tool.mypy] diff --git a/packages/submilli/src/submilli/__init__.py b/packages/submilli/src/submilli/__init__.py index 66d97c1..93d81a4 100644 --- a/packages/submilli/src/submilli/__init__.py +++ b/packages/submilli/src/submilli/__init__.py @@ -24,6 +24,7 @@ InternalServerError, NotFoundError, RunNotSuspendedError, + RunStateError, ServiceUnavailableError, SessionRequiresSecretsError, SubmilliError, @@ -75,6 +76,7 @@ "RunError", "RunNotSuspendedError", "RunState", + "RunStateError", "RunningRun", "Runs", "ServiceUnavailableError", diff --git a/packages/submilli/src/submilli/_async_client.py b/packages/submilli/src/submilli/_async_client.py index 2538117..8318bdf 100644 --- a/packages/submilli/src/submilli/_async_client.py +++ b/packages/submilli/src/submilli/_async_client.py @@ -171,7 +171,10 @@ async def execute(self, session_id: str, code: str, *, idempotency_key: str | No In a session with application tools a run may answer `SuspendedRun`; `run` answers its calls for you. With an `idempotency_key`, a repeated - request answers the same run instead of running the program again. + request answers the same run instead of running the program again. On a + server with `durable_execution`, a run still queued or working after the + server's wait answers `RunningRun`, and a run with no result to return + raises `RunStateError`. """ return b.read_run_state(await self._client._send(b.execute(session_id, code, idempotency_key))) @@ -187,7 +190,8 @@ async def run( An exception from `on_tool_call` propagates and leaves the run waiting on the server. Executing again with the same `idempotency_key` picks it - up at the call it waits on. + up at the call it waits on. Raises `RunStateError` when the run has no + result to return, such as a durable run that is `stuck`. """ state = await self.execute(session_id, code, idempotency_key=idempotency_key) return await self._client.runs.complete(state, on_tool_call=on_tool_call) @@ -223,12 +227,17 @@ async def resume(self, run_id: str, results: Sequence[ToolResult]) -> RunState: """`POST /v1/runs/{id}/resume`: answer the run's pending calls, then wait for its next state. Answering a call twice changes nothing. Raises `UnknownRunError`, - `RunNotSuspendedError` and `UnknownCallError`. + `RunNotSuspendedError` and `UnknownCallError`, and `RunStateError` for + a run with no result to return. """ return b.read_run_state(await self._client._send(b.resume(run_id, results))) async def retrieve(self, run_id: str) -> RunState: - """`GET /v1/runs/{id}`: the run's state now. Raises `UnknownRunError`.""" + """`GET /v1/runs/{id}`: the run's state now. + + Raises `UnknownRunError`, and `RunStateError` for a run with no result + to return, such as a durable run that is `stuck`. + """ return b.read_run_state(await self._client._send(b.retrieve_run(run_id))) async def complete( @@ -242,10 +251,15 @@ async def complete( """Drive a run from `state` until it finishes. Each pending call goes to `on_tool_call`, sync or async, and its - `ToolResult` goes back with `resume`. A run that is `running` is polled. - Raises `UnknownRunError` when the server lost the run, as after a restart. + `ToolResult` goes back with `resume`. A run that is `running` or + `queued` is polled. Raises `UnknownRunError` when the server lost the + run, as after a restart, and `RunStateError` when the run has no result + to return, such as a durable run that is `stuck` or `cancelled`. """ delay = poll_interval + # Recovery can expose an unanswered call again if a resume was delivered + # but not committed. Resend its answer without repeating the host effect. + answered: dict[str, tuple[PendingCall, ToolResult]] = {} while not isinstance(state, ExecuteResult): if isinstance(state, RunningRun): await asyncio.sleep(delay) @@ -259,11 +273,18 @@ async def complete( ) results = [] for call in state.calls: + previous = answered.get(call.call_id) + if previous is not None: + if previous[0] != call: + raise SubmilliError(f"run changed the tool or arguments of call {call.call_id!r}") + results.append(previous[1]) + continue result = on_tool_call(call) if inspect.isawaitable(result): result = await result if not isinstance(result, ToolResult) or result.call_id != call.call_id: raise SubmilliError(f"on_tool_call must return a ToolResult for call {call.call_id!r}") + answered[call.call_id] = (call, result) results.append(result) try: state = await self.resume(state.execution_id, results) diff --git a/packages/submilli/src/submilli/_client.py b/packages/submilli/src/submilli/_client.py index d26c47f..2572028 100644 --- a/packages/submilli/src/submilli/_client.py +++ b/packages/submilli/src/submilli/_client.py @@ -162,7 +162,10 @@ def execute(self, session_id: str, code: str, *, idempotency_key: str | None = N In a session with application tools a run may answer `SuspendedRun`; `run` answers its calls for you. With an `idempotency_key`, a repeated - request answers the same run instead of running the program again. + request answers the same run instead of running the program again. On a + server with `durable_execution`, a run still queued or working after the + server's wait answers `RunningRun`, and a run with no result to return + raises `RunStateError`. """ return b.read_run_state(self._client._send(b.execute(session_id, code, idempotency_key))) @@ -178,7 +181,8 @@ def run( An exception from `on_tool_call` propagates and leaves the run waiting on the server. Executing again with the same `idempotency_key` picks it - up at the call it waits on. + up at the call it waits on. Raises `RunStateError` when the run has no + result to return, such as a durable run that is `stuck`. """ state = self.execute(session_id, code, idempotency_key=idempotency_key) return self._client.runs.complete(state, on_tool_call=on_tool_call) @@ -214,12 +218,17 @@ def resume(self, run_id: str, results: Sequence[ToolResult]) -> RunState: """`POST /v1/runs/{id}/resume`: answer the run's pending calls, then wait for its next state. Answering a call twice changes nothing. Raises `UnknownRunError`, - `RunNotSuspendedError` and `UnknownCallError`. + `RunNotSuspendedError` and `UnknownCallError`, and `RunStateError` for + a run with no result to return. """ return b.read_run_state(self._client._send(b.resume(run_id, results))) def retrieve(self, run_id: str) -> RunState: - """`GET /v1/runs/{id}`: the run's state now. Raises `UnknownRunError`.""" + """`GET /v1/runs/{id}`: the run's state now. + + Raises `UnknownRunError`, and `RunStateError` for a run with no result + to return, such as a durable run that is `stuck`. + """ return b.read_run_state(self._client._send(b.retrieve_run(run_id))) def complete( @@ -233,10 +242,15 @@ def complete( """Drive a run from `state` until it finishes. Each pending call goes to `on_tool_call`, and its - `ToolResult` goes back with `resume`. A run that is `running` is polled. - Raises `UnknownRunError` when the server lost the run, as after a restart. + `ToolResult` goes back with `resume`. A run that is `running` or + `queued` is polled. Raises `UnknownRunError` when the server lost the + run, as after a restart, and `RunStateError` when the run has no result + to return, such as a durable run that is `stuck` or `cancelled`. """ delay = poll_interval + # Recovery can expose an unanswered call again if a resume was delivered + # but not committed. Resend its answer without repeating the host effect. + answered: dict[str, tuple[PendingCall, ToolResult]] = {} while not isinstance(state, ExecuteResult): if isinstance(state, RunningRun): time.sleep(delay) @@ -250,9 +264,16 @@ def complete( ) results = [] for call in state.calls: + previous = answered.get(call.call_id) + if previous is not None: + if previous[0] != call: + raise SubmilliError(f"run changed the tool or arguments of call {call.call_id!r}") + results.append(previous[1]) + continue result = on_tool_call(call) if not isinstance(result, ToolResult) or result.call_id != call.call_id: raise SubmilliError(f"on_tool_call must return a ToolResult for call {call.call_id!r}") + answered[call.call_id] = (call, result) results.append(result) try: state = self.resume(state.execution_id, results) diff --git a/packages/submilli/src/submilli/_errors.py b/packages/submilli/src/submilli/_errors.py index 02159ba..cbd241c 100644 --- a/packages/submilli/src/submilli/_errors.py +++ b/packages/submilli/src/submilli/_errors.py @@ -6,11 +6,13 @@ the status class (`UnknownSessionError` is a `NotFoundError`). A program that fails, or that the Blueprint denies, is not an error: it is an -`ExecuteResult` whose `error` is set. +`ExecuteResult` whose `error` is set. A run with no result to return, such as +a durable run that is `stuck`, is a `RunStateError`. """ from __future__ import annotations +from collections.abc import Mapping from typing import Any import httpx @@ -27,6 +29,7 @@ "InternalServerError", "NotFoundError", "RunNotSuspendedError", + "RunStateError", "ServiceUnavailableError", "SessionRequiresSecretsError", "SubmilliError", @@ -129,6 +132,50 @@ class UnknownCallError(BadRequestError): """`400 unknown_call`: the run waits on another call.""" +# What each status without a result means, as servers with `durable_execution` +# report them in a run's lifecycle view. +_RUN_STATES = { + "stuck": "it stopped, and waits to be recovered with POST /v1/runs/{run}/recover", + "cancelled": "it was cancelled before it finished", + "aborted": "it ended before the program started", + "indeterminate": "it was interrupted, and its outcome is unknown", + "expired": "it finished, but the server no longer keeps its result", + "completed": "it finished, but the server keeps no result for it", +} + + +class RunStateError(SubmilliError): + """The server reports a run in a state with no result to return. + + Servers with `durable_execution` report these: `stuck` (stopped, and held + for `POST /v1/runs/{id}/recover`), `cancelled` (its session closed), + `aborted` (it never started), `indeterminate` (interrupted, outcome + unknown), `expired` (its result is no longer kept), and `completed` without + a kept result. A status this SDK does not know raises it too. + + Attributes: + execution_id: The run. + session_id: The session it ran in. + status: The run's `status`, such as `stuck`. + reason: Why, when the server says, such as `credentials_unavailable`. + body: The run as the server sent it, with lifecycle metadata such as `revision`. + """ + + def __init__(self, body: Mapping[str, Any]) -> None: + self.body: dict[str, Any] = dict(body) + self.execution_id: str = str(body.get("execution_id", "")) + self.session_id: str = str(body.get("session_id", "")) + self.status: str = str(body.get("status", "")) + reason = body.get("reason") + self.reason: str | None = reason if isinstance(reason, str) else None + meaning = _RUN_STATES.get(self.status, "this version of the SDK does not know that status") + because = f" ({self.reason})" if self.reason else "" + super().__init__( + f"run {self.execution_id} is {self.status}{because} and has no result to return: " + + meaning.format(run=self.execution_id) + ) + + class IdempotencyError(APIStatusError): """The server refused an `Idempotency-Key`, and ran nothing. @@ -136,7 +183,8 @@ class IdempotencyError(APIStatusError): `idempotency_incomplete` (an earlier attempt never recorded an outcome, so its effects are unknown), `idempotency_in_progress` (an earlier attempt is still running), `idempotency_unavailable` (the record could not be read or - written), or `idempotency_key_invalid`. + written), `idempotency_response_expired` (the earlier result is no longer + retained), or `idempotency_key_invalid`. """ @property @@ -151,6 +199,7 @@ def detail(self) -> str: "idempotency_in_progress", "idempotency_unavailable", "idempotency_key_invalid", + "idempotency_response_expired", } ) @@ -183,6 +232,9 @@ def status_error(response: httpx.Response) -> APIStatusError: return _BY_CODE[(response.status_code, code)](response.status_code, body) if response.status_code == 404 and code == "not_found" and "blueprint" in (_field(body, "message") or ""): return UnknownBlueprintError(response.status_code, body) + if response.status_code == 404 and body == "unknown run": + # `GET /v1/runs/{id}` answers a run it holds no record of in plain text. + return UnknownRunError(response.status_code, body) cls = _BY_STATUS.get(response.status_code) if cls is None: cls = InternalServerError if response.status_code >= 500 else APIStatusError diff --git a/packages/submilli/src/submilli/_types.py b/packages/submilli/src/submilli/_types.py index 97473f4..1c06d4c 100644 --- a/packages/submilli/src/submilli/_types.py +++ b/packages/submilli/src/submilli/_types.py @@ -5,6 +5,8 @@ from dataclasses import dataclass, field from typing import Any, Literal, NotRequired, TypeAlias, TypedDict +from submilli._errors import RunStateError + __all__ = [ "BlueprintPrompt", "CreatedSession", @@ -77,7 +79,11 @@ def _from_json(cls, body: dict[str, Any]) -> RunError: @dataclass(frozen=True, slots=True) class ExecuteResult: - """A finished run (`status: finished`).""" + """A finished run (`status: finished`). + + On a server with `durable_execution`, `GET /v1/runs/{id}` reports a finished + run as `completed`, with this result in its `response`. + """ execution_id: str session_id: str @@ -133,18 +139,27 @@ class SuspendedRun: @dataclass(frozen=True, slots=True) class RunningRun: - """A run at work (`status: running`). Only `runs.retrieve` answers it.""" + """A run at work (`status: running`), or waiting for its turn in the session (`status: queued`). + + `runs.retrieve` answers it, and on a server with `durable_execution` so does + an execute still queued or working after the server's wait. `runs.complete` + polls it. + """ execution_id: str session_id: str - status: Literal["running"] = "running" + status: Literal["queued", "running"] = "running" RunState: TypeAlias = ExecuteResult | SuspendedRun | RunningRun def run_state(body: dict[str, Any]) -> RunState: - """A run's state from an execute, resume or run body. A body without `status` is finished.""" + """A run's state from an execute, resume or run body. A body without `status` is finished. + + Raises `RunStateError` for a run with no result to return, such as a durable + run that is `stuck`, and for a status this SDK does not know. + """ status = body.get("status", "finished") execution_id = str(body.get("execution_id", "")) session_id = str(body.get("session_id", "")) @@ -161,7 +176,14 @@ def run_state(body: dict[str, Any]) -> RunState: return SuspendedRun(execution_id=execution_id, session_id=session_id, calls=calls) if status == "running": return RunningRun(execution_id=execution_id, session_id=session_id) - return ExecuteResult._from_json(body) + if status == "queued": + return RunningRun(execution_id=execution_id, session_id=session_id, status="queued") + if status == "finished": + return ExecuteResult._from_json(body) + response = body.get("response") + if status == "completed" and isinstance(response, dict) and response.get("status", "finished") == "finished": + return ExecuteResult._from_json({"execution_id": execution_id, "session_id": session_id, **response}) + raise RunStateError(body) @dataclass(frozen=True, slots=True) diff --git a/packages/submilli/tests/integration/conftest.py b/packages/submilli/tests/integration/conftest.py index ff3184a..7338a5b 100644 --- a/packages/submilli/tests/integration/conftest.py +++ b/packages/submilli/tests/integration/conftest.py @@ -40,7 +40,7 @@ def _free_port() -> int: class ManagedServer: """A submilli-server in a temporary home.""" - def __init__(self, home: Path, binary: str) -> None: + def __init__(self, home: Path, binary: str, *, durable: bool = False) -> None: self.binary = binary self.home = home self.port = _free_port() @@ -49,7 +49,10 @@ def __init__(self, home: Path, binary: str) -> None: self.process: subprocess.Popen[bytes] | None = None self.log: IO[bytes] | None = None self.rotate_store_key() - (home / "server.yaml").write_text(f"secret_store:\n key_file: {home / 'store.key'}\n") + config = f"secret_store:\n key_file: {home / 'store.key'}\n" + if durable: + config += "durable_execution: true\nmax_durable_runs: 2\n" + (home / "server.yaml").write_text(config) self.env = {**os.environ, "SUBMILLI_HOME": str(home), "SUBMILLI_SERVER_TOKEN": self.token} def rotate_store_key(self) -> None: @@ -86,6 +89,16 @@ def restart(self) -> None: self.stop() self.start() + def crash(self) -> None: + """Stop without draining or committing pending work.""" + assert self.process is not None + self.process.kill() + self.process.wait(timeout=30) + self.process = None + if self.log is not None: + self.log.close() + self.log = None + def apply(self, name: str, yaml: str) -> None: response = httpx.put( f"{self.url}/v1/blueprints/{name}", @@ -97,11 +110,11 @@ def apply(self, name: str, yaml: str) -> None: @pytest.fixture -def managed(tmp_path: Path) -> Iterator[ManagedServer]: +def managed(tmp_path: Path, request: pytest.FixtureRequest) -> Iterator[ManagedServer]: binary = _binary() if binary is None: pytest.skip("submilli-server not found (set SUBMILLI_BIN)") - server = ManagedServer(tmp_path, binary) + server = ManagedServer(tmp_path, binary, durable=getattr(request, "param", False)) server.start() try: yield server diff --git a/packages/submilli/tests/integration/test_durable.py b/packages/submilli/tests/integration/test_durable.py new file mode 100644 index 0000000..115a094 --- /dev/null +++ b/packages/submilli/tests/integration/test_durable.py @@ -0,0 +1,298 @@ +"""Crash recovery, queued admission and callback durability against a real server.""" + +from __future__ import annotations + +import asyncio +from collections import Counter +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import Event, Thread + +import pytest + +from submilli import ( + AsyncSubmilli, + ExecuteResult, + IdempotencyError, + PendingCall, + RunningRun, + RunState, + RunStateError, + Submilli, + SuspendedRun, + ToolDeclaration, + ToolResult, +) +from tests.integration.conftest import ManagedServer + +pytestmark = [pytest.mark.integration, pytest.mark.durable, pytest.mark.parametrize("managed", [True], indirect=True)] + +BLUEPRINT = """name: durable +default: allow +vfs: none +tool_call_timeout: 2m +""" + +ECHO: ToolDeclaration = { + "name": "echo", + "input_schema": {"type": "object", "properties": {"value": {"type": "number"}}, "required": ["value"]}, + "output_schema": {"type": "number"}, +} +ONE = 'import tools from "@tools"; function main(): number { return tools.echo({value: 1}); }' +TWO = """import tools from "@tools"; +function main(): number { const first = tools.echo({value: 1}); return first + tools.echo({value: first}); } +""" + + +async def next_call(client: AsyncSubmilli, state: RunState) -> SuspendedRun: + async with asyncio.timeout(15): + while isinstance(state, RunningRun): + await asyncio.sleep(0.05) + state = await client.runs.retrieve(state.execution_id) + assert isinstance(state, SuspendedRun), state + return state + + +async def test_unanswered_call_and_committed_prefix_survive_two_crashes(managed: ManagedServer) -> None: + managed.apply("durable", BLUEPRINT) + async with AsyncSubmilli(base_url=managed.url, token=managed.token) as client: + session = await client.sessions.create("durable", tools=[ECHO]) + first = await next_call(client, await client.sessions.execute(session.id, TWO, idempotency_key="k")) + [call] = first.calls + managed.crash() + managed.start() + recovered = await next_call(client, await client.runs.retrieve(first.execution_id)) + assert recovered == first + assert await client.sessions.execute(session.id, TWO, idempotency_key="k") == first + second = await next_call( + client, await client.runs.resume(first.execution_id, [ToolResult.success(call.call_id, 6)]) + ) + assert second.calls[0].args == {"value": 6} + assert second.calls[0].call_id != call.call_id + managed.crash() + managed.start() + recovered = await next_call(client, await client.runs.retrieve(first.execution_id)) + assert recovered == second + # A conflicting retry cannot replace a committed answer or answer the next call. + assert await client.runs.resume(first.execution_id, [ToolResult.success(call.call_id, 99)]) == second + done = await client.runs.complete(second, on_tool_call=lambda c: ToolResult.success(c.call_id, 7)) + assert done.error is None and done.result == "13" + managed.restart() + retained = await client.runs.retrieve(first.execution_id) + assert retained == done + assert await client.sessions.run(session.id, TWO, idempotency_key="k") == done + + +async def test_concurrent_duplicate_resumes_commit_one_answer(managed: ManagedServer) -> None: + managed.apply("durable", BLUEPRINT) + async with AsyncSubmilli(base_url=managed.url, token=managed.token) as client: + session = await client.sessions.create("durable", tools=[ECHO]) + first = await next_call(client, await client.sessions.execute(session.id, TWO, idempotency_key="k")) + answer = ToolResult.success(first.calls[0].call_id, 3) + duplicates = await asyncio.gather(*(client.runs.resume(first.execution_id, [answer]) for _ in range(16))) + assert all(state == duplicates[0] for state in duplicates) + second = await next_call(client, duplicates[0]) + assert second.calls[0].args == {"value": 3} + result = await client.runs.complete(second, on_tool_call=lambda c: ToolResult.success(c.call_id, 4)) + assert result.result == "7" and result.error is None + terminal = await asyncio.gather(*(client.runs.resume(first.execution_id, [answer]) for _ in range(8))) + assert all(state == result for state in terminal) + + +async def test_queued_http_response_is_waited_to_completion(managed: ManagedServer) -> None: + managed.apply("durable", BLUEPRINT) + async with AsyncSubmilli(base_url=managed.url, token=managed.token, timeout=45) as client: + session = await client.sessions.create("durable", tools=[ECHO]) + blocker = await next_call(client, await client.sessions.execute(session.id, ONE, idempotency_key="first")) + # The suspended run retains session admission. This POST must return + # real queued metadata after the server's 25-second observation bound. + queued = await client.sessions.execute(session.id, ONE, idempotency_key="second") + assert isinstance(queued, RunningRun) and queued.status == "queued" + assert queued.execution_id != blocker.execution_id + answers: list[str] = [] + + def answer(call: PendingCall) -> ToolResult: + answers.append(call.call_id) + return ToolResult.success(call.call_id, 9) + + waiting = asyncio.create_task(client.runs.complete(queued, on_tool_call=answer, poll_interval=0.01)) + try: + await asyncio.sleep(0.2) + assert not waiting.done() and not answers + await client.runs.complete(blocker, on_tool_call=lambda c: ToolResult.success(c.call_id, 1)) + result = await asyncio.wait_for(waiting, timeout=15) + finally: + waiting.cancel() + await asyncio.gather(waiting, return_exceptions=True) + assert result.error is None and result.result == "9" + assert len(answers) == 1 and answers[0].startswith(f"{queued.execution_id}:") + assert await client.runs.retrieve(result.execution_id) == result + + +async def test_running_http_response_waits_and_retains_the_native_outcome(managed: ManagedServer) -> None: + entered = Event() + release = Event() + hits: list[str] = [] + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + hits.append(self.path) + entered.set() + if not release.wait(timeout=45): + self.send_error(504) + return + self.send_response(200) + self.send_header("Content-Length", "4") + self.end_headers() + self.wfile.write(b"done") + + def log_message(self, format: str, *args: object) -> None: + pass + + upstream = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = Thread(target=upstream.serve_forever, daemon=True) + thread.start() + try: + managed.stop() + config = managed.home / "server.yaml" + config.write_text(config.read_text() + "network:\n allow_localhost: true\n") + managed.start() + managed.apply("durable", BLUEPRINT + "allow_insecure_http: true\n") + code = ( + 'import { get } from "submilli:http"; ' + 'function main(): string { return get("http://127.0.0.1:' + f'{upstream.server_port}/slow").body; }}' + ) + async with AsyncSubmilli(base_url=managed.url, token=managed.token, timeout=40) as client: + session = await client.sessions.create("durable") + state = await client.sessions.execute(session.id, code, idempotency_key="slow-http") + if isinstance(state, ExecuteResult): + pytest.fail(str(state.error)) + assert entered.is_set(), state + assert isinstance(state, RunningRun) and state.status == "running", state + waiting = asyncio.create_task(client.runs.complete(state, poll_interval=0.01)) + try: + await asyncio.sleep(0.1) + assert not waiting.done() + release.set() + done = await asyncio.wait_for(waiting, timeout=10) + finally: + waiting.cancel() + await asyncio.gather(waiting, return_exceptions=True) + assert done.result == "done" and done.error is None + managed.restart() + assert await client.runs.retrieve(done.execution_id) == done + assert await client.sessions.run(session.id, code, idempotency_key="slow-http") == done + assert hits == ["/slow"] + finally: + release.set() + upstream.shutdown() + upstream.server_close() + thread.join(timeout=5) + + +async def test_many_sessions_and_many_tool_calls_remain_isolated(managed: ManagedServer) -> None: + managed.apply("durable", BLUEPRINT) + source = """import tools from "@tools"; +function main(): number { + let total = 0; + for (let i = 0; i < 5; i++) { total += tools.echo({value: i}); } + return total; +} +""" + seen: Counter[str] = Counter() + async with AsyncSubmilli(base_url=managed.url, token=managed.token) as client: + + async def run(number: int) -> ExecuteResult: + session = await client.sessions.create("durable", tools=[ECHO]) + + async def answer(call: PendingCall) -> ToolResult: + seen[call.call_id] += 1 + await asyncio.sleep(0.01) + return ToolResult.success(call.call_id, number + call.args["value"]) + + result = await client.sessions.run(session.id, source, idempotency_key="same-key", on_tool_call=answer) + assert result.result == str(5 * number + 10) and result.error is None + assert await client.sessions.run(session.id, source, idempotency_key="same-key") == result + return result + + results = await asyncio.wait_for(asyncio.gather(*(run(i) for i in range(24))), timeout=60) + assert len({result.execution_id for result in results}) == 24 + assert len(seen) == 120 and set(seen.values()) == {1} + + +@pytest.mark.parametrize("outcome", ["failed", "denied", "bad-schema"]) +async def test_durable_tool_failures_remain_failures_after_restart(managed: ManagedServer, outcome: str) -> None: + managed.apply("durable", BLUEPRINT) + async with AsyncSubmilli(base_url=managed.url, token=managed.token) as client: + session = await client.sessions.create("durable", tools=[ECHO]) + + def answer(call: PendingCall) -> ToolResult: + if outcome == "failed": + return ToolResult.failed(call.call_id, "database down") + if outcome == "denied": + return ToolResult.denied(call.call_id, "user refused") + return ToolResult.success(call.call_id, "not a number") + + result = await client.sessions.run(session.id, ONE, on_tool_call=answer) + assert result.error is not None + assert result.denied == (outcome == "denied") + if outcome == "denied": + assert result.error.source == "harness" + managed.restart() + assert await client.runs.retrieve(result.execution_id) == result + + +async def test_session_deletion_cancels_a_durable_tool_wait(managed: ManagedServer) -> None: + managed.apply("durable", BLUEPRINT) + async with AsyncSubmilli(base_url=managed.url, token=managed.token) as client: + session = await client.sessions.create("durable", tools=[ECHO]) + state = await next_call(client, await client.sessions.execute(session.id, ONE)) + assert await client.sessions.delete(session.id) + for operation in ( + client.runs.retrieve(state.execution_id), + client.runs.resume(state.execution_id, [ToolResult.success(state.calls[0].call_id, 1)]), + ): + with pytest.raises(RunStateError) as raised: + await operation + assert raised.value.status == "cancelled" + + +async def test_expired_result_is_reported_without_starting_another_run(managed: ManagedServer) -> None: + managed.stop() + config = managed.home / "server.yaml" + config.write_text(config.read_text() + "idempotency_retention: 0\nidempotency_keep_last: 1\n") + managed.start() + managed.apply("durable", BLUEPRINT) + code = "function main(): number { return 1; }" + async with AsyncSubmilli(base_url=managed.url, token=managed.token) as client: + session = await client.sessions.create("durable") + first = await client.sessions.run(session.id, code, idempotency_key="first") + assert first.error is None + await client.sessions.run(session.id, code, idempotency_key="second") + # Expiry is performed by the background reaper. Startup triggers its + # first sweep without waiting for the normal reaper interval. + managed.restart() + async with asyncio.timeout(5): + while True: + try: + await client.runs.retrieve(first.execution_id) + except RunStateError as expired: + assert expired.status == "expired" + break + await asyncio.sleep(0.05) + with pytest.raises(IdempotencyError) as refused: + await client.sessions.run(session.id, code, idempotency_key="first") + assert refused.value.code == "idempotency_response_expired" + with pytest.raises(RunStateError) as raised: + await client.runs.retrieve(first.execution_id) + assert raised.value.status == "expired" + + +def test_sync_durable_tools_and_retained_results(managed: ManagedServer) -> None: + managed.apply("durable", BLUEPRINT) + with Submilli(base_url=managed.url, token=managed.token) as client: + session = client.sessions.create("durable", tools=[ECHO]) + result = client.sessions.run(session.id, TWO, on_tool_call=lambda c: ToolResult.success(c.call_id, 2)) + assert result.result == "4" and result.error is None + managed.restart() + assert client.runs.retrieve(result.execution_id) == result diff --git a/packages/submilli/tests/integration/test_server.py b/packages/submilli/tests/integration/test_server.py index 57d35fd..b090212 100644 --- a/packages/submilli/tests/integration/test_server.py +++ b/packages/submilli/tests/integration/test_server.py @@ -168,8 +168,10 @@ async def test_secrets_sealed_with_another_key_are_refused(managed: ManagedServe managed.stop() managed.rotate_store_key() managed.start() - with pytest.raises(ConflictError, match="cannot decrypt session secrets"): + with pytest.raises(ConflictError) as refused: await client.sessions.run(session.id, COUNTER) + assert refused.value.status_code == 409 + assert "session secrets" in str(refused.value) async def test_discovery_and_one_off_runs(client: AsyncSubmilli) -> None: diff --git a/packages/submilli/tests/unit/test_durable.py b/packages/submilli/tests/unit/test_durable.py new file mode 100644 index 0000000..ea2da22 --- /dev/null +++ b/packages/submilli/tests/unit/test_durable.py @@ -0,0 +1,260 @@ +"""Durable lifecycle transitions, retries, and callback replay over mocked HTTP.""" + +from __future__ import annotations + +import asyncio +import json +from collections import Counter +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest +import respx + +from submilli import AsyncSubmilli, ExecuteResult, IdempotencyError, PendingCall, Submilli, SubmilliError, ToolResult +from tests.unit.fake_server import SERVER + + +def view(status: str, run: str = "r", **extra: Any) -> dict[str, Any]: + return {"status": status, "execution_id": run, "session_id": "s", "durable": True, **extra} + + +def suspended(sequence: int, run: str = "r") -> dict[str, Any]: + return view( + "suspended", + run, + calls=[{"call_id": f"{run}:effect-{sequence}", "tool": "echo", "args": {"value": sequence}}], + ) + + +def completed(run: str = "r") -> dict[str, Any]: + return view( + "completed", run, response={"status": "finished", "result": "done", "console": ["saved"], "error": None} + ) + + +async def test_run_waits_through_long_queues_and_recovery_between_tools( + router: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + sleep = AsyncMock() + monkeypatch.setattr("submilli._async_client.asyncio.sleep", sleep) + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=view("queued")) + polls = router.get(f"{SERVER}/v1/runs/r").mock( + side_effect=[ + httpx.Response(200, json=body) + for body in [*[view("queued")] * 40, suspended(1), view("queued"), suspended(2), completed()] + ] + ) + resumes = router.post(f"{SERVER}/v1/runs/r/resume").respond(200, json=view("running")) + calls: list[PendingCall] = [] + + async def answer(call: PendingCall) -> ToolResult: + calls.append(call) + return ToolResult.success(call.call_id, call.args["value"]) + + async with AsyncSubmilli(base_url=SERVER) as client: + result = await client.sessions.run("s", "code", idempotency_key="k", on_tool_call=answer) + assert (result.result, result.console, result.execution_id, result.session_id) == ("done", ["saved"], "r", "s") + assert [call.call_id for call in calls] == ["r:effect-1", "r:effect-2"] + assert polls.call_count == 44 and resumes.call_count == 2 + assert max(args.args[0] for args in sleep.await_args_list) <= 2 + + +def test_sync_run_waits_through_queued_running_and_tools( + router: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("submilli._client.time.sleep", lambda _: None) + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=view("running")) + polls = router.get(f"{SERVER}/v1/runs/r").mock( + side_effect=[ + httpx.Response(200, json=body) for body in [view("queued"), suspended(1), view("running"), completed()] + ] + ) + router.post(f"{SERVER}/v1/runs/r/resume").respond(200, json=view("queued")) + calls: list[str] = [] + + def answer(call: PendingCall) -> ToolResult: + calls.append(call.call_id) + return ToolResult.success(call.call_id, 1) + + with Submilli(base_url=SERVER) as client: + result = client.sessions.run("s", "code", on_tool_call=answer) + assert result.result == "done" and calls == ["r:effect-1"] and polls.call_count == 4 + + +async def test_a_running_request_does_not_finish_until_the_server_finishes(router: respx.MockRouter) -> None: + polling = asyncio.Event() + release = asyncio.Event() + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=view("queued")) + + async def retrieve(_: httpx.Request) -> httpx.Response: + polling.set() + await release.wait() + return httpx.Response(200, json=completed()) + + router.get(f"{SERVER}/v1/runs/r").mock(side_effect=retrieve) + async with AsyncSubmilli(base_url=SERVER) as client: + task = asyncio.create_task(client.sessions.run("s", "code")) + try: + await asyncio.wait_for(polling.wait(), timeout=5) + assert not task.done() + release.set() + assert (await asyncio.wait_for(task, timeout=5)).result == "done" + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def test_polling_can_be_cancelled_without_resubmitting_execution(router: respx.MockRouter) -> None: + entered = asyncio.Event() + post = router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=view("running")) + + async def retrieve(_: httpx.Request) -> httpx.Response: + entered.set() + await asyncio.Event().wait() + raise AssertionError("unreachable") + + router.get(f"{SERVER}/v1/runs/r").mock(side_effect=retrieve) + async with AsyncSubmilli(base_url=SERVER) as client: + task = asyncio.create_task(client.sessions.run("s", "code", idempotency_key="k")) + await asyncio.wait_for(entered.wait(), timeout=5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert post.call_count == 1 + + +@pytest.mark.parametrize("refusal", ["run_not_suspended", "unknown_call"]) +async def test_recovered_uncommitted_call_resends_the_answer_without_running_the_tool_twice( + router: respx.MockRouter, refusal: str +) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=suspended(1)) + router.get(f"{SERVER}/v1/runs/r").respond(200, json=suspended(1)) + resume = router.post(f"{SERVER}/v1/runs/r/resume").mock( + side_effect=[ + httpx.Response(409 if refusal == "run_not_suspended" else 400, json={"error": refusal}), + httpx.Response(200, json=completed()), + ] + ) + calls: list[str] = [] + + def answer(call: PendingCall) -> ToolResult: + calls.append(call.call_id) + return ToolResult.success(call.call_id, len(calls)) + + async with AsyncSubmilli(base_url=SERVER) as client: + result = await client.sessions.run("s", "code", on_tool_call=answer) + assert result.result == "done" + assert calls == ["r:effect-1"] + assert resume.calls[0].request.content == resume.calls[1].request.content + + +async def test_lost_resume_response_retries_the_same_answer(router: respx.MockRouter) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=suspended(1)) + resume = router.post(f"{SERVER}/v1/runs/r/resume").mock( + side_effect=[ + httpx.ReadError("response lost after commit"), + httpx.Response(503, json={"error": "run_unavailable"}, headers={"Retry-After": "0"}), + httpx.Response(200, json=completed()), + ] + ) + seen: list[str] = [] + + def answer(call: PendingCall) -> ToolResult: + seen.append(call.call_id) + return ToolResult.success(call.call_id, 1) + + async with AsyncSubmilli(base_url=SERVER) as client: + assert (await client.sessions.run("s", "code", on_tool_call=answer)).result == "done" + assert seen == ["r:effect-1"] and resume.call_count == 3 + assert len({call.request.content for call in resume.calls}) == 1 + + +def test_sync_recovery_resends_the_same_tool_answer(router: respx.MockRouter) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=suspended(1)) + router.get(f"{SERVER}/v1/runs/r").respond(200, json=suspended(1)) + resume = router.post(f"{SERVER}/v1/runs/r/resume").mock( + side_effect=[ + httpx.Response(409, json={"error": "run_not_suspended"}), + httpx.Response(200, json=completed()), + ] + ) + seen: list[str] = [] + + def answer(call: PendingCall) -> ToolResult: + seen.append(call.call_id) + return ToolResult.success(call.call_id, len(seen)) + + with Submilli(base_url=SERVER) as client: + assert client.sessions.run("s", "code", on_tool_call=answer).result == "done" + assert seen == ["r:effect-1"] + assert resume.calls[0].request.content == resume.calls[1].request.content + + +async def test_recovery_cannot_change_the_arguments_of_an_answered_call(router: respx.MockRouter) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").respond(200, json=suspended(1)) + changed = suspended(1) + changed["calls"][0]["args"] = {"value": 99} + router.get(f"{SERVER}/v1/runs/r").respond(200, json=changed) + router.post(f"{SERVER}/v1/runs/r/resume").respond(409, json={"error": "run_not_suspended"}) + async with AsyncSubmilli(base_url=SERVER) as client: + with pytest.raises(SubmilliError, match="changed the tool or arguments"): + await client.sessions.run("s", "code", on_tool_call=lambda c: ToolResult.success(c.call_id, 1)) + + +async def test_an_expired_idempotency_response_raises_without_reexecution(router: respx.MockRouter) -> None: + execute = router.post(f"{SERVER}/v1/sessions/s/execute").respond( + 409, json={"error": "idempotency_response_expired", "execution_id": "r"} + ) + async with AsyncSubmilli(base_url=SERVER) as client: + with pytest.raises(IdempotencyError) as raised: + await client.sessions.run("s", "code", idempotency_key="original-key") + assert raised.value.code == "idempotency_response_expired" and execute.call_count == 1 + + +async def test_concurrent_runs_keep_polling_and_callback_answers_isolated() -> None: + positions: Counter[str] = Counter() + handlers: Counter[str] = Counter() + submitted: Counter[str] = Counter() + + def transport(request: httpx.Request) -> httpx.Response: + parts = request.url.path.split("/") + if parts[-1] == "execute": + run = json.loads(request.content)["code"] + return httpx.Response(200, json=view("queued", run)) + run = parts[3] + if parts[-1] == "resume": + [answer] = json.loads(request.content)["results"] + assert answer["call_id"] == f"{run}:effect-{submitted[run] + 1}" + assert answer["value"] == {"run": run, "sequence": submitted[run] + 1} + submitted[run] += 1 + return httpx.Response(200, json=view("running", run)) + positions[run] += 1 + if positions[run] % 3: + body = view("queued" if positions[run] % 2 else "running", run) + elif submitted[run] < 8: + body = suspended(submitted[run] + 1, run) + else: + body = completed(run) + return httpx.Response(200, json=body) + + async def answer(call: PendingCall) -> ToolResult: + handlers[call.execution_id] += 1 + await asyncio.sleep(0) + return ToolResult.success(call.call_id, {"run": call.execution_id, "sequence": call.args["value"]}) + + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(transport)) as http, + AsyncSubmilli(base_url=SERVER, http_client=http) as client, + ): + + async def run(number: int) -> ExecuteResult: + state = await client.sessions.execute("s", f"r-{number}", idempotency_key=f"k-{number}") + return await client.runs.complete(state, on_tool_call=answer, poll_interval=0) + + results = await asyncio.wait_for(asyncio.gather(*(run(i) for i in range(64))), timeout=20) + assert len({result.execution_id for result in results}) == 64 + assert all(result.result == "done" for result in results) + assert set(handlers.values()) == set(submitted.values()) == {8} + assert set(positions.values()) == {27} diff --git a/packages/submilli/tests/unit/test_runs.py b/packages/submilli/tests/unit/test_runs.py index b05d40c..ec30301 100644 --- a/packages/submilli/tests/unit/test_runs.py +++ b/packages/submilli/tests/unit/test_runs.py @@ -11,7 +11,10 @@ from submilli import ( AsyncSubmilli, + ExecuteResult, PendingCall, + RunningRun, + RunStateError, Submilli, SubmilliError, SuspendedRun, @@ -162,6 +165,169 @@ async def test_a_timed_out_call_is_looked_up_and_a_running_run_polled(router: re assert poll.call_count == 2 +# Durable runs: the shapes of crates/submilli-server/src/handlers/durable_runs.rs (`view`, `observe`). + + +def lifecycle(status: str, reason: str | None = None, **extra: Any) -> dict[str, Any]: + """A durable run's lifecycle view, as execute, resume and `GET /v1/runs/{id}` answer it.""" + return { + "execution_id": "r", + "session_id": "s", + "status": status, + "reason": reason, + "durable": True, + "revision": 3, + "attempt": 1, + "admission_order": 7, + "created_at_unix_ms": 1_761_000_000_000, + "completed_at_unix_ms": None, + **extra, + } + + +RETAINED = { + "status": "finished", + "execution_id": "r", + "session_id": "s", + "result": None, + "console": ["before"], + "error": {"kind": "runtime_error", "message": "Error: boom"}, +} + +# Each run state with no result to return: (view, status, reason). +WITHOUT_RESULT = [ + pytest.param(lifecycle("stuck", "credentials_unavailable"), "stuck", "credentials_unavailable", id="stuck"), + pytest.param(lifecycle("cancelled", "session_closed"), "cancelled", "session_closed", id="cancelled"), + pytest.param(lifecycle("aborted", "never_dispatched"), "aborted", "never_dispatched", id="aborted"), + pytest.param( + lifecycle("indeterminate", "interrupted_without_history", durable=False), + "indeterminate", + "interrupted_without_history", + id="indeterminate", + ), + pytest.param(lifecycle("expired"), "expired", None, id="expired"), + pytest.param(lifecycle("completed", durable=False), "completed", None, id="completed-without-response"), + pytest.param(lifecycle("paused", "operator"), "paused", "operator", id="unknown-status"), +] + + +def check_without_result(error: RunStateError, view: dict[str, Any], status: str, reason: str | None) -> None: + assert (error.execution_id, error.session_id, error.status, error.reason) == ("r", "s", status, reason) + assert error.body == view + assert f"run r is {status}" in str(error) + if reason: + assert reason in str(error) + + +def reply(body: dict[str, Any]) -> httpx.Response: + return httpx.Response(200, json=body) + + +async def test_a_queued_run_is_polled_until_the_result_its_durable_view_keeps(router: respx.MockRouter) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").mock(return_value=reply(lifecycle("queued"))) + poll = router.get(f"{SERVER}/v1/runs/r").mock( + side_effect=[reply(lifecycle("running")), reply(lifecycle("completed", response=RETAINED))] + ) + + async with AsyncSubmilli(base_url=SERVER, token=TOKEN) as client: + state = await client.sessions.execute("s", "code", idempotency_key="k") + assert state == RunningRun(execution_id="r", session_id="s", status="queued") + result = await client.runs.complete(state, poll_interval=0) + + assert isinstance(result, ExecuteResult) + assert (result.execution_id, result.session_id, result.result, result.console) == ("r", "s", None, ["before"]) + assert result.error is not None and result.error.kind == "runtime_error" + assert poll.call_count == 2 + + +def test_the_sync_client_polls_a_queued_run_until_its_result(router: respx.MockRouter) -> None: + retained = {**RETAINED, "result": "2", "console": [], "error": None} + router.post(f"{SERVER}/v1/sessions/s/execute").mock(return_value=reply(lifecycle("queued"))) + router.get(f"{SERVER}/v1/runs/r").mock( + side_effect=[reply(lifecycle("queued")), reply(lifecycle("completed", response=retained))] + ) + + with Submilli(base_url=SERVER, token=TOKEN) as client: + result = client.runs.complete(client.sessions.execute("s", "code"), poll_interval=0) + + assert isinstance(result, ExecuteResult) + assert (result.result, result.error) == ("2", None) + + +@pytest.mark.parametrize(("view", "status", "reason"), WITHOUT_RESULT) +async def test_an_execute_without_a_result_raises_run_state_error( + router: respx.MockRouter, view: dict[str, Any], status: str, reason: str | None +) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").mock(return_value=reply(view)) + + async with AsyncSubmilli(base_url=SERVER, token=TOKEN) as client: + with pytest.raises(RunStateError) as raised: + await client.sessions.run("s", "code", idempotency_key="k") + check_without_result(raised.value, view, status, reason) + + +@pytest.mark.parametrize(("view", "status", "reason"), WITHOUT_RESULT) +def test_the_sync_client_raises_run_state_error_while_polling( + router: respx.MockRouter, view: dict[str, Any], status: str, reason: str | None +) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").mock(return_value=reply(lifecycle("queued"))) + router.get(f"{SERVER}/v1/runs/r").mock(return_value=reply(view)) + + with Submilli(base_url=SERVER, token=TOKEN) as client: + state = client.sessions.execute("s", "code") + with pytest.raises(RunStateError) as raised: + client.runs.complete(state, poll_interval=0) + check_without_result(raised.value, view, status, reason) + + +SUSPENDED = { + "status": "suspended", + "execution_id": "r", + "session_id": "s", + "calls": [{"call_id": "r:effect-1", "tool": "t", "args": {}}], +} + + +async def test_a_resume_that_answers_a_stopped_run_raises(router: respx.MockRouter) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").mock(return_value=reply(SUSPENDED)) + stuck = lifecycle("stuck", "execution_interrupted") + router.post(f"{SERVER}/v1/runs/r/resume").mock(return_value=reply(stuck)) + + async with AsyncSubmilli(base_url=SERVER, token=TOKEN) as client: + with pytest.raises(RunStateError) as raised: + await client.sessions.run("s", "code", on_tool_call=lambda c: ToolResult.success(c.call_id, 1)) + check_without_result(raised.value, stuck, "stuck", "execution_interrupted") + + +def test_the_sync_client_raises_when_a_resume_answers_a_stopped_run(router: respx.MockRouter) -> None: + router.post(f"{SERVER}/v1/sessions/s/execute").mock(return_value=reply(SUSPENDED)) + cancelled = lifecycle("cancelled", "session_closed") + router.post(f"{SERVER}/v1/runs/r/resume").mock(return_value=reply(cancelled)) + + with Submilli(base_url=SERVER, token=TOKEN) as client, pytest.raises(RunStateError) as raised: + client.sessions.run("s", "code", on_tool_call=lambda c: ToolResult.success(c.call_id, 1)) + check_without_result(raised.value, cancelled, "cancelled", "session_closed") + + +async def test_a_completed_run_whose_response_is_no_result_raises(router: respx.MockRouter) -> None: + view = lifecycle("completed", response="legacy bytes") + router.get(f"{SERVER}/v1/runs/r").mock(return_value=reply(view)) + + async with AsyncSubmilli(base_url=SERVER, token=TOKEN) as client: + with pytest.raises(RunStateError) as raised: + await client.runs.retrieve("r") + check_without_result(raised.value, view, "completed", None) + + +async def test_a_run_the_durable_view_does_not_know_is_unknown_run(router: respx.MockRouter) -> None: + router.get(f"{SERVER}/v1/runs/r").mock(return_value=httpx.Response(404, text="unknown run")) + + async with AsyncSubmilli(base_url=SERVER, token=TOKEN) as client: + with pytest.raises(UnknownRunError) as lost: + await client.runs.retrieve("r") + assert lost.value.status_code == 404 and lost.value.body == "unknown run" + + def test_the_sync_client_answers_calls(server: FakeServer) -> None: server.programs["orders"] = [("query_orders", {"customer": "a"})] with Submilli(base_url=SERVER, token=TOKEN) as client: