Skip to content

Commit d39ff4c

Browse files
tzhouamclaude
andcommitted
Run up to STRICT_MAX_WORKERS runs concurrently in the run service
RunService drained its queue with one worker thread, so a host that reported max_strict_workers=1 could never run two Strict reviews at once, and the reviewbot's per-backend concurrency caps could not take effect. - Settings.strict_max_workers (STRICT_MAX_WORKERS, 1-32, default 1) and StrictRuntimeConfig.max_workers start that many workers; the handshake reports the configured number. Strict API 1.3.0 -> 1.4.0. - RunQueue never hands out two runs with the same conflict key (resolved checkout + PR; issue tasks key on the checkout alone). Runs on one PR head share one PR-time worktree and cursor writes its tool-bridge config to a fixed path inside it. A busy key keeps its run queued in order without holding a worker. A malformed request cannot kill a worker. The checkout is resolved through RepositoryContextResolver (frozen path, REPO_PATHS, adapter manifest), the same resolver execution uses. - A PR run whose worktree cannot be materialized degrades to the live checkout at runtime, which the queue key cannot see. The cursor transport, the only one that writes its bridge config into the session cwd, now serializes bridge-carrying sessions in a shared checkout with an flock in that checkout's git dir (across services and the CLI too); a timeout returns a loud refusal + capability_gap. - The reaper is single-flight per service, and a reused worktree is touched under its materialize lock while the reaper re-reads mtime under its own lock, closing the gap between materialize releasing the tree and the run taking its shared hold. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent fd79058 commit d39ff4c

17 files changed

Lines changed: 878 additions & 52 deletions

File tree

‎doc/architecture/SPEC/app/run_service.md‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
# app/run_service.py —— 规范
22

3-
<!-- verified-against: 2026-09-28 -->
3+
<!-- verified-against: 2026-09-30 -->
44

55
`RunService` is the durable application boundary shared by the embedded
6-
Strict SDK and MCP transport. It owns policy-checked reserve/start, one
7-
serialized worker, isolated child launch, startup/orphan reconciliation,
6+
Strict SDK and MCP transport. It owns policy-checked reserve/start, a queue
7+
drained by `STRICT_MAX_WORKERS` workers (default 1), isolated child launch, startup/orphan reconciliation,
88
readiness, bounded status/result polling, and repository-scoped knowledge
99
reads. Construction and polling require no CLI or MCP module import.
1010

@@ -39,3 +39,33 @@ run is pinned to this server's effective `knowledge_dir` (reported as `packaged`
3939
or `unverified` for a custom directory), with `KNOWLEDGE_ROOT` removed from the
4040
child's environment. Runs reserved before pinning
4141
existed have no record and use the process default.
42+
43+
## 2026-09-30 Concurrent workers
44+
`Settings.strict_max_workers` (`STRICT_MAX_WORKERS`, 1–32, default 1; embedded
45+
hosts set it through `StrictRuntimeConfig.max_workers`) starts that many worker
46+
threads, and `capabilities()` reports the same number as `max_strict_workers`.
47+
Each worker launches one child at a time, so at most N children run at once.
48+
49+
`RunQueue` never hands out two runs with the same **conflict key** at once:
50+
the checkout (resolved through the same `RepositoryContextResolver` execution
51+
uses: frozen path, `REPO_PATHS`, then the adapter manifest) plus the PR number (`None` for issue tasks, which work in
52+
the live checkout). Runs on one PR head share one PR-time worktree, and harness
53+
sessions write their tool-bridge config into their working directory (cursor's
54+
`.cursor/mcp.json` has one fixed path per tree), so overlapping them would bind
55+
one run's agent to the other run's tool scope and trace. A run whose key is busy
56+
stays queued in order without holding a worker; workers take the oldest run
57+
whose key is free, so one busy PR cannot stall other PRs. The key is computed
58+
from the persisted `request.json` at enqueue; an unreadable request shares the
59+
key `("?", None)`. A failed launch marks the run `failed` and always frees its
60+
key. `_reap()` is single-flight per service (a non-blocking lock): a worker that
61+
finishes while another is sweeping skips its sweep.
62+
63+
The key cannot see a runtime fallback: a PR run whose worktree could not be
64+
materialized degrades to the live checkout, where issue tasks and other
65+
degraded runs also work. That shared directory is guarded where the hazard
66+
is — the cursor transport serializes bridge-carrying sessions in a shared
67+
checkout with an `flock` in its git dir (`providers/cursor.md`), across
68+
services and the CLI as well. Not covered: two agent sessions *inside one run*
69+
(ensemble lenses, per-comment verification, bounded by
70+
`strict_backend_concurrency`) still share their run's own tree; that predates
71+
this change and is independent of the worker count.

