Run up to STRICT_MAX_WORKERS runs concurrently in the run service

doc/architecture/SPEC/app/run_service.md

@@ -1,10 +1,10 @@
 # app/run_service.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `RunService` is the durable application boundary shared by the embedded
-Strict SDK and MCP transport. It owns policy-checked reserve/start, one
-serialized worker, isolated child launch, startup/orphan reconciliation,
+Strict SDK and MCP transport. It owns policy-checked reserve/start, a queue
+drained by `STRICT_MAX_WORKERS` workers (default 1), isolated child launch, startup/orphan reconciliation,
 readiness, bounded status/result polling, and repository-scoped knowledge
 reads. Construction and polling require no CLI or MCP module import.
 
@@ -39,3 +39,33 @@ run is pinned to this server's effective `knowledge_dir` (reported as `packaged`
 or `unverified` for a custom directory), with `KNOWLEDGE_ROOT` removed from the
 child's environment. Runs reserved before pinning
 existed have no record and use the process default.
+
+## 2026-09-30 Concurrent workers
+`Settings.strict_max_workers` (`STRICT_MAX_WORKERS`, 1–32, default 1; embedded
+hosts set it through `StrictRuntimeConfig.max_workers`) starts that many worker
+threads, and `capabilities()` reports the same number as `max_strict_workers`.
+Each worker launches one child at a time, so at most N children run at once.
+
+`RunQueue` never hands out two runs with the same **conflict key** at once:
+the checkout (resolved through the same `RepositoryContextResolver` execution
+uses: frozen path, `REPO_PATHS`, then the adapter manifest) plus the PR number (`None` for issue tasks, which work in
+the live checkout). Runs on one PR head share one PR-time worktree, and harness
+sessions write their tool-bridge config into their working directory (cursor's
+`.cursor/mcp.json` has one fixed path per tree), so overlapping them would bind
+one run's agent to the other run's tool scope and trace. A run whose key is busy
+stays queued in order without holding a worker; workers take the oldest run
+whose key is free, so one busy PR cannot stall other PRs. The key is computed
+from the persisted `request.json` at enqueue; an unreadable request shares the
+key `("?", None)`. A failed launch marks the run `failed` and always frees its
+key. `_reap()` is single-flight per service (a non-blocking lock): a worker that
+finishes while another is sweeping skips its sweep.
+
+The key cannot see a runtime fallback: a PR run whose worktree could not be
+materialized degrades to the live checkout, where issue tasks and other
+degraded runs also work. That shared directory is guarded where the hazard
+is — the cursor transport serializes bridge-carrying sessions in a shared
+checkout with an `flock` in its git dir (`providers/cursor.md`), across
+services and the CLI as well. Not covered: two agent sessions *inside one run*
+(ensemble lenses, per-comment verification, bounded by
+`strict_backend_concurrency`) still share their run's own tree; that predates
+this change and is independent of the worker count.

doc/architecture/SPEC/config.md

@@ -17,7 +17,9 @@ metrics 与升级，提供带类型字段和安全默认值；外加 PR4d 知识
 MCP `repo_path` 授权面与 idempotency 保留期：
 `mcp_allowed_repo_roots`（`MCP_ALLOWED_REPO_ROOTS`，空 = 只有已配置的
 checkout —— 最小权限；放宽是**运维**决定，不是调用方的）与
-`idem_retention_days`（默认 30，圈住 `.idem/` 索引）；以及
+`idem_retention_days`（默认 30，圈住 `.idem/` 索引）；run service 的并发
+worker 数 `strict_max_workers`（`STRICT_MAX_WORKERS`，默认 1，启动时校验
+1–`MAX_STRICT_WORKERS_LIMIT`=32，越界即失败而非静默收敛）；以及
 知识回流落盘目录 `knowledge_intake_dir`（`KNOWLEDGE_INTAKE_DIR`，默认空 = 关闭；
 消费方见 `doc/RFC-knowledge-intake.md`）与跨主机发布目标 `knowledge_intake_issue`
 （`KNOWLEDGE_INTAKE_ISSUE`，`owner/repo#N`，默认空 = 关闭：harvest step 把同一条记录

doc/architecture/SPEC/contract.md

@@ -1,6 +1,6 @@
 # contract.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `旧版跨仓库契约兼容层 · refactor-status: compatibility-shim`
 
