From d39ff4ccde899b26a2de5f075d174e63c0c3d4a4 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Wed, 30 Sep 2026 23:36:03 +0800 Subject: [PATCH] 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 --- doc/architecture/SPEC/app/run_service.md | 36 +- doc/architecture/SPEC/config.md | 4 +- doc/architecture/SPEC/contract.md | 7 +- doc/architecture/SPEC/engine/worktrees.md | 6 +- doc/architecture/SPEC/idempotency.md | 4 +- doc/architecture/SPEC/providers/cursor.md | 13 +- doc/architecture/SPEC/sdk.md | 11 +- src/infermatrix_copilot/app/run_service.py | 138 ++++- src/infermatrix_copilot/config.py | 19 + src/infermatrix_copilot/contract.py | 8 +- src/infermatrix_copilot/engine/worktrees.py | 16 + src/infermatrix_copilot/idempotency.py | 5 + src/infermatrix_copilot/providers/cursor.py | 109 +++- src/infermatrix_copilot/sdk/v1/models.py | 8 +- src/infermatrix_copilot/sdk/v1/strict.py | 1 + test/test_run_service_concurrency.py | 543 ++++++++++++++++++++ test/test_sdk_v1.py | 2 +- 17 files changed, 878 insertions(+), 52 deletions(-) create mode 100644 test/test_run_service_concurrency.py diff --git a/doc/architecture/SPEC/app/run_service.md b/doc/architecture/SPEC/app/run_service.md index 805ec4b8f..ff3d21adb 100644 --- a/doc/architecture/SPEC/app/run_service.md +++ b/doc/architecture/SPEC/app/run_service.md @@ -1,10 +1,10 @@ # app/run_service.py —— 规范 - + `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. diff --git a/doc/architecture/SPEC/config.md b/doc/architecture/SPEC/config.md index 90d857447..003f726d6 100644 --- a/doc/architecture/SPEC/config.md +++ b/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 把同一条记录 diff --git a/doc/architecture/SPEC/contract.md b/doc/architecture/SPEC/contract.md index 6d6eb8619..1d55b8d24 100644 --- a/doc/architecture/SPEC/contract.md +++ b/doc/architecture/SPEC/contract.md @@ -1,6 +1,6 @@ # contract.py —— 规范 - + `旧版跨仓库契约兼容层 · 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)`（显式 diff --git a/doc/architecture/SPEC/engine/worktrees.md b/doc/architecture/SPEC/engine/worktrees.md index c9f9e25e3..79d1c767c 100644 --- a/doc/architecture/SPEC/engine/worktrees.md +++ b/doc/architecture/SPEC/engine/worktrees.md @@ -1,6 +1,6 @@ # engine/worktrees.py —— 规范 - + `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 恢复路径，并在树被清理后 diff --git a/doc/architecture/SPEC/idempotency.md b/doc/architecture/SPEC/idempotency.md index b44df534f..641b4cef0 100644 --- a/doc/architecture/SPEC/idempotency.md +++ b/doc/architecture/SPEC/idempotency.md @@ -1,6 +1,6 @@ # idempotency.py —— 规范 - + `LOC ~293 · 持久 idempotency 索引：每键一个 run、一次执行 · refactor-status: ok` @@ -34,6 +34,8 @@ sha256 —— 之后新增的字段自动纳入，绝不把两个不同请求静 另一个、后果大得多的变更。`reap_stale` 只清索引条目、worktree、 `refs/imx//*`。 - reaper 绝不触碰活 run 的条目（`state not in TERMINAL` → 跳过）。 +- worktree 回收：拿到树的 `LOCK_EX` 后**重读 mtime**，期间被复用 + （`worktrees.touch`）的树跳过 —— 无锁的年龄检查与加锁之间可能插入一次复用。 - worktree 回收：存活以树**自己的**共享锁判定（绝不是注册表），移除走 `git worktree remove`（绝不裸 `rmtree`），逐树限时（一棵慢树不拖垮 整个扫描）。 diff --git a/doc/architecture/SPEC/providers/cursor.md b/doc/architecture/SPEC/providers/cursor.md index 91109a4b6..cc7bcc774 100644 --- a/doc/architecture/SPEC/providers/cursor.md +++ b/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`），绝不在别人的配置下静默开跑。 diff --git a/doc/architecture/SPEC/sdk.md b/doc/architecture/SPEC/sdk.md index 98cb04a87..2a8c57a7a 100644 --- a/doc/architecture/SPEC/sdk.md +++ b/doc/architecture/SPEC/sdk.md @@ -1,6 +1,6 @@ # sdk/ —— 规范 - + `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 握手，避免只按 diff --git a/src/infermatrix_copilot/app/run_service.py b/src/infermatrix_copilot/app/run_service.py index 0718ebd14..720e93006 100644 --- a/src/infermatrix_copilot/app/run_service.py +++ b/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,16 +115,19 @@ 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. @@ -76,27 +135,64 @@ def _reap(self) -> None: 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 diff --git a/src/infermatrix_copilot/config.py b/src/infermatrix_copilot/config.py index b2527ee46..79ccfc7de 100644 --- a/src/infermatrix_copilot/config.py +++ b/src/infermatrix_copilot/config.py @@ -13,6 +13,9 @@ _PACKAGE_ROOT = Path(__file__).resolve().parent _REPO_ROOT = Path(__file__).resolve().parents[2] _USER_CONFIG = Path.home() / ".infermatrix-copilot" / ".env" +# Upper bound on concurrently executing runs in one run service; each run is a +# subprocess driving a vendor harness, so this is a sanity cap, not a target. +MAX_STRICT_WORKERS_LIMIT = 32 # Glob patterns, not fixed paths: the adapters probe must not name a specific repo # (the core stays repo-neutral — DESIGN §V2.2.1), so it asks "is there ANY adapter # directory here?" rather than naming one. A plain filename globs to itself, so one @@ -159,6 +162,11 @@ class Settings(BaseSettings): # lost-response retry return the finished run instead of re-reviewing), so # it has to be bounded explicitly. Far longer than any retry window. idem_retention_days: int = 30 + # How many reserved runs the run service executes at once (each is its + # own subprocess). Reported to consumers as `max_strict_workers`. Runs on + # the same checkout+PR never overlap regardless (run_service.py), because + # they share the PR-time worktree that harness sessions write config into. + strict_max_workers: int = 1 # Strict execution backend (doc/features/provider-registry.md): which provider # powers runs. REQUIRED for Strict — `strict_readiness` names the exact @@ -496,6 +504,17 @@ def _validate_strict_backend(cls, v): f"(or unset), got {v!r}") return v + @field_validator("strict_max_workers") + @classmethod + def _validate_strict_max_workers(cls, v): + """A zero or negative worker count would accept runs that never + execute; an absurd one is a typo, not a capacity plan.""" + if not 1 <= v <= MAX_STRICT_WORKERS_LIMIT: + raise ValueError( + f"STRICT_MAX_WORKERS must be between 1 and " + f"{MAX_STRICT_WORKERS_LIMIT}, got {v}") + return v + @model_validator(mode="after") def _validate_tier_atomicity(self): """Reject partial tier backends at startup: URL or key alone could pair diff --git a/src/infermatrix_copilot/contract.py b/src/infermatrix_copilot/contract.py index ef4901426..7f561e082 100644 --- a/src/infermatrix_copilot/contract.py +++ b/src/infermatrix_copilot/contract.py @@ -104,9 +104,11 @@ def capabilities(*, max_strict_workers: int = 1, supports_file_locking: bool = True) -> dict[str, Any]: """What this copilot can do, for a consumer's preflight. - `max_strict_workers` is reported, not assumed: the MCP server drains its - queue with a single worker, so a bot that fans out Strict requests should - know they serialize rather than infer concurrency that does not exist.""" + `max_strict_workers` is reported, not assumed: the run service drains its + queue with `STRICT_MAX_WORKERS` workers (default 1), so a bot that fans out + Strict requests should size its own admission to this rather than infer + concurrency from the fact that `start` returns immediately. Runs on the + same checkout+PR still execute one at a time whatever this says.""" # Compatibility shim: new consumers import ``infermatrix_copilot.sdk.v1``; # existing MCP and cross-repo callers keep receiving the legacy dict shape, # now with the distribution/SDK/resource identity added. diff --git a/src/infermatrix_copilot/engine/worktrees.py b/src/infermatrix_copilot/engine/worktrees.py index c0d6ec062..d4997c43b 100644 --- a/src/infermatrix_copilot/engine/worktrees.py +++ b/src/infermatrix_copilot/engine/worktrees.py @@ -143,6 +143,20 @@ def _flock_blocking(fd: int, flags: int, timeout: float) -> bool: time.sleep(0.05) +def touch(dest: Path) -> None: + """Mark a reused tree as in use NOW, while the exclusive lock is held. + + The reaper removes trees older than the retention window that no run + holds, and a run takes its shared hold only when a later step first uses + the tree — so between `materialize` releasing its lock and that hold, an + old reused tree looked abandoned. A fresh mtime closes that gap: the + reaper re-reads it under its own exclusive lock before removing.""" + try: + os.utime(dest) + except OSError: + pass # best-effort; the reaper's lock still guards a held tree + + def owned_by(repo: Path, dest: Path, sha: str, git: GitRunner) -> tuple[bool, str]: """Whether `dest` is a worktree of `repo` currently at `sha`. @@ -202,6 +216,7 @@ def materialize_mutable(repo: Path, sha: str, dest: Path, git: GitRunner, *, ok, why = owned_by_repository(repo, dest, git) if not ok: return False, f"private worktree path is not owned ({why})" + touch(dest) return True, "reused private worktree" dest.parent.mkdir(parents=True, exist_ok=True) code, out = git(repo, "worktree", "add", "--detach", str(dest), sha) @@ -232,6 +247,7 @@ def materialize(repo: Path, sha: str, dest: Path, git: GitRunner, *, if dest.exists(): ok, why = owned_by(repo, dest, sha, git) if ok: + touch(dest) return True, f"reused worktree @ {sha[:12]}" # Safe only because the name encodes repo+sha: a mismatch here # is a foreign or torn tree, never another run's live one. diff --git a/src/infermatrix_copilot/idempotency.py b/src/infermatrix_copilot/idempotency.py index 477ae14d0..902bd2e38 100644 --- a/src/infermatrix_copilot/idempotency.py +++ b/src/infermatrix_copilot/idempotency.py @@ -247,6 +247,11 @@ def _reap_worktrees(cutoff: float, worktree_root: Path | None) -> int: except OSError: continue # a run holds it try: + # Re-read under the lock: a run may have reused (and touched) + # this tree after the unlocked check above, then released its + # materialize lock before taking its hold. + if dest.stat().st_mtime >= cutoff: + continue code, out = _git(dest, "rev-parse", "--git-common-dir", timeout=30) owner = Path(out.strip()) if code == 0 else None diff --git a/src/infermatrix_copilot/providers/cursor.py b/src/infermatrix_copilot/providers/cursor.py index 4835dc028..ccf043df1 100644 --- a/src/infermatrix_copilot/providers/cursor.py +++ b/src/infermatrix_copilot/providers/cursor.py @@ -18,9 +18,12 @@ from __future__ import annotations import json +import os import shutil import subprocess import tempfile +import time +from contextlib import contextmanager from pathlib import Path from ..agent_loop import AgentOutcome @@ -35,6 +38,75 @@ from .registry import PROVIDERS +def _session_lock_path(cwd: Path) -> Path | None: + """Where to serialize cursor sessions in `cwd`, or None when no other + run can be in it. + + The bridge config lives at ONE fixed path per working directory + (`.cursor/mcp.json`), so two sessions of different runs in the same cwd + overwrite each other's config and one agent binds to the other run's tool + scope and trace. A managed PR-time worktree is keyed by repo+PR+sha and the + run service never runs two runs on one PR at once, and a run directory is + one run's own, so neither needs anything here. A checkout does: issue + tasks and PR runs whose worktree could not be materialized all work in the + live checkout. The lock lives in that checkout's git dir, beside git's own + locks, never in the working tree.""" + from ..engine import worktrees + + if worktrees.is_managed_dest(cwd): + return None + dot_git = cwd / ".git" + if dot_git.is_dir(): + return dot_git / "imx-cursor-session.lock" + if dot_git.is_file(): # a linked worktree: `gitdir: ` + try: + line = dot_git.read_text(encoding="utf-8").strip() + except OSError: + return None + if line.startswith("gitdir:"): + admin = Path(line.split(":", 1)[1].strip()) + admin = admin if admin.is_absolute() else cwd / admin + if admin.is_dir(): + return admin / "imx-cursor-session.lock" + return None + + +@contextmanager +def _exclusive_cwd(cwd: Path, timeout_s: float): + """Hold `cwd` exclusively for one session when other runs may share it + (see `_session_lock_path`). An `flock` serializes across threads, + processes and services. Yields False when the hold timed out.""" + from ..engine.lifecycle import fcntl + + lock = _session_lock_path(cwd) + if lock is None or fcntl is None: + yield True + return + fd = os.open(str(lock), os.O_RDWR | os.O_CREAT, 0o644) + try: + deadline = time.monotonic() + timeout_s + while True: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except OSError: + if time.monotonic() >= deadline: + yield False + return + time.sleep(0.2) + try: + yield True + finally: + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + + +@contextmanager +def _no_guard(): + yield True + + class CursorTransport(HarnessTransport): """cursor-agent CLI as a Strict backend.""" @@ -168,18 +240,31 @@ def run_session(self, req: AgentSessionRequest) -> AgentOutcome: cwd = Path(req.scope.root or req.run_dir) created: list[Path] = [] - if req.bridge_spec_path is not None: - created = self._write_mcp_config(cwd, req.bridge_spec_path) - try: - events, timed_out = self._run( - f"{req.system}\n\n{req.prompt}", cwd=str(cwd), - timeout_s=req.timeout_s, model=req.model) - finally: - for path in created: - if path.is_dir(): - shutil.rmtree(path, ignore_errors=True) - else: - path.unlink(missing_ok=True) + # Waiting may outlast one session in the cwd, never an unbounded queue. + wait_s = max(60.0, float(req.timeout_s) + 60.0) + with (_exclusive_cwd(cwd, wait_s) if req.bridge_spec_path is not None + else _no_guard()) as held: + if not held: + reason = (f"working directory {cwd} stayed busy with another " + f"run's cursor session for {wait_s:.0f}s") + if req.trace is not None: + req.trace.record("capability_gap", + capability="cursor.exclusive_cwd", + step=req.step_name, detail=reason) + return AgentOutcome(text="", iterations=0, tool_calls=0, + truncated=True, refusals=[reason]) + if req.bridge_spec_path is not None: + created = self._write_mcp_config(cwd, req.bridge_spec_path) + try: + events, timed_out = self._run( + f"{req.system}\n\n{req.prompt}", cwd=str(cwd), + timeout_s=req.timeout_s, model=req.model) + finally: + for path in created: + if path.is_dir(): + shutil.rmtree(path, ignore_errors=True) + else: + path.unlink(missing_ok=True) audit = audit_events(events, roots=(str(cwd), str(req.run_dir)), read_only=req.scope.read_only, cwd=str(cwd)) usage = self._usage(events) diff --git a/src/infermatrix_copilot/sdk/v1/models.py b/src/infermatrix_copilot/sdk/v1/models.py index d8ad8f9d4..d09994769 100644 --- a/src/infermatrix_copilot/sdk/v1/models.py +++ b/src/infermatrix_copilot/sdk/v1/models.py @@ -7,7 +7,7 @@ SDK_API_VERSION = "1.0.0" DIRECT_API_VERSION = "1.1.0" -STRICT_API_VERSION = "1.3.0" # explicit carried-finding rechecks +STRICT_API_VERSION = "1.4.0" # StrictRuntimeConfig.max_workers QUALITY_API_VERSION = "1.0.0" KNOWLEDGE_API_VERSION = "1.1.0" @@ -315,12 +315,18 @@ class StrictRuntimeConfig(_Serializable): allowed_root: str backend: str = "" run_root: str = "" + # Runs this host executes at once; reported back as `max_strict_workers`. + max_workers: int = 1 def __post_init__(self) -> None: if not self.repository.alias or self.repository.full_name.count("/") != 1: raise InvalidRequestError("Strict runtime needs a repository alias and full name") if not self.checkout_path or not self.allowed_root: raise InvalidRequestError("Strict runtime needs a checkout and allowed root") + if (isinstance(self.max_workers, bool) + or not isinstance(self.max_workers, int) + or self.max_workers < 1): + raise InvalidRequestError("Strict runtime max_workers must be a positive integer") @dataclass(frozen=True) diff --git a/src/infermatrix_copilot/sdk/v1/strict.py b/src/infermatrix_copilot/sdk/v1/strict.py index 6990bbec8..7cad7f0e6 100644 --- a/src/infermatrix_copilot/sdk/v1/strict.py +++ b/src/infermatrix_copilot/sdk/v1/strict.py @@ -123,6 +123,7 @@ def __init__( overrides["strict_backend"] = config.backend if config.run_root: overrides["run_root"] = config.run_root + overrides["strict_max_workers"] = config.max_workers else: overrides = settings_overrides or {} if overrides.get("_env_file", object()) is None: diff --git a/test/test_run_service_concurrency.py b/test/test_run_service_concurrency.py new file mode 100644 index 000000000..640f82500 --- /dev/null +++ b/test/test_run_service_concurrency.py @@ -0,0 +1,543 @@ +"""The run service executes up to `STRICT_MAX_WORKERS` runs at once, never two +on the same checkout+PR, and its housekeeping never removes a tree a run is +about to use. Real threads, fake launches; all offline.""" + +from __future__ import annotations + +import json +import os +import subprocess +import threading +import time +from pathlib import Path + +import pytest + +from infermatrix_copilot import idempotency as idem +from infermatrix_copilot import run_status as rs +from infermatrix_copilot.app.run_service import UNKEYED, RunQueue, RunService +from infermatrix_copilot.config import Settings + +WAIT = 5.0 + + +def _service(settings, workers: int) -> RunService: + settings.strict_max_workers = workers + return RunService(settings) + + +def _queue_run(core: RunService, run_id: str, *, pr, repo: str = "vllm-omni", + repo_path: str = "") -> str: + """A reserved-looking run: the persisted request is what the queue keys + on, and the queued status is what a failed launch marks.""" + run_dir = core.run_root / run_id + run_dir.mkdir(parents=True, exist_ok=True) + request = {"kind": "pr_review" if pr is not None else "issue_filter", + "repo": repo, "pr": pr, "repo_path": repo_path} + (run_dir / "request.json").write_text(json.dumps(request), encoding="utf-8") + rs.init_queued(run_dir, run_id=run_id, owner_server_id=core.server_id, + owner_server_pid=core.pid) + return run_id + + +class _Recorder: + """A fake `_launch` that blocks each run until released and records how + many ran at once, overall and per PR.""" + + def __init__(self): + self.lock = threading.Lock() + self.running: dict[str, int] = {} + self.started: list[str] = [] + self.peak = 0 + self.gates: dict[str, threading.Event] = {} + self.started_event = threading.Condition(self.lock) + + def gate(self, run_id: str) -> threading.Event: + with self.lock: + return self.gates.setdefault(run_id, threading.Event()) + + def launch(self, run_id: str, *, strict_compat: bool = False) -> None: + with self.lock: + self.running[run_id] = 1 + self.started.append(run_id) + self.peak = max(self.peak, len(self.running)) + self.started_event.notify_all() + try: + assert self.gate(run_id).wait(WAIT), f"{run_id} never released" + finally: + with self.lock: + self.running.pop(run_id, None) + + def wait_started(self, *run_ids: str) -> None: + deadline = time.monotonic() + WAIT + with self.lock: + while not set(run_ids) <= set(self.started): + left = deadline - time.monotonic() + assert left > 0, f"not started: {set(run_ids) - set(self.started)}" + self.started_event.wait(left) + + def release(self, run_id: str) -> None: + self.gate(run_id).set() + + +def _wait_until(predicate, what: str) -> None: + deadline = time.monotonic() + WAIT + while not predicate(): + assert time.monotonic() < deadline, what + time.sleep(0.01) + + +# ── configuration and handshake ─────────────────────────────────────────────── +def test_default_is_one_worker_and_the_handshake_says_so(settings): + core = RunService(settings) + assert core.capabilities()["max_strict_workers"] == 1 + assert len(core._workers) == 1 + + +def test_configured_workers_are_started_and_reported(settings): + core = _service(settings, 4) + assert core.capabilities()["max_strict_workers"] == 4 + assert len(core._workers) == 4 + assert all(worker.is_alive() for worker in core._workers) + + +@pytest.mark.parametrize("bad", [0, -1, 33]) +def test_out_of_range_worker_counts_fail_at_startup(tmp_path, bad): + with pytest.raises(ValueError, match="STRICT_MAX_WORKERS"): + Settings(_env_file=None, run_root=tmp_path / "runs", + strict_max_workers=bad) + + +def test_worker_count_is_read_from_the_environment(tmp_path, monkeypatch): + monkeypatch.setenv("STRICT_MAX_WORKERS", "6") + assert Settings(_env_file=None, run_root=tmp_path).strict_max_workers == 6 + + +# ── parallelism ─────────────────────────────────────────────────────────────── +def test_runs_on_different_prs_execute_concurrently(settings, monkeypatch): + core = _service(settings, 3) + rec = _Recorder() + monkeypatch.setattr(core, "_launch", rec.launch) + ids = [_queue_run(core, f"run-20260930-000000-00000{i}", pr=10 + i) + for i in range(3)] + for run_id in ids: + core._q.put((run_id, True)) + + rec.wait_started(*ids) # all three in flight at once + assert rec.peak == 3 + for run_id in ids: + rec.release(run_id) + _wait_until(lambda: not rec.running, "runs never finished") + + +def test_never_more_runs_than_workers(settings, monkeypatch): + core = _service(settings, 2) + rec = _Recorder() + monkeypatch.setattr(core, "_launch", rec.launch) + ids = [_queue_run(core, f"run-20260930-000001-00000{i}", pr=20 + i) + for i in range(4)] + for run_id in ids: + core._q.put((run_id, True)) + + rec.wait_started(ids[0], ids[1]) + time.sleep(0.2) # give an over-eager third worker the chance to exist + assert rec.peak == 2 and len(rec.started) == 2 + for run_id in ids: + rec.release(run_id) + _wait_until(lambda: len(rec.started) == 4 and not rec.running, + "queued runs never drained") + assert rec.peak == 2 + + +def test_runs_on_the_same_pr_never_overlap(settings, monkeypatch): + """They share the PR-time worktree, and a harness writes its tool-bridge + config into that tree — overlapping would bind one run's agent to the + other's tool scope.""" + core = _service(settings, 3) + rec = _Recorder() + monkeypatch.setattr(core, "_launch", rec.launch) + first = _queue_run(core, "run-20260930-000002-000001", pr=7) + second = _queue_run(core, "run-20260930-000002-000002", pr=7) + other = _queue_run(core, "run-20260930-000002-000003", pr=8) + for run_id in (first, second, other): + core._q.put((run_id, True)) + + # the busy PR does not stall the queue behind it + rec.wait_started(first, other) + time.sleep(0.2) + assert second not in rec.started + queued, busy = core._q.snapshot() + assert queued == [second] + + rec.release(first) + rec.wait_started(second) + for run_id in (second, other): + rec.release(run_id) + _wait_until(lambda: not rec.running, "runs never finished") + + +def test_issue_tasks_on_one_checkout_serialize(settings, monkeypatch): + """Issue tasks carry no PR and work in the live checkout itself.""" + core = _service(settings, 2) + rec = _Recorder() + monkeypatch.setattr(core, "_launch", rec.launch) + a = _queue_run(core, "run-20260930-000003-000001", pr=None) + b = _queue_run(core, "run-20260930-000003-000002", pr=None) + for run_id in (a, b): + core._q.put((run_id, False)) + + rec.wait_started(a) + time.sleep(0.2) + assert b not in rec.started + rec.release(a) + rec.wait_started(b) + rec.release(b) + _wait_until(lambda: not rec.running, "runs never finished") + + +def test_conflict_key_is_the_checkout_plus_pr(settings, tmp_path): + settings.repo_paths = {"vllm-omni": str(tmp_path / "clone")} + core = RunService(settings) + a = _queue_run(core, "run-20260930-000004-000001", pr=5) + b = _queue_run(core, "run-20260930-000004-000002", pr=5, + repo_path=str(tmp_path / "clone" / ".")) + c = _queue_run(core, "run-20260930-000004-000003", pr=5, + repo_path=str(tmp_path / "other-clone")) + # an explicit path naming the configured checkout is the same checkout + assert core._conflict_key(a) == core._conflict_key(b) + # the same PR number in a different clone is a different tree + assert core._conflict_key(a) != core._conflict_key(c) + # an unreadable request still gets a key, shared with its kind + assert core._conflict_key("run-20260930-000004-999999") == UNKEYED + + +def test_a_malformed_request_still_enqueues_and_never_kills_a_worker( + settings, monkeypatch): + """A tampered `pr` (unhashable) must not raise inside `take()` — that would + kill the worker holding the queue's condition.""" + core = _service(settings, 1) + rec = _Recorder() + monkeypatch.setattr(core, "_launch", rec.launch) + bad = _queue_run(core, "run-20260930-000006-000001", pr=[1, 2]) + boom = "run-20260930-000006-000002" + _queue_run(core, boom, pr=3) + monkeypatch.setattr(core, "_conflict_key", + lambda run_id: (_ for _ in ()).throw(RuntimeError("x")) + if run_id == boom else RunService._conflict_key(core, run_id)) + core._q.put((bad, True)) + core._q.put((boom, True)) # the key function itself blew up + rec.wait_started(bad) + rec.release(bad) + rec.wait_started(boom) # enqueued under UNKEYED, still executed + rec.release(boom) + _wait_until(lambda: not rec.running and core._workers[0].is_alive(), + "the worker died on a malformed request") + + +# ── failure isolation ───────────────────────────────────────────────────────── +def test_a_failed_launch_marks_the_run_and_frees_its_pr(settings, monkeypatch): + core = _service(settings, 2) + rec = _Recorder() + boom = "run-20260930-000005-000001" + after = "run-20260930-000005-000002" + + def launch(run_id, *, strict_compat=False): + if run_id == boom: + raise RuntimeError("spawn failed") + rec.launch(run_id, strict_compat=strict_compat) + + monkeypatch.setattr(core, "_launch", launch) + for run_id in (boom, after): + _queue_run(core, run_id, pr=9) + core._q.put((run_id, True)) + + rec.wait_started(after) # same PR: only possible once the key was freed + status = rs.read_status(core.run_root / boom) or {} + assert status.get("state") == rs.FAILED + assert "spawn failed" in status.get("note", "") + rec.release(after) + _wait_until(lambda: all(w.is_alive() for w in core._workers) and not rec.running, + "a worker died with its launch") + + +# ── the queue itself ────────────────────────────────────────────────────────── +def test_queue_hands_out_the_oldest_run_whose_key_is_free(): + keys = {"a1": "A", "a2": "A", "b1": "B"} + q = RunQueue(keys.__getitem__) + for run_id in ("a1", "a2", "b1"): + q.put((run_id, False)) + assert q.take()[0] == "a1" + assert q.take()[0] == "b1" # a2 waits behind a1's key, b1 skips ahead + got: list[str] = [] + waiter = threading.Thread(target=lambda: got.append(q.take()[0])) + waiter.start() + time.sleep(0.1) + assert got == [] # a2's key is still busy + q.done("A") + waiter.join(WAIT) + assert got == ["a2"] + + +# ── housekeeping vs live runs ───────────────────────────────────────────────── +def test_only_one_sweep_runs_at_a_time(settings, monkeypatch): + core = RunService(settings) + calls = [] + monkeypatch.setattr(idem, "reap_stale", lambda *a, **k: calls.append(1)) + with core._reap_lock: + core._reap() # another worker is already sweeping + assert calls == [] + core._reap() + assert calls == [1] + + +def _sha(repo: Path) -> str: + return subprocess.run(["git", "rev-parse", "HEAD"], cwd=repo, + capture_output=True, text=True).stdout.strip() + + +def test_reusing_an_old_tree_keeps_the_reaper_off_it(settings, tmp_path, + git_repo): + """A run takes its shared hold only when a later step first uses the tree, + so reuse itself must mark the tree live.""" + from infermatrix_copilot.engine import worktrees as wt + from infermatrix_copilot.engine.steps._common import git as _git + + root = tmp_path / "worktrees" + sha = _sha(git_repo) + dest = wt.dest_for(git_repo, 1, sha, root=root) + assert wt.materialize(git_repo, sha, dest, _git)[0] + os.utime(dest, (0, 0)) # older than any retention window + + ok, detail = wt.materialize(git_repo, sha, dest, _git) + assert ok and detail.startswith("reused") + idem.reap_stale(settings.run_root, worktree_root=root) + assert dest.exists() + + +def test_reaper_rechecks_age_under_its_lock(settings, tmp_path, git_repo, + monkeypatch): + """A reuse can land between the reaper's unlocked age check and its + exclusive lock; the removal decision is re-read under the lock.""" + from infermatrix_copilot.engine import worktrees as wt + from infermatrix_copilot.engine.steps._common import git as _git + + root = tmp_path / "worktrees" + sha = _sha(git_repo) + dest = wt.dest_for(git_repo, 3, sha, root=root) + assert wt.materialize(git_repo, sha, dest, _git)[0] + os.utime(dest, (0, 0)) + + real_flock = idem.fcntl.flock + + def flock(fd, flags): + if flags & idem.fcntl.LOCK_EX: + os.utime(dest) # a run reused the tree just before this lock + return real_flock(fd, flags) + + monkeypatch.setattr(idem.fcntl, "flock", flock) + counts = idem.reap_stale(settings.run_root, worktree_root=root) + assert dest.exists() + assert counts["worktrees"] == 0 + + +# ── the embedded SDK binding ────────────────────────────────────────────────── +def test_sdk_config_sets_the_worker_count_and_the_handshake_reports_it( + tmp_path, monkeypatch): + from infermatrix_copilot.sdk.v1 import (RepositoryRef, StrictRuntime, + StrictRuntimeConfig) + + # an explicit config beats whatever the host environment says + monkeypatch.setenv("STRICT_MAX_WORKERS", "1") + config = StrictRuntimeConfig( + repository=RepositoryRef("demo", "owner/demo"), + checkout_path=str(tmp_path), allowed_root=str(tmp_path), + run_root=str(tmp_path / "runs"), max_workers=5) + with StrictRuntime(config=config) as runtime: + assert runtime.capabilities().max_strict_workers == 5 + assert len(runtime._core._workers) == 5 + + +@pytest.mark.parametrize("bad", [0, -2, True, "3"]) +def test_sdk_config_refuses_a_bad_worker_count(tmp_path, bad): + from infermatrix_copilot.sdk.v1 import (InvalidRequestError, RepositoryRef, + StrictRuntimeConfig) + + with pytest.raises(InvalidRequestError, match="max_workers"): + StrictRuntimeConfig( + repository=RepositoryRef("demo", "owner/demo"), + checkout_path=str(tmp_path), allowed_root=str(tmp_path), + max_workers=bad) + + +# ── the checkout key goes through the execution path's resolver ────────────── +_ADAPTER_MANIFEST = """\ +name: vllm_omni +status: active +repo: + path: {repo_path} + default_branch: main +push: + default_remote: origin + protected_branches: [main] +""" + + +def test_adapter_backed_checkout_keys_implicit_and_explicit_alike( + settings, git_repo): + """No REPO_PATHS entry: the adapter manifest supplies the checkout, and a + request naming that checkout explicitly is the same tree.""" + adapter = settings.adapters_dir / "vllm_omni" + adapter.mkdir(parents=True) + (adapter / "manifest.yaml").write_text( + _ADAPTER_MANIFEST.format(repo_path=git_repo)) + core = RunService(settings) + implicit = _queue_run(core, "run-20260930-000007-000001", pr=11) + explicit = _queue_run(core, "run-20260930-000007-000002", pr=11, + repo_path=str(git_repo)) + assert core._conflict_key(implicit) == core._conflict_key(explicit) + assert core._conflict_key(implicit)[0] == str(git_repo.resolve()) + + +# ── cursor sessions in a shared checkout ────────────────────────────────────── +_RECORDING_CLI = """#!/usr/bin/env python3 +import json, os, sys, time, uuid +here = os.path.dirname(os.path.abspath(__file__)) +sys.stdin.read() +start = time.time() +cfg = json.load(open(os.path.join(os.getcwd(), ".cursor", "mcp.json"))) +spec = cfg["mcpServers"]["infermatrix-tools"]["args"][-1] +time.sleep(0.4) +cfg_after = json.load(open(os.path.join(os.getcwd(), ".cursor", "mcp.json"))) +with open(os.path.join(here, "calls", uuid.uuid4().hex + ".json"), "w") as f: + json.dump({"start": start, "end": time.time(), "spec": spec, + "spec_after": cfg_after["mcpServers"]["infermatrix-tools"]["args"][-1]}, f) +print(json.dumps({"type": "result", "result": "OK " + spec})) +""" + + +def _recording_transport(tmp_path: Path): + import stat + + from infermatrix_copilot.providers.cursor import CursorTransport + + cli = tmp_path / "bin" / "cursor-agent" + (tmp_path / "bin" / "calls").mkdir(parents=True) + cli.write_text(_RECORDING_CLI, encoding="utf-8") + cli.chmod(cli.stat().st_mode | stat.S_IXUSR) + return CursorTransport(Settings(_env_file=None, strict_backend="cursor", + strict_backend_cli=str(cli))) + + +def _session(tmp_path: Path, cwd: Path, run: str, timeout_s: float = 30.0): + from infermatrix_copilot.providers.base import AgentSessionRequest + from infermatrix_copilot.scopes import read_only_scope + from infermatrix_copilot.tool_bridge import write_bridge_spec + + run_dir = tmp_path / run + run_dir.mkdir() + base = read_only_scope() + scope = type(base)(name=base.name, allowed_tools=base.allowed_tools, + read_only=True, root=str(cwd)) + bridge = write_bridge_spec(run_dir=run_dir, step_name="agent.answer", + scope=scope, repo="vllm-omni") + return AgentSessionRequest( + system="S", prompt="P", scope=scope, model="", max_iters=4, + timeout_s=timeout_s, run_dir=run_dir, step_name="agent.answer", + bridge_spec_path=bridge, trace=None), str(bridge) + + +def test_cursor_sessions_of_two_runs_in_one_checkout_never_overlap( + tmp_path, git_repo): + """The live checkout is where issue tasks and degraded PR runs work, and + `.cursor/mcp.json` has one path per directory: overlapping sessions would + each read the other run's tool-bridge spec.""" + transport = _recording_transport(tmp_path) + requests = [_session(tmp_path, git_repo, f"run{i}") for i in range(3)] + outcomes: list = [None] * 3 + + def one(i): + outcomes[i] = transport.run_session(requests[i][0]) + + threads = [threading.Thread(target=one, args=(i,)) for i in range(3)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(WAIT * 4) + + calls = sorted((json.loads(p.read_text()) for p in + (tmp_path / "bin" / "calls").glob("*.json")), + key=lambda c: c["start"]) + assert len(calls) == 3 + for earlier, later in zip(calls, calls[1:]): + assert earlier["end"] <= later["start"] # strictly one at a time + assert all(c["spec"] == c["spec_after"] for c in calls) + assert sorted(c["spec"] for c in calls) == sorted(r[1] for r in requests) + assert all(o.text.startswith("OK ") for o in outcomes) + assert not (git_repo / ".cursor").exists() # the tree is restored + + +def test_a_busy_checkout_times_out_loudly(tmp_path, git_repo): + import fcntl + + from infermatrix_copilot.providers.cursor import _session_lock_path + + transport = _recording_transport(tmp_path) + req, _ = _session(tmp_path, git_repo, "run0", timeout_s=0.0) + req = type(req)(**{**req.__dict__, "trace": _Trace()}) + fd = os.open(str(_session_lock_path(git_repo)), os.O_RDWR | os.O_CREAT) + fcntl.flock(fd, fcntl.LOCK_EX) + try: + # another run's session holds it for longer than we may wait + import infermatrix_copilot.providers.cursor as cursor_mod + + real = cursor_mod._exclusive_cwd + outcome = None + with pytest.MonkeyPatch.context() as mp: + mp.setattr(cursor_mod, "_exclusive_cwd", + lambda cwd, _t: real(cwd, 0.3)) + outcome = transport.run_session(req) + finally: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + assert outcome.text == "" and outcome.truncated + assert "busy" in outcome.refusals[0] + assert any(e["kind"] == "capability_gap" for e in req.trace.events) + assert list((tmp_path / "bin" / "calls").glob("*.json")) == [] + + +class _Trace: + def __init__(self): + self.events = [] + + def record(self, kind, **fields): + self.events.append({"kind": kind, **fields}) + + +def test_only_shared_checkouts_are_locked(tmp_path, git_repo): + from infermatrix_copilot.engine import worktrees as wt + from infermatrix_copilot.providers.cursor import _session_lock_path + + # a managed PR-time tree is one PR head's, serialized by the run queue + sha = subprocess.run(["git", "rev-parse", "HEAD"], cwd=git_repo, + capture_output=True, text=True).stdout.strip() + managed = wt.dest_for(git_repo, 4, sha, root=tmp_path / "worktrees") + assert wt.materialize(git_repo, sha, managed, _git_runner())[0] + assert _session_lock_path(managed) is None + # a run's own directory is nobody else's + plain = tmp_path / "run-dir" + plain.mkdir() + assert _session_lock_path(plain) is None + # a clone, and an unmanaged linked worktree, are shared checkouts + assert _session_lock_path(git_repo) == git_repo / ".git" / "imx-cursor-session.lock" + linked = tmp_path / "linked" + subprocess.run(["git", "worktree", "add", "-q", "--detach", str(linked)], + cwd=git_repo, check=True) + lock = _session_lock_path(linked) + assert lock is not None and lock.parent.is_dir() + assert not str(lock).startswith(str(linked) + os.sep) # never in the tree + + +def _git_runner(): + from infermatrix_copilot.engine.steps._common import git as _git + + return _git diff --git a/test/test_sdk_v1.py b/test/test_sdk_v1.py index 22af6a8cf..66133a38f 100644 --- a/test/test_sdk_v1.py +++ b/test/test_sdk_v1.py @@ -406,7 +406,7 @@ def test_strict_runtime_public_capabilities_smoke(tmp_path): caps = runtime.capabilities() assert caps.distribution_version == "0.2.0" - assert caps.strict_api_version == "1.3.0" + assert caps.strict_api_version == "1.4.0" assert caps.knowledge_api_version == KNOWLEDGE_API_VERSION assert caps.supports_knowledge_curation is True