Handle durable run states instead of returning empty results

.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:

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.
 

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

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.

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.
 

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 = ["."]
 

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)
 

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}")

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

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

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)

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]