Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions doc/architecture/SPEC/app/core.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 20 additions & 0 deletions doc/architecture/SPEC/app/reservation.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
# app/reservation.py — durable run reservation

<!-- verified-against: 2026-09-26 -->

`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.
5 changes: 5 additions & 0 deletions doc/architecture/SPEC/app/run_service.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
120 changes: 11 additions & 109 deletions src/infermatrix_copilot/app/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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-<ts>-<uuid6>`; 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
Expand Down
114 changes: 114 additions & 0 deletions src/infermatrix_copilot/app/reservation.py
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading