diff --git a/doc/architecture/SPEC/app/run_service.md b/doc/architecture/SPEC/app/run_service.md index 805ec4b8..ff3d21ad 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 90d85744..003f726d 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 6d6eb861..1d55b8d2 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 c9f9e25e..79d1c767 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 b44df534..641b4cef 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 91109a4b..cc7bcc77 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 98cb04a8..2a8c57a7 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 0718ebd1..720e9300 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 b2527ee4..79ccfc7d 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 ef490142..7f561e08 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 c0d6ec06..d4997c43 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 477ae14d..902bd2e3 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 4835dc02..ccf043df 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 d8ad8f9d..d0999476 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 6990bbec..7cad7f0e 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 00000000..640f8250 --- /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 22af6a8c..66133a38 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