‎doc/architecture/SPEC/config.md‎

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

‎doc/architecture/SPEC/contract.md‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# contract.py —— 规范
22

3-
<!-- verified-against: 2026-09-28 -->
3+
<!-- verified-against: 2026-09-30 -->
44

55
`旧版跨仓库契约兼容层 · refactor-status: compatibility-shim`
66

@@ -13,14 +13,15 @@
1313

1414
## 公开契约(`__all__`)
1515
`SDK_API_VERSION` / `DIRECT_API_VERSION` / `STRICT_API_VERSION` /
16-
`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`
16+
`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`
1717
取唯一值)。`capabilities(max_strict_workers=1,
1818
supports_file_locking=True) -> dict` 委托 SDK typed handshake 再投影为兼容
1919
dict;它包含 distribution/SDK/Direct/Strict/Quality/Knowledge 版本、resource
2020
revision、supported repositories,以及 `supports_expected_head`、
2121
`supports_structured_result`、`supports_post_false`、`supports_file_locking`、
2222
`supports_idempotent_strict_start`、`supports_knowledge_curation`、
23-
`max_strict_workers`。其余兼容导出包括 `build_review_result(run_dir) -> dict`(结构化评审
23+
`max_strict_workers`(服务真实的并发 worker 数,即 `STRICT_MAX_WORKERS`,
24+
同一 checkout+PR 的 run 仍逐个执行)。其余兼容导出包括 `build_review_result(run_dir) -> dict`(结构化评审
2425
结果:`contract_version`、`run_id`、`state`、`reviewed_head_sha`、`verdict`、
2526
`summary_markdown`、`comments`、`finding_dispositions`、`findings`、
2627
`stale`/`expected_head_sha`/`actual_head_sha`、`diagnostics`);`unknown_run_result(run_id)`(显式

‎doc/architecture/SPEC/engine/worktrees.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# engine/worktrees.py —— 规范
22

3-
<!-- verified-against: 2026-09-26 -->
3+
<!-- verified-against: 2026-09-30 -->
44

55
`LOC ~294 · PR-time worktree:身份、物化、持有 · refactor-status: ok`
66

@@ -38,6 +38,10 @@ worktree,允许本地 rebase,又不会改配置的 checkout 或复用评审
3838
恰是 head 门要防的"评错树"。
3939
- 复用验证:`--git-common-dir` 必须解析回请求方仓库自己的 git dir **且**
4040
HEAD == sha;force-remove 只对外来/撕裂的树成立,绝不对同身份的活树。
41+
- 复用时在 `LOCK_EX` 内 `touch(dest)` 刷新 mtime:run 的共享持有要等到
42+
后续 step 首次使用树时才取,`materialize` 释放锁到那一刻之间,一棵超过
43+
保留期的旧树看起来无人使用;新 mtime 让 reaper(它在自己的 `LOCK_EX` 下
44+
重读 mtime)放过它。
4145
- 可变树按 run_dir 分键,复用时只核对仓库归属(rebase 会改变 HEAD),
4246
拒绝 symlink、重定向到配置 checkout 的路径及外来仓库;调用方仅在 checkout
4347
step 重试时可重置此树。后续 step 经 checkpoint 恢复路径,并在树被清理后

‎doc/architecture/SPEC/idempotency.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# idempotency.py —— 规范
22

3-
<!-- verified-against: 2026-08-28 -->
3+
<!-- verified-against: 2026-09-30 -->
44

55
`LOC ~293 · 持久 idempotency 索引:每键一个 run、一次执行 · refactor-status: ok`
66

@@ -34,6 +34,8 @@ sha256 —— 之后新增的字段自动纳入,绝不把两个不同请求静
3434
另一个、后果大得多的变更。`reap_stale` 只清索引条目、worktree、
3535
`refs/imx/<run_id>/*`。
3636
- reaper 绝不触碰活 run 的条目(`state not in TERMINAL` → 跳过)。
37+
- worktree 回收:拿到树的 `LOCK_EX` 后**重读 mtime**,期间被复用
38+
(`worktrees.touch`)的树跳过 —— 无锁的年龄检查与加锁之间可能插入一次复用。
3739
- worktree 回收:存活以树**自己的**共享锁判定(绝不是注册表),移除走
3840
`git worktree remove`(绝不裸 `rmtree`),逐树限时(一棵慢树不拖垮
3941
整个扫描)。

‎doc/architecture/SPEC/providers/cursor.md‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ headless `cursor-agent --print --force --output-format stream-json`,
3333
stdlib + `.base` + `.registry` + `..agent_loop.AgentOutcome` + `..llm` 的类型。
3434

3535
## 测试
36-
`test_provider_cursor.py`。
36+
`test_provider_cursor.py`;共享 checkout 串行见 `test_run_service_concurrency.py`。
3737

3838
## 重构备注
3939
这套调用形状是由 Composer 评测臂(`eval/dataset/run_cursor_arm.py`)验证出来的;
@@ -51,3 +51,14 @@ scope 约束,native 调用只被**记录**。
5151

5252
## 2026-09-30
5353
`complete()` 接受并忽略 `max_budget_usd`;`stops_at_spend` 为 False(CLI 无花费阈值)。
54+
55+
### 共享 checkout 上的会话串行(2026-09-30)
56+
桥配置写在会话 cwd 的**固定路径** `.cursor/mcp.json`,同一目录里两个不同 run 的会话
57+
会互相覆盖配置,令一方的 agent 绑定到另一个 run 的工具 scope 与 trace。
58+
带桥的会话若 cwd 是**共享 checkout**(`.git` 为目录的 clone,或未托管的 linked
59+
worktree),就在该 checkout 的 git dir 里 `imx-cursor-session.lock` 上持
60+
`flock LOCK_EX` 直到会话结束并清理配置,跨线程/进程/服务串行——issue 任务与
61+
worktree 物化失败而降级到 live checkout 的 PR run 都在这里。托管的 PR-time
62+
worktree(repo+PR+sha 分键,run queue 不会让同 PR 的两个 run 并发)与 run 自己的
63+
目录不加锁。等待上限为会话超时 + 60 s;超时则返回空文本、`truncated`、refusal,
64+
并记 `capability_gap`(`cursor.exclusive_cwd`),绝不在别人的配置下静默开跑。

‎doc/architecture/SPEC/sdk.md‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# sdk/ —— 规范
22

3-
<!-- verified-against: 2026-09-29 -->
3+
<!-- verified-against: 2026-09-30 -->
44

55

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

‎src/infermatrix_copilot/app/run_service.py‎

Lines changed: 117 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,13 @@
88

99
import json
1010
import os
11-
import queue
1211
import subprocess
1312
import sys
1413
import threading
1514
import uuid
15+
from collections import deque
1616
from pathlib import Path
17-
from typing import Any
17+
from typing import Any, Callable
1818

1919
from .. import contract
2020
from .. import idempotency as idem
@@ -25,7 +25,7 @@
2525
PolicyError, authorize_repo_path, enforce_mcp_policy,
2626
enforce_quality_review_policy, enforce_strict_review_policy,
2727
)
28-
from ..task_spec import READ_ONLY_KINDS
28+
from ..task_spec import READ_ONLY_KINDS, TaskSpec
2929
from .core import Copilot
3030
from .reservation import RunReservation
3131

@@ -42,10 +42,66 @@ def read_knowledge_pin(run_dir: str | Path) -> dict:
4242
return data if isinstance(data, dict) else {}
4343

4444

45+
# The conflict key of a run whose request cannot be read or keyed: such runs
46+
# fail at launch anyway, and sharing one key keeps them from overlapping each
47+
# other rather than guessing what they touch.
48+
UNKEYED = ("?", None)
49+
50+
51+
class RunQueue:
52+
"""FIFO run queue for N workers that never hands out two runs with the
53+
same conflict key at once.
54+
55+
A run whose key is busy stays queued, in order, and does not hold a
56+
worker: the worker takes the oldest run whose key is free, so one busy PR
57+
cannot stall reviews of other PRs behind it. Keys are computed at enqueue
58+
(the request is persisted before any run is enqueued)."""
59+
60+
def __init__(self, key_for: Callable[[str], Any]):
61+
self._key_for = key_for
62+
self._cond = threading.Condition()
63+
self._pending: deque[tuple[str, bool, Any]] = deque()
64+
self._busy: set[Any] = set()
65+
66+
def put(self, item: tuple[str, bool]) -> None:
67+
run_id, strict_compat = item
68+
try:
69+
key = self._key_for(run_id)
70+
hash(key)
71+
except Exception: # noqa: BLE001 - a reserved run must still enqueue
72+
key = UNKEYED
73+
with self._cond:
74+
self._pending.append((run_id, strict_compat, key))
75+
self._cond.notify_all()
76+
77+
def take(self) -> tuple[str, bool, Any]:
78+
"""Block until a run with a free key is queued; claim its key."""
79+
with self._cond:
80+
while True:
81+
for index, item in enumerate(self._pending):
82+
if item[2] not in self._busy:
83+
del self._pending[index]
84+
self._busy.add(item[2])
85+
return item
86+
self._cond.wait()
87+
88+
def done(self, key: Any) -> None:
89+
"""Release a key claimed by `take`, waking workers blocked on it."""
90+
with self._cond:
91+
self._busy.discard(key)
92+
self._cond.notify_all()
93+
94+
def snapshot(self) -> tuple[list[str], set[Any]]:
95+
"""Queued run ids and busy keys (tests/diagnostics)."""
96+
with self._cond:
97+
return [item[0] for item in self._pending], set(self._busy)
98+
99+
45100
class RunService:
46-
"""The server core: a serialized run queue over isolated subprocesses, plus
47-
ownership-aware reconciliation. Framework-agnostic (no `mcp` import) so it is
48-
unit-testable without a live protocol connection."""
101+
"""The server core: a run queue drained by `STRICT_MAX_WORKERS` workers
102+
over isolated subprocesses, plus ownership-aware reconciliation.
103+
Framework-agnostic (no `mcp` import) so it is unit-testable without a live
104+
protocol connection."""
49105

50106
def __init__(self, settings: Settings | None = None):
51107
"""Wire reservation and workflow services, register this server's
@@ -59,44 +115,84 @@ def __init__(self, settings: Settings | None = None):
59115
self.pid = os.getpid()
60116
rs.register_server(self.run_root, self.server_id, self.pid)
61117
rs.startup_reconcile(self.run_root)
118+
# One sweep at a time per service: with several workers finishing
119+
# runs, overlapping sweeps would only race each other for the same
120+
# entries and refs.
121+
self._reap_lock = threading.Lock()
62122
self._reap()
63-
self._q: queue.Queue[tuple[str, bool]] = queue.Queue()
64-
self._worker = threading.Thread(target=self._worker_loop, daemon=True,
65-
name="imx-mcp-worker")
66-
self._worker.start()
67-
68-
# one worker thread drains the queue, so Strict runs are strictly serial.
69-
# The handshake reports this rather than letting a bot infer concurrency
70-
# from the fact that `start` returns immediately.
71-
MAX_STRICT_WORKERS = 1
123+
self.max_strict_workers = int(self.settings.strict_max_workers)
124+
self._q = RunQueue(self._conflict_key)
125+
self._workers = [
126+
threading.Thread(target=self._worker_loop, daemon=True,
127+
name=f"imx-mcp-worker-{i}")
128+
for i in range(self.max_strict_workers)]
129+
for worker in self._workers:
130+
worker.start()
72131

73132
def _reap(self) -> None:
74133
"""Bound what the idempotency index and PR-time worktrees accumulate.
75134
76135
Best-effort by construction: this runs at startup and after each
77136
terminal run, and a sweep that cannot complete must never stop a server
78137
from serving or a run from finishing."""
138+
if not self._reap_lock.acquire(blocking=False):
139+
return # a sweep is already running; the next terminal run retries
79140
try:
80141
idem.reap_stale(self.run_root,
81142
retention_days=self.settings.idem_retention_days,
82143
repo_paths=list(self.settings.repo_paths.values()))
83144
except Exception: # noqa: BLE001 — housekeeping, never load-bearing
84145
pass
146+
finally:
147+
self._reap_lock.release()
85148

86149
def capabilities(self) -> dict:
87150
"""The version/capability handshake — see `contract.capabilities`."""
88151
from ..engine.lifecycle import fcntl as _fcntl
89152

90153
return contract.capabilities(
91-
max_strict_workers=self.MAX_STRICT_WORKERS,
154+
max_strict_workers=self.max_strict_workers,
92155
supports_file_locking=_fcntl is not None)
93156

94-
# -- worker: one run at a time, each an isolated subprocess ---------------
157+
# -- workers: up to N runs at a time, each an isolated subprocess --------
158+
def _conflict_key(self, run_id: str) -> tuple[str, Any]:
159+
"""The resource a run must not share with a concurrently executing one:
160+
its checkout plus its PR (``None`` for issue tasks, which work in the
161+
live checkout itself).
162+
163+
Runs on one PR head share one PR-time worktree (engine/worktrees.py
164+
keys it by repo+PR+sha), and harness sessions write their tool-bridge
165+
config into their working directory — cursor's `.cursor/mcp.json` has
166+
one fixed path per tree — so two such runs overlapping would bind one
167+
run's agent to the other run's tool scope and trace. Different PRs get
168+
different trees and may run side by side."""
169+
try:
170+
spec = json.loads((self.run_root / run_id / "request.json")
171+
.read_text(encoding="utf-8"))
172+
except (OSError, ValueError):
173+
spec = None
174+
if not isinstance(spec, dict):
175+
return UNKEYED
176+
repo = str(spec.get("repo") or self.settings.default_repo)
177+
try:
178+
# the resolver execution itself uses (frozen path, REPO_PATHS,
179+
# then the adapter manifest), so an implicit request and an
180+
# explicit one naming the same checkout get the same key
181+
checkout = self.copilot.repository_context.repo_path_for(
182+
TaskSpec.model_validate(spec))
183+
except Exception: # noqa: BLE001 - such a run fails at launch anyway
184+
checkout = str(spec.get("repo_path") or "")
185+
where = str(Path(checkout).expanduser().resolve()) if checkout \
186+
else f"repo:{repo}"
187+
pr = spec.get("pr")
188+
return (where, None if pr is None else str(pr))
189+
95190
def _worker_loop(self) -> None:
96-
"""Drain the queue forever, launching one run subprocess at a time. A
97-
launch failure marks the run failed but never kills the worker."""
191+
"""Drain the queue forever, launching one run subprocess at a time per
192+
worker. A launch failure marks the run failed but never kills the
193+
worker, and always frees the run's conflict key."""
98194
while True:
99-
run_id, strict_compat = self._q.get()
195+
run_id, strict_compat, key = self._q.take()
100196
try:
101197
self._launch(run_id, strict_compat=strict_compat)
102198
except Exception as exc: # noqa: BLE001 - worker must survive
@@ -106,7 +202,7 @@ def _worker_loop(self) -> None:
106202
except Exception:
107203
pass
108204
finally:
109-
self._q.task_done()
205+
self._q.done(key)
110206

111207
def _launch(self, run_id: str, *, strict_compat: bool = False) -> None:
112208
"""Run one reserved run as `python -m infermatrix_copilot --execute-reserved

0 commit comments

Comments
 (0)