[Fix] Keep long s1a-mcp runs inside host tool timeouts: progress, fail-fast busy error, clean cancellation, documented settings

CHANGELOG.md

@@ -48,6 +48,12 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); ver
 
 ### Fixed
 
+- `s1a-mcp` long runs: `run_agent` reports progress per episode or browser step when the request has a
+  `progressToken`; a call made while another runs fails at once with a tool error instead of queueing past the
+  host's timeout; a cancelled run finishes its browser and model teardown (it was cut at the first `await`) and
+  records the cancellation in the job `result.json` or `answer.json` even when no episode finished; cancelled
+  seeds count in `n_total_trials`. The server instructions put serialization and the timeout advice in their
+  first 512 characters, and docs/skills.md gives the Codex and Claude Code settings.
 - `decide` rejects repeated `--option` keys before model setup instead of silently replacing an earlier choice
   description. The usage error names the duplicate key without printing either description.
 - Malformed ticket-router JSONL now names the dataset, physical line and JSON column without printing ticket contents.

README.md

@@ -132,7 +132,9 @@ folder with the score. Without a key, `--model cua` answers in process after `uv
 codex mcp add s1a -- uv run --project /path/to/system1-agents s1a-mcp
 ```
 
-It serves three tools, `list_agents`, `run_agent` and `decide`, to any host that speaks MCP.
+It serves three tools, `list_agents`, `run_agent` and `decide`, to any host that speaks MCP. Run `uv sync` in the
+checkout once before registering it, and raise the host's tool timeout: a browser run takes minutes, beyond Codex's
+60 s default. The settings for Codex and Claude Code: [docs/skills.md](docs/skills.md#host-timeouts).
 
 ## Build your own
 

docs/skills.md

@@ -80,8 +80,37 @@ or `cua`; callers that omit it keep using Jev. `clm` needs `CLM_URL` (a running
 `LAYA_SERVED_URL`. For local decisions, install the matching extra in the server's checkout (`uv sync
 --extra laya` or `uv sync --extra cua`) and pass `model="laya"` or `model="cua"`; no Jev API key is needed.
 The first local call may download the checkpoint. Each call loads and closes its model; `ms` measures the
-decision, not model loading. Agent runs and decisions are serialized, and model output stays off the stdio
-protocol stream.
+decision, not model loading. Model output stays off the stdio protocol stream.
+
+The server runs one call at a time. A `run_agent` or `decide` that arrives while another call runs fails at once
+with a tool error naming the running call and how long it has run; it does not queue. When the request carries a
+`progressToken`, `run_agent` sends `notifications/progress` once per finished episode (`episode 3/20`) or decided
+browser step (`step 7/100`), at most one a second. A cancelled call (`notifications/cancelled`) releases the slot,
+closes the browser and the decision model it started, and records the cancellation: a tool agent's job
+`result.json` has `"cancelled": true` and counts the unfinished episodes in `n_cancelled_trials`; a browser agent's
+`answer.json` has `"cancelled": true`.
+
+#### Host timeouts
+
+`flights` budgets 180 s and `allrecipes` 300 s, and a tool agent's run grows with `--episodes`. Run `uv sync` in the
+checkout once before registering the server, so the first start does not spend minutes resolving the `openjiuwen`
+git pin inside the host's startup wait.
+
+[Codex](https://developers.openai.com/codex/mcp) stops a tool call after 60 s and a server start after 10 s by
+default. Raise both for this server in `~/.codex/config.toml`
+([config reference](https://developers.openai.com/codex/config-reference)):
+
+```toml
+[mcp_servers.s1a]
+command = "uv"
+args = ["run", "--project", "/path/to/system1-agents", "s1a-mcp"]
+startup_timeout_sec = 60
+tool_timeout_sec = 600
+```
+
+[Claude Code](https://code.claude.com/docs/en/mcp) has no short tool limit by default. To set one, add
+`"timeout": 600000` (milliseconds) to the server's `.mcp.json` entry or export `MCP_TOOL_TIMEOUT=600000`; give the
+first start more room with `MCP_TIMEOUT=60000`.
 
 Each tool carries a `title` and MCP behavior hints, which hosts such as Codex (`default_tools_approval_mode =
 "writes"`) and Claude use to decide what to auto-approve. `list_agents` is read-only, idempotent and closed-world: it

pyproject.toml

@@ -10,12 +10,13 @@ requires-python = ">=3.11"
 dependencies = [
     # Tag on the jiuwen-jev branch of the ThinkFlowLab fork; CONTRIBUTING.md § The openjiuwen pin lists what it adds over upstream develop.
     "openjiuwen @ git+https://github.com/ThinkFlowLab/agent-core@jj-0.1.0",
-    # Imported directly; openjiuwen depends on all three, so they add nothing to the install.
+    # Imported directly; openjiuwen (and mcp, for anyio) already depends on each, so they add nothing to the install.
     "httpx>=0.28",
     "python-dotenv>=1.0",
     # MCP Python SDK 1.x: openjiuwen 0.1.18 needs fastmcp, whose 3.x line requires mcp<2 (fastmcp 4.x needs mcp>=2).
     # Lift the bound with the 2.x port; docs/architecture.md § MCP SDK 2.x lists the call sites and the unblock condition.
     "mcp>=1.26,<2",
+    "anyio>=4",
     "pillow>=10",
 ]
 

s1a/browser/browse.py

@@ -16,6 +16,7 @@
 from urllib.parse import urlparse
 from uuid import uuid4
 
+import anyio
 from openjiuwen.core.common.logging import logger
 from openjiuwen.core.foundation.llm import Model
 from openjiuwen.core.runner import Runner
@@ -211,8 +212,9 @@ async def run_task(agent: DeepAgent, goal: str, *, timeout_s: float, logs_dir: P
         try:
             await save_final_screenshot(agent, answer, logs_dir)  # before the cleanup closes the page
         finally:  # a cancellation during the screenshot still releases the browser and the session
-            await agent.cleanup_task_resources()
-            await Runner.release(conversation_id)
+            with anyio.CancelScope(shield=True):  # a cancelled task (an MCP host's cancel) must not cut the teardown
+                await agent.cleanup_task_resources()
+                await Runner.release(conversation_id)
     answer["final"] = str(result.get("output") or "")
     answer["ok"] = result.get("result_type") == "answer" and bool(answer["final"])
     if not answer["ok"]:
@@ -253,7 +255,9 @@ async def browse(
         case "jev" | "clm" | "laya" | "laya-served" | "cua" | "omnijev":
             if decision_model is None:
                 raise RuntimeError(f"--model {model_name} needs a decision model")
-            slot_model = BrowserDecisionModel(spec, policy, counted, decision_model=decision_model, value_model=None)
+            slot_model = BrowserDecisionModel(
+                spec, policy, counted, decision_model=decision_model, value_model=None, max_steps=max_steps
+            )
             agent = create_browser_agent(
                 slot_model,
                 language=spec.language,
@@ -396,7 +400,9 @@ def policy_from_args(args: argparse.Namespace) -> BrowserPolicy:
 
 
 async def play(spec: BrowserAgentSpec, args: argparse.Namespace) -> Answer:
-    """One task inside an already started Runner: the answer without the ticks, which ``decision_ticks.json`` holds."""
+    """One task inside an already started Runner: the answer without the ticks, which ``decision_ticks.json`` holds.
+
+    A cancelled task writes ``answer.json`` with ``"cancelled": true`` and re-raises."""
     if args.profile_out is not None and args.model == "llm":
         raise RuntimeError("--profile-out needs a decision model: the profiler times the slot model's turns")
     chat = chat_model_from_env()
@@ -415,9 +421,15 @@ async def play(spec: BrowserAgentSpec, args: argparse.Namespace) -> Answer:
             chat=chat,
             decision_model=decision_model,
         )
+    except asyncio.CancelledError:  # a host cancelled the call: answer.json says so instead of keeping a stale answer
+        args.logs_dir.mkdir(parents=True, exist_ok=True)
+        cancelled = {"ok": False, "cancelled": True, "error": "cancelled", "agent": spec.name, "model": args.model}
+        (args.logs_dir / "answer.json").write_text(json.dumps(cancelled, indent=1), encoding="utf-8")
+        raise
     finally:
-        if decision_model is not None:
-            await decision_model.close()
+        with anyio.CancelScope(shield=True):  # a cancelled task still closes its decision model
+            if decision_model is not None:
+                await decision_model.close()
         if profiler is not None:
             profiler.detach()
     result = {key: value for key, value in answer.items() if key != "ticks"}

s1a/browser/decision_model.py

@@ -32,6 +32,7 @@
 from openjiuwen.core.foundation.llm import AssistantMessage, AssistantMessageChunk, ToolCall
 from openjiuwen.core.foundation.llm.model import Model
 from openjiuwen.harness.tools.browser_move.utils.parsing import extract_json_object
+from s1a import progress
 from s1a.browser import prompts
 from s1a.browser.action_space import (
     NONE_VALUE,
@@ -172,6 +173,7 @@ def __init__(
         *,
         decision_model: DecisionModel,
         value_model: Model | None,
+        max_steps: int | None = None,
     ) -> None:
         """``policy.goal_value_cache`` offers goal-extracted values to Jev as a choice head; off means every typed
         value comes from the chat model given the field and page context. ``policy.prefetch_values`` makes that
@@ -180,7 +182,8 @@ def __init__(
         sends each action and the next probe as one ``browser_run_code_unsafe`` call when that tool is
         offered, skipping the runtime's target validation and one transport round trip per step.
         ``value_model`` answers the typed-value calls when given; the fallback chat model otherwise.
-        ``decision_model`` decides every browser step; its ``name`` (``jev``, ``laya``) is the ``--model`` value."""
+        ``decision_model`` decides every browser step; its ``name`` (``jev``, ``laya``) is the ``--model`` value.
+        Each decided step is reported to ``s1a.progress``, out of ``max_steps`` when given."""
         super().__init__(fallback.model_client_config, fallback.model_config)
         self._spec = spec
         self._fallback = fallback
@@ -190,6 +193,7 @@ def __init__(
         self._batch_actions = policy.batch_actions
         self._rethink_on = policy.rethink_on
         self._recovery_limits = policy.recovery_limits
+        self._max_steps = max_steps
         self._language = spec.language
         self._decision_model = decision_model
         # the chat model writes the bounded recovery's plan, as the tool front's RethinkRail does
@@ -365,6 +369,8 @@ async def _decide_message(self, messages: Any) -> AssistantMessage:
             if screenshot_ms is not None:  # an image-reading model: the capture is part of the step's cost
                 record["screenshot_ms"] = screenshot_ms
             run.ticks.append(record)
+            step = len(run.ticks)
+            await progress.report(step, self._max_steps, f"step {step}/{self._max_steps or '?'}")
             if move.operation == "WAIT":
                 run.consecutive_waits += 1
                 wait_entry = {"action": "wait", "kind": "wait", "text": None, "page_changed": None}

s1a/jobs.py

@@ -89,12 +89,28 @@ def summarize(episodes: list[Episode]) -> dict[str, Any]:
     }
 
 
-def write_job(eval_name: str, episodes: list[Episode], *, results_dir: Path) -> Path:
+def write_job(
+    eval_name: str,
+    episodes: list[Episode],
+    *,
+    results_dir: Path,
+    cancelled: int = 0,
+    policy: str | None = None,
+) -> Path:
     """One Harbor job folder per run: config.json, result.json, summary.json, one trial folder per episode.
 
+    ``cancelled`` counts the planned trials a cancellation stopped before they finished; a job with any has
+    ``"cancelled": true`` in its result.json. ``n_total_trials`` (and the dataset ``n_tasks``) include those
+    cancelled trials so Harbor-style readers see a consistent total. A cancellation that finished no episode
+    still writes the job when ``cancelled`` is positive; pass ``policy`` so the folder can name the agent.
+
     The trial shape follows agent-eval's runner so Nullius reads both roots alike.
     """
-    policy = episodes[0].policy
+    if not episodes and cancelled <= 0:
+        raise ValueError("write_job needs at least one episode, or a positive cancelled count")
+    if not episodes and policy is None:
+        raise ValueError("write_job with no episodes needs policy")
+    policy = policy if policy is not None else episodes[0].policy
     agent_name = f"s1a-evals/{policy}"
     job_dir = (
         results_dir / eval_name / f"{datetime.now():%Y-%m-%d__%H-%M-%S-%f}__{policy}"
@@ -152,28 +168,32 @@ def write_job(eval_name: str, episodes: list[Episode], *, results_dir: Path) ->
         }
         (trial_dir / "result.json").write_text(json.dumps(trial, ensure_ascii=False, indent=2), encoding="utf-8")
     finished_at = now_iso()
+    started_at = episodes[0].started_at if episodes else finished_at
     errored = sum(episode.error is not None for episode in episodes)
+    n_total = len(episodes) + cancelled
     job = {
         "id": str(uuid.uuid4()),
-        "started_at": episodes[0].started_at,
+        "started_at": started_at,
         "updated_at": finished_at,
         "finished_at": finished_at,
-        "n_total_trials": len(episodes),
+        "n_total_trials": n_total,
         "stats": {
             "n_completed_trials": len(episodes) - errored,
             "n_errored_trials": errored,
             "n_running_trials": 0,
             "n_pending_trials": 0,
-            "n_cancelled_trials": 0,
+            "n_cancelled_trials": cancelled,
             "n_retries": 0,
         },
+        "cancelled": cancelled > 0,
     }
     (job_dir / "result.json").write_text(json.dumps(job, indent=2), encoding="utf-8")
     config = {
         "job_name": job_dir.name,
         "agents": [{"name": agent_name, "model_name": policy}],
-        "datasets": [{"name": eval_name, "n_tasks": len(episodes)}],  # the dict shape Nullius's parse_job reads
+        "datasets": [{"name": eval_name, "n_tasks": n_total}],  # the dict shape Nullius's parse_job reads
     }
     (job_dir / "config.json").write_text(json.dumps(config, indent=2), encoding="utf-8")
-    (job_dir / "summary.json").write_text(json.dumps(summarize(episodes), indent=2), encoding="utf-8")
+    summary = summarize(episodes) if episodes else {"episodes": 0, "scored": 0, "errors": 0}
+    (job_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
     return job_dir