Provider registry M2+M3: claude-code and codex Strict backends

doc/RFC-provider-registry.md

@@ -1,8 +1,13 @@
 # RFC — Provider registry: subscription-auth harness backends for Strict
 
-- Status: accepted (grilling session 2026-08-14); M1 — registry, api
-  parity, tool bridge, cursor transport — implemented on this branch;
-  claude-code (M2) and codex (M3) declared but unshipped
+- Status: accepted (grilling session 2026-08-14); M1 (registry, api
+  parity, tool bridge, cursor transport) merged in PR #81 and live-smoked;
+  M2 claude-code and M3 codex implemented on this branch — claude-code
+  live-smoked on subscription auth, codex offline-tested only (no ChatGPT
+  login on the dev machine; readiness reports the login gap). The bridge
+  additionally serves the on-demand `repo_map`; skill/memory search tools
+  remain in-process only (cross-process candidate writes deliberately not
+  opened).
 - Owner: LLM/backend layer (`llm.py`, `config.py`), agent runtime
   (`engine/agent_runtime/runner.py`), new `src/infermatrix_copilot/providers/`
 - Prior art studied: Hermes Agent `api_mode` transports

src/infermatrix_copilot/cli/doctor.py

@@ -64,6 +64,9 @@ def _check_strict_backend(settings) -> tuple[bool, str]:
         return False, (f"backend {backend} selected but its CLI is missing — "
                        "fix: install it or set STRICT_BACKEND_CLI=/path/to/"
                        "cli in ~/.infermatrix-copilot/.env")
+    gap = transport.auth_gap()
+    if gap:
+        return False, f"backend {backend}: {gap}"
     return True, f"backend {backend} via {cli}"
 
 
@@ -137,7 +140,7 @@ def _check_playbooks(settings) -> tuple[bool, str]:
 }
 
 
-def _tier_targets(settings) -> list[tuple[str, "object"]]:
+def _tier_targets(settings) -> list[tuple[str, object]]:
     """(label, ResolvedTarget) per configured tier; performance omitted (not
     an error) when deferred/unconfigured."""
     from ..config import TierNotConfiguredError

src/infermatrix_copilot/mcp_server.py

@@ -31,8 +31,9 @@
 import sys
 import threading
 import uuid
+from collections.abc import Callable
 from pathlib import Path
-from typing import Any, Callable, Literal, Optional
+from typing import Any, Literal
 
 from . import run_status as rs
 from .config import Settings
@@ -50,7 +51,7 @@ class CopilotMCP:
     ownership-aware reconciliation. Framework-agnostic (no `mcp` import) so it is
     unit-testable without a live protocol connection."""
 
-    def __init__(self, settings: Optional[Settings] = None):
+    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."""
@@ -64,7 +65,7 @@ def __init__(self, settings: Optional[Settings] = None):
         self.pid = os.getpid()
         rs.register_server(self.run_root, self.server_id, self.pid)
         rs.startup_reconcile(self.run_root)
-        self._q: "queue.Queue[tuple[str, bool]]" = queue.Queue()
+        self._q: queue.Queue[tuple[str, bool]] = queue.Queue()
         self._worker = threading.Thread(target=self._worker_loop, daemon=True,
                                         name="omni-mcp-worker")
         self._worker.start()
@@ -202,6 +203,10 @@ def strict_readiness(self, repo: str) -> list[str]:
                         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)
         repo_path = self.copilot._resolve_repo_path(repo)
         if not repo_path or not Path(repo_path).is_dir():
             missing.append(
@@ -312,7 +317,7 @@ def _guard(fn: Callable[[], dict]) -> dict:
         return {"error": str(exc)}
 
 
-def build_mcp(settings: Optional[Settings] = None):
+def build_mcp(settings: Settings | None = None):
     """Build the FastMCP server with the V1 read-only tools bound to a
     `CopilotMCP`. Importing FastMCP here keeps the `mcp` dependency out of the
     core import path (it lives behind the `[mcp]` extra)."""

src/infermatrix_copilot/providers/base.py

@@ -19,13 +19,28 @@
 
 from __future__ import annotations
 
+import os
 import shutil
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Literal
 
 from ..scopes import ToolScope
 
+# Env a harness CLI subprocess keeps. Everything else — API keys, base URLs,
+# gh tokens, host markers like CLAUDECODE — is dropped: subscription auth
+# lives in HOME state, and an inherited ANTHROPIC_BASE_URL (a gateway on
+# this class of machine) would silently reroute a vendor CLI's traffic.
+_ENV_KEEP = {"PATH", "HOME", "TERM", "COLORTERM", "LANG", "USER", "LOGNAME",
+             "SHELL", "TMPDIR"}
+_ENV_KEEP_PREFIXES = ("LC_", "XDG_")
+
+
+def sanitized_env() -> dict[str, str]:
+    """The allowlisted environment for spawning a harness CLI."""
+    return {k: v for k, v in os.environ.items()
+            if k in _ENV_KEEP or k.startswith(_ENV_KEEP_PREFIXES)}
+
 
 @dataclass(frozen=True)
 class ProviderSpec:
@@ -109,6 +124,13 @@ def require_cli(self) -> str:
                 "STRICT_BACKEND_CLI=/path/to/cli in ~/.infermatrix-copilot/.env")
         return cli
 
+    def auth_gap(self) -> str | None:
+        """A one-line auth problem with its fix, or None when unknown/fine.
+        Cheap enough for `strict_readiness` (one fast CLI status call at
+        most); transports without a cheap check return None and let the run
+        surface auth errors loudly."""
+        return None
+
     # -- contract ------------------------------------------------------------
     def run_session(self, req: AgentSessionRequest):
         """Run one delegated agent step; returns `agent_loop.AgentOutcome`."""

src/infermatrix_copilot/providers/claude_code.py

@@ -0,0 +1,205 @@
+"""Claude Code harness transport — Strict on a Claude subscription (M2).
+
+The cleanest harness citizen (probed live 2026-08-14 on claude 2.1.232):
+headless ``claude -p --output-format json`` returns ONE JSON object with
+``result``, ``num_turns``, ``stop_reason``, ``total_cost_usd``, ``usage``
+and per-model ``modelUsage``; ``--max-turns`` maps our iteration budget
+natively; ``--system-prompt`` (+ ``--exclude-dynamic-system-prompt-
+sections``) replaces the vendor system prompt with our step contract.
+
+Tool governance is fully PREVENTIVE here, unlike cursor: built-in tools are
+denied wholesale via ``--disallowedTools``, and only the MCP tool bridge is
+allowed (``--mcp-config`` + ``--strict-mcp-config`` + ``--allowedTools
+mcp__infermatrix-tools``), so every tool call flows through
+``tools.dispatch`` — no native-tool audit needed. Session tool counts come
+from the bridge trace delta.
+
+Env is the shared allowlist (`base.sanitized_env`): the CLI must use its
+own subscription auth from HOME, never an inherited ANTHROPIC_API_KEY /
+ANTHROPIC_BASE_URL, and never see the host's CLAUDECODE marker.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+from ..agent_loop import AgentOutcome
+from ..llm import Block, Reply
+from .base import (
+    AgentSessionRequest,
+    HarnessTransport,
+    SessionUsage,
+    flatten_messages,
+    sanitized_env,
+)
+from .registry import PROVIDERS
+
+# Built-ins denied for every session: the bridge is the only tool surface.
+_BUILTIN_DENY = ("Bash,Read,Write,Edit,Glob,Grep,WebFetch,WebSearch,"
+                 "NotebookEdit,Task,TodoWrite")
+_BRIDGE_SERVER = "infermatrix-tools"
+
+
+class ClaudeCodeTransport(HarnessTransport):
+    """claude CLI (headless) as a Strict backend."""
+
+    spec = PROVIDERS["claude-code"]
+
+    # -- process plumbing ----------------------------------------------------
+    def _run(self, prompt_text: str, *, system: str, cwd: str,
+             timeout_s: float, max_turns: int, model: str = "",
+             mcp_config: Path | None = None) -> tuple[dict, bool]:
+        """One CLI invocation → (parsed result object, timed_out). Prompt on
+        stdin (argv has a 128KiB per-arg limit; evidence packs exceed it)."""
+        cmd = [self.require_cli(), "-p", "--output-format", "json",
+               "--max-turns", str(max_turns),
+               "--disallowedTools", _BUILTIN_DENY]
+        if system:
+            cmd += ["--system-prompt", system,
+                    "--exclude-dynamic-system-prompt-sections"]
+        selected = model or self.settings.strict_backend_model
+        if selected:
+            cmd += ["--model", selected]
+        if mcp_config is not None:
+            cmd += ["--mcp-config", str(mcp_config), "--strict-mcp-config",
+                    "--allowedTools", f"mcp__{_BRIDGE_SERVER}"]
+        try:
+            proc = subprocess.run(
+                cmd, input=prompt_text, cwd=cwd, env=sanitized_env(),
+                capture_output=True, text=True, encoding="utf-8",
+                errors="replace", timeout=timeout_s, check=False)
+            stdout = proc.stdout or ""
+        except subprocess.TimeoutExpired as exc:
+            raw = exc.stdout or b""
+            stdout = raw.decode("utf-8", "replace") if isinstance(raw, bytes) \
+                else str(raw)
+            return self._parse(stdout), True
+        return self._parse(stdout), False
+
+    @staticmethod
+    def _parse(stdout: str) -> dict:
+        """The single JSON object from -p json output; tolerant of stray
+        warning lines before it."""
+        start = stdout.find("{")
+        if start < 0:
+            return {}
+        try:
+            data = json.loads(stdout[start:])
+            return data if isinstance(data, dict) else {}
+        except json.JSONDecodeError:
+            return {}
+
+    @staticmethod
+    def _usage(data: dict) -> SessionUsage:
+        raw = data.get("usage") or {}
+        usage = SessionUsage(
+            input_tokens=int(raw.get("input_tokens") or 0),
+            output_tokens=int(raw.get("output_tokens") or 0),
+            cost_usd=(float(data["total_cost_usd"])
+                      if data.get("total_cost_usd") is not None else None))
+        # served model = the modelUsage entry that did the main work (max
+        # cost); helper models (haiku sidecars) lose that comparison
+        models = data.get("modelUsage") or {}
+        if isinstance(models, dict) and models:
+            usage.served_model = max(
+                models, key=lambda m: float(
+                    (models[m] or {}).get("costUSD") or 0))
+        return usage
+
+    def _write_mcp_config(self, spec_path: Path) -> Path:
+        """The --mcp-config file, next to the bridge spec (never in the
+        worktree — claude takes the config by flag, so nothing litters the
+        session tree)."""
+        package_root = Path(__file__).resolve().parents[2]
+        config = spec_path.with_suffix(".mcp.json")
+        config.write_text(json.dumps({"mcpServers": {_BRIDGE_SERVER: {
+            "command": sys.executable,
+            "args": ["-m", "infermatrix_copilot.tool_bridge",
+                     "--spec", str(spec_path)],
+            "env": {"PYTHONPATH": str(package_root)},
+        }}}, indent=2), encoding="utf-8")
+        return config
+
+    @staticmethod
+    def _bridge_activity(run_dir: Path, since_line: int) -> tuple[int, list[str]]:
+        """(tool_calls, tools_used) from the bridge trace delta — with
+        built-ins denied, bridged calls ARE the session's tool activity."""
+        trace = run_dir / "bridge_trace.jsonl"
+        if not trace.exists():
+            return 0, []
+        tools: list[str] = []
+        for line in trace.read_text(encoding="utf-8").splitlines()[since_line:]:
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if event.get("kind") == "tool_call":
+                tools.append(str(event.get("tool") or "?"))
+        return len(tools), tools
+
+    @staticmethod
+    def _trace_lines(run_dir: Path) -> int:
+        trace = run_dir / "bridge_trace.jsonl"
+        return len(trace.read_text(encoding="utf-8").splitlines()) \
+            if trace.exists() else 0
+
+    # -- transport contract --------------------------------------------------
+    def run_session(self, req: AgentSessionRequest) -> AgentOutcome:
+        cwd = str(req.scope.root or req.run_dir)
+        mcp_config = (self._write_mcp_config(req.bridge_spec_path)
+                      if req.bridge_spec_path is not None else None)
+        before = self._trace_lines(req.run_dir)
+        data, timed_out = self._run(
+            req.prompt, system=req.system, cwd=cwd, timeout_s=req.timeout_s,
+            max_turns=req.max_iters, model=req.model, mcp_config=mcp_config)
+        usage = self._usage(data)
+        tool_calls, tools_used = self._bridge_activity(req.run_dir, before)
+        is_error = bool(data.get("is_error"))
+        if req.trace is not None:
+            req.trace.record(
+                "harness_session", provider=self.spec.id, step=req.step_name,
+                num_turns=data.get("num_turns"), is_error=is_error,
+                timed_out=timed_out, cost_usd=usage.cost_usd,
+                bridge_tool_calls=tool_calls,
+                served_model=usage.served_model)
+        return AgentOutcome(
+            text=str(data.get("result") or ""),
+            iterations=int(data.get("num_turns") or 0),
+            tool_calls=tool_calls,
+            truncated=timed_out or data.get("stop_reason") == "max_turns",
+            refusals=[],
+            input_tokens=usage.input_tokens,
+            output_tokens=usage.output_tokens,
+            tools_used=tools_used[:40])
+
+    def complete(self, *, system: str, messages: list[dict],
+                 model: str = "", max_tokens: int | None = None,
+                 role: str = "") -> Reply:
+        """Tool-less one-shot: built-ins denied, no MCP config, two turns
+        (one to think, the cap as a backstop). Runs in the run-less scratch
+        of the process cwd — with every tool denied there is nothing to
+        contain."""
+        import tempfile
+
+        with tempfile.TemporaryDirectory(prefix="imc-claude-oneshot-") as td:
+            data, timed_out = self._run(
+                flatten_messages("", messages), system=system, cwd=td,
+                timeout_s=self.settings.strict_backend_timeout_s,
+                max_turns=2, model=model)
+        usage = self._usage(data)
+        text = str(data.get("result") or "")
+        return Reply(
+            blocks=[Block(type="text", text=text)] if text else [],
+            stop_reason="max_tokens" if timed_out else "end_turn",
+            usage={"input_tokens": usage.input_tokens,
+                   "output_tokens": usage.output_tokens,
+                   "cache_read_input_tokens": int(
+                       (data.get("usage") or {}).get(
+                           "cache_read_input_tokens") or 0),
+                   "cache_creation_input_tokens": int(
+                       (data.get("usage") or {}).get(
+                           "cache_creation_input_tokens") or 0)},
+            model=usage.served_model)