From d0cdafaec91e640b86b13915407ba4bf4d3f29d1 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Sat, 26 Sep 2026 17:32:27 +0800 Subject: [PATCH] Move durable run reservation below Copilot orchestration --- doc/architecture/SPEC/app/core.md | 8 +- doc/architecture/SPEC/app/reservation.md | 20 ++++ doc/architecture/SPEC/app/run_service.md | 5 + src/infermatrix_copilot/app/core.py | 120 ++------------------- src/infermatrix_copilot/app/reservation.py | 114 ++++++++++++++++++++ src/infermatrix_copilot/app/run_service.py | 19 ++-- test/test_idempotency.py | 16 +++ test/test_mcp.py | 9 +- test/test_step_registration_boundary.py | 19 ++++ 9 files changed, 207 insertions(+), 123 deletions(-) create mode 100644 doc/architecture/SPEC/app/reservation.md create mode 100644 src/infermatrix_copilot/app/reservation.py diff --git a/doc/architecture/SPEC/app/core.md b/doc/architecture/SPEC/app/core.md index 20ac6611f..3b29433e8 100644 --- a/doc/architecture/SPEC/app/core.md +++ b/doc/architecture/SPEC/app/core.md @@ -8,10 +8,12 @@ executes or resumes a playbook, and reserves/executes durable runs. The CLI imports it through `cli.copilot` for compatibility; the application never imports CLI or MCP transports. +`Copilot` keeps reservation and run-path methods as compatibility delegates +to `app.reservation.RunReservation`. It owns workflow resolution and child +execution; it does not own idempotency indexing or run ID validation. + The pre-execution gate runs before a normal CLI run directory is created. -Reserved runs persist a canonical request and ownership/idempotency state -before child execution. The child rechecks the read-only policy and checkout -binding; a run ID must stay contained under the configured run root. Existing +The child rechecks the read-only policy and checkout binding. Existing checkpoint keys, run IDs, and blocked exit codes remain stable. `GateOutcome`, `_mode_review_context`, and `Copilot` retain their import path diff --git a/doc/architecture/SPEC/app/reservation.md b/doc/architecture/SPEC/app/reservation.md new file mode 100644 index 000000000..4e625ccc9 --- /dev/null +++ b/doc/architecture/SPEC/app/reservation.md @@ -0,0 +1,20 @@ +# app/reservation.py — durable run reservation + + + +`RunReservation` owns run ID generation and containment, explicit checkout +authorization, request persistence, initial queued status, and idempotent +retry. Importing it does not load the workflow orchestrator, CLI, or MCP +transport. It holds settings but no worker, planner, executor, or model. + +The persisted `request.json` and status schema, run ID format, and 0600 +request permission remain compatible. Keyed reservations lock the key and +return `(run_id, created)`: a resolvable run is reused, while a queued run +orphaned before launch may be reclaimed once. A key with a different spec +fails; an unkeyed request creates a new run. The child validates the frozen +request again before executing it. + +`contained_run_dir` rejects malformed IDs and paths escaping `run_root`. +Polling may request `must_exist=False` to report a valid but unknown ID; +execution requires an existing directory. `RunService` calls this service +directly, and `Copilot` preserves its old reservation methods as delegates. diff --git a/doc/architecture/SPEC/app/run_service.md b/doc/architecture/SPEC/app/run_service.md index fd9243639..0a8ad9d97 100644 --- a/doc/architecture/SPEC/app/run_service.md +++ b/doc/architecture/SPEC/app/run_service.md @@ -8,6 +8,11 @@ serialized worker, 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. +Reservation and run ID containment go directly through `RunReservation`, +not the workflow `Copilot` facade. `Copilot` remains for playbook readiness +and workflow execution; it does not mediate the service's reserve or poll +paths. + Reservations return `(run_id, created)` and enqueue only newly created runs. The worker launches an isolated subprocess with posting and pushing disabled, then reconciles the persisted status. A run can survive host restart; closing diff --git a/src/infermatrix_copilot/app/core.py b/src/infermatrix_copilot/app/core.py index 465ca3156..65aeb59af 100644 --- a/src/infermatrix_copilot/app/core.py +++ b/src/infermatrix_copilot/app/core.py @@ -11,13 +11,11 @@ import asyncio import json import os -import re import time import uuid from pathlib import Path from typing import NamedTuple -from .. import idempotency as idem from .. import run_status as rs from ..config import Settings, TierNotConfiguredError from ..engine.executor import Executor @@ -35,6 +33,7 @@ from ..task_spec import TaskSpec from ..ui import style from ..metrics import format_metrics_line +from .reservation import RunReservation class GateOutcome(NamedTuple): @@ -70,6 +69,7 @@ def __init__(self, settings: Settings | None = None): configured dir, and the planner over both. `last_run_dir` starts unset and is filled by the first execution.""" self.settings = settings or Settings() + self.reservations = RunReservation(self.settings) # provider-registry seam: a real llm.LLM under backend "api" # (byte-identical), a HarnessLLM adapter under a harness backend from ..providers import llm_for @@ -433,121 +433,23 @@ def resume_last(self) -> int: print("no resumable run found") return 1 - # -- MCP surface (reserve + child execute) -------------------------------- - # These support the start/poll MCP server (mcp_server.py). The CLI path - # (run_task/run_playbook) is deliberately untouched: it still gates BEFORE - # creating a run dir, so an aborted plan leaves no directory. Reservation - # (dir before plan) is an MCP-only shape whose blocked/failed outcome is a - # terminal poll record, not litter. - _RUN_ID_RE = re.compile(r"^run-\d{8}-\d{6}-[0-9a-f]{6}$") - + # -- MCP child execution and compatibility reservation facade ----------- @staticmethod def _new_run_id() -> str: - """A fresh unique run id (`run--`; same format the CLI uses).""" - return f"run-{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" + return RunReservation.new_run_id() def _contained_run_dir(self, run_id: str, *, must_exist: bool = True) -> Path: - """Validate `run_id` and resolve it to a directory strictly contained - under `run_root` — rejecting path traversal from an untrusted MCP arg. - Raises ValueError on a bad pattern, an escape, or (unless - `must_exist=False`) a missing run. - - `must_exist=False` is for the poll path, which must distinguish a - well-formed id whose run this server has never heard of — answered as - `state: unknown` so a client can tell "lost" from "still running" — from - a malformed or escaping id, which stays an error.""" - if not self._RUN_ID_RE.match(run_id or ""): - raise ValueError(f"invalid run_id: {run_id!r}") - root = self.settings.run_root.resolve() - run_dir = (self.settings.run_root / run_id).resolve() - if run_dir.parent != root: - raise ValueError(f"run_id escapes run_root: {run_id!r}") - if must_exist and not run_dir.exists(): - raise ValueError(f"no such run: {run_id!r}") - return run_dir + return self.reservations.contained_run_dir(run_id, must_exist=must_exist) def reserve_run(self, spec: TaskSpec, *, owner_server_id: str, owner_server_pid: int, idempotency_key: str = "") -> tuple[str, bool]: - """MCP-only: reserve a run and return `(run_id, created)` **without** - planning or executing (no LLM), so the tool call returns in ms and the - caller polls. Persists `request.json` (0600) + an initial `queued` - `run_status.json` stamped with the owning server; planning happens later - in the child. - - `created` is what the caller enqueues on. Deduplicating the id alone - would not deduplicate execution — the start path would hand back an - existing id and still launch a second child for it. - - With an `idempotency_key`, a retry that lost the original response gets - the same run — including a finished one, whose result it can then read - instead of re-reviewing the same commit. See `idempotency.py` for why a - spec hash alone cannot do this and why the scope is Strict-only. - - A `repo_path` on the spec is authorized and canonicalized HERE, so the - persisted request names one immutable checkout for the life of the run. - The child re-authorizes it anyway — `request.json` is untrusted — but - freezing it at reservation is what stops two concurrent per-call repos - from racing through shared settings, which is how the deleted - `configure_strict_repo` worked.""" - from .request_policy import authorize_repo_path - - if spec.repo_path: - spec = spec.model_copy(update={"repo_path": authorize_repo_path( - spec.repo, spec.repo_path, self.settings)}) - key = idem.validate_key(idempotency_key) - if not key: - return self._reserve_new(spec, owner_server_id, owner_server_pid), True - with idem.key_lock(self.settings.run_root, key): - return self._reserve_keyed(spec, key, owner_server_id, - owner_server_pid) - - def _reserve_keyed(self, spec: TaskSpec, key: str, owner_server_id: str, - owner_server_pid: int) -> tuple[str, bool]: - """Resolve or create the run for `key`. Caller holds the key lock.""" - fingerprint = idem.spec_fingerprint(spec.model_dump()) - entry = idem.read_entry(self.settings.run_root, key) - if entry: - run_dir = self.settings.run_root / str(entry.get("run_id") or "") - # The index is a cache; the persisted request is the authority. A - # key presented with a DIFFERENT spec is a caller error — serving - # the old run would quietly review the wrong thing under a name the - # caller believes means something else. - recorded = str(entry.get("spec_fingerprint") or "") - if run_dir.exists() and recorded and recorded != fingerprint: - raise idem.IdempotencyError( - f"idempotency_key {key!r} was already used for a different " - "request; use a new key for a new attempt") - if run_dir.exists() and idem.resolvable(run_dir): - return run_dir.name, False - if run_dir.exists() and idem.relaunchable(run_dir): - # Reserved but never launched — the owner died between - # publishing this entry and starting a child. Re-arm and report - # it as created so the caller enqueues it; the re-stamp is what - # stops a SECOND retry from enqueueing it again. - if rs.reclaim_queued(run_dir, owner_server_id=owner_server_id, - owner_server_pid=owner_server_pid): - return run_dir.name, True - return run_dir.name, False - run_id = self._reserve_new(spec, owner_server_id, owner_server_pid) - idem.write_entry(self.settings.run_root, key, run_id, fingerprint) - return run_id, True - - def _reserve_new(self, spec: TaskSpec, owner_server_id: str, - owner_server_pid: int) -> str: - """Create the run directory, request and initial status.""" - run_id = self._new_run_id() - run_dir = self.settings.run_root / run_id - run_dir.mkdir(parents=True, exist_ok=True) - req = run_dir / "request.json" - req.write_text(json.dumps(spec.model_dump(), indent=2), encoding="utf-8") - try: # least-privilege perms; advisory only against same-user tampering - os.chmod(req, 0o600) - except OSError: - pass - rs.init_queued(run_dir, run_id=run_id, owner_server_id=owner_server_id, - owner_server_pid=owner_server_pid) - return run_id + """Compatibility entry for callers that still hold a ``Copilot``.""" + return self.reservations.reserve( + spec, owner_server_id=owner_server_id, + owner_server_pid=owner_server_pid, + idempotency_key=idempotency_key, + ) def execute_reserved(self, run_id: str) -> int: """MCP-only child entry (subprocess, stdout -> console.log). Writes its diff --git a/src/infermatrix_copilot/app/reservation.py b/src/infermatrix_copilot/app/reservation.py new file mode 100644 index 000000000..b8cd6135e --- /dev/null +++ b/src/infermatrix_copilot/app/reservation.py @@ -0,0 +1,114 @@ +"""Durable run identity, reservation, and idempotent retry policy. + +This service owns the on-disk reservation contract. It has no planner, +executor, transport, or long-lived worker state. +""" + +from __future__ import annotations + +import json +import os +import re +import time +import uuid +from pathlib import Path + +from .. import idempotency as idem +from .. import run_status as rs +from ..config import Settings +from ..task_spec import TaskSpec +from .request_policy import authorize_repo_path + + +class RunReservation: + _RUN_ID_RE = re.compile(r"^run-\d{8}-\d{6}-[0-9a-f]{6}$") + + def __init__(self, settings: Settings): + self.settings = settings + + @staticmethod + def new_run_id() -> str: + """Create a unique id in the persisted run-directory format.""" + return f"run-{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" + + def contained_run_dir(self, run_id: str, *, must_exist: bool = True) -> Path: + """Resolve an untrusted id strictly inside the configured run root. + + A well-formed missing id may be returned for polling so callers can + distinguish an unknown run from an invalid or escaping id. + """ + if not self._RUN_ID_RE.match(run_id or ""): + raise ValueError(f"invalid run_id: {run_id!r}") + root = self.settings.run_root.resolve() + run_dir = (self.settings.run_root / run_id).resolve() + if run_dir.parent != root: + raise ValueError(f"run_id escapes run_root: {run_id!r}") + if must_exist and not run_dir.exists(): + raise ValueError(f"no such run: {run_id!r}") + return run_dir + + def reserve( + self, spec: TaskSpec, *, owner_server_id: str, + owner_server_pid: int, idempotency_key: str = "", + ) -> tuple[str, bool]: + """Persist a queued run and return ``(run_id, newly_created)``. + + An explicit checkout is authorized and frozen before persistence. + Keyed retries re-use a resolvable run, or re-arm a queued run whose + owner died before it launched, under the idempotency-key lock. + """ + if spec.repo_path: + spec = spec.model_copy(update={"repo_path": authorize_repo_path( + spec.repo, spec.repo_path, self.settings)}) + key = idem.validate_key(idempotency_key) + if not key: + return self._reserve_new(spec, owner_server_id, owner_server_pid), True + with idem.key_lock(self.settings.run_root, key): + return self._reserve_keyed(spec, key, owner_server_id, + owner_server_pid) + + def _reserve_keyed( + self, spec: TaskSpec, key: str, owner_server_id: str, + owner_server_pid: int, + ) -> tuple[str, bool]: + """Resolve or create the run for a key while its lock is held.""" + fingerprint = idem.spec_fingerprint(spec.model_dump()) + entry = idem.read_entry(self.settings.run_root, key) + if entry: + run_dir = self.settings.run_root / str(entry.get("run_id") or "") + # The index is a cache; the persisted request is the authority. + # A reused key for a different spec must not serve the old run. + recorded = str(entry.get("spec_fingerprint") or "") + if run_dir.exists() and recorded and recorded != fingerprint: + raise idem.IdempotencyError( + f"idempotency_key {key!r} was already used for a different " + "request; use a new key for a new attempt") + if run_dir.exists() and idem.resolvable(run_dir): + return run_dir.name, False + if run_dir.exists() and idem.relaunchable(run_dir): + # The owner died after publishing the key but before launch. + # Re-stamp the queued run so only this retry enqueues it. + if rs.reclaim_queued(run_dir, owner_server_id=owner_server_id, + owner_server_pid=owner_server_pid): + return run_dir.name, True + return run_dir.name, False + run_id = self._reserve_new(spec, owner_server_id, owner_server_pid) + idem.write_entry(self.settings.run_root, key, run_id, fingerprint) + return run_id, True + + def _reserve_new( + self, spec: TaskSpec, owner_server_id: str, owner_server_pid: int, + ) -> str: + """Create the run directory, request, and initial queued status.""" + run_id = self.new_run_id() + run_dir = self.settings.run_root / run_id + run_dir.mkdir(parents=True, exist_ok=True) + req = run_dir / "request.json" + req.write_text(json.dumps(spec.model_dump(), indent=2), encoding="utf-8") + try: # least-privilege perms; advisory only against same-user tampering + os.chmod(req, 0o600) + except OSError: + pass + rs.init_queued(run_dir, run_id=run_id, owner_server_id=owner_server_id, + owner_server_pid=owner_server_pid) + return run_id diff --git a/src/infermatrix_copilot/app/run_service.py b/src/infermatrix_copilot/app/run_service.py index e756e0d15..744bb1ac0 100644 --- a/src/infermatrix_copilot/app/run_service.py +++ b/src/infermatrix_copilot/app/run_service.py @@ -27,6 +27,7 @@ ) from ..task_spec import READ_ONLY_KINDS from .core import Copilot +from .reservation import RunReservation class RunService: @@ -35,10 +36,10 @@ class RunService: unit-testable without a live protocol connection.""" def __init__(self, settings: Settings | None = None): - """Wire settings + a `Copilot` (for `reserve_run`/`execute_reserved` path - helpers), register this server's liveness token, reconcile any runs - orphaned by a previous server, and start the single worker thread.""" + """Wire reservation and workflow services, register this server's + liveness token, reconcile orphaned runs, and start the worker.""" self.settings = settings or Settings() + self.reservations = RunReservation(self.settings) self.copilot = Copilot(self.settings) self.run_root = Path(self.settings.run_root) self.run_root.mkdir(parents=True, exist_ok=True) @@ -169,7 +170,7 @@ def start(self, spec_dict: dict) -> str: and `issue_filter`, which carry no PR and no head, so any spec-derived key would collapse every issue task in a repo onto one entry.""" spec = enforce_mcp_policy(spec_dict, allowed_repos=self.settings.mcp_allowed_repos, settings=self.settings) - run_id, created = self.copilot.reserve_run( + run_id, created = self.reservations.reserve( spec, owner_server_id=self.server_id, owner_server_pid=self.pid) if created: self._q.put((run_id, False)) @@ -194,7 +195,7 @@ def reserve_strict_review(self, spec_dict: dict) -> tuple[str, bool]: spec = enforce_strict_review_policy( spec_dict, allowed_repos=self.settings.mcp_allowed_repos, settings=self.settings) - run_id, created = self.copilot.reserve_run( + run_id, created = self.reservations.reserve( spec, owner_server_id=self.server_id, owner_server_pid=self.pid, idempotency_key=str(spec_dict.get("idempotency_key") or "")) if created: @@ -211,7 +212,7 @@ def reserve_quality_review(self, spec_dict: dict) -> tuple[str, bool]: spec = enforce_quality_review_policy( spec_dict, allowed_repos=self.settings.mcp_allowed_repos, settings=self.settings) - run_id, created = self.copilot.reserve_run( + run_id, created = self.reservations.reserve( spec, owner_server_id=self.server_id, owner_server_pid=self.pid, idempotency_key=str(spec_dict.get("idempotency_key") or "")) if created: @@ -295,7 +296,7 @@ def quality_readiness(self, repo: str, repo_path: str = "") -> list[str]: def get_status(self, run_id: str) -> dict: """Lazy-reconcile then return `run_status.json` + `progress.json` (when present — queued/planning runs have none).""" - run_dir = self.copilot._contained_run_dir(run_id) + run_dir = self.reservations.contained_run_dir(run_id) rs.reconcile_if_dead(run_dir, self.run_root) status = rs.read_status(run_dir) or {} progress = None @@ -320,7 +321,7 @@ def get_result(self, run_id: str, offset: int = 0) -> dict: # A well-formed id this server has never heard of is answered, not # raised, so a bot holding a stale id can tell "lost" from "still # running". A malformed or escaping id is still an error. - run_dir = self.copilot._contained_run_dir(run_id, must_exist=False) + run_dir = self.reservations.contained_run_dir(run_id, must_exist=False) if not run_dir.exists(): return {"run_id": run_id, "state": "unknown", "note": "", "report": None, "report_path": None, "next_offset": None, @@ -350,7 +351,7 @@ def get_result(self, run_id: str, offset: int = 0) -> dict: def get_quality_result(self, run_id: str) -> dict: """Poll one quality run, returning its dedicated typed contract.""" - run_dir = self.copilot._contained_run_dir(run_id, must_exist=False) + run_dir = self.reservations.contained_run_dir(run_id, must_exist=False) if not run_dir.exists(): return { "run_id": run_id, diff --git a/test/test_idempotency.py b/test/test_idempotency.py index 862848839..3ae537b5d 100644 --- a/test/test_idempotency.py +++ b/test/test_idempotency.py @@ -12,6 +12,7 @@ from infermatrix_copilot import idempotency as idem from infermatrix_copilot import run_status as rs +from infermatrix_copilot.app.reservation import RunReservation from infermatrix_copilot.cli.copilot import Copilot from infermatrix_copilot.task_spec import TaskSpec @@ -63,6 +64,21 @@ def test_same_key_returns_the_same_run_and_creates_once(settings): assert created_first is True and created_second is False +def test_headless_reservation_and_copilot_facade_share_keyed_state(settings): + spec = _spec() + run_id, created = RunReservation(settings).reserve( + spec, owner_server_id="S1", owner_server_pid=1234, + idempotency_key=KEY, + ) + again, reused = _reserve(Copilot(settings), spec) + + assert created is True and reused is False and again == run_id + run_dir = Path(settings.run_root) / run_id + assert json.loads((run_dir / "request.json").read_text()) == spec.model_dump() + assert rs.read_status(run_dir)["state"] == rs.QUEUED + assert len(list(Path(settings.run_root).glob("run-*"))) == 1 + + def test_same_key_still_dedupes_after_the_run_is_terminal(settings): """The whole point: a retry that lost the response must be handed the FINISHED run and its result, not a second review of the same commit.""" diff --git a/test/test_mcp.py b/test/test_mcp.py index 4e6915c85..93e4be86a 100644 --- a/test/test_mcp.py +++ b/test/test_mcp.py @@ -325,9 +325,14 @@ def test_strict_reservation_reports_idempotent_reuse_accurately( core = _core(settings) queued = [] monkeypatch.setattr(core._q, "put", queued.append) + + def wrong_reservation_boundary(*_args, **_kwargs): + raise AssertionError("RunService called the Copilot facade") + + monkeypatch.setattr(core.copilot, "reserve_run", wrong_reservation_boundary) monkeypatch.setattr( - core.copilot, - "reserve_run", + core.reservations, + "reserve", lambda *_args, **_kwargs: ("run-20260829-010101-abcdef", False), ) diff --git a/test/test_step_registration_boundary.py b/test/test_step_registration_boundary.py index 5b8e740e9..316136fae 100644 --- a/test/test_step_registration_boundary.py +++ b/test/test_step_registration_boundary.py @@ -30,3 +30,22 @@ def test_builtin_steps_load_only_when_registry_is_assembled(): capture_output=True, text=True, timeout=30, ) assert result.returncode == 0, result.stderr + + +def test_run_reservation_import_does_not_load_orchestrator_or_transports(): + source = Path(__file__).resolve().parents[1] / "src" + code = """ +import sys +from infermatrix_copilot.app.reservation import RunReservation +for name in ('infermatrix_copilot.app.core', + 'infermatrix_copilot.app.run_service', + 'infermatrix_copilot.cli.entry', + 'infermatrix_copilot.mcp_server'): + assert name not in sys.modules, name +""" + result = subprocess.run( + [sys.executable, "-c", code], + env={**os.environ, "PYTHONPATH": str(source)}, + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr