From 2608b25e99d08013443586ed50f77d3cc7a40a3f Mon Sep 17 00:00:00 2001 From: tzhouam Date: Sat, 26 Sep 2026 04:20:48 +0800 Subject: [PATCH] Move durable runs behind a headless application service --- doc/architecture/SPEC/app/core.md | 18 + doc/architecture/SPEC/app/run_service.md | 22 + doc/architecture/SPEC/cli.md | 7 +- doc/architecture/SPEC/mcp_server.md | 58 +- doc/architecture/SPEC/metrics.md | 3 +- doc/architecture/SPEC/sdk.md | 10 +- src/infermatrix_copilot/app/__init__.py | 1 + src/infermatrix_copilot/app/core.py | 792 ++++++++++++++++++++ src/infermatrix_copilot/app/run_service.py | 413 +++++++++++ src/infermatrix_copilot/cli/copilot.py | 793 +-------------------- src/infermatrix_copilot/cli/utils.py | 10 +- src/infermatrix_copilot/mcp_server.py | 431 +---------- src/infermatrix_copilot/metrics.py | 11 + src/infermatrix_copilot/sdk/v1/__init__.py | 8 + src/infermatrix_copilot/sdk/v1/models.py | 50 ++ src/infermatrix_copilot/sdk/v1/strict.py | 119 +++- test/test_mcp.py | 2 + test/test_planner_playbooks.py | 4 +- test/test_sdk_v1.py | 55 ++ 19 files changed, 1526 insertions(+), 1281 deletions(-) create mode 100644 doc/architecture/SPEC/app/core.md create mode 100644 doc/architecture/SPEC/app/run_service.md create mode 100644 src/infermatrix_copilot/app/__init__.py create mode 100644 src/infermatrix_copilot/app/core.py create mode 100644 src/infermatrix_copilot/app/run_service.py diff --git a/doc/architecture/SPEC/app/core.md b/doc/architecture/SPEC/app/core.md new file mode 100644 index 000000000..035aa0417 --- /dev/null +++ b/doc/architecture/SPEC/app/core.md @@ -0,0 +1,18 @@ +# app/core.py —— 规范 + + + +`Copilot` is the headless workflow application. It resolves a TaskSpec, +enforces plan review and confirmation, 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. + +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 +checkpoint keys, run IDs, and blocked exit codes remain stable. + +`GateOutcome`, `_mode_review_context`, and `Copilot` retain their import path +through the compatibility module. Only orchestration belongs here; command +argument parsing and stdio protocol registration remain transport concerns. diff --git a/doc/architecture/SPEC/app/run_service.md b/doc/architecture/SPEC/app/run_service.md new file mode 100644 index 000000000..fd9243639 --- /dev/null +++ b/doc/architecture/SPEC/app/run_service.md @@ -0,0 +1,22 @@ +# 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, +readiness, bounded status/result polling, and repository-scoped knowledge +reads. Construction and polling require no CLI or MCP module import. + +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 +a service only removes its liveness token and must not cancel a durable child. +The child rechecks repository/path/read-only policy against the same explicit +roots as the parent. Polling an unknown valid run ID reports `unknown`; +malformed or escaping IDs fail validation. Structured results and capped +report pages remain available after terminal status. + +`mcp_server.CopilotMCP` remains a compatibility alias. The MCP server owns +tool registration and protocol error projection, while the SDK owns typed +request/result projection. Neither transport owns the run queue. diff --git a/doc/architecture/SPEC/cli.md b/doc/architecture/SPEC/cli.md index 952c14ecf..fe046f4bb 100644 --- a/doc/architecture/SPEC/cli.md +++ b/doc/architecture/SPEC/cli.md @@ -1,6 +1,6 @@ # cli/ —— 规范 - + `LOC ~1420（6 个文件） · 接口 + 编排门面 · refactor-status: ok` @@ -12,11 +12,10 @@ flag CLI 与 `Copilot` 门面：解析 → 过门 → 执行；并持有 run 目 ## 包内布局（一个文件一个关注点） - `__init__.py` —— 只再导出 `Copilot`、`main`（见下方公开面）；无逻辑。 - `__main__.py` —— `python -m infermatrix_copilot.cli` 的对等入口。 -- `copilot.py` —— `Copilot` 编排器（resolve/run_task/run_playbook/run_queue/ - resume_last/_execute + 内置命令）。 +- `copilot.py` —— 兼容导出 `app.core.Copilot`；编排实现不再属于 CLI。 - `entry.py` —— `argparse`、`_handle_line`、`main`（把 argv/stdin 变成对 `Copilot` 的调用）。 -- `utils.py` —— 纯格式化器：`parse_task_params`、`format_metrics_line`。 +- `utils.py` —— `parse_task_params` 和共享 metrics 格式化器的兼容导出。 - `doctor.py` —— 预检诊断（2026-07 新增）：逐项 ✓/✗，每个失败给出**唯一**确切的修复命令。 - 子命令：`doctor` 与 `migrate-knowledge`（PR4d 部署期知识迁移；**显式 owner 动作，零 LLM**，需 `--repo `，支持 report-only；见 RUNBOOK）。 diff --git a/doc/architecture/SPEC/mcp_server.md b/doc/architecture/SPEC/mcp_server.md index 26f3b9f85..64a57ae76 100644 --- a/doc/architecture/SPEC/mcp_server.md +++ b/doc/architecture/SPEC/mcp_server.md @@ -1,12 +1,12 @@ # mcp_server.py —— 规范 - + -`LOC ~506 · Strict 后台机器（reserve/start/poll） · refactor-status: ok` +`MCP 协议适配器 · refactor-status: ok` ## 职责 -为 MCP 宿主运行 Strict 工作流：预约、拉起、跟踪、供给结果 —— -**且绝不阻塞一次同步工具调用**。 +把 MCP 工具请求映射到 `app.run_service.RunService`。预约、worker、执行与 +轮询的应用实现由该服务持有；这里仅注册工具并投影协议错误。 ## 功能 评审要跑 5–12 分钟，任何同步 MCP 调用都会超时，所以接口是**start/poll 工具对**： @@ -15,7 +15,7 @@ 和 `get_status`（`run_status` + `progress`）负责轮询。 ## 公开契约 -`CopilotMCP`、`build_mcp()`、那组 start/poll 工具，外加 +`CopilotMCP` 是 `RunService` 的兼容别名；`build_mcp()` 注册 start/poll 工具，外加 `get_capabilities()`（包 `contract.capabilities`，含 `MAX_STRICT_WORKERS=1` 与文件锁能力上报）。`start_review` 接受 `expected_head_sha`；`start_strict_review` 接受 `idempotency_key` 并把 @@ -28,50 +28,18 @@ `pr-quality` playbook。 ## 不变量（**C2**、**C3**、**E1**） -- **安全是结构性的，不是"信任宿主"。** `enforce_mcp_policy` 在这里跑一次， - **并且**在子进程里（权威地）再跑一次，因此只有 `READ_ONLY_KINDS` 会被执行、 - `post` 恒为 `False`、只有 allowlist 内的仓库可达 —— **无论被篡改的 - `request.json` 声称什么**。 -- **每次 run 都是一个隔离子进程** - （`python -m infermatrix_copilot --execute-reserved `）。两个理由都很关键： - copilot 的 stdout 必须离开本进程的 JSON-RPC stdio 通道（子进程 stdout → - `/console.log`），且进程级全局 tracer / `last_run_dir` **由此天然按 run 隔离**。 -- **run 由单个 worker 线程串行执行。** worker 等待子进程后把退出码交给对账： - 退出码 3 且无终态即 lock-loser 签名（真正 BLOCKED 的子进程会先写终态再退出）， - 以 `suspect_lock_loser` 传入 `reconcile_after_wait`。 -- 跨重启、跨多个并发 server 的轮询正确性，来自 `run_status.py` 的持久记录 + 按属主 - 对账，**而不是内存状态**。 -- **`mcp` SDK 是可选 import**，藏在 `[mcp]` extra 之后，且**绝不能**被核心包 import - —— 纯 CLI 安装保持零依赖。 -- `build_mcp()` 暴露 V1 工具面（autonomous 工作流执行器）；它**默认不注册**。 -- **子进程环境闸恒关**：`_launch` 对**每个**子进程（含 Strict）无条件写 - `ALLOW_POST="0"` —— 给策略的"拒绝而非恢复"（`mcp_policy.md`）系上第二道 - 背带；同时转发 `MCP_ALLOWED_REPO_ROOTS`，让子进程的 - `authorize_repo_path` 复检对着父进程判过的同一组根。 -- **入队只发生在 `created` 为真时**：`reserve_run` 现在返回 - `(run_id, created)` —— 重试命中既有键只返回既有 run，绝不第二次入队 - （配合 `run_status.claim_for_execution` 的 CAS，双保险关掉重复执行）。 -- **`configure_strict_repo` 已删除**（进程全局 `settings.repo_paths` 突变）： - 替代是 `strict_readiness(repo, repo_path="")` 按调用校验 —— 两个并发 - Strict 请求各带各的 checkout，再无共享状态可互踩。 -- **回收是尽力而为、绝不拦路**：启动时与每次 run 终局后跑 - `idempotency.reap_stale`；一次清不完的扫描绝不阻止 server 服务或 run - 完成。 -- **轮询对未知 id 给结构化答案**：`get_result` 对形状合法但不存在的 - run_id 返回 `contract.unknown_run_result`（`state: unknown`）而不是抛错 - —— 丢响应和还在跑可区分；终局响应同时附带 - `contract.build_review_result` 的结构化 `result`（分页 `report` 保留）。 -- `pr_quality` 与 Strict 共用同一个串行 worker 和隔离子进程，但用普通的 - `--execute-reserved` 权威复检；结果由 `get_quality_result` 装配为独立质量契约， - 不会把 code-review findings 数量误当质量结论。 +- `build_mcp()` 只注册协议工具和转换拒绝；不创建第二个执行器。 +- `mcp` SDK 在 `build_mcp()` 内才加载；SDK 和核心包不依赖可选 MCP 安装。 +- `CopilotMCP` 兼容别名指向 `RunService`，所以旧调用仍共享同一策略、 + 幂等与轮询语义。队列、隔离子进程、双重策略复检及恢复规则详见 + [`app/run_service.md`](app/run_service.md)。 ## 边界 —— 不属于这里 不含 Direct 模式逻辑（`thin_mcp_server.py`）；不定义策略（`mcp_policy.py`）； 不定义状态文件格式（`run_status.py`）。 ## 依赖（允许） -stdlib + `mcp`（可选 extra）+ `.mcp_policy` + `.run_status` + `.cli` -+ `.contract` + `.idempotency`。 +stdlib + `mcp`（仅构造时可选加载）+ `.app.run_service` + `.mcp_policy`。 ## 测试 `test_mcp.py`（篡改防御、单写者对账、分页、只读工具集、快照绑定转发、 @@ -81,5 +49,5 @@ stdlib + `mcp`（可选 extra）+ `.mcp_policy` + `.run_status` + `.cli` 第二个子进程 no-op、post 拒绝）。 ## 重构备注 -预约形状（先建 run 目录，再规划）是 **MCP 专属**的；CLI 主路径仍然在**建目录之前** -过门，所以被放弃的计划不会留下目录。**不要在没有保住这个差别的前提下把两者统一。** +预约形状（先建 run 目录，再规划）由 MCP 和 embedded Strict SDK 共用；CLI +主路径仍在建目录之前过门。不要把这两种不同的执行入口混为一谈。 diff --git a/doc/architecture/SPEC/metrics.md b/doc/architecture/SPEC/metrics.md index a91e146d2..74cf2d7c4 100644 --- a/doc/architecture/SPEC/metrics.md +++ b/doc/architecture/SPEC/metrics.md @@ -1,6 +1,6 @@ # metrics.py —— 规范 - + `LOC ~493 · 跨切（测量） · refactor-status: oversized` @@ -14,6 +14,7 @@ S（由 incident 得出的安全系数）、C（RQS3e 式的对数成本，基 ## 公开契约 `collect_run_metrics(run_dir, settings, status) -> {quality, risk, cost, catq}`。 +`format_metrics_line` 为应用和 CLI 共用的纯格式化函数，不参与评分计算。 ## 不变量 - **E3**：metrics 是关于一次 run 的**事实**，且**绝不能把这次 run 搞坏** —— diff --git a/doc/architecture/SPEC/sdk.md b/doc/architecture/SPEC/sdk.md index 6c2639d79..4d08c1919 100644 --- a/doc/architecture/SPEC/sdk.md +++ b/doc/architecture/SPEC/sdk.md @@ -19,12 +19,15 @@ - Direct：`DirectReviewRequest` → `DirectClient.plan()` → `DirectReviewPlan`； `read_document(document_id, offset, max_bytes)`； `DirectCompletionRequest` → `validate()` → `DirectCompletionDecision`。 -- Strict：`StrictRuntime(settings_overrides=...)`，以及 +- Strict：`StrictRuntime(config=StrictRuntimeConfig(...))`，以及 `capabilities` / `readiness` / `reserve_review`（`start_review` 别名）/ - `get_status` / `get_result` / `close`。 + `get_status` / `get_result` / `close`。旧 `settings_overrides` 暂作兼容入口； + 新配置显式绑定完整仓库名、checkout、允许根、backend 和可选 run root。 + `StrictPollResult.review` 提供类型化结构结果，`payload` 保持旧 wire 兼容。 - Quality：同一 `StrictRuntime` 接受 typed `QualityReviewRequest`，提供 `quality_readiness` / `reserve_quality_review`（`start_quality_review` 别名）/ `get_quality_result`；机械信号只是 bounded hints，结果固定绑定 expected head。 + `QualityPollResult.review` 投影类型化质量结果；非法结构触发 `ResultDecodeError`。 - Knowledge：宿主把不可信输入投影成 `KnowledgeEvidenceEvent` / `KnowledgeEvidenceBatch`， 再依次调用 `KnowledgeCurator.build_prompt()`、自己的 model adapter、 `validate_proposals()` 与 `apply()`。返回值为 typed @@ -36,7 +39,8 @@ - **公开 import 无副作用**：单纯 import `sdk.v1` 不加载 `config`、legacy `contract`、`direct_routing` 或任一 server；Direct provider 实现在调用 - `plan`/`validate` 时才加载，Strict server 在构造 `StrictRuntime` 时才加载。 + `plan`/`validate` 时才加载；构造 `StrictRuntime` 加载 headless RunService， + 不加载 CLI 或 MCP transport。 - **只跨 document ID，不跨 provider filesystem path**：知识入口、route、guide、 map 与 fallback 均为相对 knowledge-root 的 ID；`read_document` 拒绝 absolute、 traversal 与 bundle 外路径。Strict result 丢弃私有 `report_path`，保留 report diff --git a/src/infermatrix_copilot/app/__init__.py b/src/infermatrix_copilot/app/__init__.py new file mode 100644 index 000000000..0ceec65ae --- /dev/null +++ b/src/infermatrix_copilot/app/__init__.py @@ -0,0 +1 @@ +"""Headless application services shared by SDK and transports.""" diff --git a/src/infermatrix_copilot/app/core.py b/src/infermatrix_copilot/app/core.py new file mode 100644 index 000000000..756f1dc13 --- /dev/null +++ b/src/infermatrix_copilot/app/core.py @@ -0,0 +1,792 @@ +"""The `Copilot` orchestrator: NL-resolved TaskSpec → plan (reuse > adapt > +generate) → plan-review gate → executor, plus the compound-command queue, +resume, and the /status /logs /playbooks built-ins. + +This is the orchestration core; the argparse/REPL wiring lives in `entry.py` and +metrics formatting is shared with `metrics.py`. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import re +import time +import uuid +from pathlib import Path +from typing import NamedTuple + +import yaml + +from .. import idempotency as idem +from .. import run_status as rs +from ..config import Settings, TierNotConfiguredError +from ..engine.executor import Executor +from ..engine.lifecycle import RunLock, RunLockHeld, run_guarded +from ..engine.planner import Planner, PlanningError, Resolution +from ..engine.registry import StepRegistry +from ..engine.steps import register_builtin_steps +from ..llm import LLM +from ..notify import BLOCKED_EXIT, Notifier +from ..playbooks.store import PlaybookStore, parse_playbook, playbook_to_doc +from ..push import PushPolicy +from ..review.reviewer import run_plan_review +from ..run_trace import RunTrace +from ..task_spec import TaskSpec +from ..ui import style +from ..metrics import format_metrics_line + + +class GateOutcome(NamedTuple): + """Result of the pre-execution gates. `proceed` says whether to run; when + False, `exit_code` is the process code the caller should return. Read at + the call site as: ``if not gate.proceed: return gate.exit_code``.""" + + proceed: bool + exit_code: int = 0 + + @classmethod + def go(cls) -> "GateOutcome": + """Gates passed — run the task.""" + return cls(proceed=True) + + @classmethod + def stop(cls, exit_code: int) -> "GateOutcome": + """A gate halted the run — return this exit code.""" + return cls(proceed=False, exit_code=exit_code) + + +def _mode_review_context(playbook, spec) -> str: + """Plan-review context for MODE-AWARE playbooks: the reviewer sees the + raw yaml's FULL step list, but the `when:` gates resolve against the + ALREADY-RESOLVED mode (`resolve_effective_mode` runs before the review + gate) — without this context a report-only plan looks like it runs its + write/push steps and gets spuriously blocked. The active-step set is + the same mechanical truth the executor computes, never prose.""" + if not getattr(playbook, "mode_aware", False): + return "" + mode = str((getattr(spec, "params", None) or {}) + .get("rebase_mode", "") or "") + if not mode: + return "" + from ..engine.executor import _eval_when + from ..rebase_engine.modes import mode_state_flags + + flags = {"task_spec": {}, **mode_state_flags(mode)} + active = [s.get("id", s.get("step", "?")) + for s in playbook_to_doc(playbook).get("steps", []) + if "when" not in s or _eval_when(s["when"], flags)] + repo = str(getattr(spec, "repo", "") or "") + repo_line = (f"\nTarget repo (authoritative): {repo!r} — bound at " + "runtime from the TaskSpec; the yaml `repos:` list is a " + "planner RECALL FILTER where empty means repo-neutral, " + "never untargeted." if repo else "") + return (f"\n\nResolved mode context (authoritative): " + f"rebase_mode={mode}. Under this mode the `when:` gates run " + f"ONLY these steps: {active}. Every other listed step is " + "statically gated OFF for this run — judge the plan for THIS " + "mode's step set." + + repo_line + + "\nWrite/push governance (authoritative): the mode's own " + "push/CI steps are governed at runtime by the push-gate " + "ruling, guard_push, and the ALLOW_PUSH env double-gate — " + "the task tier does not forbid steps this mode activates.") + + +class Copilot: + """Orchestration core: resolves a TaskSpec to a playbook (reuse > adapt > + generate), runs it through the plan-review gate + confirmation into the + Executor, and owns the compound-command queue, resume, and the /status + /logs /playbooks built-ins. Holds the long-lived collaborators (LLM, + step registry, playbook store, planner) and tracks `last_run_dir` for the + built-ins.""" + + def __init__(self, settings: Settings | None = None): + """Wire up the collaborators from `settings` (default `Settings()`): + the LLM, the built-in step registry, the playbook store rooted at the + configured dir, and the planner over both. `last_run_dir` starts unset + and is filled by the first execution.""" + self.settings = settings or 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 + + self.llm = llm_for(self.settings) + self.registry = register_builtin_steps(StepRegistry()) + self.store = PlaybookStore(self.settings.playbooks_dir, self.registry) + self.planner = Planner(self.store, self.registry) + self.last_run_dir: Path | None = None + # why the last execution stopped, when it stopped blocked — read by + # `_execute_reserved_locked`, which owns the terminal run_status write + self.last_blocked_reason: str = "" + + # -- planning --------------------------------------------------------------- + def resolve(self, spec: TaskSpec) -> Resolution: + """Resolve `spec` to a Resolution via the planner, passing the repo's + capability set so the planner only reuses playbooks the target supports. + Capabilities come from the repo's adapter (if any), plus `repo.path` when + a path is resolvable even without a adapter (REPO_PATHS works adapter-less).""" + # Only genuine adapter ABSENCE takes the unknown-capabilities + # compatibility path — a malformed/unreadable KNOWN adapter must fail + # closed here, not fail open into capabilities=None and recall a + # playbook whose requirements were never established. + from ..adapters.base import AdapterError, AdapterNotFound, AdapterRegistry + adapter_name = spec.repo.replace("-", "_") + try: + adapter = AdapterRegistry(self.settings.adapters_dir).resolve( + name=adapter_name) + except AdapterNotFound: + # the registry resolves by DECLARED manifest name and skips + # manifest-less directories — so "not found" alone does not + # prove absence. Only a genuinely absent directory is the + # v1-compatible path; an existing directory that failed to + # load/resolve (deleted manifest, wrong name:) fails closed. + if (Path(self.settings.adapters_dir) / adapter_name).exists(): + raise AdapterError( + f"adapter directory {adapter_name!r} exists but did not " + "load/resolve (missing manifest.yaml or mismatched " + "name:) — refusing to plan with unknown capabilities") + adapter = None # absence: v1-compatible + except FileNotFoundError: + adapter = None # no adapters directory at all + if adapter is None: + # No adapter means capabilities are UNKNOWN, not zero — the + # store's requires-filter (now covering exact-repo playbooks too) + # skips on None, keeping adapter-less setups v1-compatible + # instead of silently dropping every playbook with a `requires:`. + if self._repo_path_for(spec): + return self.planner.resolve(spec, capabilities=None) + return self.planner.resolve(spec, capabilities=set()) + capabilities = set(adapter.capabilities) + if self._repo_path_for(spec): # REPO_PATHS works adapter-less + capabilities.add("repo.path") + return self.planner.resolve(spec, capabilities=capabilities) + + def _plan_review_gate(self, resolution: Resolution, spec: TaskSpec, + assume_yes: bool) -> bool: + """Inline Plan-Review for adapted/generated plans. LLM verdict shown in + the session; block stops. A non-`lgtm` verdict is only ever *surfaced* + — the human `[y/N]` is what actually gates it — so when `--yes` removes + that human, the same verdict must stop the run instead + (`ReviewVerdict.passing`, SPEC C6: only `lgtm` passes).""" + if not resolution.requires_review: + return True + doc = yaml.safe_dump(playbook_to_doc(resolution.playbook), sort_keys=False) + task_text = spec.describe() + _mode_review_context( + resolution.playbook, spec) + verdict = run_plan_review(self.llm, playbook_doc=doc, + task=task_text, + model=self.settings.reviewer) + if verdict.verdict != "unavailable": + print(f" plan review: {verdict.verdict}" + + (f" — {verdict.critiques}" if verdict.critiques else "")) + if verdict.verdict == "block": + print("✋ plan blocked by reviewer.") + return False + if verdict.passing: + return True + # Everything below is non-passing: `revise`, or `unavailable`. Both used + # to return True on the strength of a confirmation that `--yes` had + # already deleted, so an unattended run executed an unvetted plan on an + # unread verdict. Measured on the release matrix: an unparseable review + # let three of four backends run a pr-rebase plan through to its push + # gate, while the one backend whose reviewer parsed cleanly BLOCKED the + # very same plan. + if assume_yes: + reason = ("no reviewer LLM" if verdict.verdict == "unavailable" + else f"plan review returned {verdict.verdict}") + print(f"✋ {reason} and --yes leaves no human to gate it — blocked.") + return False + if verdict.verdict == "unavailable": + print(" ⚠ no reviewer LLM — your confirmation is the plan-review gate") + return True # revise/unavailable, surfaced to the user before their confirm + + def _gate_and_confirm(self, resolution: Resolution, spec: TaskSpec, + assume_yes: bool, *, prompt: str = "Proceed?", + force_confirm: bool = False) -> GateOutcome: + """Plan-review gate + [y/N] confirm (concision K6). Returns + `GateOutcome.go()` to run, or `GateOutcome.stop(code)` to halt with a + process exit code (BLOCKED_EXIT on a reviewer block, 1 on a user + abort). Confirm fires for confirm_required or a review-requiring/ + explicit plan, unless assume_yes.""" + if not self._plan_review_gate(resolution, spec, assume_yes): + return GateOutcome.stop(BLOCKED_EXIT) # reviewer blocked the plan + need = force_confirm or spec.confirm_required or resolution.requires_review + if need and not assume_yes: + if input(f"{prompt} [y/N] ").strip().lower() not in ("y", "yes"): + print("aborted.") + return GateOutcome.stop(1) # user declined + return GateOutcome.go() + + # -- execution ----------------------------------------------------------------- + def run_task(self, spec: TaskSpec, *, assume_yes: bool = False, + plan_only: bool = False) -> int: + """Full path for one task: resolve → print plan → gate/confirm → + persist task.json → execute. Returns a process exit code — 0 done, + 1 failed/aborted, BLOCKED_EXIT when planning fails or a gate blocks. + `plan_only` prints the resolved plan and returns 0 without running; + `assume_yes` skips the interactive confirm.""" + try: + resolution = self.resolve(spec) + except PlanningError as exc: + print(style("✋ cannot plan: ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + # tier preflight (plan v2): backend availability is DEPLOYMENT state, + # so it is enforced here at resolution — before the echo, the gate, + # and the run dir — never discovered as a BLOCKED step minutes in. + try: + tier_target = self.settings.tier_target(spec.mode) + except TierNotConfiguredError as exc: + print(style("✋ cannot run: ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + + # Mode governance (Rev 8 §2.1): for mode-aware playbooks the ONE + # authority is params.rebase_mode, resolved + WRITTEN BACK before the + # plan echo and the confirmation gate (spec.report_only reflects the + # canonical mode, so every TaskSpec-derived consumer sees one truth). + # The locked delegating playbook does not declare mode_aware and is + # untouched — byte-identical behavior. + if getattr(resolution.playbook, "mode_aware", False): + from ..rebase_engine.modes import (ModeConflictError, + resolve_effective_mode) + try: + resolve_effective_mode(spec) + except ModeConflictError as exc: + print(style("✋ blocked: ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + + print(style("→ task: ", "bold", "cyan") + spec.describe()) + print(style("→ plan: ", "bold", "magenta") + f"{resolution.mode} {resolution.playbook.name}" + f"@{resolution.playbook.version} ({resolution.playbook.status}) " + f"steps={[s.step for s in resolution.playbook.steps]}") + for note in resolution.notes: + print(f" · {note}") + print(style("→ path: ", "bold", "cyan") + + f"{spec.mode} · agent-model={tier_target.model}" + f" @ {tier_target.host}") + if plan_only: + return 0 + + gate = self._gate_and_confirm(resolution, spec, assume_yes) + if not gate.proceed: + return gate.exit_code + + run_id = (f"run-{time.strftime('%Y%m%d-%H%M%S')}" + f"-{uuid.uuid4().hex[:6]}") # unique — same-second runs collided + run_dir = self.settings.run_root / run_id + run_dir.mkdir(parents=True, exist_ok=True) + (run_dir / "task.json").write_text(json.dumps({ + "spec": spec.model_dump(), "playbook": playbook_to_doc(resolution.playbook), + "invocation_id": os.environ.get("IMX_INVOCATION_ID", ""), + }, indent=2)) + return self._execute(resolution.playbook, spec, run_dir, + resolution_mode=resolution.mode, tier=resolution.tier) + + def run_playbook(self, name: str, *, params: dict | None = None, + report_only: bool = False, assume_yes: bool = False, + plan_only: bool = False) -> int: + """Explicit playbook override — the only way to execute a CANDIDATE + (candidates stay planner-invisible). Always treated as requiring + review + confirmation.""" + playbook = self.store.get(name) + if playbook is None: + print(f"✋ no playbook named {name!r} (see /playbooks)") + return BLOCKED_EXIT + kind = playbook.task_kinds[0] + # lift target params into first-class spec fields so the explicit + # playbook path can address a PR/issue/repo like every other surface + params = dict(params or {}) + pr = params.pop("pr", None) + issue = params.pop("issue", None) + repo = str(params.pop("repo", "") or self.settings.default_repo) + try: + pr = int(pr) if pr is not None else None + issue = int(issue) if issue is not None else None + except (TypeError, ValueError): + print("✋ pr/issue task params must be integers") + return BLOCKED_EXIT + if repo not in (self.settings.repo_paths or {repo: ""}): + print(f"✋ unknown repo alias {repo!r} " + f"(known: {', '.join(sorted(self.settings.repo_paths or {}))})") + return BLOCKED_EXIT + spec = TaskSpec(kind=kind, repo=repo, pr=pr, issue=issue, + report_only=report_only, params=params) + from ..intent import validate_spec + err = validate_spec(spec) + if err: + print(f"✋ {err} (pass e.g. --task-param pr=5134)") + return BLOCKED_EXIT + # mode governance applies to explicit invocations too — the ONLY + # way to run the candidate v3/v1 playbooks today (Rev 8 §2.1) + if getattr(playbook, "mode_aware", False): + from ..rebase_engine.modes import (ModeConflictError, + resolve_effective_mode) + try: + resolve_effective_mode(spec) + except ModeConflictError as exc: + print(style("✋ blocked: ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + print(style("→ task: ", "bold", "cyan") + spec.describe() + + style(" [explicit playbook override]", "yellow")) + print(style("→ plan: ", "bold", "magenta") + f"explicit {playbook.name}@{playbook.version} " + f"({playbook.status}) steps={[s.step for s in playbook.steps]}") + if plan_only: + return 0 + resolution = Resolution(mode="explicit", playbook=playbook, + tier=spec.tier, requires_review=True) + gate = self._gate_and_confirm( + resolution, spec, assume_yes, force_confirm=True, + prompt=f"Run {playbook.status} playbook '{name}'?") + if not gate.proceed: + return gate.exit_code + run_id = (f"run-{time.strftime('%Y%m%d-%H%M%S')}" + f"-{uuid.uuid4().hex[:6]}") # unique — same-second runs collided + run_dir = self.settings.run_root / run_id + run_dir.mkdir(parents=True, exist_ok=True) + (run_dir / "task.json").write_text(json.dumps({ + "spec": spec.model_dump(), "playbook": playbook_to_doc(playbook), + "invocation_id": os.environ.get("IMX_INVOCATION_ID", ""), + }, indent=2)) + return self._execute(playbook, spec, run_dir, + resolution_mode="explicit", tier=spec.tier) + + def _execute(self, playbook, spec: TaskSpec, run_dir: Path, *, + resolution_mode: str = "resume", tier: str = "?", + resuming: bool = False, held_lock: RunLock | None = None) -> int: + """Run a resolved `playbook` to completion in `run_dir`: init tracing + + notifier, seed the shared state (repo path, push policy, protected + branches / high-risk modules from the adapter when present), drive the + Executor, then print per-step marks, optional run metrics, and the final + status. `resuming` seeds the state so steps can pick up where they left + off. Returns the exit code (0 done, BLOCKED_EXIT blocked, else 1), and + records `last_blocked_reason` for the reserved-run caller, which has only + the code but owns the terminal `run_status.json` write.""" + self.last_run_dir = run_dir + self.last_blocked_reason = "" + lock = held_lock + if lock is None: + try: + # First, before any trace/status write: a losing concurrent + # invocation (second --resume) must leave the active run's + # artifacts completely untouched. execute_reserved holds the + # lock across its whole lifecycle and passes it in instead. + lock = RunLock(run_dir).acquire() + except RunLockHeld as exc: + print(style("✋ ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + # Per-repo knowledge lock, SHARED, held for the run's lifetime: + # never contends with other runs; it exists so the knowledge + # migration's EXCLUSIVE acquire can prove no potential store + # writer is alive (and so no run starts mid-migration). + knowledge_lock = None + if spec.repo: + from ..memory.paths import (KnowledgeLockHeld, + KnowledgePaths, + KnowledgeRunLock, + KnowledgeStateError) + try: + knowledge_lock = KnowledgeRunLock( + KnowledgePaths.resolve(self.settings, spec.repo) + .knowledge_run_lock).acquire_shared() + except (KnowledgeLockHeld, KnowledgeStateError) as exc: + # BOTH refusals take the terminal protocol: an invalid + # activation marker (KnowledgeStateError) must exit + # blocked/3 with the run lock RELEASED, never escape as + # a traceback that leaves the lock held in a long-lived + # process (PR-boundary F4) + if held_lock is None: + lock.release() + print(style("✋ ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + try: + from .. import tracing + tracing.init(run_dir.name, run_dir / "trace.jsonl") + # Stamp the workflow into the span file itself, so a trace lifted out of + # its run directory still says which playbook and task produced it. + tracing.run_meta(playbook=f"{playbook.name}@{playbook.version}", + task_kind=spec.kind, repo=spec.repo, tier=tier, + mode=spec.mode, resolution=resolution_mode, + report_only=spec.report_only, post=spec.post, + resuming=resuming, params=spec.params) + trace = RunTrace(run_dir / "run_trace.jsonl") + notifier = Notifier(self.settings, run_dir, trace, run_dir.name) + trace.record("task", spec=spec.model_dump(), resolution=resolution_mode, + playbook=playbook.name, tier=tier) + state: dict = { + "task_spec": spec.model_dump(), + "repo_path": self._repo_path_for(spec), + "push_policy": PushPolicy(), # steps may replace with a derived policy + "protected_branches": self.settings.protected_branches, + "resuming": resuming, + } + adapter = self._adapter_for(spec.repo) + if adapter is not None: + # repo knowledge from the adapter, not core settings (v2 P0 fix #5) + state["protected_branches"] = adapter.protected_branches + if adapter.high_risk_modules: + state["high_risk_modules"] = adapter.high_risk_modules + executor = Executor(self.registry, self.settings, run_dir=run_dir, + trace=trace, llm=self.llm, notifier=notifier) + # run_guarded finalizes inside the event loop: playbooks that + # register run finalizers (lifecycle.register_finalizer) get + # teardown on every exit path. Nothing registered == no-op. + outcome = asyncio.run( + run_guarded(executor.run(playbook, state), run_dir)) + + if outcome.status == "done": + notifier.resolve() + + if self.settings.metrics_enabled: + try: # metrics are facts about the run; never let them break it + from ..metrics import collect_run_metrics + m = collect_run_metrics(run_dir, self.settings, outcome.status) + print(format_metrics_line(m, run_dir)) + except Exception as exc: + trace.record("metrics_error", error=f"{type(exc).__name__}: {exc}") + + for step_id, r in outcome.step_results.items(): + mark = style("✓", "green") if r.ok else style("✗", "red", "bold") + print(f" {mark} {step_id}: {r.summary}") + print(f"run {run_dir.name}: {outcome.status} ({run_dir})") + if outcome.status == "blocked": + self.last_blocked_reason = str(outcome.blocked_reason or "") + print(style(" ⚠ ", "yellow", "bold") + f"{outcome.blocked_reason}\n see {run_dir / 'ESCALATION.md'}") + return BLOCKED_EXIT + return 0 if outcome.status == "done" else 1 + finally: + if knowledge_lock is not None: + knowledge_lock.release() + if held_lock is None: + lock.release() + + def run_queue(self, specs: list[TaskSpec], *, assume_yes: bool = False, + plan_only: bool = False) -> int: + """Ordered task queue for compound commands; stops on failure/blocked.""" + if len(specs) > 1: + print(f"⧉ queued {len(specs)} tasks:") + for i, s in enumerate(specs, 1): + print(f" {i}. {s.describe()}") + for i, spec in enumerate(specs, 1): + if len(specs) > 1: + print(f"\n── task {i}/{len(specs)} ──") + code = self.run_task(spec, assume_yes=assume_yes, plan_only=plan_only) + if code != 0: + if i < len(specs): + print(f"⏸ queue stopped: {len(specs) - i} task(s) not run") + return code + return 0 + + def resume_last(self) -> int: + """Re-enter the most recent run at its first incomplete step.""" + runs = sorted(self.settings.run_root.glob("run-*")) \ + if self.settings.run_root.exists() else [] + for run_dir in reversed(runs): + task_file = run_dir / "task.json" + if not task_file.exists(): + continue + saved = json.loads(task_file.read_text(encoding="utf-8")) + spec = TaskSpec(**saved["spec"]) + playbook = parse_playbook(saved["playbook"], str(task_file)) + try: # tier preflight holds on re-entry too (deployment state + # may have changed since the run was created) + self.settings.tier_target(spec.mode) + except TierNotConfiguredError as exc: + print(style("✋ cannot resume: ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + print(f"↻ resuming {run_dir.name}: {spec.describe()}") + return self._execute(playbook, spec, run_dir, resuming=True) + 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}$") + + @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]}" + + 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 + + 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 ..mcp_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 + + def execute_reserved(self, run_id: str) -> int: + """MCP-only child entry (subprocess, stdout -> console.log). Writes its + own pid FIRST (single-writer invariant), **re-enforces** the read-only + MCP policy on the persisted request (request.json is untrusted — a host + could have rewritten it), plans, then executes, driving + `run_status.json` planning -> running -> terminal. Returns the exit code.""" + from ..mcp_policy import enforce_mcp_policy + + return self._execute_reserved(run_id, enforce_mcp_policy) + + def execute_strict_reserved(self, run_id: str) -> int: + """Execute a reserved Strict review using the previous Eco workflow.""" + from ..mcp_policy import enforce_strict_review_policy + + return self._execute_reserved(run_id, enforce_strict_review_policy) + + def _execute_reserved(self, run_id: str, policy) -> int: + """Shared reserved-run executor with an authoritative child policy.""" + run_dir = self._contained_run_dir(run_id) + try: + # Lock the whole reserved-run lifecycle before the first status + # write: a duplicate child must not overwrite child_pid, flip + # run_status.json through PLANNING, or mark the active run + # BLOCKED while the winning process is still executing. + lock = RunLock(run_dir).acquire() + except RunLockHeld as exc: + print(style("✋ ", "red", "bold") + str(exc)) + return BLOCKED_EXIT + try: + return self._execute_reserved_locked(run_dir, lock, policy) + finally: + lock.release() + + def _execute_reserved_locked(self, run_dir: Path, lock: RunLock, + policy) -> int: + """The body of `_execute_reserved`, run while holding the run lock.""" + from ..mcp_policy import PolicyError + + # The claim, not the lock, is what makes execution at-most-once. + # `RunLock` excludes concurrent children; it says nothing about a second + # child that arrives after the first finished and released. A losing + # child exits here without planning, executing, or writing status — it + # is a duplicate, not a failure. + if not rs.claim_for_execution(run_dir, child_pid=os.getpid()): + state = (rs.read_status(run_dir) or {}).get("state", "?") + print(style("↷ ", "yellow") + + f"run {run_dir.name} is already claimed (state={state}) — " + "not executing it twice") + return 0 + try: + raw = json.loads((run_dir / "request.json").read_text(encoding="utf-8")) + spec = policy( + raw, allowed_repos=self.settings.mcp_allowed_repos, + settings=self.settings) + except (PolicyError, OSError, json.JSONDecodeError, ValueError) as exc: + rs.mark(run_dir, rs.FAILED, note=f"policy/request rejected: {exc}") + return 1 + try: + resolution = self.resolve(spec) + except PlanningError as exc: + rs.mark(run_dir, rs.BLOCKED, note=f"cannot plan: {exc}") + return BLOCKED_EXIT + try: # tier preflight is re-enforced in the MCP child (authoritative) + self.settings.tier_target(spec.mode) + except TierNotConfiguredError as exc: + rs.mark(run_dir, rs.BLOCKED, note=f"cannot run: {exc}") + return BLOCKED_EXIT + (run_dir / "task.json").write_text(json.dumps({ + "spec": spec.model_dump(), + "playbook": playbook_to_doc(resolution.playbook), + }, indent=2)) + rs.mark(run_dir, rs.RUNNING) + try: + code = self._execute(resolution.playbook, spec, run_dir, + resolution_mode=resolution.mode, tier=resolution.tier, + held_lock=lock) + except Exception as exc: # a crash still leaves a terminal record + rs.mark(run_dir, rs.FAILED, note=f"{type(exc).__name__}: {exc}") + raise + # Carry WHY it stopped into the terminal record. An MCP client only ever + # sees `run_status.json`, and a bare `blocked` with no note gives it + # nothing to distinguish a stale-head refusal from a missing tier or a + # failed checkout — the reason was previously printed to the child's + # console and discarded. + rs.mark(run_dir, {0: rs.DONE, BLOCKED_EXIT: rs.BLOCKED}.get(code, rs.FAILED), + note=self.last_blocked_reason if code == BLOCKED_EXIT else "") + return code + + def _adapter_for(self, repo: str): + """The repo's registered adapter, or None (never raises).""" + try: + from ..adapters.base import AdapterRegistry + + return AdapterRegistry(self.settings.adapters_dir).resolve( + name=repo.replace("-", "_")) + except Exception: + return None + + def _repo_path_for(self, spec: TaskSpec) -> str: + """The checkout THIS spec runs against: its frozen `repo_path` first, + else the ambient resolution by alias. + + Every place a run learns its checkout must agree. Resolution derives the + `repo.path` capability from this, and `_execute` seeds `state` from it: + with an explicit path but no ambient `REPO_PATHS` and no adapter + `repo.path`, an alias-only lookup would drop the capability and fail to + resolve a playbook that requires it — while a perfectly valid checkout + sat in the spec. With both configured and differing, planning would + evaluate one checkout and execution would run against another.""" + return spec.repo_path or self._resolve_repo_path(spec.repo) + + def _resolve_repo_path(self, repo: str) -> str: + """REPO_PATHS first; fall back to the repo's adapter manifest (adapter zero + declares repo.path), so runs work even without a .env in reach. + + Alias-only: callers that hold a TaskSpec want `_repo_path_for`, which + honors a frozen per-run path.""" + p = self.settings.repo_path(repo) + if p: + return str(p) + adapter = self._adapter_for(repo) + if adapter and adapter.repo_path: + return adapter.repo_path + return "" + + # -- built-ins --------------------------------------------------------------- + def status(self) -> str: + """Human-readable status of the current (or most recent) run: completed + steps from progress.json, plus a rebase-phase line (module/test counts, + CI result) when rebase_status.json exists. Falls back to the newest + run-* dir when no run has executed this session.""" + if not self.last_run_dir: + runs = sorted(self.settings.run_root.glob("run-*")) \ + if self.settings.run_root.exists() else [] + if not runs: + return "no runs yet" + self.last_run_dir = runs[-1] + progress = self.last_run_dir / "progress.json" + lines = [] + if progress.exists(): + done = list(json.loads(progress.read_text(encoding="utf-8")).get("completed", {})) + lines.append(f"{self.last_run_dir.name}: completed steps: {done}") + else: + lines.append(f"{self.last_run_dir.name}: no progress recorded") + rebase_status = self.last_run_dir / "rebase_status.json" + if rebase_status.exists(): + s = json.loads(rebase_status.read_text(encoding="utf-8")) + mods = s.get("modules", {}) + tests = s.get("tests", {}) + lines.append( + f" rebase: phase={s.get('phase')} modules(done={mods.get('done', 0)} " + f"failed={mods.get('failed', 0)}) tests(completed={tests.get('completed', 0)} " + f"failed={len(tests.get('failed', []))}" + + (f" current={tests.get('current')}" if tests.get("current") else "") + + f") ci={s.get('ci_result') or '-'}" + ) + return "\n".join(lines) + + def logs(self, n: int = 20) -> str: + """Return the last `n` lines of the current run's run_trace.jsonl, or a + placeholder string when there is no run / no trace yet.""" + if not self.last_run_dir: + return "no runs yet" + tracefile = self.last_run_dir / "run_trace.jsonl" + if not tracefile.exists(): + return "no trace" + return "".join(tracefile.read_text(encoding="utf-8", errors="replace").splitlines(keepends=True)[-n:]) + + def playbooks(self) -> str: + """One line per registered playbook (name@version, status, task kinds), + or "(none)" when the store is empty.""" + return "\n".join( + f"{p.name}@{p.version} [{p.status}] kinds={p.task_kinds}" + for p in self.store.all() + ) or "(none)" diff --git a/src/infermatrix_copilot/app/run_service.py b/src/infermatrix_copilot/app/run_service.py new file mode 100644 index 000000000..cd6d74014 --- /dev/null +++ b/src/infermatrix_copilot/app/run_service.py @@ -0,0 +1,413 @@ +"""Durable, headless run reservation, execution, polling, and knowledge reads. + +The MCP protocol server and embedded SDK both call this service. It owns no +transport or CLI command implementation. +""" + +from __future__ import annotations + +import json +import os +import queue +import subprocess +import sys +import threading +import uuid +from pathlib import Path +from typing import Any + +from .. import contract +from .. import idempotency as idem +from .. import run_status as rs +from ..config import Settings +from ..knowledge_docs import KnowledgeDocs, KnowledgeDocsError +from ..mcp_policy import ( + PolicyError, authorize_repo_path, enforce_mcp_policy, + enforce_quality_review_policy, enforce_strict_review_policy, +) +from ..task_spec import READ_ONLY_KINDS +from .core import Copilot + + +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.""" + + 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.""" + self.settings = settings or Settings() + self.copilot = Copilot(self.settings) + self.run_root = Path(self.settings.run_root) + self.run_root.mkdir(parents=True, exist_ok=True) + self.server_id = uuid.uuid4().hex + self.pid = os.getpid() + rs.register_server(self.run_root, self.server_id, self.pid) + rs.startup_reconcile(self.run_root) + 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 + + def _reap(self) -> None: + """Bound what the idempotency index and PR-time worktrees accumulate. + + 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.""" + 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 + + 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, + supports_file_locking=_fcntl is not None) + + # -- worker: one run at a time, each an isolated subprocess --------------- + 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.""" + while True: + run_id, strict_compat = self._q.get() + try: + self._launch(run_id, strict_compat=strict_compat) + except Exception as exc: # noqa: BLE001 - worker must survive + try: + rs.mark(self.run_root / run_id, rs.FAILED, + note=f"launch error: {type(exc).__name__}: {exc}") + except Exception: + pass + finally: + self._q.task_done() + + def _launch(self, run_id: str, *, strict_compat: bool = False) -> None: + """Run one reserved run as `python -m infermatrix_copilot --execute-reserved + `, child stdout+stderr -> console.log. No MCP child may write + outward, Strict included. After `.wait()` the child is reaped, so we + reconcile as sole writer.""" + run_dir = self.run_root / run_id + env = dict(os.environ) + # Belt to the policy's braces. Strict used to be handed ALLOW_POST=1 + # when this server allowed it, because Strict specs could carry + # post=True; the policy now refuses that, so leaving the env gate open + # would only preserve a path to a second publisher on the same PR. + env["ALLOW_POST"] = "0" + env["ALLOW_PUSH"] = "0" + # The Windows Store Python runtime otherwise inherits the machine's + # legacy console code page (commonly GBK). Reports legitimately contain + # Unicode markers such as ✓; force the isolated child's stdio to UTF-8 + # so rendering a completed report cannot fail after all work is done. + env["PYTHONUTF8"] = "1" + env["PYTHONIOENCODING"] = "utf-8" + # The child builds its own Settings() from env/.env, so make THIS server's + # effective config authoritative for it: the same run_root (to locate the + # reserved dir) and the same read-only allowlist (its policy re-check). + env["RUN_ROOT"] = str(self.run_root) + env["MCP_REPO_ALLOWLIST"] = json.dumps(self.settings.mcp_allowed_repos) + env["DEFAULT_REPO"] = self.settings.default_repo + env["REPO_PATHS"] = json.dumps(self.settings.repo_paths) + env["REPO_FULL_NAMES"] = json.dumps(self.settings.repo_full_names) + # The child re-authorizes the request's `repo_path`, so it must judge it + # against the SAME roots and identities this server did — otherwise the + # re-check silently validates against different config than the parent. + env["MCP_ALLOWED_REPO_ROOTS"] = json.dumps( + self.settings.allowed_repo_roots) + if self.settings.strict_backend: + env["STRICT_BACKEND"] = self.settings.strict_backend + popen_kwargs: dict[str, Any] = {} + if os.name == "nt": + # Codex/Claude launch the MCP server over stdio. Without a new + # Windows process group, host or transport shutdown can propagate a + # console control event into the long-running review child and turn + # it into KeyboardInterrupt. CREATE_NO_WINDOW also prevents a + # console flash for every MCP run. + popen_kwargs["creationflags"] = ( + subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW + ) + else: + # Keep the durable child alive and independently reconcilable when + # its stdio MCP parent is restarted. + popen_kwargs["start_new_session"] = True + with open(run_dir / "console.log", "ab") as log: + execute_arg = ( + "--execute-strict-reserved" + if strict_compat else "--execute-reserved" + ) + proc = subprocess.Popen( + [sys.executable, "-m", "infermatrix_copilot", execute_arg, run_id], + stdout=log, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, + cwd=str(self.run_root), env=env, + **popen_kwargs, + ) + proc.wait() + # BLOCKED_EXIT without a terminal status is the lock-loser signature: + # a genuinely blocked child writes its terminal state before exiting 3 + rs.reconcile_after_wait(run_dir, child_pid=proc.pid, + suspect_lock_loser=(proc.returncode == 3)) + self._reap() # the run is terminal; sweep what it left behind + + # -- start (reserve + enqueue) ------------------------------------------- + def start(self, spec_dict: dict) -> str: + """Boundary policy enforcement + reserve + enqueue; returns the run id. + + No idempotency key here by design: `start` also serves `issue_answer` + 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( + spec, owner_server_id=self.server_id, owner_server_pid=self.pid) + if created: + self._q.put((run_id, False)) + return run_id + + def start_strict_review(self, spec_dict: dict) -> str: + """Reserve a packaged Strict review workflow. + + Enqueues only a run this call actually created. Returning an existing id + while still enqueueing would deduplicate the *id* and not the + *execution* — the second child would review the same PR again.""" + run_id, _created = self.reserve_strict_review(spec_dict) + return run_id + + def reserve_strict_review(self, spec_dict: dict) -> tuple[str, bool]: + """Reserve Strict work and expose whether this call created the run. + + ``start_strict_review`` retains its string-only compatibility contract; + the typed SDK uses this operation so an idempotent retry is never + mislabeled as a newly created run. + """ + spec = enforce_strict_review_policy( + spec_dict, allowed_repos=self.settings.mcp_allowed_repos, + settings=self.settings) + run_id, created = self.copilot.reserve_run( + spec, owner_server_id=self.server_id, owner_server_pid=self.pid, + idempotency_key=str(spec_dict.get("idempotency_key") or "")) + if created: + self._q.put((run_id, True)) + return run_id, created + + def start_quality_review(self, spec_dict: dict) -> str: + """Reserve the dedicated, idempotent PR quality workflow.""" + run_id, _created = self.reserve_quality_review(spec_dict) + return run_id + + def reserve_quality_review(self, spec_dict: dict) -> tuple[str, bool]: + """Reserve quality work and report whether this call created it.""" + spec = enforce_quality_review_policy( + spec_dict, allowed_repos=self.settings.mcp_allowed_repos, + settings=self.settings) + run_id, created = self.copilot.reserve_run( + spec, owner_server_id=self.server_id, owner_server_pid=self.pid, + idempotency_key=str(spec_dict.get("idempotency_key") or "")) + if created: + self._q.put((run_id, False)) + return run_id, created + + def strict_readiness(self, repo: str, repo_path: str = "") -> list[str]: + """Return actionable setup gaps before reserving a Strict run. + + `repo_path` validates THAT explicit checkout. The deleted + `configure_strict_repo` used to mutate `settings.repo_paths` so this + method would see it — process-global state written per call, so two + concurrent Strict requests for different checkouts could each preflight + against the other's.""" + missing = [] + # Backend selection is EXPLICIT for Strict (doc/RFC-provider-registry + # .md): unset refuses with the exact fix, never falls back silently. + backend = self.settings.strict_backend + if not backend: + missing.append( + "STRICT_BACKEND not set; add STRICT_BACKEND=api (or cursor / " + "claude-code / codex) to ~/.infermatrix-copilot/.env") + elif backend == "api": + if not self.settings.shared_api_key: + missing.append( + "model credential missing; set ANTHROPIC_API_KEY or " + "OPENAI_API_KEY in " + "~/.infermatrix-copilot/.env" + ) + else: + from ..providers import transport_for + + try: + transport = transport_for(self.settings) + except NotImplementedError as exc: + missing.append(str(exc)) + else: + if not transport.cli_path(): + missing.append( + f"STRICT_BACKEND={backend} selected but its CLI is " + "not installed; install it or set STRICT_BACKEND_CLI " + "in ~/.infermatrix-copilot/.env") + else: + gap = transport.auth_gap() + if gap: + missing.append(gap) + if repo_path: + try: + repo_path = authorize_repo_path(repo, repo_path, self.settings) + except PolicyError as exc: + missing.append(str(exc)) + repo_path = "" + else: + repo_path = self.copilot._resolve_repo_path(repo) + if not repo_path or not Path(repo_path).is_dir(): + missing.append( + f"checkout for {repo!r} missing; run the installer with " + "--repo-path or set REPO_PATHS in " + "~/.infermatrix-copilot/.env" + ) + if self.copilot.store.get("pr-review") is None: + missing.append( + "packaged pr-review playbook missing; reinstall " + "InferMatrixCopilot" + ) + return missing + + def quality_readiness(self, repo: str, repo_path: str = "") -> list[str]: + """Return setup gaps for the quality workflow and its model backend.""" + missing = [ + item for item in self.strict_readiness(repo, repo_path) + if "packaged pr-review playbook" not in item + ] + if self.copilot.store.get("pr-quality") is None: + missing.append( + "packaged pr-quality playbook missing; reinstall " + "InferMatrixCopilot") + return missing + + # -- poll ----------------------------------------------------------------- + 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) + rs.reconcile_if_dead(run_dir, self.run_root) + status = rs.read_status(run_dir) or {} + progress = None + pf = run_dir / "progress.json" + if pf.exists(): + try: + progress = json.loads(pf.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + progress = None + return {"run_id": run_id, "status": status, "progress": progress} + + def get_result(self, run_id: str, offset: int = 0) -> dict: + """Lazy-reconcile then return the run state; once terminal, attach a + size-capped page of the report (RUN_REPORT.md, or ESCALATION.md when + blocked) with `next_offset` + `report_path` for paging — never an + unbounded dump over the protocol. + + Terminal responses also carry `result`: the structured form from + `contract.build_review_result`, so a machine consumer reads the verdict + and findings as data instead of scraping them back out of the Markdown. + The `report` paging stays for back-compat with existing hosts.""" + # 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) + if not run_dir.exists(): + return {"run_id": run_id, "state": "unknown", "note": "", + "report": None, "report_path": None, "next_offset": None, + "result": contract.unknown_run_result(run_id)} + rs.reconcile_if_dead(run_dir, self.run_root) + status = rs.read_status(run_dir) or {} + state = status.get("state") + out: dict[str, Any] = {"run_id": run_id, "state": state, + "note": status.get("note", "")} + if state not in rs.TERMINAL: + return out # still queued/planning/running — poll again + out["result"] = contract.build_review_result(run_dir) + report_path = run_dir / "RUN_REPORT.md" + if state == rs.BLOCKED and (run_dir / "ESCALATION.md").exists(): + report_path = run_dir / "ESCALATION.md" + if report_path.exists(): + text = report_path.read_text(encoding="utf-8", errors="replace") + offset = max(0, int(offset)) + cap = self.settings.mcp_report_max_bytes + out["report"] = text[offset:offset + cap] + out["report_path"] = str(report_path) + nxt = offset + cap + out["next_offset"] = nxt if nxt < len(text) else None + else: + out.update(report=None, report_path=None, next_offset=None) + return out + + 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) + if not run_dir.exists(): + return { + "run_id": run_id, + "state": "unknown", + "note": "", + "result": contract.unknown_quality_result(run_id), + } + rs.reconcile_if_dead(run_dir, self.run_root) + status = rs.read_status(run_dir) or {} + state = status.get("state") + out: dict[str, Any] = { + "run_id": run_id, + "state": state, + "note": status.get("note", ""), + } + if state in rs.TERMINAL: + out["result"] = contract.build_quality_result(run_dir) + return out + + def list_playbooks(self) -> dict: + """The read-only V1 surface: the exposed kinds and the vetted playbooks + backing them (read-only introspection; no run started).""" + pbs = [line for line in self.copilot.playbooks().splitlines() + if any(k in line for k in READ_ONLY_KINDS)] + return {"read_only_kinds": sorted(READ_ONLY_KINDS), "playbooks": pbs} + + def _docs(self, repo: str) -> KnowledgeDocs: + """Build the same repo-scoped knowledge view used by workflow agents.""" + from ..adapters.base import AdapterRegistry + + repo = repo or self.settings.default_repo + if repo not in self.settings.mcp_allowed_repos: + raise PolicyError(f"repo {repo!r} is not permitted") + try: + adapter = AdapterRegistry(self.settings.adapters_dir).resolve( + name=repo.replace("-", "_")) + except Exception as exc: + raise PolicyError(f"no knowledge adapter for repo {repo!r}") from exc + kn = adapter.manifest.get("knowledge") or {} + return KnowledgeDocs(self.settings.knowledge_dir, kn.get("repo_subdir")) + + def doc_search(self, query: str, repo: str = "", limit: int = 40) -> dict: + """Search general + the selected repo's curated Markdown knowledge.""" + selected = repo or self.settings.default_repo + hits = self._docs(selected).search(query, limit=limit) + return {"query": query, "repo": selected, "matches": hits, + "truncated": len(hits) >= max(1, min(int(limit), 100))} + + def doc_read(self, path: str, repo: str = "", offset: int = 0) -> dict: + """Read one document from general + the selected repo's slice.""" + selected = repo or self.settings.default_repo + try: + page = self._docs(selected).read(path, offset=offset) + except FileNotFoundError as exc: + raise KnowledgeDocsError(f"no such document: {path}") from exc + return {"repo": selected, **page} + + def close(self) -> None: + """Deregister this server's liveness token (best-effort, on shutdown).""" + rs.unregister_server(self.run_root, self.server_id) diff --git a/src/infermatrix_copilot/cli/copilot.py b/src/infermatrix_copilot/cli/copilot.py index e411b9526..8da2260e5 100644 --- a/src/infermatrix_copilot/cli/copilot.py +++ b/src/infermatrix_copilot/cli/copilot.py @@ -1,792 +1,5 @@ -"""The `Copilot` orchestrator: NL-resolved TaskSpec → plan (reuse > adapt > -generate) → plan-review gate → executor, plus the compound-command queue, -resume, and the /status /logs /playbooks built-ins. +"""Compatibility import for the application orchestrator.""" -This is the orchestration core; the argparse/REPL wiring lives in `entry.py` and -the pure formatters in `utils.py`. -""" +from ..app.core import Copilot, GateOutcome, _mode_review_context -from __future__ import annotations - -import asyncio -import json -import os -import re -import time -import uuid -from pathlib import Path -from typing import NamedTuple - -import yaml - -from .. import idempotency as idem -from .. import run_status as rs -from ..config import Settings, TierNotConfiguredError -from ..engine.executor import Executor -from ..engine.lifecycle import RunLock, RunLockHeld, run_guarded -from ..engine.planner import Planner, PlanningError, Resolution -from ..engine.registry import StepRegistry -from ..engine.steps import register_builtin_steps -from ..llm import LLM -from ..notify import BLOCKED_EXIT, Notifier -from ..playbooks.store import PlaybookStore, parse_playbook, playbook_to_doc -from ..push import PushPolicy -from ..review.reviewer import run_plan_review -from ..run_trace import RunTrace -from ..task_spec import TaskSpec -from ..ui import style -from .utils import format_metrics_line - - -class GateOutcome(NamedTuple): - """Result of the pre-execution gates. `proceed` says whether to run; when - False, `exit_code` is the process code the caller should return. Read at - the call site as: ``if not gate.proceed: return gate.exit_code``.""" - - proceed: bool - exit_code: int = 0 - - @classmethod - def go(cls) -> "GateOutcome": - """Gates passed — run the task.""" - return cls(proceed=True) - - @classmethod - def stop(cls, exit_code: int) -> "GateOutcome": - """A gate halted the run — return this exit code.""" - return cls(proceed=False, exit_code=exit_code) - - -def _mode_review_context(playbook, spec) -> str: - """Plan-review context for MODE-AWARE playbooks: the reviewer sees the - raw yaml's FULL step list, but the `when:` gates resolve against the - ALREADY-RESOLVED mode (`resolve_effective_mode` runs before the review - gate) — without this context a report-only plan looks like it runs its - write/push steps and gets spuriously blocked. The active-step set is - the same mechanical truth the executor computes, never prose.""" - if not getattr(playbook, "mode_aware", False): - return "" - mode = str((getattr(spec, "params", None) or {}) - .get("rebase_mode", "") or "") - if not mode: - return "" - from ..engine.executor import _eval_when - from ..rebase_engine.modes import mode_state_flags - - flags = {"task_spec": {}, **mode_state_flags(mode)} - active = [s.get("id", s.get("step", "?")) - for s in playbook_to_doc(playbook).get("steps", []) - if "when" not in s or _eval_when(s["when"], flags)] - repo = str(getattr(spec, "repo", "") or "") - repo_line = (f"\nTarget repo (authoritative): {repo!r} — bound at " - "runtime from the TaskSpec; the yaml `repos:` list is a " - "planner RECALL FILTER where empty means repo-neutral, " - "never untargeted." if repo else "") - return (f"\n\nResolved mode context (authoritative): " - f"rebase_mode={mode}. Under this mode the `when:` gates run " - f"ONLY these steps: {active}. Every other listed step is " - "statically gated OFF for this run — judge the plan for THIS " - "mode's step set." - + repo_line + - "\nWrite/push governance (authoritative): the mode's own " - "push/CI steps are governed at runtime by the push-gate " - "ruling, guard_push, and the ALLOW_PUSH env double-gate — " - "the task tier does not forbid steps this mode activates.") - - -class Copilot: - """Orchestration core: resolves a TaskSpec to a playbook (reuse > adapt > - generate), runs it through the plan-review gate + confirmation into the - Executor, and owns the compound-command queue, resume, and the /status - /logs /playbooks built-ins. Holds the long-lived collaborators (LLM, - step registry, playbook store, planner) and tracks `last_run_dir` for the - built-ins.""" - - def __init__(self, settings: Settings | None = None): - """Wire up the collaborators from `settings` (default `Settings()`): - the LLM, the built-in step registry, the playbook store rooted at the - configured dir, and the planner over both. `last_run_dir` starts unset - and is filled by the first execution.""" - self.settings = settings or 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 - - self.llm = llm_for(self.settings) - self.registry = register_builtin_steps(StepRegistry()) - self.store = PlaybookStore(self.settings.playbooks_dir, self.registry) - self.planner = Planner(self.store, self.registry) - self.last_run_dir: Path | None = None - # why the last execution stopped, when it stopped blocked — read by - # `_execute_reserved_locked`, which owns the terminal run_status write - self.last_blocked_reason: str = "" - - # -- planning --------------------------------------------------------------- - def resolve(self, spec: TaskSpec) -> Resolution: - """Resolve `spec` to a Resolution via the planner, passing the repo's - capability set so the planner only reuses playbooks the target supports. - Capabilities come from the repo's adapter (if any), plus `repo.path` when - a path is resolvable even without a adapter (REPO_PATHS works adapter-less).""" - # Only genuine adapter ABSENCE takes the unknown-capabilities - # compatibility path — a malformed/unreadable KNOWN adapter must fail - # closed here, not fail open into capabilities=None and recall a - # playbook whose requirements were never established. - from ..adapters.base import AdapterError, AdapterNotFound, AdapterRegistry - adapter_name = spec.repo.replace("-", "_") - try: - adapter = AdapterRegistry(self.settings.adapters_dir).resolve( - name=adapter_name) - except AdapterNotFound: - # the registry resolves by DECLARED manifest name and skips - # manifest-less directories — so "not found" alone does not - # prove absence. Only a genuinely absent directory is the - # v1-compatible path; an existing directory that failed to - # load/resolve (deleted manifest, wrong name:) fails closed. - if (Path(self.settings.adapters_dir) / adapter_name).exists(): - raise AdapterError( - f"adapter directory {adapter_name!r} exists but did not " - "load/resolve (missing manifest.yaml or mismatched " - "name:) — refusing to plan with unknown capabilities") - adapter = None # absence: v1-compatible - except FileNotFoundError: - adapter = None # no adapters directory at all - if adapter is None: - # No adapter means capabilities are UNKNOWN, not zero — the - # store's requires-filter (now covering exact-repo playbooks too) - # skips on None, keeping adapter-less setups v1-compatible - # instead of silently dropping every playbook with a `requires:`. - if self._repo_path_for(spec): - return self.planner.resolve(spec, capabilities=None) - return self.planner.resolve(spec, capabilities=set()) - capabilities = set(adapter.capabilities) - if self._repo_path_for(spec): # REPO_PATHS works adapter-less - capabilities.add("repo.path") - return self.planner.resolve(spec, capabilities=capabilities) - - def _plan_review_gate(self, resolution: Resolution, spec: TaskSpec, - assume_yes: bool) -> bool: - """Inline Plan-Review for adapted/generated plans. LLM verdict shown in - the session; block stops. A non-`lgtm` verdict is only ever *surfaced* - — the human `[y/N]` is what actually gates it — so when `--yes` removes - that human, the same verdict must stop the run instead - (`ReviewVerdict.passing`, SPEC C6: only `lgtm` passes).""" - if not resolution.requires_review: - return True - doc = yaml.safe_dump(playbook_to_doc(resolution.playbook), sort_keys=False) - task_text = spec.describe() + _mode_review_context( - resolution.playbook, spec) - verdict = run_plan_review(self.llm, playbook_doc=doc, - task=task_text, - model=self.settings.reviewer) - if verdict.verdict != "unavailable": - print(f" plan review: {verdict.verdict}" - + (f" — {verdict.critiques}" if verdict.critiques else "")) - if verdict.verdict == "block": - print("✋ plan blocked by reviewer.") - return False - if verdict.passing: - return True - # Everything below is non-passing: `revise`, or `unavailable`. Both used - # to return True on the strength of a confirmation that `--yes` had - # already deleted, so an unattended run executed an unvetted plan on an - # unread verdict. Measured on the release matrix: an unparseable review - # let three of four backends run a pr-rebase plan through to its push - # gate, while the one backend whose reviewer parsed cleanly BLOCKED the - # very same plan. - if assume_yes: - reason = ("no reviewer LLM" if verdict.verdict == "unavailable" - else f"plan review returned {verdict.verdict}") - print(f"✋ {reason} and --yes leaves no human to gate it — blocked.") - return False - if verdict.verdict == "unavailable": - print(" ⚠ no reviewer LLM — your confirmation is the plan-review gate") - return True # revise/unavailable, surfaced to the user before their confirm - - def _gate_and_confirm(self, resolution: Resolution, spec: TaskSpec, - assume_yes: bool, *, prompt: str = "Proceed?", - force_confirm: bool = False) -> GateOutcome: - """Plan-review gate + [y/N] confirm (concision K6). Returns - `GateOutcome.go()` to run, or `GateOutcome.stop(code)` to halt with a - process exit code (BLOCKED_EXIT on a reviewer block, 1 on a user - abort). Confirm fires for confirm_required or a review-requiring/ - explicit plan, unless assume_yes.""" - if not self._plan_review_gate(resolution, spec, assume_yes): - return GateOutcome.stop(BLOCKED_EXIT) # reviewer blocked the plan - need = force_confirm or spec.confirm_required or resolution.requires_review - if need and not assume_yes: - if input(f"{prompt} [y/N] ").strip().lower() not in ("y", "yes"): - print("aborted.") - return GateOutcome.stop(1) # user declined - return GateOutcome.go() - - # -- execution ----------------------------------------------------------------- - def run_task(self, spec: TaskSpec, *, assume_yes: bool = False, - plan_only: bool = False) -> int: - """Full path for one task: resolve → print plan → gate/confirm → - persist task.json → execute. Returns a process exit code — 0 done, - 1 failed/aborted, BLOCKED_EXIT when planning fails or a gate blocks. - `plan_only` prints the resolved plan and returns 0 without running; - `assume_yes` skips the interactive confirm.""" - try: - resolution = self.resolve(spec) - except PlanningError as exc: - print(style("✋ cannot plan: ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - # tier preflight (plan v2): backend availability is DEPLOYMENT state, - # so it is enforced here at resolution — before the echo, the gate, - # and the run dir — never discovered as a BLOCKED step minutes in. - try: - tier_target = self.settings.tier_target(spec.mode) - except TierNotConfiguredError as exc: - print(style("✋ cannot run: ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - - # Mode governance (Rev 8 §2.1): for mode-aware playbooks the ONE - # authority is params.rebase_mode, resolved + WRITTEN BACK before the - # plan echo and the confirmation gate (spec.report_only reflects the - # canonical mode, so every TaskSpec-derived consumer sees one truth). - # The locked delegating playbook does not declare mode_aware and is - # untouched — byte-identical behavior. - if getattr(resolution.playbook, "mode_aware", False): - from ..rebase_engine.modes import (ModeConflictError, - resolve_effective_mode) - try: - resolve_effective_mode(spec) - except ModeConflictError as exc: - print(style("✋ blocked: ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - - print(style("→ task: ", "bold", "cyan") + spec.describe()) - print(style("→ plan: ", "bold", "magenta") + f"{resolution.mode} {resolution.playbook.name}" - f"@{resolution.playbook.version} ({resolution.playbook.status}) " - f"steps={[s.step for s in resolution.playbook.steps]}") - for note in resolution.notes: - print(f" · {note}") - print(style("→ path: ", "bold", "cyan") - + f"{spec.mode} · agent-model={tier_target.model}" - f" @ {tier_target.host}") - if plan_only: - return 0 - - gate = self._gate_and_confirm(resolution, spec, assume_yes) - if not gate.proceed: - return gate.exit_code - - run_id = (f"run-{time.strftime('%Y%m%d-%H%M%S')}" - f"-{uuid.uuid4().hex[:6]}") # unique — same-second runs collided - run_dir = self.settings.run_root / run_id - run_dir.mkdir(parents=True, exist_ok=True) - (run_dir / "task.json").write_text(json.dumps({ - "spec": spec.model_dump(), "playbook": playbook_to_doc(resolution.playbook), - "invocation_id": os.environ.get("IMX_INVOCATION_ID", ""), - }, indent=2)) - return self._execute(resolution.playbook, spec, run_dir, - resolution_mode=resolution.mode, tier=resolution.tier) - - def run_playbook(self, name: str, *, params: dict | None = None, - report_only: bool = False, assume_yes: bool = False, - plan_only: bool = False) -> int: - """Explicit playbook override — the only way to execute a CANDIDATE - (candidates stay planner-invisible). Always treated as requiring - review + confirmation.""" - playbook = self.store.get(name) - if playbook is None: - print(f"✋ no playbook named {name!r} (see /playbooks)") - return BLOCKED_EXIT - kind = playbook.task_kinds[0] - # lift target params into first-class spec fields so the explicit - # playbook path can address a PR/issue/repo like every other surface - params = dict(params or {}) - pr = params.pop("pr", None) - issue = params.pop("issue", None) - repo = str(params.pop("repo", "") or self.settings.default_repo) - try: - pr = int(pr) if pr is not None else None - issue = int(issue) if issue is not None else None - except (TypeError, ValueError): - print("✋ pr/issue task params must be integers") - return BLOCKED_EXIT - if repo not in (self.settings.repo_paths or {repo: ""}): - print(f"✋ unknown repo alias {repo!r} " - f"(known: {', '.join(sorted(self.settings.repo_paths or {}))})") - return BLOCKED_EXIT - spec = TaskSpec(kind=kind, repo=repo, pr=pr, issue=issue, - report_only=report_only, params=params) - from ..intent import validate_spec - err = validate_spec(spec) - if err: - print(f"✋ {err} (pass e.g. --task-param pr=5134)") - return BLOCKED_EXIT - # mode governance applies to explicit invocations too — the ONLY - # way to run the candidate v3/v1 playbooks today (Rev 8 §2.1) - if getattr(playbook, "mode_aware", False): - from ..rebase_engine.modes import (ModeConflictError, - resolve_effective_mode) - try: - resolve_effective_mode(spec) - except ModeConflictError as exc: - print(style("✋ blocked: ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - print(style("→ task: ", "bold", "cyan") + spec.describe() - + style(" [explicit playbook override]", "yellow")) - print(style("→ plan: ", "bold", "magenta") + f"explicit {playbook.name}@{playbook.version} " - f"({playbook.status}) steps={[s.step for s in playbook.steps]}") - if plan_only: - return 0 - resolution = Resolution(mode="explicit", playbook=playbook, - tier=spec.tier, requires_review=True) - gate = self._gate_and_confirm( - resolution, spec, assume_yes, force_confirm=True, - prompt=f"Run {playbook.status} playbook '{name}'?") - if not gate.proceed: - return gate.exit_code - run_id = (f"run-{time.strftime('%Y%m%d-%H%M%S')}" - f"-{uuid.uuid4().hex[:6]}") # unique — same-second runs collided - run_dir = self.settings.run_root / run_id - run_dir.mkdir(parents=True, exist_ok=True) - (run_dir / "task.json").write_text(json.dumps({ - "spec": spec.model_dump(), "playbook": playbook_to_doc(playbook), - "invocation_id": os.environ.get("IMX_INVOCATION_ID", ""), - }, indent=2)) - return self._execute(playbook, spec, run_dir, - resolution_mode="explicit", tier=spec.tier) - - def _execute(self, playbook, spec: TaskSpec, run_dir: Path, *, - resolution_mode: str = "resume", tier: str = "?", - resuming: bool = False, held_lock: RunLock | None = None) -> int: - """Run a resolved `playbook` to completion in `run_dir`: init tracing + - notifier, seed the shared state (repo path, push policy, protected - branches / high-risk modules from the adapter when present), drive the - Executor, then print per-step marks, optional run metrics, and the final - status. `resuming` seeds the state so steps can pick up where they left - off. Returns the exit code (0 done, BLOCKED_EXIT blocked, else 1), and - records `last_blocked_reason` for the reserved-run caller, which has only - the code but owns the terminal `run_status.json` write.""" - self.last_run_dir = run_dir - self.last_blocked_reason = "" - lock = held_lock - if lock is None: - try: - # First, before any trace/status write: a losing concurrent - # invocation (second --resume) must leave the active run's - # artifacts completely untouched. execute_reserved holds the - # lock across its whole lifecycle and passes it in instead. - lock = RunLock(run_dir).acquire() - except RunLockHeld as exc: - print(style("✋ ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - # Per-repo knowledge lock, SHARED, held for the run's lifetime: - # never contends with other runs; it exists so the knowledge - # migration's EXCLUSIVE acquire can prove no potential store - # writer is alive (and so no run starts mid-migration). - knowledge_lock = None - if spec.repo: - from ..memory.paths import (KnowledgeLockHeld, - KnowledgePaths, - KnowledgeRunLock, - KnowledgeStateError) - try: - knowledge_lock = KnowledgeRunLock( - KnowledgePaths.resolve(self.settings, spec.repo) - .knowledge_run_lock).acquire_shared() - except (KnowledgeLockHeld, KnowledgeStateError) as exc: - # BOTH refusals take the terminal protocol: an invalid - # activation marker (KnowledgeStateError) must exit - # blocked/3 with the run lock RELEASED, never escape as - # a traceback that leaves the lock held in a long-lived - # process (PR-boundary F4) - if held_lock is None: - lock.release() - print(style("✋ ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - try: - from .. import tracing - tracing.init(run_dir.name, run_dir / "trace.jsonl") - # Stamp the workflow into the span file itself, so a trace lifted out of - # its run directory still says which playbook and task produced it. - tracing.run_meta(playbook=f"{playbook.name}@{playbook.version}", - task_kind=spec.kind, repo=spec.repo, tier=tier, - mode=spec.mode, resolution=resolution_mode, - report_only=spec.report_only, post=spec.post, - resuming=resuming, params=spec.params) - trace = RunTrace(run_dir / "run_trace.jsonl") - notifier = Notifier(self.settings, run_dir, trace, run_dir.name) - trace.record("task", spec=spec.model_dump(), resolution=resolution_mode, - playbook=playbook.name, tier=tier) - state: dict = { - "task_spec": spec.model_dump(), - "repo_path": self._repo_path_for(spec), - "push_policy": PushPolicy(), # steps may replace with a derived policy - "protected_branches": self.settings.protected_branches, - "resuming": resuming, - } - adapter = self._adapter_for(spec.repo) - if adapter is not None: - # repo knowledge from the adapter, not core settings (v2 P0 fix #5) - state["protected_branches"] = adapter.protected_branches - if adapter.high_risk_modules: - state["high_risk_modules"] = adapter.high_risk_modules - executor = Executor(self.registry, self.settings, run_dir=run_dir, - trace=trace, llm=self.llm, notifier=notifier) - # run_guarded finalizes inside the event loop: playbooks that - # register run finalizers (lifecycle.register_finalizer) get - # teardown on every exit path. Nothing registered == no-op. - outcome = asyncio.run( - run_guarded(executor.run(playbook, state), run_dir)) - - if outcome.status == "done": - notifier.resolve() - - if self.settings.metrics_enabled: - try: # metrics are facts about the run; never let them break it - from ..metrics import collect_run_metrics - m = collect_run_metrics(run_dir, self.settings, outcome.status) - print(format_metrics_line(m, run_dir)) - except Exception as exc: - trace.record("metrics_error", error=f"{type(exc).__name__}: {exc}") - - for step_id, r in outcome.step_results.items(): - mark = style("✓", "green") if r.ok else style("✗", "red", "bold") - print(f" {mark} {step_id}: {r.summary}") - print(f"run {run_dir.name}: {outcome.status} ({run_dir})") - if outcome.status == "blocked": - self.last_blocked_reason = str(outcome.blocked_reason or "") - print(style(" ⚠ ", "yellow", "bold") + f"{outcome.blocked_reason}\n see {run_dir / 'ESCALATION.md'}") - return BLOCKED_EXIT - return 0 if outcome.status == "done" else 1 - finally: - if knowledge_lock is not None: - knowledge_lock.release() - if held_lock is None: - lock.release() - - def run_queue(self, specs: list[TaskSpec], *, assume_yes: bool = False, - plan_only: bool = False) -> int: - """Ordered task queue for compound commands; stops on failure/blocked.""" - if len(specs) > 1: - print(f"⧉ queued {len(specs)} tasks:") - for i, s in enumerate(specs, 1): - print(f" {i}. {s.describe()}") - for i, spec in enumerate(specs, 1): - if len(specs) > 1: - print(f"\n── task {i}/{len(specs)} ──") - code = self.run_task(spec, assume_yes=assume_yes, plan_only=plan_only) - if code != 0: - if i < len(specs): - print(f"⏸ queue stopped: {len(specs) - i} task(s) not run") - return code - return 0 - - def resume_last(self) -> int: - """Re-enter the most recent run at its first incomplete step.""" - runs = sorted(self.settings.run_root.glob("run-*")) \ - if self.settings.run_root.exists() else [] - for run_dir in reversed(runs): - task_file = run_dir / "task.json" - if not task_file.exists(): - continue - saved = json.loads(task_file.read_text(encoding="utf-8")) - spec = TaskSpec(**saved["spec"]) - playbook = parse_playbook(saved["playbook"], str(task_file)) - try: # tier preflight holds on re-entry too (deployment state - # may have changed since the run was created) - self.settings.tier_target(spec.mode) - except TierNotConfiguredError as exc: - print(style("✋ cannot resume: ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - print(f"↻ resuming {run_dir.name}: {spec.describe()}") - return self._execute(playbook, spec, run_dir, resuming=True) - 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}$") - - @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]}" - - 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 - - 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 ..mcp_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 - - def execute_reserved(self, run_id: str) -> int: - """MCP-only child entry (subprocess, stdout -> console.log). Writes its - own pid FIRST (single-writer invariant), **re-enforces** the read-only - MCP policy on the persisted request (request.json is untrusted — a host - could have rewritten it), plans, then executes, driving - `run_status.json` planning -> running -> terminal. Returns the exit code.""" - from ..mcp_policy import enforce_mcp_policy - - return self._execute_reserved(run_id, enforce_mcp_policy) - - def execute_strict_reserved(self, run_id: str) -> int: - """Execute a reserved Strict review using the previous Eco workflow.""" - from ..mcp_policy import enforce_strict_review_policy - - return self._execute_reserved(run_id, enforce_strict_review_policy) - - def _execute_reserved(self, run_id: str, policy) -> int: - """Shared reserved-run executor with an authoritative child policy.""" - run_dir = self._contained_run_dir(run_id) - try: - # Lock the whole reserved-run lifecycle before the first status - # write: a duplicate child must not overwrite child_pid, flip - # run_status.json through PLANNING, or mark the active run - # BLOCKED while the winning process is still executing. - lock = RunLock(run_dir).acquire() - except RunLockHeld as exc: - print(style("✋ ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - try: - return self._execute_reserved_locked(run_dir, lock, policy) - finally: - lock.release() - - def _execute_reserved_locked(self, run_dir: Path, lock: RunLock, - policy) -> int: - """The body of `_execute_reserved`, run while holding the run lock.""" - from ..mcp_policy import PolicyError - - # The claim, not the lock, is what makes execution at-most-once. - # `RunLock` excludes concurrent children; it says nothing about a second - # child that arrives after the first finished and released. A losing - # child exits here without planning, executing, or writing status — it - # is a duplicate, not a failure. - if not rs.claim_for_execution(run_dir, child_pid=os.getpid()): - state = (rs.read_status(run_dir) or {}).get("state", "?") - print(style("↷ ", "yellow") + - f"run {run_dir.name} is already claimed (state={state}) — " - "not executing it twice") - return 0 - try: - raw = json.loads((run_dir / "request.json").read_text(encoding="utf-8")) - spec = policy( - raw, allowed_repos=self.settings.mcp_allowed_repos, - settings=self.settings) - except (PolicyError, OSError, json.JSONDecodeError, ValueError) as exc: - rs.mark(run_dir, rs.FAILED, note=f"policy/request rejected: {exc}") - return 1 - try: - resolution = self.resolve(spec) - except PlanningError as exc: - rs.mark(run_dir, rs.BLOCKED, note=f"cannot plan: {exc}") - return BLOCKED_EXIT - try: # tier preflight is re-enforced in the MCP child (authoritative) - self.settings.tier_target(spec.mode) - except TierNotConfiguredError as exc: - rs.mark(run_dir, rs.BLOCKED, note=f"cannot run: {exc}") - return BLOCKED_EXIT - (run_dir / "task.json").write_text(json.dumps({ - "spec": spec.model_dump(), - "playbook": playbook_to_doc(resolution.playbook), - }, indent=2)) - rs.mark(run_dir, rs.RUNNING) - try: - code = self._execute(resolution.playbook, spec, run_dir, - resolution_mode=resolution.mode, tier=resolution.tier, - held_lock=lock) - except Exception as exc: # a crash still leaves a terminal record - rs.mark(run_dir, rs.FAILED, note=f"{type(exc).__name__}: {exc}") - raise - # Carry WHY it stopped into the terminal record. An MCP client only ever - # sees `run_status.json`, and a bare `blocked` with no note gives it - # nothing to distinguish a stale-head refusal from a missing tier or a - # failed checkout — the reason was previously printed to the child's - # console and discarded. - rs.mark(run_dir, {0: rs.DONE, BLOCKED_EXIT: rs.BLOCKED}.get(code, rs.FAILED), - note=self.last_blocked_reason if code == BLOCKED_EXIT else "") - return code - - def _adapter_for(self, repo: str): - """The repo's registered adapter, or None (never raises).""" - try: - from ..adapters.base import AdapterRegistry - - return AdapterRegistry(self.settings.adapters_dir).resolve( - name=repo.replace("-", "_")) - except Exception: - return None - - def _repo_path_for(self, spec: TaskSpec) -> str: - """The checkout THIS spec runs against: its frozen `repo_path` first, - else the ambient resolution by alias. - - Every place a run learns its checkout must agree. Resolution derives the - `repo.path` capability from this, and `_execute` seeds `state` from it: - with an explicit path but no ambient `REPO_PATHS` and no adapter - `repo.path`, an alias-only lookup would drop the capability and fail to - resolve a playbook that requires it — while a perfectly valid checkout - sat in the spec. With both configured and differing, planning would - evaluate one checkout and execution would run against another.""" - return spec.repo_path or self._resolve_repo_path(spec.repo) - - def _resolve_repo_path(self, repo: str) -> str: - """REPO_PATHS first; fall back to the repo's adapter manifest (adapter zero - declares repo.path), so runs work even without a .env in reach. - - Alias-only: callers that hold a TaskSpec want `_repo_path_for`, which - honors a frozen per-run path.""" - p = self.settings.repo_path(repo) - if p: - return str(p) - adapter = self._adapter_for(repo) - if adapter and adapter.repo_path: - return adapter.repo_path - return "" - - # -- built-ins --------------------------------------------------------------- - def status(self) -> str: - """Human-readable status of the current (or most recent) run: completed - steps from progress.json, plus a rebase-phase line (module/test counts, - CI result) when rebase_status.json exists. Falls back to the newest - run-* dir when no run has executed this session.""" - if not self.last_run_dir: - runs = sorted(self.settings.run_root.glob("run-*")) \ - if self.settings.run_root.exists() else [] - if not runs: - return "no runs yet" - self.last_run_dir = runs[-1] - progress = self.last_run_dir / "progress.json" - lines = [] - if progress.exists(): - done = list(json.loads(progress.read_text(encoding="utf-8")).get("completed", {})) - lines.append(f"{self.last_run_dir.name}: completed steps: {done}") - else: - lines.append(f"{self.last_run_dir.name}: no progress recorded") - rebase_status = self.last_run_dir / "rebase_status.json" - if rebase_status.exists(): - s = json.loads(rebase_status.read_text(encoding="utf-8")) - mods = s.get("modules", {}) - tests = s.get("tests", {}) - lines.append( - f" rebase: phase={s.get('phase')} modules(done={mods.get('done', 0)} " - f"failed={mods.get('failed', 0)}) tests(completed={tests.get('completed', 0)} " - f"failed={len(tests.get('failed', []))}" - + (f" current={tests.get('current')}" if tests.get("current") else "") - + f") ci={s.get('ci_result') or '-'}" - ) - return "\n".join(lines) - - def logs(self, n: int = 20) -> str: - """Return the last `n` lines of the current run's run_trace.jsonl, or a - placeholder string when there is no run / no trace yet.""" - if not self.last_run_dir: - return "no runs yet" - tracefile = self.last_run_dir / "run_trace.jsonl" - if not tracefile.exists(): - return "no trace" - return "".join(tracefile.read_text(encoding="utf-8", errors="replace").splitlines(keepends=True)[-n:]) - - def playbooks(self) -> str: - """One line per registered playbook (name@version, status, task kinds), - or "(none)" when the store is empty.""" - return "\n".join( - f"{p.name}@{p.version} [{p.status}] kinds={p.task_kinds}" - for p in self.store.all() - ) or "(none)" +__all__ = ["Copilot", "GateOutcome", "_mode_review_context"] diff --git a/src/infermatrix_copilot/cli/utils.py b/src/infermatrix_copilot/cli/utils.py index dfa47f06e..7576ce8b9 100644 --- a/src/infermatrix_copilot/cli/utils.py +++ b/src/infermatrix_copilot/cli/utils.py @@ -22,12 +22,4 @@ def parse_task_params(kvs: list[str]) -> dict: return params -def format_metrics_line(m: dict, run_dir: Path) -> str: - """One-line run-cost/quality summary from a collected metrics dict.""" - cost, risk, catq = m["cost"], m["risk"], m["catq"] - catq_str = (f" CATQ={catq:.3f}" + ("*" if m["quality"]["partial"] else "") - if catq is not None else "") - partial = "+? (unpriced calls)" if cost.get("cost_partial") else "" - return (f" metrics: usd≈{cost['usd']:.2f}{partial} {cost['minutes']:.1f}min " - f"S={risk['safety_multiplier']:.2f}{catq_str}" - f" ({run_dir / 'metrics.json'})") +from ..metrics import format_metrics_line diff --git a/src/infermatrix_copilot/mcp_server.py b/src/infermatrix_copilot/mcp_server.py index 49e0de5ac..fbddadf99 100644 --- a/src/infermatrix_copilot/mcp_server.py +++ b/src/infermatrix_copilot/mcp_server.py @@ -1,437 +1,22 @@ -"""MCP stdio server — the copilot's read-only V1 surface for Claude Code / Codex. +"""MCP stdio transport over the headless durable RunService. -Both hosts speak MCP, so this is the one portable boundary. Knowledge lookup is -synchronous through repo-scoped `doc_search` / `doc_read`. A review takes 5–12 -min, which would blow a synchronous host-call timeout, so workflow tasks use a -**start + poll** pair: `start_*` reserves a run and returns a `run_id` -immediately; `get_result`/`get_status` poll it. The heavy workflow runs in an -**isolated subprocess** (`python -m infermatrix_copilot --execute-reserved `), which -(a) keeps the copilot's stdout out of this process's JSON-RPC stdio channel — -child stdout goes to `/console.log` — and (b) makes the process-global -tracer / `last_run_dir` per-run. - -Safety is structural, not host-trusted: `enforce_mcp_policy` runs here AND -(authoritatively) in the child, so only the three `READ_ONLY_KINDS` ever run, -`post` is always False, and only allow-listed repos are reachable — regardless of -what a tampered `request.json` claims. Runs are serialized through one worker -thread; the durable `run_status.json` + ownership-aware reconciliation -(run_status.py) make polling correct across restarts and multiple servers. - -This module imports the optional `mcp` SDK; it is gated behind the `[mcp]` extra -and never imported by the core package. +Protocol wiring belongs here. Run reservation, policy, execution, and polling +belong to app.run_service and are shared with the embedded SDK. The optional +MCP dependency is imported only while constructing the protocol server. """ from __future__ import annotations -import json import os -import queue import signal -import subprocess +import subprocess # compatibility: older tests/hosts patch this shared module import sys -import threading -import uuid from collections.abc import Callable -from pathlib import Path -from typing import Any, Literal +from typing import Literal -from . import contract -from . import idempotency as idem -from . import run_status as rs from .config import Settings -from .knowledge_docs import KnowledgeDocs, KnowledgeDocsError -from .mcp_policy import ( - PolicyError, - authorize_repo_path, - enforce_mcp_policy, - enforce_quality_review_policy, - enforce_strict_review_policy, -) -from .task_spec import READ_ONLY_KINDS - - -class CopilotMCP: - """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.""" - - 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.""" - from .cli.copilot import Copilot - - self.settings = settings or Settings() - self.copilot = Copilot(self.settings) - self.run_root = Path(self.settings.run_root) - self.run_root.mkdir(parents=True, exist_ok=True) - self.server_id = uuid.uuid4().hex - self.pid = os.getpid() - rs.register_server(self.run_root, self.server_id, self.pid) - rs.startup_reconcile(self.run_root) - 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 - - def _reap(self) -> None: - """Bound what the idempotency index and PR-time worktrees accumulate. - - 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.""" - 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 - - 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, - supports_file_locking=_fcntl is not None) - - # -- worker: one run at a time, each an isolated subprocess --------------- - 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.""" - while True: - run_id, strict_compat = self._q.get() - try: - self._launch(run_id, strict_compat=strict_compat) - except Exception as exc: # noqa: BLE001 - worker must survive - try: - rs.mark(self.run_root / run_id, rs.FAILED, - note=f"launch error: {type(exc).__name__}: {exc}") - except Exception: - pass - finally: - self._q.task_done() - - def _launch(self, run_id: str, *, strict_compat: bool = False) -> None: - """Run one reserved run as `python -m infermatrix_copilot --execute-reserved - `, child stdout+stderr -> console.log. No MCP child may write - outward, Strict included. After `.wait()` the child is reaped, so we - reconcile as sole writer.""" - run_dir = self.run_root / run_id - env = dict(os.environ) - # Belt to the policy's braces. Strict used to be handed ALLOW_POST=1 - # when this server allowed it, because Strict specs could carry - # post=True; the policy now refuses that, so leaving the env gate open - # would only preserve a path to a second publisher on the same PR. - env["ALLOW_POST"] = "0" - env["ALLOW_PUSH"] = "0" - # The Windows Store Python runtime otherwise inherits the machine's - # legacy console code page (commonly GBK). Reports legitimately contain - # Unicode markers such as ✓; force the isolated child's stdio to UTF-8 - # so rendering a completed report cannot fail after all work is done. - env["PYTHONUTF8"] = "1" - env["PYTHONIOENCODING"] = "utf-8" - # The child builds its own Settings() from env/.env, so make THIS server's - # effective config authoritative for it: the same run_root (to locate the - # reserved dir) and the same read-only allowlist (its policy re-check). - env["RUN_ROOT"] = str(self.run_root) - env["MCP_REPO_ALLOWLIST"] = json.dumps(self.settings.mcp_allowed_repos) - env["DEFAULT_REPO"] = self.settings.default_repo - env["REPO_PATHS"] = json.dumps(self.settings.repo_paths) - env["REPO_FULL_NAMES"] = json.dumps(self.settings.repo_full_names) - # The child re-authorizes the request's `repo_path`, so it must judge it - # against the SAME roots and identities this server did — otherwise the - # re-check silently validates against different config than the parent. - env["MCP_ALLOWED_REPO_ROOTS"] = json.dumps( - self.settings.allowed_repo_roots) - popen_kwargs: dict[str, Any] = {} - if os.name == "nt": - # Codex/Claude launch the MCP server over stdio. Without a new - # Windows process group, host or transport shutdown can propagate a - # console control event into the long-running review child and turn - # it into KeyboardInterrupt. CREATE_NO_WINDOW also prevents a - # console flash for every MCP run. - popen_kwargs["creationflags"] = ( - subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW - ) - else: - # Keep the durable child alive and independently reconcilable when - # its stdio MCP parent is restarted. - popen_kwargs["start_new_session"] = True - with open(run_dir / "console.log", "ab") as log: - execute_arg = ( - "--execute-strict-reserved" - if strict_compat else "--execute-reserved" - ) - proc = subprocess.Popen( - [sys.executable, "-m", "infermatrix_copilot", execute_arg, run_id], - stdout=log, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, - cwd=str(self.run_root), env=env, - **popen_kwargs, - ) - proc.wait() - # BLOCKED_EXIT without a terminal status is the lock-loser signature: - # a genuinely blocked child writes its terminal state before exiting 3 - rs.reconcile_after_wait(run_dir, child_pid=proc.pid, - suspect_lock_loser=(proc.returncode == 3)) - self._reap() # the run is terminal; sweep what it left behind - - # -- start (reserve + enqueue) ------------------------------------------- - def start(self, spec_dict: dict) -> str: - """Boundary policy enforcement + reserve + enqueue; returns the run id. - - No idempotency key here by design: `start` also serves `issue_answer` - 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( - spec, owner_server_id=self.server_id, owner_server_pid=self.pid) - if created: - self._q.put((run_id, False)) - return run_id - - def start_strict_review(self, spec_dict: dict) -> str: - """Reserve a packaged Strict review workflow. - - Enqueues only a run this call actually created. Returning an existing id - while still enqueueing would deduplicate the *id* and not the - *execution* — the second child would review the same PR again.""" - run_id, _created = self.reserve_strict_review(spec_dict) - return run_id - - def reserve_strict_review(self, spec_dict: dict) -> tuple[str, bool]: - """Reserve Strict work and expose whether this call created the run. - - ``start_strict_review`` retains its string-only compatibility contract; - the typed SDK uses this operation so an idempotent retry is never - mislabeled as a newly created run. - """ - spec = enforce_strict_review_policy( - spec_dict, allowed_repos=self.settings.mcp_allowed_repos, - settings=self.settings) - run_id, created = self.copilot.reserve_run( - spec, owner_server_id=self.server_id, owner_server_pid=self.pid, - idempotency_key=str(spec_dict.get("idempotency_key") or "")) - if created: - self._q.put((run_id, True)) - return run_id, created - - def start_quality_review(self, spec_dict: dict) -> str: - """Reserve the dedicated, idempotent PR quality workflow.""" - run_id, _created = self.reserve_quality_review(spec_dict) - return run_id - - def reserve_quality_review(self, spec_dict: dict) -> tuple[str, bool]: - """Reserve quality work and report whether this call created it.""" - spec = enforce_quality_review_policy( - spec_dict, allowed_repos=self.settings.mcp_allowed_repos, - settings=self.settings) - run_id, created = self.copilot.reserve_run( - spec, owner_server_id=self.server_id, owner_server_pid=self.pid, - idempotency_key=str(spec_dict.get("idempotency_key") or "")) - if created: - self._q.put((run_id, False)) - return run_id, created - - def strict_readiness(self, repo: str, repo_path: str = "") -> list[str]: - """Return actionable setup gaps before reserving a Strict run. - - `repo_path` validates THAT explicit checkout. The deleted - `configure_strict_repo` used to mutate `settings.repo_paths` so this - method would see it — process-global state written per call, so two - concurrent Strict requests for different checkouts could each preflight - against the other's.""" - missing = [] - # Backend selection is EXPLICIT for Strict (doc/RFC-provider-registry - # .md): unset refuses with the exact fix, never falls back silently. - backend = self.settings.strict_backend - if not backend: - missing.append( - "STRICT_BACKEND not set; add STRICT_BACKEND=api (or cursor / " - "claude-code / codex) to ~/.infermatrix-copilot/.env") - elif backend == "api": - if not self.settings.shared_api_key: - missing.append( - "model credential missing; set ANTHROPIC_API_KEY or " - "OPENAI_API_KEY in " - "~/.infermatrix-copilot/.env" - ) - else: - from .providers import transport_for - - try: - transport = transport_for(self.settings) - except NotImplementedError as exc: - missing.append(str(exc)) - else: - if not transport.cli_path(): - missing.append( - f"STRICT_BACKEND={backend} selected but its CLI is " - "not installed; install it or set STRICT_BACKEND_CLI " - "in ~/.infermatrix-copilot/.env") - else: - gap = transport.auth_gap() - if gap: - missing.append(gap) - if repo_path: - try: - repo_path = authorize_repo_path(repo, repo_path, self.settings) - except PolicyError as exc: - missing.append(str(exc)) - repo_path = "" - else: - repo_path = self.copilot._resolve_repo_path(repo) - if not repo_path or not Path(repo_path).is_dir(): - missing.append( - f"checkout for {repo!r} missing; run the installer with " - "--repo-path or set REPO_PATHS in " - "~/.infermatrix-copilot/.env" - ) - if self.copilot.store.get("pr-review") is None: - missing.append( - "packaged pr-review playbook missing; reinstall " - "InferMatrixCopilot" - ) - return missing - - def quality_readiness(self, repo: str, repo_path: str = "") -> list[str]: - """Return setup gaps for the quality workflow and its model backend.""" - missing = [ - item for item in self.strict_readiness(repo, repo_path) - if "packaged pr-review playbook" not in item - ] - if self.copilot.store.get("pr-quality") is None: - missing.append( - "packaged pr-quality playbook missing; reinstall " - "InferMatrixCopilot") - return missing - - # -- poll ----------------------------------------------------------------- - 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) - rs.reconcile_if_dead(run_dir, self.run_root) - status = rs.read_status(run_dir) or {} - progress = None - pf = run_dir / "progress.json" - if pf.exists(): - try: - progress = json.loads(pf.read_text(encoding="utf-8")) - except (json.JSONDecodeError, OSError): - progress = None - return {"run_id": run_id, "status": status, "progress": progress} - - def get_result(self, run_id: str, offset: int = 0) -> dict: - """Lazy-reconcile then return the run state; once terminal, attach a - size-capped page of the report (RUN_REPORT.md, or ESCALATION.md when - blocked) with `next_offset` + `report_path` for paging — never an - unbounded dump over the protocol. - - Terminal responses also carry `result`: the structured form from - `contract.build_review_result`, so a machine consumer reads the verdict - and findings as data instead of scraping them back out of the Markdown. - The `report` paging stays for back-compat with existing hosts.""" - # 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) - if not run_dir.exists(): - return {"run_id": run_id, "state": "unknown", "note": "", - "report": None, "report_path": None, "next_offset": None, - "result": contract.unknown_run_result(run_id)} - rs.reconcile_if_dead(run_dir, self.run_root) - status = rs.read_status(run_dir) or {} - state = status.get("state") - out: dict[str, Any] = {"run_id": run_id, "state": state, - "note": status.get("note", "")} - if state not in rs.TERMINAL: - return out # still queued/planning/running — poll again - out["result"] = contract.build_review_result(run_dir) - report_path = run_dir / "RUN_REPORT.md" - if state == rs.BLOCKED and (run_dir / "ESCALATION.md").exists(): - report_path = run_dir / "ESCALATION.md" - if report_path.exists(): - text = report_path.read_text(encoding="utf-8", errors="replace") - offset = max(0, int(offset)) - cap = self.settings.mcp_report_max_bytes - out["report"] = text[offset:offset + cap] - out["report_path"] = str(report_path) - nxt = offset + cap - out["next_offset"] = nxt if nxt < len(text) else None - else: - out.update(report=None, report_path=None, next_offset=None) - return out - - 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) - if not run_dir.exists(): - return { - "run_id": run_id, - "state": "unknown", - "note": "", - "result": contract.unknown_quality_result(run_id), - } - rs.reconcile_if_dead(run_dir, self.run_root) - status = rs.read_status(run_dir) or {} - state = status.get("state") - out: dict[str, Any] = { - "run_id": run_id, - "state": state, - "note": status.get("note", ""), - } - if state in rs.TERMINAL: - out["result"] = contract.build_quality_result(run_dir) - return out - - def list_playbooks(self) -> dict: - """The read-only V1 surface: the exposed kinds and the vetted playbooks - backing them (read-only introspection; no run started).""" - pbs = [line for line in self.copilot.playbooks().splitlines() - if any(k in line for k in READ_ONLY_KINDS)] - return {"read_only_kinds": sorted(READ_ONLY_KINDS), "playbooks": pbs} - - def _docs(self, repo: str) -> KnowledgeDocs: - """Build the same repo-scoped knowledge view used by workflow agents.""" - from .adapters.base import AdapterRegistry - - repo = repo or self.settings.default_repo - if repo not in self.settings.mcp_allowed_repos: - raise PolicyError(f"repo {repo!r} is not permitted") - try: - adapter = AdapterRegistry(self.settings.adapters_dir).resolve( - name=repo.replace("-", "_")) - except Exception as exc: - raise PolicyError(f"no knowledge adapter for repo {repo!r}") from exc - kn = adapter.manifest.get("knowledge") or {} - return KnowledgeDocs(self.settings.knowledge_dir, kn.get("repo_subdir")) - - def doc_search(self, query: str, repo: str = "", limit: int = 40) -> dict: - """Search general + the selected repo's curated Markdown knowledge.""" - selected = repo or self.settings.default_repo - hits = self._docs(selected).search(query, limit=limit) - return {"query": query, "repo": selected, "matches": hits, - "truncated": len(hits) >= max(1, min(int(limit), 100))} - - def doc_read(self, path: str, repo: str = "", offset: int = 0) -> dict: - """Read one document from general + the selected repo's slice.""" - selected = repo or self.settings.default_repo - try: - page = self._docs(selected).read(path, offset=offset) - except FileNotFoundError as exc: - raise KnowledgeDocsError(f"no such document: {path}") from exc - return {"repo": selected, **page} - - def close(self) -> None: - """Deregister this server's liveness token (best-effort, on shutdown).""" - rs.unregister_server(self.run_root, self.server_id) +from .mcp_policy import PolicyError +from .app.run_service import RunService as CopilotMCP def _guard(fn: Callable[[], dict]) -> dict: diff --git a/src/infermatrix_copilot/metrics.py b/src/infermatrix_copilot/metrics.py index 2c437acec..80386f056 100644 --- a/src/infermatrix_copilot/metrics.py +++ b/src/infermatrix_copilot/metrics.py @@ -582,3 +582,14 @@ def collect_run_metrics(run_dir: Path, settings: "Settings", except OSError: pass return metrics + + +def format_metrics_line(m: dict, run_dir: Path) -> str: + """One-line run-cost/quality summary from a collected metrics dict.""" + cost, risk, catq = m["cost"], m["risk"], m["catq"] + catq_str = (f" CATQ={catq:.3f}" + ("*" if m["quality"]["partial"] else "") + if catq is not None else "") + partial = "+? (unpriced calls)" if cost.get("cost_partial") else "" + return (f" metrics: usd≈{cost['usd']:.2f}{partial} {cost['minutes']:.1f}min " + f"S={risk['safety_multiplier']:.2f}{catq_str}" + f" ({run_dir / 'metrics.json'})") diff --git a/src/infermatrix_copilot/sdk/v1/__init__.py b/src/infermatrix_copilot/sdk/v1/__init__.py index 29a6cad13..ae83b91c9 100644 --- a/src/infermatrix_copilot/sdk/v1/__init__.py +++ b/src/infermatrix_copilot/sdk/v1/__init__.py @@ -32,13 +32,17 @@ KnowledgeRuleProposal, KnowledgeValidatorResult, QualityPollResult, + QualityReviewResult, QualityReviewRequest, QualityRunHandle, RepositoryRef, + ResultDecodeError, SDKError, StrictPollResult, + StrictReviewResult, StrictReviewRequest, StrictRunHandle, + StrictRuntimeConfig, UnsupportedRepositoryError, ) from .strict import StrictRuntime @@ -76,13 +80,17 @@ "KnowledgeValidatorError", "KnowledgeValidatorResult", "QualityPollResult", + "QualityReviewResult", "QualityReviewRequest", "QualityRunHandle", "RepositoryRef", + "ResultDecodeError", "SDKError", "StrictPollResult", + "StrictReviewResult", "StrictReviewRequest", "StrictRunHandle", + "StrictRuntimeConfig", "StrictRuntime", "UnsupportedRepositoryError", "get_capabilities", diff --git a/src/infermatrix_copilot/sdk/v1/models.py b/src/infermatrix_copilot/sdk/v1/models.py index 202d54c23..e54d4d7dd 100644 --- a/src/infermatrix_copilot/sdk/v1/models.py +++ b/src/infermatrix_copilot/sdk/v1/models.py @@ -45,6 +45,10 @@ class KnowledgeCurationError(SDKError): code = "knowledge_curation_error" +class ResultDecodeError(SDKError): + code = "result_decode_error" + + @dataclass(frozen=True) class RepositoryRef(_Serializable): alias: str @@ -292,11 +296,44 @@ class StrictRunHandle(_Serializable): created: bool +@dataclass(frozen=True) +class StrictRuntimeConfig(_Serializable): + """Explicit repository and worker binding for an embedded Strict host.""" + + repository: RepositoryRef + checkout_path: str + allowed_root: str + backend: str = "" + run_root: str = "" + + 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") + + +@dataclass(frozen=True) +class StrictReviewResult(_Serializable): + contract_version: str + reviewed_head_sha: str + verdict: str + summary_markdown: str + comments: tuple[dict[str, Any], ...] + findings: tuple[dict[str, Any], ...] + finding_rechecks: tuple[FindingRecheck, ...] + rechecks_complete: bool + recheck_missing: tuple[str, ...] + stale: bool + diagnostics: dict[str, Any] + + @dataclass(frozen=True) class StrictPollResult(_Serializable): run_id: str state: str payload: dict[str, Any] + review: StrictReviewResult | None = None @property def terminal(self) -> bool: @@ -319,11 +356,24 @@ class QualityRunHandle(_Serializable): created: bool +@dataclass(frozen=True) +class QualityReviewResult(_Serializable): + contract_version: str + reviewed_head_sha: str + verdict: str + confidence: str + summary: str + reasons: tuple[dict[str, Any], ...] + stale: bool + diagnostics: dict[str, Any] + + @dataclass(frozen=True) class QualityPollResult(_Serializable): run_id: str state: str payload: dict[str, Any] + review: QualityReviewResult | None = None @property def terminal(self) -> bool: diff --git a/src/infermatrix_copilot/sdk/v1/strict.py b/src/infermatrix_copilot/sdk/v1/strict.py index 2517f1b9e..6990bbec8 100644 --- a/src/infermatrix_copilot/sdk/v1/strict.py +++ b/src/infermatrix_copilot/sdk/v1/strict.py @@ -2,30 +2,139 @@ from __future__ import annotations +from pathlib import Path from typing import Any, Self from .direct import get_capabilities from .models import ( Capabilities, QualityPollResult, + QualityReviewResult, QualityReviewRequest, QualityRunHandle, + FindingRecheck, + InvalidRequestError, + ResultDecodeError, StrictPollResult, + StrictReviewResult, StrictReviewRequest, StrictRunHandle, + StrictRuntimeConfig, ) +def _object_rows(raw: Any, field: str) -> tuple[dict[str, Any], ...]: + if raw is None: + return () + if not isinstance(raw, list) or any(not isinstance(row, dict) for row in raw): + raise ResultDecodeError(f"{field} must be an array of objects") + return tuple(raw) + + +def _diagnostics(raw: Any) -> dict[str, Any]: + if raw is None: + return {} + if not isinstance(raw, dict): + raise ResultDecodeError("diagnostics must be an object") + return raw + + +def _review_result(raw: Any) -> StrictReviewResult | None: + if raw is None: + return None + if not isinstance(raw, dict): + raise ResultDecodeError("Strict result must be an object") + rechecks = [] + for row in _object_rows(raw.get("finding_rechecks"), "finding_rechecks"): + outcome = row.get("outcome") + if not isinstance(outcome, str) or outcome not in {"fixed", "still_affected", "unverified"}: + raise ResultDecodeError("finding recheck outcome is invalid") + try: + rechecks.append(FindingRecheck(**row)) + except (TypeError, ValueError) as exc: + raise ResultDecodeError(f"invalid finding recheck: {exc}") from exc + missing = raw.get("recheck_missing") or [] + if not isinstance(missing, list) or any(not isinstance(item, str) for item in missing): + raise ResultDecodeError("recheck_missing must be an array of strings") + complete = raw.get("rechecks_complete", False) + if not isinstance(complete, bool): + raise ResultDecodeError("rechecks_complete must be a boolean") + return StrictReviewResult( + contract_version=str(raw.get("contract_version") or ""), + reviewed_head_sha=str(raw.get("reviewed_head_sha") or ""), + verdict=str(raw.get("verdict") or ""), + summary_markdown=str(raw.get("summary_markdown") or ""), + comments=_object_rows(raw.get("comments"), "comments"), + findings=_object_rows(raw.get("findings"), "findings"), + finding_rechecks=tuple(rechecks), + rechecks_complete=complete, + recheck_missing=tuple(missing), + stale=bool(raw.get("stale", False)), + diagnostics=_diagnostics(raw.get("diagnostics")), + ) + + +def _quality_result(raw: Any) -> QualityReviewResult | None: + if raw is None: + return None + if not isinstance(raw, dict): + raise ResultDecodeError("quality result must be an object") + return QualityReviewResult( + contract_version=str(raw.get("contract_version") or ""), + reviewed_head_sha=str(raw.get("reviewed_head_sha") or ""), + verdict=str(raw.get("verdict") or ""), + confidence=str(raw.get("confidence") or ""), + summary=str(raw.get("summary") or ""), + reasons=_object_rows(raw.get("reasons"), "reasons"), + stale=bool(raw.get("stale", False)), + diagnostics=_diagnostics(raw.get("diagnostics")), + ) + + class StrictRuntime: """Own a durable local Strict runtime without exposing MCP internals.""" - def __init__(self, *, settings_overrides: dict[str, Any] | None = None): + def __init__( + self, *, config: StrictRuntimeConfig | None = None, + settings_overrides: dict[str, Any] | None = None, + ): # Lazy imports keep Direct-only consumers independent of server startup # and make this module, not the consumer, the implementation boundary. from ...config import Settings - from ...mcp_server import CopilotMCP - - self._core = CopilotMCP(Settings(**(settings_overrides or {}))) + from ...app.run_service import RunService + + if config is not None and settings_overrides is not None: + raise InvalidRequestError("choose config or legacy settings_overrides") + if config is not None: + checkout = Path(config.checkout_path).resolve() + root = Path(config.allowed_root).resolve() + if not checkout.is_relative_to(root): + raise InvalidRequestError("checkout is outside the allowed root") + alias = config.repository.alias + overrides: dict[str, Any] = { + "_env_file": None, + "repo_paths": {alias: str(checkout)}, + "repo_full_names": {alias: config.repository.full_name}, + "mcp_allowed_repo_roots": [str(root)], + "mcp_repo_allowlist": [alias], + "default_repo": alias, + } + if config.backend: + overrides["strict_backend"] = config.backend + if config.run_root: + overrides["run_root"] = config.run_root + else: + overrides = settings_overrides or {} + if overrides.get("_env_file", object()) is None: + # Settings also reads adapter variables through model_config; a + # constructor-only _env_file override does not cover that path. + class EmbeddedSettings(Settings): + model_config = {**Settings.model_config, "env_file": None} + + settings = EmbeddedSettings(**overrides) + else: + settings = Settings(**overrides) + self._core = RunService(settings) def capabilities(self) -> Capabilities: raw = dict(self._core.capabilities()) @@ -103,6 +212,7 @@ def get_result(self, run_id: str, *, offset: int = 0) -> StrictPollResult: run_id=run_id, state=str(payload.get("state") or "unknown"), payload=payload, + review=_review_result(payload.get("result")), ) def get_quality_result(self, run_id: str) -> QualityPollResult: @@ -112,6 +222,7 @@ def get_quality_result(self, run_id: str) -> QualityPollResult: run_id=run_id, state=str(payload.get("state") or "unknown"), payload=payload, + review=_quality_result(payload.get("result")), ) def close(self) -> None: diff --git a/test/test_mcp.py b/test/test_mcp.py index 2cbed6ff1..4e6915c85 100644 --- a/test/test_mcp.py +++ b/test/test_mcp.py @@ -557,6 +557,7 @@ def test_child_launch_never_opens_the_post_gate( core = _core(settings) core.settings.allow_post = allow_post + core.settings.strict_backend = "codex" run_id, _ = core.copilot.reserve_run( TaskSpec(kind="pr_review", repo="vllm-omni", pr=3), owner_server_id=core.server_id, owner_server_pid=core.pid) @@ -579,6 +580,7 @@ def fake_popen(argv, **kwargs): assert captured["env"]["ALLOW_POST"] == "0" assert captured["env"]["ALLOW_PUSH"] == "0" + assert captured["env"]["STRICT_BACKEND"] == "codex" expected_arg = ( "--execute-strict-reserved" if strict_compat else "--execute-reserved") assert expected_arg in captured["argv"] diff --git a/test/test_planner_playbooks.py b/test/test_planner_playbooks.py index e87d4996c..67531684d 100644 --- a/test/test_planner_playbooks.py +++ b/test/test_planner_playbooks.py @@ -141,7 +141,7 @@ def _gate(settings, verdict_kind, *, assume_yes, monkeypatch): resolution that requires review (adapted/generated plans and explicit --playbook overrides; exact reuse never reaches this gate).""" from infermatrix_copilot.cli import Copilot - from infermatrix_copilot.cli import copilot as copilot_mod + from infermatrix_copilot.app import core as copilot_mod from infermatrix_copilot.engine.planner import Resolution from infermatrix_copilot.review.reviewer import ReviewVerdict from infermatrix_copilot.task_spec import TaskSpec @@ -194,7 +194,7 @@ def test_interactive_run_surfaces_and_defers_to_the_confirm( def _gate_and_confirm(settings, verdict_kind, *, assume_yes, monkeypatch, answer="y"): from infermatrix_copilot.cli import Copilot - from infermatrix_copilot.cli import copilot as copilot_mod + from infermatrix_copilot.app import core as copilot_mod from infermatrix_copilot.engine.planner import Resolution from infermatrix_copilot.review.reviewer import ReviewVerdict from infermatrix_copilot.task_spec import TaskSpec diff --git a/test/test_sdk_v1.py b/test/test_sdk_v1.py index 0bb0a4d06..9c3cc4589 100644 --- a/test/test_sdk_v1.py +++ b/test/test_sdk_v1.py @@ -315,6 +315,61 @@ def test_strict_facade_preserves_idempotency_and_hides_private_paths(): assert runtime.capabilities().distribution_version == "0.2.0" +def test_strict_typed_result_and_decode_error(): + from infermatrix_copilot.sdk.v1 import ResultDecodeError + + runtime = object.__new__(StrictRuntime) + core = _FakeStrictCore() + runtime._core = core + result = runtime.get_result("run-20260829-010101-abcdef") + assert result.review.verdict == "APPROVE" + assert result.review.findings == () + + core.get_result = lambda *_args, **_kwargs: {"state": "done", "result": {"comments": "invalid"}} + with pytest.raises(ResultDecodeError, match="comments"): + runtime.get_result("run-20260829-010101-abcdef") + + +def test_typed_strict_config_rejects_checkout_outside_root(tmp_path): + from infermatrix_copilot.sdk.v1 import StrictRuntimeConfig, InvalidRequestError + + config = StrictRuntimeConfig( + repository=RepositoryRef("demo", "owner/demo"), + checkout_path=str(tmp_path.parent / "elsewhere"), + allowed_root=str(tmp_path), + ) + with pytest.raises(InvalidRequestError, match="outside the allowed root"): + StrictRuntime(config=config) + + +def test_headless_strict_runtime_does_not_import_cli_or_mcp(): + script = """ +import os +import sys +import tempfile +from pathlib import Path +from infermatrix_copilot.sdk.v1 import RepositoryRef, StrictRuntime, StrictRuntimeConfig +with tempfile.TemporaryDirectory() as root: + Path(root, '.env').write_text('STRICT_BACKEND=cursor\\n') + os.chdir(root) + config = StrictRuntimeConfig( + repository=RepositoryRef('demo', 'owner/demo'), + checkout_path=root, allowed_root=root, + backend='codex', + run_root=str(Path(root) / 'runs'), + ) + with StrictRuntime(config=config) as runtime: + assert runtime.capabilities().sdk_api_version == '1.0.0' + assert runtime._core.settings.strict_backend == 'codex' + assert runtime._core.settings.model_config.get('env_file') is None + assert not {'infermatrix_copilot.mcp_server', + 'infermatrix_copilot.cli.copilot', + 'infermatrix_copilot.cli.entry'} & sys.modules.keys() +""" + env = {**os.environ, "PYTHONPATH": str(Path(__file__).parents[1] / "src")} + subprocess.run([sys.executable, "-c", script], check=True, env=env) + + def test_quality_facade_is_typed_idempotent_and_head_bound(): runtime = object.__new__(StrictRuntime) core = _FakeStrictCore()