diff --git a/openswarm-runner/runner/workflow_run.py b/openswarm-runner/runner/workflow_run.py index b196f1ac..2e516f90 100644 --- a/openswarm-runner/runner/workflow_run.py +++ b/openswarm-runner/runner/workflow_run.py @@ -6,6 +6,7 @@ behave exactly as they do on a laptop. """ import json +import logging import time from typing import Any, Callable, Dict, List, Optional @@ -15,9 +16,17 @@ from typeguard import typechecked from runner.boot.backend_process import BackendProcess +logger = logging.getLogger(__name__) + TERMINAL_STATUSES = ("success", "failure", "ran_late", "skipped") POLL_INTERVAL_SECONDS = 1.0 TRANSCRIPT_MAX_CHARS = 14000 +# The backend is one process on one event loop, and a heavy tool call can hold it: MCP registry +# work has been measured keeping it from answering for well over a minute. A reply that is late +# therefore means busy, not dead, so the budget is generous and lateness is never fatal. The only +# thing that ends a run early is the process actually exiting. +REQUEST_TIMEOUT_SECONDS = 60.0 +TRIGGER_RETRY_SECONDS = 5.0 class WorkflowRunFailed(RuntimeError): @@ -127,14 +136,49 @@ def p_get_json(client: httpx.Client, backend: BackendProcess, path: str) -> Dict @typechecked -def trigger_run(client: httpx.Client, backend: BackendProcess, workflow_id: str) -> str: - response = client.post( - f"{backend.base_url}/api/workflows/{workflow_id}/run", - headers=backend.headers(), - json={}, - ) - response.raise_for_status() - body = response.json() +def p_started_run_id(client: httpx.Client, backend: BackendProcess, workflow_id: str) -> Optional[str]: + """The newest run on record, or None. Safe to adopt: the runs file ships empty in every + container, so anything in this list was started by the POST we just made.""" + try: + body = p_get_json(client, backend, f"/api/workflows/{workflow_id}/runs?limit=1") + except httpx.HTTPError: + return None + for record in body.get("runs") or []: + if isinstance(record, dict) and record.get("id"): + return str(record["id"]) + return None + + +@typechecked +def trigger_run(client: httpx.Client, backend: BackendProcess, workflow_id: str, deadline: float) -> str: + """Start the run, waiting out a backend too busy to answer instead of failing the job. + + A POST whose reply never arrived may still have started the run, so a retry looks for that + run before firing again. Posting blind would either execute the workflow twice or come back + "Previous run still active", and both are worse than waiting. + """ + while True: + try: + response = client.post( + f"{backend.base_url}/api/workflows/{workflow_id}/run", + headers=backend.headers(), + json={}, + ) + response.raise_for_status() + body = response.json() + break + except httpx.HTTPError as exc: + if not backend.is_alive(): + raise WorkflowRunFailed(f"backend died before the run could start: {exc}") from exc + adopted = p_started_run_id(client, backend, workflow_id) + if adopted: + logger.warning("trigger reply never arrived (%s); adopting the run it started", exc) + return adopted + if time.monotonic() >= deadline: + raise WorkflowRunFailed(f"backend never accepted the run trigger: {exc}") from exc + logger.warning("trigger did not answer (%s); backend is busy, retrying", exc) + time.sleep(TRIGGER_RETRY_SECONDS) + run_id = str(body.get("run_id") or "") if not run_id: raise WorkflowRunFailed( @@ -184,13 +228,18 @@ def execute_workflow( Blowing the deadline stops the run and reports `timed_out`; the caller still gets whatever the agent produced before the wall came down. """ - with httpx.Client(timeout=30.0) as client: - run_id = trigger_run(client, backend, workflow_id) + with httpx.Client(timeout=REQUEST_TIMEOUT_SECONDS) as client: + run_id = trigger_run(client, backend, workflow_id, deadline) record: Dict[str, Any] = {} timed_out = False while True: - record = p_find_run(client, backend, workflow_id, run_id) or record + try: + record = p_find_run(client, backend, workflow_id, run_id) or record + except httpx.HTTPError as exc: + # A poll that goes unanswered says the backend is busy, and the run it is busy + # with is this one. Crashing here used to throw away a run that then finished fine. + logger.warning("poll for run %s went unanswered (%s); still waiting", run_id, exc) status = str(record.get("status") or "running") if on_progress is not None: on_progress(RunProgress( diff --git a/openswarm-runner/tests/test_workflow_run.py b/openswarm-runner/tests/test_workflow_run.py index 612a23f9..ce1f00eb 100644 --- a/openswarm-runner/tests/test_workflow_run.py +++ b/openswarm-runner/tests/test_workflow_run.py @@ -1,6 +1,23 @@ """Reading a finished session correctly, including the failures the backend calls success.""" -from runner.workflow_run import final_answer, render_transcript, system_notices +import subprocess +import sys +import time +from typing import Any, Dict, Iterator + +import httpx +import pytest + +from runner import workflow_run +from runner.boot.backend_process import BackendProcess +from runner.workflow_run import ( + WorkflowRunFailed, + execute_workflow, + final_answer, + render_transcript, + system_notices, + trigger_run, +) # Shape taken verbatim from a real container run whose provider token was rejected. REJECTED_TOKEN_SESSION = [ @@ -34,3 +51,73 @@ def test_the_transcript_keeps_tool_calls_and_drops_hidden_turns() -> None: assert "[tool Bash]" in transcript assert "PONG" in transcript assert "draft" not in transcript + + +# A backend busy enough to miss a reply is the normal case, not a broken one: a single MCP +# registry call has been measured holding its event loop past a minute. These pin that a late +# reply never costs the user the run. + +RUN_ROW = {"id": "run_1", "status": "success", "session_id": "sess_1", "cost_usd": 0.0} + + +@pytest.fixture +def backend() -> Iterator[BackendProcess]: + """A real BackendProcess around a process that just sits there, so is_alive() is honest.""" + process = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + try: + yield BackendProcess(process=process, base_url="http://backend.test", token="t") + finally: + process.kill() + process.wait(timeout=10) + + +def p_client(stalls: int) -> httpx.Client: + """A client whose first `stalls` requests time out, exactly like a starved event loop.""" + state = {"left": stalls} + + def handle(request: httpx.Request) -> httpx.Response: + if state["left"] > 0: + state["left"] -= 1 + raise httpx.ReadTimeout("timed out", request=request) + if request.url.path.endswith("/run"): + return httpx.Response(200, json={"run_id": "run_1", "status": "running"}) + if request.url.path.endswith("/runs"): + return httpx.Response(200, json={"runs": [RUN_ROW]}) + return httpx.Response(404, json={}) + + return httpx.Client(transport=httpx.MockTransport(handle), timeout=1.0) + + +def test_a_stalled_trigger_adopts_the_run_it_already_started(backend: BackendProcess) -> None: + # One unanswered POST, then the run it started is visible. Posting again would either run the + # workflow twice or come back "Previous run still active". + assert trigger_run(p_client(stalls=1), backend, "wf_1", time.monotonic() + 5.0) == "run_1" + + +def test_a_dead_backend_fails_the_trigger_instead_of_waiting(backend: BackendProcess) -> None: + backend.process.kill() + backend.process.wait(timeout=10) + with pytest.raises(WorkflowRunFailed, match="backend died"): + trigger_run(p_client(stalls=99), backend, "wf_1", time.monotonic() + 5.0) + + +def test_a_stalled_poll_does_not_throw_away_a_run_that_finishes( + backend: BackendProcess, monkeypatch: pytest.MonkeyPatch +) -> None: + polls = {"left": 2} + + def p_find(*_args: Any, **_kwargs: Any) -> Dict[str, Any]: + if polls["left"] > 0: + polls["left"] -= 1 + raise httpx.ReadTimeout("timed out") + return RUN_ROW + + monkeypatch.setattr(workflow_run, "trigger_run", lambda *_a, **_k: "run_1") + monkeypatch.setattr(workflow_run, "p_find_run", p_find) + monkeypatch.setattr(workflow_run, "p_collect_session", lambda *_a, **_k: ANSWERED_SESSION) + monkeypatch.setattr(workflow_run, "POLL_INTERVAL_SECONDS", 0.01) + + outcome = execute_workflow(backend, "wf_1", time.monotonic() + 10.0) + assert outcome.status == "success" + assert outcome.answer == "PONG" + assert polls["left"] == 0