@@ -13,14 +13,15 @@
 
 ## 公开契约（`__all__`）
 `SDK_API_VERSION` / `DIRECT_API_VERSION` / `STRICT_API_VERSION` /
-`QUALITY_API_VERSION` / `KNOWLEDGE_API_VERSION`（SDK、Quality 为 `"1.0.0"`，Direct 为 `"1.1.0"`，Strict 为 `"1.3.0"`，Knowledge 为 `"1.1.0"`，且都从 `sdk.v1.models`
+`QUALITY_API_VERSION` / `KNOWLEDGE_API_VERSION`（SDK、Quality 为 `"1.0.0"`，Direct 为 `"1.1.0"`，Strict 为 `"1.4.0"`，Knowledge 为 `"1.1.0"`，且都从 `sdk.v1.models`
 取唯一值）。`capabilities(max_strict_workers=1,
 supports_file_locking=True) -> dict` 委托 SDK typed handshake 再投影为兼容
 dict；它包含 distribution/SDK/Direct/Strict/Quality/Knowledge 版本、resource
 revision、supported repositories，以及 `supports_expected_head`、
 `supports_structured_result`、`supports_post_false`、`supports_file_locking`、
 `supports_idempotent_strict_start`、`supports_knowledge_curation`、
-`max_strict_workers`。其余兼容导出包括 `build_review_result(run_dir) -> dict`（结构化评审
+`max_strict_workers`（服务真实的并发 worker 数，即 `STRICT_MAX_WORKERS`，
+同一 checkout+PR 的 run 仍逐个执行）。其余兼容导出包括 `build_review_result(run_dir) -> dict`（结构化评审
 结果：`contract_version`、`run_id`、`state`、`reviewed_head_sha`、`verdict`、
 `summary_markdown`、`comments`、`finding_dispositions`、`findings`、
 `stale`/`expected_head_sha`/`actual_head_sha`、`diagnostics`）；`unknown_run_result(run_id)`（显式

doc/architecture/SPEC/engine/worktrees.md

@@ -1,6 +1,6 @@
 # engine/worktrees.py —— 规范
 
-<!-- verified-against: 2026-09-26 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~294 · PR-time worktree：身份、物化、持有 · refactor-status: ok`
 
@@ -38,6 +38,10 @@ worktree，允许本地 rebase，又不会改配置的 checkout 或复用评审
   恰是 head 门要防的"评错树"。
 - 复用验证：`--git-common-dir` 必须解析回请求方仓库自己的 git dir **且**
   HEAD == sha；force-remove 只对外来/撕裂的树成立，绝不对同身份的活树。
+- 复用时在 `LOCK_EX` 内 `touch(dest)` 刷新 mtime：run 的共享持有要等到
+  后续 step 首次使用树时才取，`materialize` 释放锁到那一刻之间，一棵超过
+  保留期的旧树看起来无人使用；新 mtime 让 reaper（它在自己的 `LOCK_EX` 下
+  重读 mtime）放过它。
 - 可变树按 run_dir 分键，复用时只核对仓库归属（rebase 会改变 HEAD），
   拒绝 symlink、重定向到配置 checkout 的路径及外来仓库；调用方仅在 checkout
   step 重试时可重置此树。后续 step 经 checkpoint 恢复路径，并在树被清理后

doc/architecture/SPEC/idempotency.md

@@ -1,6 +1,6 @@
 # idempotency.py —— 规范
 
-<!-- verified-against: 2026-08-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~293 · 持久 idempotency 索引：每键一个 run、一次执行 · refactor-status: ok`
 
@@ -34,6 +34,8 @@ sha256 —— 之后新增的字段自动纳入，绝不把两个不同请求静
   另一个、后果大得多的变更。`reap_stale` 只清索引条目、worktree、
   `refs/imx/<run_id>/*`。
 - reaper 绝不触碰活 run 的条目（`state not in TERMINAL` → 跳过）。
+- worktree 回收：拿到树的 `LOCK_EX` 后**重读 mtime**，期间被复用
+  （`worktrees.touch`）的树跳过 —— 无锁的年龄检查与加锁之间可能插入一次复用。
 - worktree 回收：存活以树**自己的**共享锁判定（绝不是注册表），移除走
   `git worktree remove`（绝不裸 `rmtree`），逐树限时（一棵慢树不拖垮
   整个扫描）。

doc/architecture/SPEC/providers/cursor.md

@@ -33,7 +33,7 @@ headless `cursor-agent --print --force --output-format stream-json`，
 stdlib + `.base` + `.registry` + `..agent_loop.AgentOutcome` + `..llm` 的类型。
 
 ## 测试
-`test_provider_cursor.py`。
+`test_provider_cursor.py`；共享 checkout 串行见 `test_run_service_concurrency.py`。
 
 ## 重构备注
 这套调用形状是由 Composer 评测臂（`eval/dataset/run_cursor_arm.py`）验证出来的；
@@ -51,3 +51,14 @@ scope 约束，native 调用只被**记录**。
 
 ## 2026-09-30
 `complete()` 接受并忽略 `max_budget_usd`；`stops_at_spend` 为 False（CLI 无花费阈值）。
+
+### 共享 checkout 上的会话串行（2026-09-30）
+桥配置写在会话 cwd 的**固定路径** `.cursor/mcp.json`，同一目录里两个不同 run 的会话
+会互相覆盖配置，令一方的 agent 绑定到另一个 run 的工具 scope 与 trace。
+带桥的会话若 cwd 是**共享 checkout**（`.git` 为目录的 clone，或未托管的 linked
+worktree），就在该 checkout 的 git dir 里 `imx-cursor-session.lock` 上持
+`flock LOCK_EX` 直到会话结束并清理配置，跨线程/进程/服务串行——issue 任务与
+worktree 物化失败而降级到 live checkout 的 PR run 都在这里。托管的 PR-time
+worktree（repo+PR+sha 分键，run queue 不会让同 PR 的两个 run 并发）与 run 自己的
+目录不加锁。等待上限为会话超时 + 60 s；超时则返回空文本、`truncated`、refusal，
+并记 `capability_gap`（`cursor.exclusive_cwd`），绝不在别人的配置下静默开跑。

doc/architecture/SPEC/sdk.md

@@ -1,6 +1,6 @@
 # sdk/ —— 规范
 
-<!-- verified-against: 2026-09-29 -->
+<!-- verified-against: 2026-09-30 -->
 
 
 `Python SDK v1 · 跨仓库唯一 typed 边界 · refactor-status: ok`
@@ -22,7 +22,9 @@
 - Strict：`StrictRuntime(config=StrictRuntimeConfig(...))`，以及
   `capabilities` / `readiness` / `reserve_review`（`start_review` 别名）/
   `get_status` / `get_result` / `close`。旧 `settings_overrides` 暂作兼容入口；
-  新配置显式绑定完整仓库名、checkout、允许根、backend 和可选 run root。
+  新配置显式绑定完整仓库名、checkout、允许根、backend、可选 run root 与
+  `max_workers`（默认 1，正整数；即宿主并发执行的 run 数，经 handshake 的
+  `max_strict_workers` 回报）。
   `StrictPollResult.review` 提供类型化结构结果，`payload` 保持旧 wire 兼容。
 - Quality：同一 `StrictRuntime` 接受 typed `QualityReviewRequest`，提供
   `quality_readiness` / `reserve_quality_review`（`start_quality_review` 别名）/
@@ -86,8 +88,9 @@
   push、开 PR 或 schedule。ReviewBot 必须向 `KnowledgeCurator` 传 dedicated work
   checkout，并继续拥有重试、artifact 与本地补丁导出；SDK 也绝不写 packaged
   knowledge tree。
-- SDK、Quality API 版本常量为 `1.0.0`；Direct 为 `1.1.0`，Strict 为 `1.3.0`
-  （1.1 新增 `finding_dispositions`；1.2 新增 `findings`；1.3 新增显式 recheck），Knowledge 为 `1.1.0`
+- SDK、Quality API 版本常量为 `1.0.0`；Direct 为 `1.1.0`，Strict 为 `1.4.0`
+  （1.1 新增 `finding_dispositions`；1.2 新增 `findings`；1.3 新增显式 recheck；
+  1.4 新增 `StrictRuntimeConfig.max_workers`），Knowledge 为 `1.1.0`
   （1.1 向后兼容地新增 `catalog_entries()` / `KnowledgeCatalogEntry` 与容量预检），
   distribution 为 `0.2.0`；`Capabilities.knowledge_api_version` 与
   `supports_knowledge_curation` 组成 ReviewBot 的 paired-release 握手，避免只按

src/infermatrix_copilot/app/run_service.py

@@ -8,13 +8,13 @@
 
 import json
 import os
-import queue
 import subprocess
 import sys
 import threading
 import uuid
+from collections import deque
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 
 from .. import contract
 from .. import idempotency as idem
@@ -25,7 +25,7 @@
     PolicyError, authorize_repo_path, enforce_mcp_policy,
     enforce_quality_review_policy, enforce_strict_review_policy,
 )
-from ..task_spec import READ_ONLY_KINDS
+from ..task_spec import READ_ONLY_KINDS, TaskSpec
 from .core import Copilot
 from .reservation import RunReservation
 
@@ -42,10 +42,66 @@ def read_knowledge_pin(run_dir: str | Path) -> dict:
     return data if isinstance(data, dict) else {}
 
 
+# The conflict key of a run whose request cannot be read or keyed: such runs
+# fail at launch anyway, and sharing one key keeps them from overlapping each
+# other rather than guessing what they touch.
+UNKEYED = ("?", None)
+
+
+class RunQueue:
+    """FIFO run queue for N workers that never hands out two runs with the
+    same conflict key at once.
+
+    A run whose key is busy stays queued, in order, and does not hold a
+    worker: the worker takes the oldest run whose key is free, so one busy PR
+    cannot stall reviews of other PRs behind it. Keys are computed at enqueue
+    (the request is persisted before any run is enqueued)."""
+
+    def __init__(self, key_for: Callable[[str], Any]):
+        self._key_for = key_for
+        self._cond = threading.Condition()
+        self._pending: deque[tuple[str, bool, Any]] = deque()
+        self._busy: set[Any] = set()
+
+    def put(self, item: tuple[str, bool]) -> None:
+        run_id, strict_compat = item
+        try:
+            key = self._key_for(run_id)
+            hash(key)
+        except Exception:  # noqa: BLE001 - a reserved run must still enqueue
+            key = UNKEYED
+        with self._cond:
+            self._pending.append((run_id, strict_compat, key))
+            self._cond.notify_all()
+
+    def take(self) -> tuple[str, bool, Any]:
+        """Block until a run with a free key is queued; claim its key."""
+        with self._cond:
+            while True:
+                for index, item in enumerate(self._pending):
+                    if item[2] not in self._busy:
+                        del self._pending[index]
+                        self._busy.add(item[2])
+                        return item
+                self._cond.wait()
+
+    def done(self, key: Any) -> None:
+        """Release a key claimed by `take`, waking workers blocked on it."""
+        with self._cond:
+            self._busy.discard(key)
+            self._cond.notify_all()
+
+    def snapshot(self) -> tuple[list[str], set[Any]]:
+        """Queued run ids and busy keys (tests/diagnostics)."""
+        with self._cond:
+            return [item[0] for item in self._pending], set(self._busy)
+
+
 class RunService:
-    """The server core: a serialized run queue over isolated subprocesses, plus
-    ownership-aware reconciliation. Framework-agnostic (no `mcp` import) so it is
-    unit-testable without a live protocol connection."""
+    """The server core: a run queue drained by `STRICT_MAX_WORKERS` workers
+    over isolated subprocesses, plus ownership-aware reconciliation.
+    Framework-agnostic (no `mcp` import) so it is unit-testable without a live
+    protocol connection."""
 
     def __init__(self, settings: Settings | None = None):
         """Wire reservation and workflow services, register this server's
@@ -59,44 +115,84 @@ def __init__(self, settings: Settings | None = None):
         self.pid = os.getpid()
         rs.register_server(self.run_root, self.server_id, self.pid)
         rs.startup_reconcile(self.run_root)
+        # One sweep at a time per service: with several workers finishing
+        # runs, overlapping sweeps would only race each other for the same
+        # entries and refs.
+        self._reap_lock = threading.Lock()
         self._reap()
-        self._q: queue.Queue[tuple[str, bool]] = queue.Queue()
-        self._worker = threading.Thread(target=self._worker_loop, daemon=True,
-                                        name="imx-mcp-worker")
-        self._worker.start()
-
-    # one worker thread drains the queue, so Strict runs are strictly serial.
-    # The handshake reports this rather than letting a bot infer concurrency
-    # from the fact that `start` returns immediately.
-    MAX_STRICT_WORKERS = 1
+        self.max_strict_workers = int(self.settings.strict_max_workers)
+        self._q = RunQueue(self._conflict_key)
+        self._workers = [
+            threading.Thread(target=self._worker_loop, daemon=True,
+                             name=f"imx-mcp-worker-{i}")
+            for i in range(self.max_strict_workers)]
+        for worker in self._workers:
+            worker.start()
 
     def _reap(self) -> None:
         """Bound what the idempotency index and PR-time worktrees accumulate.
 
         Best-effort by construction: this runs at startup and after each
         terminal run, and a sweep that cannot complete must never stop a server
         from serving or a run from finishing."""
+        if not self._reap_lock.acquire(blocking=False):
+            return  # a sweep is already running; the next terminal run retries
         try:
             idem.reap_stale(self.run_root,
                             retention_days=self.settings.idem_retention_days,
                             repo_paths=list(self.settings.repo_paths.values()))
         except Exception:  # noqa: BLE001 — housekeeping, never load-bearing
             pass
+        finally:
+            self._reap_lock.release()
 
     def capabilities(self) -> dict:
         """The version/capability handshake — see `contract.capabilities`."""
         from ..engine.lifecycle import fcntl as _fcntl
 
         return contract.capabilities(
-            max_strict_workers=self.MAX_STRICT_WORKERS,
+            max_strict_workers=self.max_strict_workers,
             supports_file_locking=_fcntl is not None)
 
-    # -- worker: one run at a time, each an isolated subprocess ---------------
+    # -- workers: up to N runs at a time, each an isolated subprocess --------
+    def _conflict_key(self, run_id: str) -> tuple[str, Any]:
+        """The resource a run must not share with a concurrently executing one:
+        its checkout plus its PR (``None`` for issue tasks, which work in the
+        live checkout itself).
+
+        Runs on one PR head share one PR-time worktree (engine/worktrees.py
+        keys it by repo+PR+sha), and harness sessions write their tool-bridge
+        config into their working directory — cursor's `.cursor/mcp.json` has
+        one fixed path per tree — so two such runs overlapping would bind one
+        run's agent to the other run's tool scope and trace. Different PRs get
+        different trees and may run side by side."""
+        try:
+            spec = json.loads((self.run_root / run_id / "request.json")
+                              .read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            spec = None
+        if not isinstance(spec, dict):
+            return UNKEYED
+        repo = str(spec.get("repo") or self.settings.default_repo)
+        try:
+            # the resolver execution itself uses (frozen path, REPO_PATHS,
+            # then the adapter manifest), so an implicit request and an
+            # explicit one naming the same checkout get the same key
+            checkout = self.copilot.repository_context.repo_path_for(
+                TaskSpec.model_validate(spec))
+        except Exception:  # noqa: BLE001 - such a run fails at launch anyway
+            checkout = str(spec.get("repo_path") or "")
+        where = str(Path(checkout).expanduser().resolve()) if checkout \
+            else f"repo:{repo}"
+        pr = spec.get("pr")
+        return (where, None if pr is None else str(pr))
+
     def _worker_loop(self) -> None:
-        """Drain the queue forever, launching one run subprocess at a time. A
-        launch failure marks the run failed but never kills the worker."""
+        """Drain the queue forever, launching one run subprocess at a time per
+        worker. A launch failure marks the run failed but never kills the
+        worker, and always frees the run's conflict key."""
         while True:
-            run_id, strict_compat = self._q.get()
+            run_id, strict_compat, key = self._q.take()
             try:
                 self._launch(run_id, strict_compat=strict_compat)
             except Exception as exc:  # noqa: BLE001 - worker must survive
@@ -106,7 +202,7 @@ def _worker_loop(self) -> None:
                 except Exception:
                     pass
             finally:
-                self._q.task_done()
+                self._q.done(key)
 
     def _launch(self, run_id: str, *, strict_compat: bool = False) -> None:
         """Run one reserved run as `python -m infermatrix_copilot --execute-reserved