feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610 )

Master-parity previously compared only the daemon's startup commit against the local on-disk HEAD. When the checkout was not pulled, parity reported green even though the live remote master had advanced, so a stale daemon could claim a mutation-safe result while running outdated capability gates (observed during PR #592 recovery, where the resolver correctly required restart but parity said in_parity=true). Changes: - master_parity_gate.assess_master_parity() gains an optional live_remote_head and reports the three commits distinctly (daemon_start_head, local_head, live_remote_head) plus live_known / live_stale / mutation_safe. A result is mutation_safe only when daemon, local checkout, and live remote all agree. - parity_block_reasons() now blocks mutations on live-staleness too; read-only operations remain unblocked (non-goal: never block diagnostics offline). - New parity_resolver_disagreement(): typed fail-closed blocker naming the capability resolver as authoritative when it requires restart but parity looks locally green. - New read_remote_master_head(): best-effort `git ls-remote` for the live target, cached with a 60s TTL (bounded offline latency, no network probe per gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic. - Server: _current_master_parity() reads the live remote head; gitea_assess_master_parity and runtime_context surface the distinguished SHAs, mutation_safe, and resolver-authoritative guidance. Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker, remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8 remaining failures (test_config TestAuthIntegration, test_credentials TestGetCredentials) are baseline-proven keychain/env failures identical on the unmodified base. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-09 18:52:59 -04:00
22 changed files with 527 additions and 6015 deletions
@@ -1,610 +0,0 @@
 """Controller-owned work allocator policy (#600).
 Builds on the #613 control-plane DB substrate (``ControlPlaneDB.assign_and_lease``).
 Workers must not self-select exclusive work under the standard multi-LLM
 workflow. They call ``gitea_allocate_next_work`` which:
 1. Inspects candidate Gitea issues/PRs (never raw monitoring incidents).
 2. Applies ADR routing policy (role, terminal path, leases, blocked, deps).
 3. Atomically assigns + leases the selected item in one DB transaction.
 This module is pure selection + substrate orchestration. Gitea I/O for live
 inventory lives in the MCP tool wrapper so tests can inject candidates.
 #612 remains downstream: bridge-created Gitea issues become candidates only
 after they exist as normal issues; this module never assigns incidents.
 """
 from __future__ import annotations
 import os
 import uuid
 from dataclasses import dataclass, field
 from typing import Any, Sequence
 from control_plane_db import (
    ControlPlaneDB,
    ControlPlaneError,
    InvalidWorkKindError,
    LeaseRequiredError,
    WORK_KINDS,
 )
 # Outcomes required by #600 / ADR §5.
 OUTCOME_ASSIGNED = "assigned_work"
 OUTCOME_WAIT = "wait"
 OUTCOME_BLOCKED_TERMINAL = "blocked_by_terminal_path"
 OUTCOME_BLOCKED_LEASE = "blocked_by_active_lease"
 OUTCOME_NEEDS_CONTROLLER = "needs_controller"
 OUTCOME_NO_SAFE = "no_safe_work"
 OUTCOME_ROLE_INELIGIBLE = "role_ineligible"
 OUTCOME_PREVIEW = "preview"  # dry-run only (apply=false)
 ROLE_AUTHOR = "author"
 ROLE_REVIEWER = "reviewer"
 ROLE_MERGER = "merger"
 ROLE_RECONCILER = "reconciler"
 ROLE_CONTROLLER = "controller"
 VALID_ROLES = frozenset(
    {ROLE_AUTHOR, ROLE_REVIEWER, ROLE_MERGER, ROLE_RECONCILER, ROLE_CONTROLLER}
 )
 # Default action matrices by role (mutation gate will re-check).
 ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
    ROLE_AUTHOR: (
        ("implement", "comment", "push", "create_pr"),
        ("approve", "merge", "request_changes", "self_select_without_assignment"),
    ),
    ROLE_REVIEWER: (
        ("review", "comment", "approve", "request_changes"),
        ("merge", "push", "create_pr", "self_select_without_assignment"),
    ),
    ROLE_MERGER: (
        ("merge", "comment"),
        ("approve", "request_changes", "push", "create_pr", "self_select_without_assignment"),
    ),
    ROLE_RECONCILER: (
        ("comment", "diagnose", "cleanup"),
        ("approve", "merge", "push", "create_pr", "self_select_without_assignment"),
    ),
    ROLE_CONTROLLER: (
        ("comment", "diagnose", "allocate"),
        ("approve", "merge", "push", "create_pr"),
    ),
 }
@dataclass
 class WorkCandidate:
    """One assignable Gitea issue or PR presented to the allocator."""
    kind: str  # issue | pr
    number: int
    state: str = "open"
    labels: tuple[str, ...] = ()
    title: str = ""
    priority: int = 0
    head_sha: str | None = None
    # Routing signals (callers derive from Gitea / review feedback).
    request_changes_current_head: bool = False
    approval_on_current_head: bool = False
    approval_stale: bool = False
    approval_contaminated: bool = False
    mergeable: bool = False
    blocked: bool = False
    dependency_unmet: bool = False
    dependency_reason: str | None = None
    already_claimed_elsewhere: bool = False
    def __post_init__(self) -> None:
        self.kind = (self.kind or "").strip().lower()
        self.state = (self.state or "open").strip().lower()
        self.labels = tuple(
            str(x).strip().lower() for x in (self.labels or ()) if str(x).strip()
        )
        if self.kind not in WORK_KINDS:
            raise InvalidWorkKindError(
                f"candidate kind '{self.kind}' is not assignable; only "
                f"{sorted(WORK_KINDS)} (never raw incidents)"
            )
    def as_dict(self) -> dict[str, Any]:
        return {
            "kind": self.kind,
            "number": self.number,
            "state": self.state,
            "labels": list(self.labels),
            "title": self.title,
            "priority": self.priority,
            "head_sha": self.head_sha,
            "request_changes_current_head": self.request_changes_current_head,
            "approval_on_current_head": self.approval_on_current_head,
            "approval_stale": self.approval_stale,
            "approval_contaminated": self.approval_contaminated,
            "mergeable": self.mergeable,
            "blocked": self.blocked,
            "dependency_unmet": self.dependency_unmet,
            "dependency_reason": self.dependency_reason,
        }
@dataclass
 class SkipRecord:
    kind: str
    number: int
    reason: str
    def as_dict(self) -> dict[str, Any]:
        return {"kind": self.kind, "number": self.number, "reason": self.reason}
 def normalize_role(role: str | None, *, profile_name: str | None = None) -> str:
    """Map profile/role strings to a canonical allocator role."""
    raw = (role or "").strip().lower()
    if raw in VALID_ROLES:
        return raw
    prof = (profile_name or "").strip().lower()
    for token in VALID_ROLES:
        if token in prof or prof.endswith(f"-{token}"):
            return token
    if "author" in raw:
        return ROLE_AUTHOR
    if "review" in raw:
        return ROLE_REVIEWER
    if "merg" in raw:
        return ROLE_MERGER
    if "reconcil" in raw:
        return ROLE_RECONCILER
    if "control" in raw:
        return ROLE_CONTROLLER
    raise ControlPlaneError(
        f"unknown allocator role '{role}' (profile={profile_name!r}); "
        f"expected one of {sorted(VALID_ROLES)}"
    )
 def expected_role_for_candidate(c: WorkCandidate) -> str:
    """ADR §5.3 routing: which role should take this work next."""
    if c.kind == "pr":
        if c.approval_contaminated:
            return ROLE_RECONCILER
        if c.request_changes_current_head:
            return ROLE_AUTHOR
        if c.approval_stale:
            return ROLE_REVIEWER
        if c.approval_on_current_head and c.mergeable:
            return ROLE_MERGER
        # Open PR without terminal verdict → reviewer
        return ROLE_REVIEWER
    # Issues: ready work → author by default; blocked stays controller/none
    labels = set(c.labels)
    if "status:blocked" in labels or c.blocked:
        return ROLE_CONTROLLER
    return ROLE_AUTHOR
 def role_actions(role: str) -> tuple[tuple[str, ...], tuple[str, ...]]:
    return ROLE_ACTIONS.get(role, ROLE_ACTIONS[ROLE_AUTHOR])
 def classify_skip(
    c: WorkCandidate,
    *,
    role: str,
    terminal_pr: int | None,
 ) -> str | None:
    """Return skip reason, or None if candidate is selectable for *role*."""
    if c.state in ("merged", "closed"):
        return f"{c.kind}#{c.number} is {c.state}; never assign"
    if c.blocked or "status:blocked" in c.labels:
        return f"{c.kind}#{c.number} is blocked"
    if c.dependency_unmet:
        return (
            c.dependency_reason
            or f"{c.kind}#{c.number} has unmet dependencies"
        )
    if c.already_claimed_elsewhere:
        return f"{c.kind}#{c.number} already claimed elsewhere"
    if c.kind == "pr" and not (c.head_sha or "").strip():
        return f"pr#{c.number} missing head_sha pin"
    # Terminal path first: when an active terminal PR exists, only that PR
    # (or controller diagnosis) is assignable for review-path roles.
    if terminal_pr is not None and c.kind == "pr" and c.number != terminal_pr:
        if role in (ROLE_REVIEWER, ROLE_MERGER):
            return (
                f"pr#{c.number} skipped: active terminal-review lock on "
                f"PR #{terminal_pr} must be resolved first"
            )
    expected = expected_role_for_candidate(c)
    if role == ROLE_CONTROLLER:
        # Controller may inspect anything but only assigns diagnosis targets
        # when contaminated / blocked.
        if expected == ROLE_RECONCILER or c.blocked:
            return None
        return f"{c.kind}#{c.number} does not require controller (expected {expected})"
    if role != expected:
        return (
            f"{c.kind}#{c.number} expects role '{expected}', active role is '{role}'"
        )
    # Ready-gate for issues: prefer status:ready when labels present.
    if c.kind == "issue" and c.labels:
        if "status:ready" not in c.labels and "status:in-progress" not in c.labels:
            # Allow unlabeled open issues; only skip explicit non-ready states.
            if any(l.startswith("status:") for l in c.labels):
                return f"issue#{c.number} not status:ready ({','.join(c.labels)})"
    return None
 def sort_candidates(candidates: Sequence[WorkCandidate]) -> list[WorkCandidate]:
    """Higher priority first; then lower number (older issues) for stability."""
    return sorted(
        candidates,
        key=lambda c: (-int(c.priority), c.kind != "pr", int(c.number)),
    )
 def allocate_next_work(
    db: ControlPlaneDB,
    *,
    session_id: str,
    role: str,
    remote: str,
    org: str,
    repo: str,
    candidates: Sequence[WorkCandidate],
    apply: bool = False,
    profile_name: str | None = None,
    username: str | None = None,
    lease_ttl_seconds: int | None = None,
 ) -> dict[str, Any]:
    """Select and optionally reserve the next work unit via control-plane DB.
    *apply=False* (default): dry-run selection only — no lease/assignment.
    *apply=True*: atomic ``assign_and_lease`` for the selected candidate.
    Never uses file locks or comment-only leases as the assignment source.
    """
    if db is None:
        return {
            "success": False,
            "outcome": OUTCOME_NO_SAFE,
            "reasons": [
                "control-plane DB substrate unavailable (fail closed, #600/#613)"
            ],
            "skipped": [],
            "assignment": None,
            "substrate": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
        }
    try:
        role_norm = normalize_role(role, profile_name=profile_name)
    except ControlPlaneError as exc:
        return {
            "success": False,
            "outcome": OUTCOME_ROLE_INELIGIBLE,
            "reasons": [str(exc)],
            "skipped": [],
            "assignment": None,
            "substrate": "control_plane_db",
        }
    session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
    try:
        db.upsert_session(
            session_id=session_id,
            role=role_norm,
            profile=profile_name,
            pid=os.getpid(),
        )
    except Exception as exc:  # noqa: BLE001 — surface structured
        return {
            "success": False,
            "outcome": OUTCOME_NO_SAFE,
            "reasons": [
                f"failed to register session in control-plane DB: {exc} "
                "(fail closed, #613)"
            ],
            "skipped": [],
            "assignment": None,
            "substrate": "control_plane_db",
        }
    # Expire stale leases globally before selection.
    try:
        db.expire_stale_leases()
    except Exception as exc:  # noqa: BLE001
        return {
            "success": False,
            "outcome": OUTCOME_NO_SAFE,
            "reasons": [f"lease expiry failed: {exc} (fail closed)"],
            "skipped": [],
            "assignment": None,
            "substrate": "control_plane_db",
        }
    terminal = None
    try:
        terminal = db.get_active_terminal_lock(remote=remote, org=org, repo=repo)
    except Exception as exc:  # noqa: BLE001
        return {
            "success": False,
            "outcome": OUTCOME_NO_SAFE,
            "reasons": [f"terminal lock lookup failed: {exc} (fail closed)"],
            "skipped": [],
            "assignment": None,
            "substrate": "control_plane_db",
        }
    terminal_pr = int(terminal["terminal_pr"]) if terminal else None
    skipped: list[SkipRecord] = []
    ordered = sort_candidates(list(candidates))
    selected: WorkCandidate | None = None
    for c in ordered:
        reason = classify_skip(c, role=role_norm, terminal_pr=terminal_pr)
        if reason:
            skipped.append(SkipRecord(c.kind, c.number, reason))
            continue
        selected = c
        break
    if selected is None:
        # If terminal lock blocks all review work, surface that explicitly.
        if terminal_pr is not None and role_norm in (ROLE_REVIEWER, ROLE_MERGER):
            outcome = OUTCOME_BLOCKED_TERMINAL
            reasons = [
                f"no safe work for role '{role_norm}': active terminal-review "
                f"lock on PR #{terminal_pr} (resolve terminal path first, #332/#600)"
            ]
        else:
            outcome = OUTCOME_NO_SAFE
            reasons = [
                f"no safe assignable work for role '{role_norm}' "
                f"among {len(ordered)} candidates"
            ]
        return {
            "success": True,
            "outcome": outcome,
            "apply": bool(apply),
            "role": role_norm,
            "profile_name": profile_name,
            "username": username,
            "session_id": session_id,
            "remote": remote,
            "org": org,
            "repo": repo,
            "selected": None,
            "expected_role_next": None,
            "reasons": reasons,
            "skipped": [s.as_dict() for s in skipped],
            "terminal_pr": terminal_pr,
            "assignment": None,
            "substrate": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
            "downstream_note": (
                "#612 incident bridge remains downstream of #600; "
                "allocator never assigns raw monitoring incidents"
            ),
        }
    expected_role = expected_role_for_candidate(selected)
    allowed, forbidden = role_actions(role_norm)
    selection = {
        "kind": selected.kind,
        "number": selected.number,
        "title": selected.title,
        "labels": list(selected.labels),
        "head_sha": selected.head_sha,
        "priority": selected.priority,
        "expected_role_next": expected_role,
        "reason_selected": (
            f"highest-priority candidate for role '{role_norm}' "
            f"(expected_role={expected_role})"
        ),
    }
    if not apply:
        return {
            "success": True,
            "outcome": OUTCOME_PREVIEW,
            "apply": False,
            "role": role_norm,
            "profile_name": profile_name,
            "username": username,
            "session_id": session_id,
            "remote": remote,
            "org": org,
            "repo": repo,
            "selected": selection,
            "expected_role_next": expected_role,
            "reasons": [
                "dry-run only (apply=false); no assignment/lease created — "
                "call again with apply=true to reserve via control-plane DB"
            ],
            "skipped": [s.as_dict() for s in skipped],
            "terminal_pr": terminal_pr,
            "assignment": None,
            "substrate": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
            "downstream_note": (
                "#612 incident bridge remains downstream of #600; "
                "allocator never assigns raw monitoring incidents"
            ),
        }
    # Atomic reserve via #613 substrate.
    ttl = lease_ttl_seconds if lease_ttl_seconds is not None else None
    try:
        kwargs: dict[str, Any] = {
            "session_id": session_id,
            "role": role_norm,
            "remote": remote,
            "org": org,
            "repo": repo,
            "kind": selected.kind,
            "number": selected.number,
            "expected_head_sha": selected.head_sha,
            "allowed_actions": allowed,
            "forbidden_actions": forbidden,
            "phase": "allocated",
        }
        if ttl is not None:
            kwargs["lease_ttl_seconds"] = int(ttl)
        result = db.assign_and_lease(**kwargs)
    except (InvalidWorkKindError, LeaseRequiredError, ControlPlaneError) as exc:
        return {
            "success": False,
            "outcome": OUTCOME_NO_SAFE,
            "apply": True,
            "role": role_norm,
            "session_id": session_id,
            "remote": remote,
            "org": org,
            "repo": repo,
            "selected": selection,
            "expected_role_next": expected_role,
            "reasons": [f"atomic assign+lease failed: {exc} (fail closed, #613)"],
            "skipped": [s.as_dict() for s in skipped],
            "terminal_pr": terminal_pr,
            "assignment": None,
            "substrate": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
        }
    if result.outcome == "wait":
        return {
            "success": True,
            "outcome": OUTCOME_WAIT,
            "apply": True,
            "role": role_norm,
            "session_id": session_id,
            "remote": remote,
            "org": org,
            "repo": repo,
            "selected": selection,
            "expected_role_next": expected_role,
            "reasons": [result.reason or "foreign active lease"],
            "skipped": [s.as_dict() for s in skipped],
            "terminal_pr": terminal_pr,
            "assignment": result.as_dict(),
            "owner_session_id": result.owner_session_id,
            "substrate": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
        }
    if result.outcome == "no_safe_work":
        return {
            "success": True,
            "outcome": OUTCOME_NO_SAFE,
            "apply": True,
            "role": role_norm,
            "session_id": session_id,
            "remote": remote,
            "org": org,
            "repo": repo,
            "selected": selection,
            "expected_role_next": expected_role,
            "reasons": [result.reason or "no_safe_work"],
            "skipped": [s.as_dict() for s in skipped],
            "terminal_pr": terminal_pr,
            "assignment": result.as_dict(),
            "substrate": "control_plane_db",
            "file_lock_only": False,
            "comment_lease_only": False,
        }
    # assigned
    return {
        "success": True,
        "outcome": OUTCOME_ASSIGNED,
        "apply": True,
        "role": role_norm,
        "profile_name": profile_name,
        "username": username,
        "session_id": session_id,
        "remote": remote,
        "org": org,
        "repo": repo,
        "selected": selection,
        "expected_role_next": expected_role,
        "reasons": [
            selection["reason_selected"],
            result.reason or "atomic assign+lease created",
        ],
        "skipped": [s.as_dict() for s in skipped],
        "terminal_pr": terminal_pr,
        "assignment": result.as_dict(),
        "lease_proof": {
            "assignment_id": result.assignment_id,
            "lease_id": result.lease_id,
            "expires_at": result.expires_at,
            "expected_head_sha": result.expected_head_sha,
            "allowed_actions": list(result.allowed_actions),
            "forbidden_actions": list(result.forbidden_actions),
            "source": "control_plane_db.assign_and_lease",
        },
        "next_valid_command": _next_command(role_norm, selected),
        "substrate": "control_plane_db",
        "file_lock_only": False,
        "comment_lease_only": False,
        "downstream_note": (
            "#612 incident bridge remains downstream of #600; "
            "allocator never assigns raw monitoring incidents"
        ),
    }
 def _next_command(role: str, c: WorkCandidate) -> str:
    if role == ROLE_AUTHOR and c.kind == "issue":
        return f"implement issue #{c.number} under a branches/ worktree; open PR when ready"
    if role == ROLE_AUTHOR and c.kind == "pr":
        return (
            f"address REQUEST_CHANGES on PR #{c.number} at head "
            f"{(c.head_sha or '')[:12]} and push fixes"
        )
    if role == ROLE_REVIEWER:
        return (
            f"review PR #{c.number} pinned at head {(c.head_sha or '')[:12]} "
            "via full reviewer workflow"
        )
    if role == ROLE_MERGER:
        return (
            f"merge PR #{c.number} only with explicit operator MERGE "
            f"authorization at head {(c.head_sha or '')[:12]}"
        )
    if role == ROLE_RECONCILER:
        return f"diagnose contested state for {c.kind}#{c.number}"
    return f"proceed on {c.kind}#{c.number} under role {role}"
 def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
    """Build a WorkCandidate from a plain dict (tests / MCP inventory)."""
    return WorkCandidate(
        kind=str(data.get("kind") or "issue"),
        number=int(data["number"]),
        state=str(data.get("state") or "open"),
        labels=tuple(data.get("labels") or ()),
        title=str(data.get("title") or ""),
        priority=int(data.get("priority") or 0),
        head_sha=data.get("head_sha"),
        request_changes_current_head=bool(data.get("request_changes_current_head")),
        approval_on_current_head=bool(data.get("approval_on_current_head")),
        approval_stale=bool(data.get("approval_stale")),
        approval_contaminated=bool(data.get("approval_contaminated")),
        mergeable=bool(data.get("mergeable")),
        blocked=bool(data.get("blocked")),
        dependency_unmet=bool(data.get("dependency_unmet")),
        dependency_reason=data.get("dependency_reason"),
        already_claimed_elsewhere=bool(data.get("already_claimed_elsewhere")),
    )
@@ -1,63 +0,0 @@
 # Control-plane DB substrate (#613)
 **Status:** Implemented (SQLite single-writer MVP)
 **ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md)
 **Module:** `control_plane_db.py`
 ## Architecture statement
 > **DB coordinates, Gitea records, Sentry/GlitchTip observe; the bridge is the only path that turns observations into Gitea work.**
 ## What this ships
 | Capability | Notes |
 |------------|--------|
 | Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` |
 | Atomic assign+lease | `ControlPlaneDB.assign_and_lease` — one `BEGIN IMMEDIATE` transaction |
 | Mutation gate | `require_valid_assignment` — live lease + allowed action + non-terminal work + non-stale head |
 | Heartbeat / release / expire | Lease lifecycle helpers |
 | Terminal-lock index | Routing signal for #600 (terminal path first) |
 | `incident_links` | Provider-neutral link model for #612 — **not** assignable work; scope keys NULL-safe |
 ## Hard rules (enforced in code)
 1. Assignable `work_items.kind` ∈ {`issue`, `pr`} only — **never** raw Sentry/GlitchTip incidents.
 2. Two concurrent sessions cannot both receive an active assignment on the same open work item (second gets `wait`).
 3. Merged/closed work items return `no_safe_work` at assign time, and `require_valid_assignment` fails closed if the work item becomes terminal later.
 4. Assignments pin `expected_head_sha`; mutations fail closed if the work item head drifts.
 5. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6).
 6. `incident_links` optional scope fields are stored as empty strings (never NULL) so UNIQUE is canonical across minimal upserts.
 7. Legacy `incident_links` migration collapses NULL-scope duplicates **only** when Gitea targets **and** all meaningful observation metadata agree (fingerprint, status, event_count, permalink, timestamps, linked PRs, etc.). Conflicting metadata fails closed — no silent discard.
 ## Dependency chain
 ```text
 #613 control-plane DB (this)  →  #600 allocator API  →  #612 incident bridge
 ```
 - **#600** must call this substrate (not file locks / comment-only leases alone) for completion.
 - **#612** must write `incident_links` here and create **Gitea issues**; the allocator assigns those issues, not raw incidents.
 ## Allocator API (#600)
 Module: `allocator_service.py` · MCP tool: `gitea_allocate_next_work`
 - Workers call `gitea_allocate_next_work(apply=false|true)` instead of self-selecting work.
 - `apply=false` returns a dry-run selection (`outcome=preview`) with skip reasons.
 - `apply=true` reserves via `ControlPlaneDB.assign_and_lease` (atomic assignment+lease).
 - Coordination source is **always** the control-plane DB — not file locks or comment-only leases.
 - Routing follows ADR §5.3 (REQUEST_CHANGES → author, terminal path first, foreign lease → wait).
 - Raw monitoring incidents are never candidates (#612 remains downstream).
 ## Non-goals (intentionally deferred)
 - Sentry/GlitchTip provider adapters and auto-watchdog — **#612**
 - Gitea comment/label mirror writers (may be added by allocator tooling later)
 ## Tests
 ```bash
 python3 -m pytest tests/test_control_plane_db.py -q
 ```
@@ -1,301 +0,0 @@
 # ADR: MCP allocator, control-plane DB, and Sentry/GlitchTip incident bridge architecture
 - **Status:** Accepted (implementation pending; blocks code for #600 / #612 / #613)
 - **Date:** 2026-07-09
 - **Tracking issues:**
  - [#613](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/613) — control-plane DB (first)
  - [#600](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/600) — allocator API (second)
  - [#612](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/612) — Sentry/GlitchTip incident bridge (third)
 - **Related:** existing GlitchTip contracts (`glitchtip-to-gitea-workflow-design.md`, `glitchtip-gitea-deduplication-linking-design.md`), trust boundaries (`tool-boundaries.md`, `safety-model.md`), current leases (`pr_work_lease.py`, `issue_lock_store.py`)
 ## 1. Context
 Multiple LLM sessions can independently inspect the Gitea queue and start the same issue or PR. Existing coordination (Gitea comment leases, local issue-lock files, labels) often detects collisions **after** work has started. Parallel issues #600, #612, and #613 each describe part of a fix; without one ADR they risk overlapping stores, wrong dependency order, and trust-boundary violations.
 This ADR is the canonical architecture decision **before** implementing those issues.
 ## 2. Decision summary (core)
 | Layer | Owns | Must not |
 |-------|------|----------|
 | **Control-plane DB** | Sessions, atomic assignment, leases, heartbeats, terminal-lock index, events, incident links | Bypass Gitea workflow gates or replace issue/PR history |
 | **Gitea** | Durable work record: issues, PRs, comments, labels, reviews, merges | Be the only concurrency lock under multi-session load |
 | **Sentry / GlitchTip** | Incidents, events, provider UI | Assign work, approve/merge/close, or mutate Gitea outside the bridge |
 | **Incident bridge** | Provider adapters, reconcile/create Gitea issues, link storage upsert, optional provider writeback | Hand raw provider incidents to the allocator as work items |
 **One-liner:** **DB coordinates. Gitea records. Sentry/GlitchTip observe. The bridge is the only path that turns observations into Gitea work. The allocator assigns only Gitea issues/PRs, never raw monitoring incidents.**
 ## 3. Dependency order
 Implementation **must** follow:
 ```text
 #613 control-plane DB  →  #600 allocator API  →  #612 incident bridge
   (atomic substrate)      (routing policy)       (feeds Gitea work)
 ```
 | Issue | Role | Depends on |
 |-------|------|------------|
 | **#613** | Durable coordination DB + lease/assignment transactions | — |
 | **#600** | `gitea_allocate_next_work` policy and tool surface | **#613** (hard) |
 | **#612** | Multi-project Sentry/GlitchTip → Gitea bridge | **#613** for `incident_links` index; **#600** before treating bridge-created issues as allocator feed in multi-worker prod |
 **Hard rules:**
 1. Do **not** implement #600 on file locks / comment-only leases and call it done.
 2. Do **not** ship #612 as a second assignment system for raw incidents.
 3. Partial previews (read-only list tools, schema stubs) may land earlier if they do not claim “allocator complete” or “bridge complete.”
 ## 4. Authority boundaries
 ### 4.1 Gitea (durable source of truth for work)
 Gitea remains authoritative for:
 - Issue and PR identity, titles, bodies, state (open/closed/merged)
 - Comments, reviews, approvals, REQUEST_CHANGES
 - Labels and workflow status labels (`status:ready`, `status:in-progress`, …)
 - Merges, closes, branch refs as recorded by Gitea/git hosting
 Operators and auditors read **history** from Gitea.
 ### 4.2 Control-plane DB (coordination / index)
 The control-plane DB is authoritative for **live multi-session coordination**:
 - Which session holds which assignment/lease
 - Heartbeat freshness and expiry
 - Terminal-lock index for routing
 - Event log of allocation/lease transitions
 - `incident_links` index (see §8)
 It is **not** a substitute for Gitea history. When DB and Gitea disagree on durable work state (e.g. PR already merged), **Gitea wins**; the DB is reconciled.
 ### 4.3 Sentry / GlitchTip (observe)
 Providers own incident lifecycle and raw event data. They:
 - Must **not** bypass Gitea gates
 - Must **not** assign LLM work
 - May receive **writeback** of resolution status only through the bridge, when configured and supported
 ### 4.4 Trust boundaries for credentials
 Aligned with `docs/tool-boundaries.md` and `docs/safety-model.md`:
 - **One MCP server process per trust boundary** remains the default.
 - Monitor API tokens (Sentry/GlitchTip) live in the **bridge/orchestrator boundary** (or a dedicated observability write/read profile), **not** blindly inside every Gitea MCP author/reviewer/merger process.
 - Gitea tokens stay on Gitea MCP profiles only.
 - Orchestrators compose services; they must not become a single credential pool that silently mixes Gitea write + monitor tokens into every worker.
 Tool names such as `gitea_observability_*` may exist as a **namespace façade** only if the runtime still enforces separate credential scopes (e.g. bridge process vs pure Gitea mutation process).
 ## 5. Allocator rules (#600 on top of #613)
 ### 5.1 Worker contract
 - Workers **do not self-select** work under the standard multi-LLM workflow.
 - Workers call **`gitea_allocate_next_work`** (with `apply=true` only when taking work).
 - Mutations that claim exclusive work require a **valid assignment and lease** from the control-plane DB.
 - Return values include at least: role, issue/PR number, expected head SHA (when applicable), lease ID, expiry, allowed actions, forbidden actions — or a non-assignment outcome (`WAIT`, terminal-path block, no safe work, needs controller, etc.).
 ### 5.2 Atomic reserve
 - **Assignment creation and lease creation happen in one DB transaction.**
 - Two concurrent sessions **must not** receive the same issue/PR as assigned work.
 - On contention: second session gets **WAIT** or **owner-resume**, never a duplicate lease.
 ### 5.3 Routing policy
 | Condition | Route |
 |-----------|--------|
 | Current-head **REQUEST_CHANGES** | **Author** (not reviewer/merger) |
 | **Stale** approval (approval not on current head) | **Reviewer** |
 | **Clean** approval on current head, mergeable | **Merger** |
 | Contested / contaminated approval | **Diagnosis / reconciler** (controller path) |
 | Active **foreign** lease | **WAIT** or **owner-resume** |
 | **Terminal-review** lock present | **Terminal-path resolution first** before downstream review work |
 | Merged / closed PR | **Never assign** |
 | Issue locked by sanctioned lease | **Never assign** to another worker unless lease cleanup/adoption is sanctioned |
 ### 5.4 Relationship to Gitea mirrors
 After a successful assignment, the allocator (or sanctioned tooling) may update Gitea comments/labels so humans and audits see state. Those mirrors **are not** the sole lock source.
 ## 6. Control-plane DB topology (#613)
 ### 6.1 Preferred architecture (multi-daemon / Proxmox)
 For true multi-session concurrency (multiple MCP daemons, multiple hosts, or Proxmox VMs):
 **Prefer either:**
 1. **Shared Postgres** used by all Gitea MCP profiles / allocator callers, **or**
 2. A **single allocator daemon** (sole writer to the coordination store) that all workers call over MCP/RPC.
 Both satisfy: one transactional authority for “who has this work item.”
 ### 6.2 SQLite MVP
 SQLite is acceptable **only** for a **single-writer MVP**:
 - One process owns writes (allocator daemon or single co-located MCP server), **or**
 - Documented single-host single-daemon experiments with file locking and no multi-host claims.
 SQLite is **not** sufficient to declare multi-session production readiness when four independent LLM sessions talk to four MCP processes without a shared writer.
 ### 6.3 Suggested entities (normative shape)
 Minimum logical entities (names may vary; semantics must not):
 1. **sessions** — session_id, role/profile, namespace, pid, started_at, last_heartbeat_at, status  
 2. **work_items** — remote/org/repo, kind ∈ {`issue`, `pr`}, number, state, priority, current_head_sha, updated_at  
 3. **leases** — lease_id, work_item_id, session_id, role, phase, expires_at, heartbeat_at, status  
 4. **assignments** — assignment_id, work_item_id, session_id, allowed_action(s), expected_head_sha, status  
 5. **terminal_locks** — repo/org, terminal_pr, review_id, decision, status, cleanup_state  
 6. **events** — event_id, work_item_id, type, message, created_at  
 7. **incident_links** — provider-neutral link model (see §8)
 **Explicit non-kind:** do **not** use `work_items.kind = sentry_incident` (or glitchtip_incident) as an assignable work unit. Incidents become work only after the bridge creates/links a **Gitea issue**.
 ## 7. Lease migration (avoid split-brain)
 ### 7.1 Target state
 - **Primary** for live coordination: **control-plane DB** leases and assignments.
 - **Gitea comment leases** (e.g. `<!-- mcp-review-lease:v1 -->`) and **local issue-lock files** become:
  - **mirrors** of DB state for human visibility / recovery, and/or
  - written **only** through the allocator (or sanctioned lease tools that update DB + mirror in one workflow).
 ### 7.2 Migration rules
 1. New exclusive claims go through DB assignment+lease first.
 2. Comment lease / file lock writers that skip the DB are **deprecated** once #613+#600 land.
 3. Reconciler tooling heals: expired DB lease, orphan Gitea comment, orphan local file — fail closed when ambiguous.
 4. **Split-brain is a defect:** “DB free + Gitea comment leased” or “DB leased + Gitea free” must be detectable and reconcilable; production paths must not rely on two independent writers.
 ### 7.3 Heartbeats
 - Heartbeats update the **DB**.
 - Optional Gitea summaries must avoid comment spam (periodic summary or edit-in-place policy).
 ## 8. Observability bridge (#612)
 ### 8.1 Role
 The bridge:
 1. Scans configured Sentry and/or GlitchTip projects (multi-project mappings).
 2. Deduplicates against existing links / Gitea issues.
 3. Creates or updates **normal Gitea issues** (labels, sanitized body, provider URL).
 4. Upserts **one** canonical `incident_links` row.
 5. Optionally writes resolution status back to the provider when supported.
 6. Never approves, merges, closes, or otherwise bypasses Gitea workflow gates.
 ### 8.2 Allocator interaction
 - The allocator sees observability work **only after** a Gitea issue exists (typically `status:ready` + observability labels).
 - **Do not assign raw Sentry/GlitchTip incidents** as work items.
 - Bridge AC “allocator can see linked issues” means: **as ordinary Gitea issues**, not a special incident queue.
 ### 8.3 Provider adapters and trust
 - Separate **Sentry** and **GlitchTip** adapters; document API differences; do not assume writeback parity.
 - Self-hosted base URLs supported (e.g. `https://sentry.prgs.cc`).
 - Tokens from environment / secret store only.
 - Prefer phase-1 **explicit reconcile / dry-run** before unsupervised watchdog auto-filing, consistent with existing GlitchTip filing safety contracts.
 ### 8.4 Home of implementation
 Filing/orchestration may live in:
 - a control-plane / bridge package or profile, and/or  
 - Gitea-Tools as operator-facing tools that **delegate** to that boundary,
 but must not violate §4.4 credential isolation.
 ## 9. Incident link storage (single source of truth)
 ### 9.1 Canonical model: `incident_links` (provider-neutral)
 Prefer a **provider-neutral** model over Sentry-only `sentry_links`:
 | Field (logical) | Purpose |
 |-----------------|---------|
 | provider | `sentry` \| `glitchtip` \| … |
 | provider_base_url | Self-hosted base |
 | provider_org / provider_project | Mapping key |
 | provider_issue_id / provider_short_id | Provider identity |
 | provider_permalink | Human URL |
 | fingerprint | Stable dedupe key when available |
 | gitea_org / gitea_repo / gitea_issue_number | Linked work |
 | linked_pr_numbers | Optional PR association |
 | first_seen / last_seen / event_count | Summary metrics (safe) |
 | status | link lifecycle |
 | release_resolved_at / last_sync_at | Sync bookkeeping |
 ### 9.2 Gitea as human mirror
 - Issue body markers / structured comments (e.g. HTML comment metadata) remain the **human- and audit-visible mirror**.
 - They must stay consistent with `incident_links` but are **not** a second independent write path for inventing links without the bridge.
 ### 9.3 One truth
 - **Do not** maintain two independent systems of record (e.g. full mapping only in #612 storage **and** a separate #613 `sentry_links` with different semantics).
 - #612 implementation **must** use the same `incident_links` model introduced under #613 (or a single agreed store both reference).
 ## 10. Security and redaction
 Mandatory for bridge payloads, Gitea issue bodies/comments, DB-stored event summaries, and logs:
 - No API tokens, DSNs, passwords, cookies, auth headers
 - No keychain IDs, private config contents, raw session-state files, full prompt bodies
 - Sanitize stack locals and request data; store only safe tags/context
 - Logs must never print provider tokens or DSNs
 - Redaction tests are required for #612
 ## 11. Non-goals
 - Replace Gitea as the durable workflow record  
 - Let Sentry/GlitchTip assign work or mutate Gitea workflow state outside sanctioned tools  
 - Let the bridge approve, merge, close, release, or bypass Gitea gates  
 - Implement #600 on file locks / comments alone and call allocator complete  
 - Treat raw monitoring incidents as `work_items` for the allocator  
 - Put monitor tokens into every Gitea MCP worker process by default  
 ## 12. Consequences
 ### Positive
 - Clear implementation order and ownership  
 - Multi-session safety becomes testable at the DB layer  
 - Observability becomes assignable work without special-casing the allocator  
 - Trust boundaries stay compatible with existing control-plane docs  
 ### Costs / risks
 - Migration from comment/file leases requires reconciler work  
 - Shared Postgres or single allocator daemon is operational cost  
 - Bridge must be careful not to spam Gitea issues (dedupe, caps, policy modes)  
 ### Follow-ups (documentation only until implemented)
 1. Update #600, #612, and #613 bodies to **link this ADR** and restate hard dependencies.  
 2. Implement #613 schema + atomic assign/lease.  
 3. Implement #600 against the DB.  
 4. Implement #612 against `incident_links` + Gitea create/update.  
 5. Deprecate dual writers for leases.
 ## 13. Acceptance criteria for *this* ADR
 This document is accepted when:
 1. It is merged into `docs/architecture/` on the default branch.  
 2. #600 / #612 / #613 (or their PRs) reference this path as the architecture source of truth.  
 3. No implementation PR for those issues claims completion without conforming to §§2–11.
 ## 14. Document history
 | Date | Change |
 |------|--------|
 | 2026-07-09 | Initial ADR: dependency order, authority model, topology, lease migration, bridge, `incident_links`, security, non-goals |
@@ -1,118 +0,0 @@
 # Recovering from `client is closing: EOF` on a Gitea MCP namespace (#543)
 ## Symptom
 A tool call through a Gitea MCP namespace — `gitea-author`, `gitea-reviewer`,
 `gitea-merger`, or the shared `gitea-tools` namespace — fails immediately with:
 ```
 client is closing: EOF
 ```
 Every subsequent call to that same namespace returns the same error, including
 cheap read tools such as `gitea_whoami` and `gitea_list_profiles`. Other MCP
 servers registered with the same client (for example `context7`) keep working,
 so this is **not** a global MCP-client outage.
 ## Why this is not a code defect
 This failure is a **transport-level** condition in the IDE / MCP client manager,
 not a missing or broken tool:
 - The tool can be present and registered in the Python `FastMCP` tool manager.
 - Direct Python inspection of the server confirms the tool exists.
 - Running the server manually and sending JSON-RPC over stdio works fine
  (offline spawn) — that path does **not** prove the IDE namespace is healthy.
 The client manager entered a closed state after the backing subprocess for that
 namespace terminated (or was killed) behind its back. Once closed, the client
 does **not** re-spawn the child on the next tool call — it just replays
 `client is closing: EOF`. The OS process may even still be alive if a parent
 language-server process is holding the stdio pipes open.
 This is the canonical "registered in FastMCP ≠ callable through the namespace"
 false-ready state. It is distinct from the **stale-runtime** family in #531 /
 #544, where the process is reachable but running behind `master`; that case is
 detected by the `ps`-based `_check_mcp_runtimes_diagnostics` in
 `gitea_mcp_server.py`. The EOF case is a dead/closed transport, not a stale one,
 so the `ps` check alone will not surface it.
 ## Recovery path (canonical — client reconnect only)
 Do the steps in order. Stop as soon as a live **client-namespace** call succeeds.
 1. **Confirm the blast radius.** Call a cheap read tool on the failing namespace
   (`gitea_whoami` or `gitea_list_profiles`). Then call the same tool on a
   different MCP server (e.g. `context7`).
   - Only the Gitea namespace fails → single-namespace transport close. Continue.
   - Every server fails → restart the whole MCP client, not just one namespace.
 2. **Reconnect the namespace through the client, not the shell.** Use the IDE /
   client MCP-reconnect action for that server entry (in Claude Code:
   `/mcp` → reconnect the affected `gitea-*` server). Reconnecting forces the
   client to spawn a fresh subprocess and re-open the pipe. This clears the
   closed-client state that a bare `kill`/respawn from a terminal does **not**.
 3. **Do not "fix" it by importing the server or poking the process.** Reaching
   for `python -c 'import gitea_mcp_server ...'`, raw JSON-RPC from a shell,
   killing PIDs to force a respawn, or touching MCP config mtimes does **not**
   restore the *client's* view of the namespace and violates the daemon-import
   guard (#558, `docs/mcp-daemon-import-guard.md`). The only sanctioned repair
   is a **client reconnect / relaunch**.
 4. **Verify through the same path the workflow will use.** After reconnect, call
   the specific tool the blocked workflow needs — not just any tool — through
   the target namespace. For a merge that means calling the merger-authorized
   adoption/merge tool through `gitea-merger`. A green `gitea_whoami` on one
   namespace does **not** prove another namespace or another tool is callable.
   Record success with:
   ```text
   gitea_assess_mcp_namespace_health(..., probe_source="client_namespace")
   ```
 5. **If reconnect does not clear it,** relaunch the client entirely, then repeat
   step 4. If EOF persists after a full relaunch, the backing subprocess is
   failing to start — inspect its stderr / launch config (command path, venv,
   `*_MCP_CONFIG`, `*_MCP_PROFILE` env) rather than retrying the call. Still
   do not use PID kill or config-touch as the primary recovery.
 ## Diagnostics to capture when reporting EOF
 Include all of these so the failure is actionable and reproducible:
 - **Namespace name** that returned EOF (`gitea-author` / `gitea-reviewer` /
  `gitea-merger` / `gitea-tools`).
 - **Tool** that was called and the **exact** error string.
 - **PID** of the backing process (if any) and whether it was still alive
  (informational only — not a recovery action).
 - **Profile / env** for that namespace (execution profile, `*_MCP_PROFILE`,
  worktree binding such as `GITEA_AUTHOR_WORKTREE`).
 - **Config path** the client launched the server from.
 - Result of the **cross-server control** call (did `context7` succeed?).
 ## Offline spawn probe (non-authoritative)
 `test_mcp_conn.py` performs a full JSON-RPC handshake against a **fresh
 subprocess** (`initialize` → `initialized` → `tools/list` → `tools/call`) and
 classifies with `probe_source=offline_spawn`. That is useful for offline
 launch/registration debugging. It is **not** proof the IDE-managed namespace is
 healthy. See `docs/mcp-namespace-health.md`.
 ## Do-not list during EOF recovery
 - Do **not** retry a blocked merge/adoption until the required tool is confirmed
  callable through the merger-authorized **client** namespace (see #543).
 - Do **not** clean, reset, or rebind a **foreign** worktree to work around the
  error.
 - Do **not** bypass the namespace with direct imports, raw API/curl, or
  in-memory state restoration.
 - Do **not** kill MCP PIDs or touch config mtimes as a substitute for client
  reconnect.
 ## Related
 - #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
 - #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
 - `docs/mcp-client-registration.md` — per-server registration contract.
 - `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
@@ -1,88 +0,0 @@
 # MCP namespace health diagnostics (#543)
 Gitea MCP tools can be registered in the Python FastMCP server while the IDE's
 live MCP namespace is still unusable. The failure usually appears as
 `client is closing: EOF`, `transport closed`, or an empty response when calling
 a tool such as `gitea_whoami`.
 Do not treat static tool registration as proof that review or merge workflows
 can proceed. A reviewer or merger flow must have **client-namespace** evidence
 that the required tool is callable through the configured IDE MCP namespace.
 ## Probe sources (do not confuse them)
 | Source | How obtained | Proves IDE namespace? |
 | --- | --- | --- |
 | `client_namespace` | Tool call through the IDE-managed MCP client | **Yes** |
 | `offline_spawn` | `test_mcp_conn.py` subprocess JSON-RPC handshake | **No** (offline only) |
 `gitea_assess_mcp_namespace_health` accepts `probe_source` and only sets
 `ide_namespace_proven=true` for `client_namespace` success. Offline spawn
 success never clears review/merge mutation gates.
 ## Client-namespace health check (canonical)
 1. Through the IDE client, call a cheap tool on the target namespace
   (`gitea_whoami` or `gitea_list_profiles`).
 2. Feed the live result into:
 ```text
 gitea_assess_mcp_namespace_health(
    namespace="gitea-merger",          # or reviewer / author / tools
    registered_tools=[...],            # optional static list
    probe_result={"success": true, "result": {...}},
    probe_source="client_namespace",
 )
 ```
 3. A healthy client-namespace assessment is recorded in the MCP session and
   consulted by **live** `gitea_submit_pr_review` / `gitea_merge_pr` gates.
 4. If the probe fails with EOF, recover via **client reconnect only** — see
   `docs/mcp-namespace-eof-recovery.md`. Do **not** kill PIDs or touch MCP
   config mtimes as a recovery procedure.
 ## Offline spawn probe (non-authoritative)
 ```bash
 python3 test_mcp_conn.py --config ~/.gemini/config/mcp_config.json
 ```
 This script spawns a **separate** server process from config, performs
 JSON-RPC `initialize` → `tools/list` → `tools/call`, and classifies the
 result with `probe_source=offline_spawn`. Use it for offline debugging of
 launch command / registration. It does **not** prove the IDE-managed
 namespace is healthy.
 By default the script checks:
 | Namespace | Required tool |
 | --- | --- |
 | `gitea-author` | `gitea_whoami` |
 | `gitea-reviewer` | `gitea_whoami` |
 | `gitea-merger` | `gitea_whoami` |
 | `gitea-tools` | `gitea_list_profiles` |
 ## Recovery (canonical)
 When a namespace returns EOF, follow
 `docs/mcp-namespace-eof-recovery.md` in order:
 1. Confirm blast radius (Gitea namespace vs all MCP servers).
 2. **Reconnect the namespace through the client** (IDE reconnect / relaunch).
 3. Do not repair via shell imports, raw JSON-RPC, PID kills, or config mtime
   touches — those do not restore the client's closed transport.
 4. Re-verify the **specific** required tool through the target namespace.
 5. Resume review/merge only after a successful `client_namespace` assessment.
 ## Enforcement
 1. **State machine (read-only):** feed `blocks_merge_workflow` from a
   `client_namespace` assessment into
   `gitea_assess_review_merge_state_machine(live_namespace_broken=...)`.
 2. **Live mutations:** `gitea_submit_pr_review` and `gitea_merge_pr` call
   `_live_namespace_health_gate` and fail closed when the session has a
   recorded unhealthy or non-client probe for the required namespace
   (`gitea-reviewer` for review, `gitea-merger` for merge).
 When blocked, repair the IDE namespace and re-record a healthy
 `client_namespace` assessment before retrying the mutation.
@@ -24,12 +24,26 @@ from __future__ import annotations
 import os
 import subprocess
 import time
 # Live-remote head cache: the parity gate runs on every mutation and every
 # runtime-context read, so the ``git ls-remote`` result is cached briefly to
 # avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
 _REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
 _REMOTE_HEAD_TTL = 60.0
 def _clear_remote_head_cache() -> None:
    """Reset the live-remote head cache (test isolation / forced refresh)."""
    _REMOTE_HEAD_CACHE.clear()
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
 # GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
 ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
 def read_git_head(root: str) -> str | None:
@@ -58,6 +72,57 @@ def read_git_head(root: str) -> str | None:
    return (res.stdout or "").strip() or None
 def read_remote_master_head(
    root: str,
    remote: str = "origin",
    branch: str = "master",
    ttl: float = _REMOTE_HEAD_TTL,
 ) -> str | None:
    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
    a daemon that is behind the live remote master apart from one whose local
    checkout simply hasn't been pulled. ``None`` means the live head could not
    be resolved (offline, no such remote, git unavailable, error) -- callers
    must treat unknown live state as *not mutation-safe* while never blocking
    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
    precedence so the wiring can be exercised deterministically and offline.
    The result is cached for *ttl* seconds per (root, remote, branch) so the
    gate does not run a network probe on every mutation/read (``ttl=0`` forces
    a live probe). Both hits and ``None`` misses are cached to bound offline
    latency; the env override bypasses the cache and the subprocess entirely.
    """
    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
    if forced is not None:
        return forced.strip() or None
    if not root:
        return None
    key = (root, remote, branch)
    now = time.monotonic()
    if ttl > 0:
        cached = _REMOTE_HEAD_CACHE.get(key)
        if cached is not None and (now - cached[0]) < ttl:
            return cached[1]
    sha: str | None = None
    try:
        res = subprocess.run(
            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5,
        )
        if res.returncode == 0:
            lines = (res.stdout or "").strip().splitlines()
            if lines:
                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
    except Exception:
        sha = None
    _REMOTE_HEAD_CACHE[key] = (now, sha)
    return sha
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
    """Capture the process source-tree baseline once at server startup.
@@ -72,18 +137,38 @@ def _short(sha: str | None) -> str:
    return sha[:12] if sha else "unknown"
-def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
+def assess_master_parity(
    startup: dict | None,
    current_head: str | None,
    live_remote_head: str | None = None,
 ) -> dict:
    """Compare the startup baseline against the current on-disk ``HEAD``.
-    Pure: both HEADs are supplied by the caller. Returns a structured result:
+    Pure: all HEADs are supplied by the caller. Returns a structured result:
    - ``in_parity``    -- server code matches the on-disk master (or parity
      could not be determined, which is not treated as stale).
    - ``stale``        -- the on-disk master has definitively advanced past the
      running process.
-    - ``restart_required`` -- alias of ``stale``; the recovery action.
+    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
-    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``determinable`` -- whether both local HEADs were known well enough to
      compare.
    - ``startup_head`` / ``current_head`` / ``reasons``.
    #610 adds live-remote awareness so a daemon that is stale relative to the
    *live* remote master cannot report a mutation-safe result even when the
    local checkout HEAD still matches the daemon's startup commit:
    - ``daemon_start_head`` -- the commit the running process started at
      (alias of ``startup_head``, named for clarity in reports).
    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
      could not be fetched.
    - ``live_known`` -- whether the live remote target was resolved.
    - ``live_stale`` -- the live remote master has advanced past the running
      process (daemon is behind live master) even if local parity is green.
    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
      target all agree; the only state in which a mutation may rely on parity.
    """
    startup_head = (startup or {}).get("startup_head")
    reasons: list[str] = []
@@ -91,32 +176,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
    if startup_head is None:
        reasons.append(
            "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
                       live_remote_head, False, reasons)
    if current_head is None:
        reasons.append(
            "current workspace HEAD could not be read; code parity cannot be "
            "enforced")
-        return _result(True, False, False, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
                       live_remote_head, False, reasons)
-    if startup_head == current_head:
+    local_in_parity = startup_head == current_head
-        return _result(True, False, True, startup_head, current_head, reasons)
+    local_stale = not local_in_parity
    if local_stale:
        reasons.append(
            f"MCP server started at commit {_short(startup_head)} but the "
            f"workspace master is now {_short(current_head)}; restart the "
            f"server to load the current capability gates")
-    reasons.append(
+    live_known = live_remote_head is not None
-        f"MCP server started at commit {_short(startup_head)} but the workspace "
+    live_stale = live_known and live_remote_head != startup_head
-        f"master is now {_short(current_head)}; restart the server to load the "
+    if live_stale:
-        f"current capability gates")
+        reasons.append(
-    return _result(False, True, True, startup_head, current_head, reasons)
+            f"live remote master is {_short(live_remote_head)} but the MCP "
            f"server started at {_short(startup_head)}; the daemon is stale "
            f"relative to live master -- restart/reconnect before mutating")
    return _result(
        local_in_parity, local_stale, True, startup_head, current_head,
        live_remote_head, live_stale, reasons)
-def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
+def _result(in_parity, stale, determinable, startup_head, current_head,
            live_remote_head, live_stale, reasons):
    live_known = live_remote_head is not None
    mutation_safe = (
        determinable and in_parity and live_known and not live_stale)
    return {
        "in_parity": in_parity,
        "stale": stale,
-        "restart_required": stale,
+        "restart_required": stale or live_stale,
        "determinable": determinable,
        "startup_head": startup_head,
        "current_head": current_head,
        # #610 distinguished signals:
        "daemon_start_head": startup_head,
        "local_head": current_head,
        "live_remote_head": live_remote_head,
        "live_known": live_known,
        "live_stale": live_stale,
        "mutation_safe": mutation_safe,
        "reasons": list(reasons),
    }
@@ -130,11 +239,13 @@ def parity_block_reasons(assessment: dict) -> list[str]:
    """Block reasons for a mutation gate (empty when the mutation may proceed).
    A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons; only a definitively stale server blocks.
+    reasons. A definitively stale server blocks, and (#610) a daemon that is
    stale relative to the *live* remote master blocks even when the local
    checkout HEAD still matches the daemon's startup commit.
    """
    if gate_disabled():
        return []
-    if assessment.get("stale"):
+    if assessment.get("stale") or assessment.get("live_stale"):
        return list(assessment.get("reasons") or
                    ["server code is stale relative to master (fail closed)"])
    return []
@@ -147,6 +258,10 @@ def parity_report(assessment: dict) -> dict:
        "restart_required": True,
        "startup_head": assessment.get("startup_head"),
        "current_head": assessment.get("current_head"),
        # #610: name the live remote target so the report distinguishes a
        # local-code stale from a daemon-behind-live-master stale.
        "live_remote_head": assessment.get("live_remote_head"),
        "live_stale": bool(assessment.get("live_stale")),
        "reasons": list(assessment.get("reasons") or []),
        "recovery": [
            "The running MCP server is executing code older than the current "
@@ -157,6 +272,45 @@ def parity_report(assessment: dict) -> dict:
    }
 def parity_resolver_disagreement(
    assessment: dict,
    resolver_restart_required: bool,
 ) -> dict | None:
    """Typed blocker when the resolver requires restart but parity looks green.
    The capability resolver (``gitea_resolve_task_capability``) detects stale
    runtime authoritatively for mutation safety (#610). When it requires a
    restart, local-only parity must never override it: this returns a typed,
    fail-closed blocker that names the resolver as authoritative. Returns
    ``None`` when the resolver does not require a restart.
    """
    if not resolver_restart_required:
        return None
    parity_optimistic = bool(assessment.get("in_parity")) and not (
        assessment.get("stale") or assessment.get("live_stale"))
    return {
        "kind": "parity_resolver_disagreement",
        "restart_required": True,
        "resolver_authoritative": True,
        "parity_optimistic": parity_optimistic,
        "daemon_start_head": assessment.get("daemon_start_head"),
        "local_head": assessment.get("local_head"),
        "live_remote_head": assessment.get("live_remote_head"),
        "reasons": [
            "The capability resolver requires a restart/reconnect (stale "
            "runtime) but master-parity reported local code as in-parity. "
            "The resolver is authoritative for mutation safety; do not mutate "
            "on local parity alone. Restart/reconnect the Gitea MCP server "
            "and re-verify before mutating.",
        ],
        "recovery": [
            "Trust the resolver: treat this session as stale.",
            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
            "current master and live target state, then re-run preflight.",
        ],
    }
 def format_parity(assessment: dict) -> str:
    """One-line human summary for logs / runtime context."""
    if assessment.get("stale"):
@@ -1,306 +0,0 @@
 """Assess live MCP namespace health without trusting static registration.
 The IDE/client namespace can fail with EOF even when this Python process still
 registers the Gitea tools with FastMCP.  These helpers keep that distinction
 explicit so reviewer/merger flows can fail closed on the live path.
 Probe sources
 -------------
 * ``client_namespace`` — evidence from the IDE-managed MCP client path (the
  only source that can prove the workflow namespace is healthy).
 * ``offline_spawn`` — a separate ``subprocess.Popen`` JSON-RPC handshake
  (e.g. ``test_mcp_conn.py``). Useful offline, but **never** proves the
  IDE-managed namespace is callable.
 * ``unknown`` — legacy/unspecified; treated as not IDE-proven.
 """
 from __future__ import annotations
 from typing import Any
 REQUIRED_NAMESPACE_TOOLS = {
    "gitea-author": "gitea_whoami",
    "gitea-reviewer": "gitea_whoami",
    "gitea-merger": "gitea_whoami",
    "gitea-tools": "gitea_list_profiles",
 }
 DEFAULT_NAMESPACES = tuple(REQUIRED_NAMESPACE_TOOLS)
 # Namespaces that must be healthy for a given mutation task.
 TASK_REQUIRED_NAMESPACES = {
    "review_pr": "gitea-reviewer",
    "submit_review": "gitea-reviewer",
    "merge_pr": "gitea-merger",
 }
 PROBE_SOURCE_CLIENT = "client_namespace"
 PROBE_SOURCE_OFFLINE = "offline_spawn"
 PROBE_SOURCE_UNKNOWN = "unknown"
 VALID_PROBE_SOURCES = frozenset(
    {PROBE_SOURCE_CLIENT, PROBE_SOURCE_OFFLINE, PROBE_SOURCE_UNKNOWN}
 )
 EOF_PATTERNS = (
    "client is closing: eof",
    "transport closed",
    "connection closed",
    "broken pipe",
    "end of file",
    "eof",
 )
 SAFE_ENV_KEYS = (
    "GITEA_MCP_PROFILE",
    "GITEA_PROFILE_NAME",
    "GITEA_SERVICE",
    "GITEA_EXECUTION_ROLE",
    "GITEA_MCP_CONFIG",
 )
 def _as_list(value: Any) -> list[str] | None:
    if value is None:
        return None
    if isinstance(value, (list, tuple, set)):
        return [str(v) for v in value]
    return [str(value)]
 def _contains_eof(text: str | None) -> bool:
    lowered = (text or "").lower()
    return any(pattern in lowered for pattern in EOF_PATTERNS)
 def _normalize_probe_source(probe_source: str | None) -> str:
    raw = (probe_source or PROBE_SOURCE_UNKNOWN).strip().lower()
    if raw in VALID_PROBE_SOURCES:
        return raw
    return PROBE_SOURCE_UNKNOWN
 def _safe_env_summary(process: dict[str, Any] | None) -> dict[str, str]:
    if not process:
        return {}
    env = process.get("env") or process.get("environment") or {}
    if not isinstance(env, dict):
        return {}
    return {
        key: str(env[key])
        for key in SAFE_ENV_KEYS
        if key in env and env[key] not in (None, "")
    }
 def classify_namespace_probe(
    namespace: str,
    *,
    required_tool: str | None = None,
    registered_tools: list[str] | tuple[str, ...] | set[str] | None = None,
    probe_result: dict[str, Any] | None = None,
    process: dict[str, Any] | None = None,
    config_path: str | None = None,
    profile: str | None = None,
    configured: bool = True,
    probe_source: str | None = None,
 ) -> dict[str, Any]:
    """Classify whether a required tool is callable through a live namespace.
    ``registered_tools`` is static/server-side evidence. ``probe_result`` is
    live invocation evidence. Only ``probe_source=client_namespace`` proves the
    IDE-managed path; ``offline_spawn`` is an offline subprocess check only.
    """
    ns = (namespace or "").strip()
    tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami"
    source = _normalize_probe_source(probe_source)
    registered_list = _as_list(registered_tools)
    registered = None if registered_list is None else tool in registered_list
    probe = probe_result or {}
    probe_success = bool(probe.get("success"))
    error_message = str(
        probe.get("error")
        or probe.get("message")
        or probe.get("stderr")
        or probe.get("exception")
        or ""
    )
    error_type = str(probe.get("error_type") or "").strip()
    if not error_type and error_message:
        if _contains_eof(error_message):
            error_type = "namespace_eof"
        elif "timeout" in error_message.lower():
            error_type = "namespace_timeout"
        else:
            error_type = "namespace_call_failed"
    if not configured:
        error_type = "namespace_not_configured"
    elif registered is False:
        error_type = "tool_missing"
    elif not probe_result:
        error_type = "live_probe_missing"
    elif not probe_success and not error_type:
        error_type = "namespace_call_failed"
    callable_live = bool(configured and probe_result and probe_success)
    # Probe-path health (spawn or client). IDE-proven only for client path.
    healthy = bool(configured and registered is not False and callable_live)
    ide_namespace_proven = bool(healthy and source == PROBE_SOURCE_CLIENT)
    process_pid = process.get("pid") if isinstance(process, dict) else None
    profile_name = profile or (
        process.get("profile") if isinstance(process, dict) else None
    )
    env_summary = _safe_env_summary(process)
    reasons: list[str] = []
    if not configured:
        reasons.append(f"MCP namespace '{ns}' is not configured.")
    if registered is False:
        reasons.append(
            f"Required tool '{tool}' is not registered in namespace '{ns}'."
        )
    if error_type == "live_probe_missing":
        reasons.append(
            f"No live client invocation proof was supplied for '{ns}.{tool}'."
        )
    elif error_type == "namespace_eof":
        reasons.append(
            f"Live MCP namespace '{ns}' returned EOF while invoking '{tool}'."
        )
    elif error_type == "namespace_timeout":
        reasons.append(
            f"Live MCP namespace '{ns}' timed out while invoking '{tool}'."
        )
    elif error_type == "namespace_call_failed":
        reasons.append(
            f"Live MCP namespace '{ns}' failed while invoking '{tool}'."
        )
    if source == PROBE_SOURCE_OFFLINE:
        reasons.append(
            "Probe source is offline_spawn (subprocess JSON-RPC); this does "
            "not prove the IDE-managed MCP namespace is healthy."
        )
    elif source == PROBE_SOURCE_UNKNOWN and probe_result:
        reasons.append(
            "Probe source unspecified; treat as not IDE-namespace proof unless "
            "re-supplied with probe_source='client_namespace'."
        )
    remediation = []
    if not healthy or not ide_namespace_proven:
        remediation.append(
            "Reconnect the IDE MCP client namespace (client reconnect / "
            f"relaunch), then invoke '{tool}' through namespace '{ns}' and "
            "record the result with probe_source='client_namespace'."
        )
        remediation.append(
            "Do not treat offline subprocess probes (test_mcp_conn.py) or "
            "shell kill/PID respawn as proof the IDE namespace is repaired."
        )
        if process_pid and source != PROBE_SOURCE_OFFLINE:
            remediation.append(
                f"Diagnostics may include PID {process_pid}; process details "
                "are informational only — recovery is client-layer reconnect."
            )
    else:
        remediation.append(
            f"IDE-managed namespace '{ns}' can invoke '{tool}' "
            f"(probe_source={source})."
        )
    # Client-namespace broken health blocks review/merge. Offline probes never
    # authorize mutations and only block when they report unhealthy (still
    # fail-closed for known bad spawn evidence).
    blocks = False
    if source == PROBE_SOURCE_CLIENT:
        blocks = namespace_health_blocks_task("merge_pr", healthy)
    elif source == PROBE_SOURCE_OFFLINE:
        # Offline never proves IDE health; never unblock. Unhealthy offline
        # still surfaces as a soft diagnostic, not a mutation-ledger block.
        blocks = False
    else:
        # Unknown source: only block when evidence is unhealthy (fail closed
        # on bad data without treating success as IDE proof).
        blocks = namespace_health_blocks_task("merge_pr", healthy)
    return {
        "success": healthy,
        "healthy": healthy,
        "namespace": ns,
        "required_tool": tool,
        "configured": configured,
        "registered_tools_checked": registered_list is not None,
        "required_tool_registered": registered,
        "required_tool_callable": callable_live,
        "probe_source": source,
        "ide_namespace_proven": ide_namespace_proven,
        "error_type": None if healthy else error_type,
        "error_message": error_message or None,
        "reasons": reasons,
        "remediation": remediation,
        "diagnostics": {
            "namespace": ns,
            "required_tool": tool,
            "process_pid": process_pid,
            "profile": profile_name,
            "env": env_summary,
            "config_path": config_path,
            "probe_source": source,
        },
        "blocks_merge_workflow": blocks,
    }
 def namespace_health_blocks_task(task: str, healthy: bool) -> bool:
    """Return whether a broken namespace must block a workflow task."""
    if healthy:
        return False
    return (task or "").strip() in {"merge_pr", "review_pr", "submit_review"}
 def required_namespace_for_task(task: str) -> str | None:
    """Map a mutation task to the MCP namespace that must be healthy."""
    return TASK_REQUIRED_NAMESPACES.get((task or "").strip())
 def mutation_gate_from_session(
    task: str,
    session_health: dict[str, dict[str, Any]] | None,
 ) -> list[str]:
    """Fail-closed gate using recorded client-namespace health assessments.
    * Missing session entry → no gate (caller has not assessed yet).
    * Client-namespace unhealthy / not IDE-proven → block mutation.
    * Offline-only session entries never authorize mutations.
    """
    ns = required_namespace_for_task(task)
    if not ns:
        return []
    store = session_health or {}
    entry = store.get(ns)
    if not entry:
        return []
    source = _normalize_probe_source(entry.get("probe_source"))
    if source != PROBE_SOURCE_CLIENT:
        return [
            f"recorded namespace health for '{ns}' is probe_source={source}, "
            "not client_namespace; re-probe through the IDE-managed path "
            f"before {(task or 'mutation')}"
        ]
    if entry.get("ide_namespace_proven") and entry.get("healthy"):
        return []
    if entry.get("blocks_merge_workflow") or not entry.get("healthy"):
        detail = entry.get("error_type") or "unhealthy"
        return [
            f"live MCP namespace '{ns}' is recorded {detail} "
            f"(probe_source={source}); repair the IDE namespace before "
            f"{(task or 'mutation')} (fail closed, #543)"
        ]
    if not entry.get("ide_namespace_proven"):
        return [
            f"live MCP namespace '{ns}' is not IDE-proven; supply a "
            "client_namespace probe before mutation (fail closed, #543)"
        ]
    return []
@@ -6,9 +6,6 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill)
 import os
 import sys
 sys.stderr = open("/tmp/mcp_server_stderr.log", "a", buffering=1)
 sys.stderr.write(f"\n--- MCP SERVER STARTUP (PID {os.getpid()}) ---\n")
 from role_session_router import (
    python_bytes_have_conflict_markers,
    skip_python_scan_walk_root,
@@ -30,9 +30,6 @@ DEFAULT_TTL_HOURS = 4.0
 KIND_WORKFLOW_LOAD = "review_workflow_load"
 KIND_DECISION_LOCK = "review_decision_lock"
 KIND_REVIEW_DRAFT = "review_draft"
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
 SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
@@ -93,16 +93,8 @@ def assess_workflow_blockers(
    capability_blocked: bool = False,
    mcp_reconnect_failed: bool = False,
    stale_capability_state: bool = False,
    live_namespace_broken: bool = False,
 ) -> dict[str, Any]:
-    """Return hard blockers that forbid all PR queue work (#290 AC3).
+    """Return hard blockers that forbid all PR queue work (#290 AC3)."""
    ``live_namespace_broken`` fails the merge/review path closed when the live
    MCP namespace call path is unusable (e.g. ``client is closing: EOF``) even
    though the tool is registered in FastMCP (#543 AC5). Feed it the
    ``blocks_merge_workflow`` verdict from
    ``mcp_namespace_health.classify_namespace_probe``.
    """
    reasons: list[str] = []
    if infra_stop:
        reasons.append("infra_stop is active; PR selection/review/merge is forbidden")
@@ -112,12 +104,6 @@ def assess_workflow_blockers(
        reasons.append("MCP reconnect failed; stale session state cannot be reused")
    if stale_capability_state:
        reasons.append("stale MCP capability state detected after reconnect failure")
    if live_namespace_broken:
        reasons.append(
            "live MCP namespace call path is broken (registered in FastMCP but "
            "not callable through the namespace); repair the namespace before "
            "review/merge"
        )
    return {
        "block": bool(reasons),
        "reasons": reasons,
@@ -132,19 +118,16 @@ def assess_state_advancement(
    state_completion: dict[str, bool] | None,
    *,
    target_state: str,
-    **blocker_kwargs,
+    infra_stop: bool = False,
    capability_blocked: bool = False,
 ) -> dict[str, Any]:
-    """Fail closed when *target_state* is requested before upstream gates pass.
+    """Fail closed when *target_state* is requested before upstream gates pass."""
    Forwards every blocker flag (``infra_stop``, ``capability_blocked``,
    ``mcp_reconnect_failed``, ``stale_capability_state``,
    ``live_namespace_broken``) to :func:`assess_workflow_blockers` so
    ``can_approve``/``can_merge`` honor the full blocker set, not just
    infra_stop/capability_blocked (#543 AC5).
    """
    completion = dict(state_completion or {})
    target = _clean(target_state).upper()
-    blockers = assess_workflow_blockers(**blocker_kwargs)
+    blockers = assess_workflow_blockers(
        infra_stop=infra_stop,
        capability_blocked=capability_blocked,
    )
    reasons = list(blockers["reasons"])
    try:
@@ -1,4 +1,4 @@
-"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620).
+"""Stale / moot #332 review-decision lock detection and cleanup policy (#594).
 #332 correctly hard-stops a reviewer session after a terminal live review
 mutation. After #559 those locks are durable on disk, so they can outlive the
@@ -6,12 +6,6 @@ work they protect: when the referenced PR is already merged or closed, no
 same-PR merge sequence remains, yet the durable ledger still blocks unrelated
 new terminal reviews.
 #620 scopes the terminal boundary by **reviewed head SHA**: a prior
 REQUEST_CHANGES (or APPROVE) on head A must not block a fresh formal decision
 on head B of the **same open PR**. Same-head duplicates remain fail-closed.
 Open-PR lock *cleanup* remains forbidden unless the PR is truly merged/closed
 (#594); new-head re-review is allowed without deleting the durable lock.
 This module is pure policy (no Gitea I/O). The MCP tool
 ``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
 helpers, and only then clears durable state when cleanup is allowed.
@@ -29,45 +23,6 @@ from typing import Any
 TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
 def normalize_head_sha(value: Any) -> str | None:
    """Normalize a git SHA for equality comparison; None if empty/invalid."""
    if value is None:
        return None
    text = str(value).strip().lower()
    if not text:
        return None
    return text
 def heads_equal(a: Any, b: Any) -> bool:
    na = normalize_head_sha(a)
    nb = normalize_head_sha(b)
    if not na or not nb:
        return False
    return na == nb
 def mutation_head_sha(mutation: dict | None, lock: dict | None = None) -> str | None:
    """Head SHA that bounds a live mutation (#620).
    Prefers fields recorded on the mutation itself. For *legacy* mutations
    that predate head-scoping, falls back to the lock's
    ``ready_expected_head_sha`` only when that ready binding is for the same
    PR number (the frozen durable PR #619-style ledger).
    """
    if not isinstance(mutation, dict):
        return None
    for key in ("head_sha", "expected_head_sha", "reviewed_head_sha"):
        head = normalize_head_sha(mutation.get(key))
        if head:
            return head
    if not isinstance(lock, dict):
        return None
    if lock.get("ready_pr_number") != mutation.get("pr_number"):
        return None
    return normalize_head_sha(lock.get("ready_expected_head_sha"))
 def last_terminal_mutation(lock: dict | None) -> dict | None:
    """Return the last terminal live mutation on *lock*, or None."""
    if not lock:
@@ -80,125 +35,18 @@ def last_terminal_mutation(lock: dict | None) -> dict | None:
    return terminals[-1] if terminals else None
 def prior_live_mutations_block_boundary(
    lock: dict | None,
    *,
    pr_number: int,
    expected_head_sha: str | None,
 ) -> bool:
    """True when prior live mutations block a new decision for PR+head (#620).
    Historical terminals on a **different head of the same PR** do not block.
    Any prior mutation on another PR, the same head, or without a comparable
    head SHA fails closed (blocks).
    Head resolution uses ``mutation_head_sha(m, lock)`` so pre-#620 ledgers
    that only stored ``ready_expected_head_sha`` still compare correctly
    (PR #619-style).
    """
    if not lock:
        return False
    if lock.get("correction_authorized"):
        return False
    prior = list(lock.get("live_mutations") or [])
    if not prior:
        return False
    target = normalize_head_sha(expected_head_sha)
    for m in prior:
        if not isinstance(m, dict):
            return True
        m_pr = m.get("pr_number")
        m_head = mutation_head_sha(m, lock)
        if (
            m_pr == pr_number
            and target
            and m_head
            and not heads_equal(m_head, target)
        ):
            # Same open PR, different reviewed head — historical boundary only.
            continue
        return True
    return False
 def terminal_boundary_allows_fresh_decision(
    lock: dict | None,
    *,
    pr_number: int,
    expected_head_sha: str | None,
    operation: str,
 ) -> bool:
    """Whether #332 hard-stop should yield for a new PR+head boundary (#620).
    Only for ``mark_ready`` / ``review`` / ``resume`` on the **same PR** when
    the requested head differs from the last terminal's head. Merge is never
    reopened by head change (approved head merge path is separate).
    """
    if operation not in ("mark_ready", "review", "resume"):
        return False
    if not lock:
        return False
    if lock.get("correction_authorized"):
        return True
    last = last_terminal_mutation(lock)
    if last is None:
        return True
    if last.get("pr_number") != pr_number:
        return False
    locked_head = mutation_head_sha(last, lock)
    target = normalize_head_sha(expected_head_sha)
    if not locked_head or not target:
        return False
    return not heads_equal(locked_head, target)
 def backfill_terminal_heads_from_ready(lock: dict) -> dict:
    """Stamp head_sha onto legacy terminal mutations before overwriting ready_*.
    Called when marking a fresh decision on a new head so historical rows keep
    their original boundary after ``ready_expected_head_sha`` advances (#620).
    """
    if not isinstance(lock, dict):
        return lock
    ready_pr = lock.get("ready_pr_number")
    ready_head = normalize_head_sha(lock.get("ready_expected_head_sha"))
    if ready_pr is None or not ready_head:
        return lock
    mutations = list(lock.get("live_mutations") or [])
    changed = False
    for m in mutations:
        if not isinstance(m, dict):
            continue
        if m.get("action") not in TERMINAL_REVIEW_ACTIONS:
            continue
        if m.get("pr_number") != ready_pr:
            continue
        if mutation_head_sha(m):
            continue
        m["head_sha"] = ready_head
        changed = True
    if changed:
        lock["live_mutations"] = mutations
    return lock
 def classify_pr_live_state(pr_live: dict | None) -> dict[str, Any]:
    """Normalize a Gitea PR payload into merge/closed flags."""
    pr = pr_live or {}
    merged = bool(pr.get("merged") or pr.get("merged_at"))
    state = (pr.get("state") or "").strip().lower() or None
    closed = state == "closed"
    head = pr.get("head") if isinstance(pr.get("head"), dict) else {}
    head_sha = normalize_head_sha(
        pr.get("head_sha") or pr.get("head_commit_sha") or head.get("sha")
    )
    return {
        "pr_state": state,
        "pr_merged": merged,
        "pr_merged_or_closed": merged or closed,
        "merge_commit_sha": pr.get("merge_commit_sha")
        or pr.get("merged_commit_sha"),
        "current_pr_head_sha": head_sha,
    }
@@ -208,7 +56,6 @@ def lock_summary(lock: dict | None) -> dict[str, Any] | None:
        return None
    last = last_terminal_mutation(lock)
    mutations = list(lock.get("live_mutations") or [])
    last_head = mutation_head_sha(last, lock) if last else None
    return {
        "task": lock.get("task"),
        "remote": lock.get("remote"),
@@ -220,21 +67,16 @@ def lock_summary(lock: dict | None) -> dict[str, Any] | None:
        "session_profile": lock.get("session_profile"),
        "ready_pr_number": lock.get("ready_pr_number"),
        "ready_action": lock.get("ready_action"),
        "ready_expected_head_sha": normalize_head_sha(
            lock.get("ready_expected_head_sha")
        ),
        "live_mutations_count": len(mutations),
        "last_terminal": (
            {
                "pr_number": last.get("pr_number"),
                "action": last.get("action"),
                "review_id": last.get("review_id"),
                "head_sha": last_head,
            }
            if last
            else None
        ),
        "locked_head_sha": last_head,
        "correction_authorized": bool(lock.get("correction_authorized")),
        "updated_at": lock.get("updated_at") or lock.get("recorded_at"),
    }
@@ -246,16 +88,13 @@ def assess_stale_review_decision_lock(
    pr_live: dict | None = None,
    pr_lookup_error: str | None = None,
    active_profile_identity: str | None = None,
    current_pr_head_sha: str | None = None,
 ) -> dict[str, Any]:
-    """Assess whether a durable review-decision lock is stale/moot (#594/#620).
+    """Assess whether a durable review-decision lock is stale/moot (#594).
    *pr_live* must be the live Gitea payload for the **last terminal
    mutation's PR**. When no lock / no terminal mutation exists, cleanup is
    not applicable. When the PR is still open (or lookup fails), cleanup is
-    forbidden (#594). Open PR + different live head reports
+    forbidden and #332 hard-stop remains in force.
    ``stale_by_head`` / ``fresh_review_on_current_head_allowed`` (#620)
    without enabling cleanup.
    """
    summary = lock_summary(lock)
    result: dict[str, Any] = {
@@ -263,10 +102,6 @@ def assess_stale_review_decision_lock(
        "lock_summary": summary,
        "last_terminal_pr": None,
        "last_terminal_action": None,
        "locked_head_sha": None,
        "current_pr_head_sha": normalize_head_sha(current_pr_head_sha),
        "stale_by_head": False,
        "fresh_review_on_current_head_allowed": False,
        "is_moot": False,
        "cleanup_allowed": False,
        "profile_match": True,
@@ -306,10 +141,8 @@ def assess_stale_review_decision_lock(
    pr_number = last.get("pr_number")
    action = last.get("action")
    locked_head = mutation_head_sha(last, lock)
    result["last_terminal_pr"] = pr_number
    result["last_terminal_action"] = action
    result["locked_head_sha"] = locked_head
    if pr_number is None:
        result["reasons"].append(
@@ -332,46 +165,25 @@ def assess_stale_review_decision_lock(
        return result
    live = classify_pr_live_state(pr_live)
    current_head = normalize_head_sha(
        current_pr_head_sha or live.get("current_pr_head_sha")
    )
    result.update(
        {
            "pr_state": live["pr_state"],
            "pr_merged": live["pr_merged"],
            "pr_merged_or_closed": live["pr_merged_or_closed"],
            "merge_commit_sha": live["merge_commit_sha"],
            "current_pr_head_sha": current_head,
        }
    )
    if not live["pr_merged_or_closed"]:
-        stale_by_head = bool(
+        result["reasons"].append(
-            locked_head and current_head and not heads_equal(locked_head, current_head)
+            f"terminal review mutation on open PR #{pr_number} is still active "
            f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
        )
        result["stale_by_head"] = stale_by_head
        # Fresh formal decision is allowed on the new head without cleanup (#620).
        result["fresh_review_on_current_head_allowed"] = stale_by_head
        if stale_by_head:
            result["reasons"].append(
                f"terminal {action} on PR #{pr_number} is bound to head "
                f"{locked_head[:12]}…; live head is {current_head[:12]}… — "
                "fresh formal review on current head is allowed without "
                "cleanup (fail closed for cleanup, #620); #332 same-head "
                "duplicate protection remains"
            )
        else:
            result["reasons"].append(
                f"terminal review mutation on open PR #{pr_number} is still active "
                f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
            )
        return result
    # Merged or closed: lock is moot. Same-PR merge continuation is impossible.
    result["is_moot"] = True
    result["cleanup_allowed"] = True
    result["stale_by_head"] = False
    result["fresh_review_on_current_head_allowed"] = False
    result["reasons"].append(
        f"terminal mutation ({action} on PR #{pr_number}) is moot: PR is "
        f"{'merged' if live['pr_merged'] else 'closed'}; canonical cleanup allowed (#594)"
@@ -139,17 +139,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.pr.create",
        "role": "author",
    },
    # #600: controller-owned allocator — any authenticated profile may call;
    # routing enforces role match to selected work. Uses control-plane DB (#613).
    "allocate_next_work": {
        "permission": "gitea.read",
        "role": "author",
    },
    "gitea_allocate_next_work": {
        "permission": "gitea.read",
        "role": "author",
    },
    "reconcile_landed_pr": {
        "permission": "gitea.read",
        "role": "author",
@@ -1,45 +1,20 @@
 #!/usr/bin/env python3
-"""Offline-only MCP namespace spawn probe (NOT IDE-namespace proof).
+"""Live health-check script to verify Gitea MCP namespace connections.
-Spawns a *separate* MCP server process from config via subprocess.Popen,
+Spawns the MCP server processes as defined in the IDE's global config,
-performs the JSON-RPC handshake, verifies the required tool is registered,
+performs the JSON-RPC handshake, and queries the tools list to verify
-and invokes that tool. Results are classified with
+that the connection is fully operational and doesn't return EOF.
 ``probe_source=offline_spawn``.
 This path is useful for offline launch/registration debugging. It does
 **not** prove the IDE-managed MCP client namespace is healthy (#543). For
 workflow gates, pass live IDE call evidence with
 ``probe_source=client_namespace`` to ``gitea_assess_mcp_namespace_health``.
 """
 import argparse
 import json
 import os
 import subprocess
 import sys
-from mcp_namespace_health import REQUIRED_NAMESPACE_TOOLS, classify_namespace_probe
+def run_connection_test(name, config):
 def _read_json_line(proc):
    line = proc.stdout.readline()
    if not line:
        stderr_content = proc.stderr.read()
        return None, stderr_content
    return json.loads(line), None
 def _write_message(proc, payload):
    proc.stdin.write(json.dumps(payload) + "\n")
    proc.stdin.flush()
 def run_connection_test(name, config, *, required_tool=None, config_path=None):
    print(f"Testing MCP connection for '{name}'...")
    command = config.get("command")
    args = config.get("args", [])
    env = config.get("env", {})
    tool_name = required_tool or REQUIRED_NAMESPACE_TOOLS.get(name) or "gitea_whoami"
    # Merge current environment
    run_env = os.environ.copy()
@@ -56,17 +31,8 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
            bufsize=1,
            env=run_env
        )
-    except Exception as exc:
+    except Exception as e:
-        print(f"  [FAIL] Failed to spawn process: {exc}")
+        print(f"  [FAIL] Failed to spawn process: {e}")
        assessment = classify_namespace_probe(
            name,
            required_tool=tool_name,
            probe_result={"success": False, "error": str(exc)},
            process={"profile": env.get("GITEA_MCP_PROFILE"), "env": env},
            config_path=config_path,
                probe_source="offline_spawn",
        )
        print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
        return False
    # Send initialize request
@@ -82,36 +48,26 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
    }
    try:
-        _write_message(proc, init_req)
+        proc.stdin.write(json.dumps(init_req) + "\n")
        proc.stdin.flush()
        # Read response
-        res, stderr_content = _read_json_line(proc)
+        line = proc.stdout.readline()
-        if res is None:
+        if not line:
            stderr_content = proc.stderr.read()
            print(f"  [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
            assessment = classify_namespace_probe(
                name,
                required_tool=tool_name,
                probe_result={"success": False, "error": stderr_content or "EOF"},
                process={
                    "pid": proc.pid,
                    "profile": env.get("GITEA_MCP_PROFILE"),
                    "env": env,
                },
                config_path=config_path,
                probe_source="offline_spawn",
            )
            print(f"  remediation: {' '.join(assessment['remediation'])}")
            proc.terminate()
            return False
-        print(f"  [OK] Received initialize response: {str(res)[:150]}...")
+        print(f"  [OK] Received initialize response: {line.strip()[:150]}...")
        # Send initialized notification
        init_notif = {
            "jsonrpc": "2.0",
            "method": "notifications/initialized"
        }
-        _write_message(proc, init_notif)
+        proc.stdin.write(json.dumps(init_notif) + "\n")
        proc.stdin.flush()
        # Send tools/list request
        list_req = {
@@ -120,27 +76,16 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
            "params": {},
            "id": 2
        }
-        _write_message(proc, list_req)
+        proc.stdin.write(json.dumps(list_req) + "\n")
        proc.stdin.flush()
-        res, stderr_content = _read_json_line(proc)
+        line = proc.stdout.readline()
-        if res is None:
+        if not line:
            print("  [FAIL] Received EOF on tools/list request.")
            assessment = classify_namespace_probe(
                name,
                required_tool=tool_name,
                probe_result={"success": False, "error": stderr_content or "EOF"},
                process={
                    "pid": proc.pid,
                    "profile": env.get("GITEA_MCP_PROFILE"),
                    "env": env,
                },
                config_path=config_path,
                probe_source="offline_spawn",
            )
            print(f"  remediation: {' '.join(assessment['remediation'])}")
            proc.terminate()
            return False
        res = json.loads(line)
        if "error" in res:
            print(f"  [FAIL] Server returned error: {res['error']}")
            proc.terminate()
@@ -149,115 +94,16 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
        tools = res.get("result", {}).get("tools", [])
        tool_names = [t.get("name") for t in tools]
        print(f"  [OK] Successfully retrieved {len(tool_names)} tools: {tool_names[:5]}...")
        if tool_name not in tool_names:
            assessment = classify_namespace_probe(
                name,
                required_tool=tool_name,
                registered_tools=tool_names,
                probe_result={"success": False, "error": "required tool missing"},
                process={
                    "pid": proc.pid,
                    "profile": env.get("GITEA_MCP_PROFILE"),
                    "env": env,
                },
                config_path=config_path,
                probe_source="offline_spawn",
            )
            print(f"  [FAIL] Required tool '{tool_name}' is not registered.")
            print(f"  remediation: {' '.join(assessment['remediation'])}")
            proc.terminate()
            return False
        call_req = {
            "jsonrpc": "2.0",
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": {}},
            "id": 3,
        }
        _write_message(proc, call_req)
        call_res, stderr_content = _read_json_line(proc)
        if call_res is None:
            assessment = classify_namespace_probe(
                name,
                required_tool=tool_name,
                registered_tools=tool_names,
                probe_result={"success": False, "error": stderr_content or "EOF"},
                process={
                    "pid": proc.pid,
                    "profile": env.get("GITEA_MCP_PROFILE"),
                    "env": env,
                },
                config_path=config_path,
                probe_source="offline_spawn",
            )
            print(f"  [FAIL] Received EOF on {tool_name} invocation.")
            print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
            print(f"  remediation: {' '.join(assessment['remediation'])}")
            proc.terminate()
            return False
        if "error" in call_res:
            assessment = classify_namespace_probe(
                name,
                required_tool=tool_name,
                registered_tools=tool_names,
                probe_result={"success": False, "error": call_res["error"]},
                process={
                    "pid": proc.pid,
                    "profile": env.get("GITEA_MCP_PROFILE"),
                    "env": env,
                },
                config_path=config_path,
                probe_source="offline_spawn",
            )
            print(f"  [FAIL] {tool_name} invocation returned error: {call_res['error']}")
            print(f"  remediation: {' '.join(assessment['remediation'])}")
            proc.terminate()
            return False
        assessment = classify_namespace_probe(
            name,
            required_tool=tool_name,
            registered_tools=tool_names,
            probe_result={"success": True, "result": call_res.get("result")},
            process={
                "pid": proc.pid,
                "profile": env.get("GITEA_MCP_PROFILE"),
                "env": env,
            },
            config_path=config_path,
                probe_source="offline_spawn",
        )
        print(f"  [OK] Successfully invoked required tool '{tool_name}'.")
        print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
        proc.terminate()
        return True
-    except Exception as exc:
+    except Exception as e:
-        print(f"  [FAIL] Error during handshake: {exc}")
+        print(f"  [FAIL] Error during handshake: {e}")
        proc.terminate()
        return False
 def main():
-    parser = argparse.ArgumentParser()
+    config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
    parser.add_argument(
        "--config",
        default=os.environ.get(
            "MCP_CONFIG_PATH",
            os.path.expanduser("~/.gemini/config/mcp_config.json"),
        ),
        help="Path to MCP config JSON.",
    )
    parser.add_argument(
        "--namespace",
        action="append",
        dest="namespaces",
        help="Namespace to test. May be repeated.",
    )
    args = parser.parse_args()
    config_path = args.config
    try:
        with open(config_path) as f:
            mcp_config = json.load(f)
@@ -266,16 +112,10 @@ def main():
        sys.exit(1)
    servers = mcp_config.get("mcpServers", {})
    namespaces = args.namespaces or list(REQUIRED_NAMESPACE_TOOLS)
    failed = False
-    for name in namespaces:
+    for name in ["gitea-author", "gitea-reviewer"]:
        if name in servers:
-            if not run_connection_test(
+            if not run_connection_test(name, servers[name]):
                name,
                servers[name],
                required_tool=REQUIRED_NAMESPACE_TOOLS.get(name),
                config_path=config_path,
            ):
                failed = True
        else:
            print(f"Server '{name}' not found in mcp_config.json")
@@ -66,7 +66,6 @@ def _reset_mutation_authority(monkeypatch):
    monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
    monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
    monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
    monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
    monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
    monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
    monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
@@ -1,366 +0,0 @@
 """Tests for controller-owned allocator (#600) on control-plane DB (#613)."""
 from __future__ import annotations
 import os
 import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from allocator_service import (
    OUTCOME_ASSIGNED,
    OUTCOME_BLOCKED_TERMINAL,
    OUTCOME_NO_SAFE,
    OUTCOME_PREVIEW,
    OUTCOME_WAIT,
    WorkCandidate,
    allocate_next_work,
    candidate_from_dict,
    classify_skip,
    expected_role_for_candidate,
 )
 from control_plane_db import ControlPlaneDB, InvalidWorkKindError
 class AllocatorServiceTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
        self.db = ControlPlaneDB(self.db_path)
    def tearDown(self) -> None:
        self._tmp.cleanup()
    def _alloc(self, **kwargs):
        defaults = dict(
            db=self.db,
            session_id="s-test",
            role="author",
            remote="prgs",
            org="org",
            repo="repo",
            candidates=[],
            apply=False,
            profile_name="prgs-author",
            username="jcwalker3",
        )
        defaults.update(kwargs)
        return allocate_next_work(**defaults)
    def test_selects_ready_issue_for_author(self) -> None:
        cands = [
            WorkCandidate(
                kind="issue",
                number=612,
                labels=("status:ready",),
                title="bridge",
                dependency_unmet=True,
                dependency_reason="downstream of #600",
            ),
            WorkCandidate(
                kind="issue",
                number=600,
                labels=("status:ready", "type:feature"),
                title="allocator",
                priority=50,
            ),
            WorkCandidate(
                kind="issue",
                number=601,
                labels=("status:blocked",),
                title="blocked",
                blocked=True,
            ),
        ]
        res = self._alloc(candidates=cands, apply=False)
        self.assertTrue(res["success"])
        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
        self.assertEqual(res["selected"]["number"], 600)
        self.assertEqual(res["selected"]["kind"], "issue")
        # When 600 is highest priority valid work, lower-priority blocked/dep
        # candidates are not visited. Prove they are skipped when they sort first.
        res2 = self._alloc(
            candidates=[
                WorkCandidate(
                    kind="issue",
                    number=612,
                    labels=("status:ready",),
                    priority=99,
                    dependency_unmet=True,
                    dependency_reason="downstream of #600",
                ),
                WorkCandidate(
                    kind="issue",
                    number=601,
                    labels=("status:blocked",),
                    priority=98,
                    blocked=True,
                ),
                WorkCandidate(
                    kind="issue",
                    number=600,
                    labels=("status:ready",),
                    priority=1,
                ),
            ],
            apply=False,
        )
        skipped_nums = {s["number"] for s in res2["skipped"]}
        self.assertIn(612, skipped_nums)
        self.assertIn(601, skipped_nums)
        self.assertEqual(res2["selected"]["number"], 600)
        self.assertIsNone(res["assignment"])
        self.assertFalse(res["file_lock_only"])
        self.assertFalse(res["comment_lease_only"])
    def test_atomic_assign_and_lease_on_apply(self) -> None:
        cands = [
            WorkCandidate(
                kind="issue",
                number=600,
                labels=("status:ready",),
                priority=10,
            )
        ]
        res = self._alloc(candidates=cands, apply=True, session_id="s-a")
        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
        asn = res["assignment"]
        self.assertEqual(asn["outcome"], "assigned")
        self.assertIsNotNone(asn["assignment_id"])
        self.assertIsNotNone(asn["lease_id"])
        self.assertEqual(asn["work_number"], 600)
        self.assertIn("implement", asn["allowed_actions"])
        self.assertIn("merge", asn["forbidden_actions"])
        proof = res["lease_proof"]
        self.assertEqual(proof["source"], "control_plane_db.assign_and_lease")
    def test_concurrent_allocators_no_double_assign(self) -> None:
        cand = WorkCandidate(
            kind="pr",
            number=100,
            head_sha="a" * 40,
            priority=10,
        )
        # Seed work item so both race the same target
        self.db.upsert_work_item(
            remote="prgs",
            org="org",
            repo="repo",
            kind="pr",
            number=100,
            current_head_sha="a" * 40,
        )
        results = []
        lock = threading.Lock()
        def worker(sid: str):
            r = allocate_next_work(
                self.db,
                session_id=sid,
                role="reviewer",
                remote="prgs",
                org="org",
                repo="repo",
                candidates=[cand],
                apply=True,
                profile_name="prgs-reviewer",
            )
            with lock:
                results.append(r)
        with ThreadPoolExecutor(max_workers=2) as pool:
            futs = [pool.submit(worker, f"s-{i}") for i in range(2)]
            for f in as_completed(futs):
                f.result()
        outcomes = [r["outcome"] for r in results]
        self.assertEqual(outcomes.count(OUTCOME_ASSIGNED), 1, outcomes)
        self.assertEqual(outcomes.count(OUTCOME_WAIT), 1, outcomes)
    def test_blocked_and_dependency_skipped(self) -> None:
        cands = [
            WorkCandidate(kind="issue", number=1, blocked=True, labels=("status:blocked",)),
            WorkCandidate(
                kind="issue",
                number=612,
                labels=("status:ready",),
                dependency_unmet=True,
                dependency_reason="downstream of #600",
            ),
        ]
        res = self._alloc(candidates=cands, apply=True, role="author")
        self.assertEqual(res["outcome"], OUTCOME_NO_SAFE)
        self.assertIsNone(res["selected"])
        reasons = " ".join(s["reason"] for s in res["skipped"])
        self.assertIn("blocked", reasons.lower())
        self.assertIn("600", reasons)
    def test_already_leased_returns_wait(self) -> None:
        self.db.upsert_session(session_id="owner", role="author")
        self.db.assign_and_lease(
            session_id="owner",
            role="author",
            remote="prgs",
            org="org",
            repo="repo",
            kind="issue",
            number=50,
        )
        res = self._alloc(
            session_id="other",
            candidates=[
                WorkCandidate(kind="issue", number=50, labels=("status:ready",))
            ],
            apply=True,
        )
        self.assertEqual(res["outcome"], OUTCOME_WAIT)
        self.assertEqual(res["owner_session_id"], "owner")
    def test_role_ineligible_skipped(self) -> None:
        # PR with current-head REQUEST_CHANGES expects author, not reviewer.
        cand = WorkCandidate(
            kind="pr",
            number=9,
            head_sha="b" * 40,
            request_changes_current_head=True,
        )
        self.assertEqual(expected_role_for_candidate(cand), "author")
        res = self._alloc(
            role="reviewer",
            profile_name="prgs-reviewer",
            candidates=[cand],
            apply=False,
        )
        self.assertEqual(res["outcome"], OUTCOME_NO_SAFE)
        self.assertTrue(any("expects role" in s["reason"] for s in res["skipped"]))
    def test_terminal_lock_blocks_downstream_reviewer_prs(self) -> None:
        self.db.set_terminal_lock(
            remote="prgs",
            org="org",
            repo="repo",
            terminal_pr=10,
            decision="request_changes",
            status="active",
        )
        cands = [
            WorkCandidate(kind="pr", number=11, head_sha="c" * 40, priority=5),
            WorkCandidate(kind="pr", number=10, head_sha="d" * 40, priority=1),
        ]
        res = self._alloc(
            role="reviewer",
            profile_name="prgs-reviewer",
            candidates=cands,
            apply=False,
        )
        # Terminal PR #10 is selectable; #11 skipped for terminal path.
        self.assertEqual(res["selected"]["number"], 10)
        self.assertTrue(
            any(
                s["number"] == 11 and "terminal-review lock" in s["reason"]
                for s in res["skipped"]
            )
        )
    def test_terminal_lock_blocks_all_when_only_downstream(self) -> None:
        self.db.set_terminal_lock(
            remote="prgs",
            org="org",
            repo="repo",
            terminal_pr=10,
            decision="request_changes",
            status="active",
        )
        res = self._alloc(
            role="reviewer",
            profile_name="prgs-reviewer",
            candidates=[
                WorkCandidate(kind="pr", number=99, head_sha="e" * 40),
            ],
            apply=False,
        )
        self.assertEqual(res["outcome"], OUTCOME_BLOCKED_TERMINAL)
    def test_no_work_structured(self) -> None:
        res = self._alloc(candidates=[], apply=True)
        self.assertEqual(res["outcome"], OUTCOME_NO_SAFE)
        self.assertIsNone(res["selected"])
        self.assertTrue(res["reasons"])
    def test_rejects_incident_kind(self) -> None:
        with self.assertRaises(InvalidWorkKindError):
            WorkCandidate(kind="sentry_incident", number=1)
    def test_612_downstream_marker_in_result(self) -> None:
        res = self._alloc(
            candidates=[
                WorkCandidate(kind="issue", number=600, labels=("status:ready",))
            ],
            apply=True,
        )
        self.assertIn("612", res.get("downstream_note", ""))
    def test_substrate_not_file_or_comment_lease(self) -> None:
        res = self._alloc(
            candidates=[
                WorkCandidate(kind="issue", number=1, labels=("status:ready",))
            ],
            apply=True,
        )
        self.assertEqual(res["substrate"], "control_plane_db")
        self.assertFalse(res["file_lock_only"])
        self.assertFalse(res["comment_lease_only"])
        self.assertEqual(
            res["lease_proof"]["source"], "control_plane_db.assign_and_lease"
        )
    def test_candidate_from_dict(self) -> None:
        c = candidate_from_dict(
            {
                "kind": "pr",
                "number": 7,
                "head_sha": "f" * 40,
                "approval_on_current_head": True,
                "mergeable": True,
            }
        )
        self.assertEqual(expected_role_for_candidate(c), "merger")
    def test_merger_gets_clean_approval(self) -> None:
        c = WorkCandidate(
            kind="pr",
            number=3,
            head_sha="1" * 40,
            approval_on_current_head=True,
            mergeable=True,
            priority=100,
        )
        res = self._alloc(
            role="merger",
            profile_name="prgs-merger",
            candidates=[c],
            apply=True,
            session_id="merger-1",
        )
        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
        self.assertEqual(res["assignment"]["work_number"], 3)
        self.assertIn("merge", res["assignment"]["allowed_actions"])
    def test_db_unavailable_fails_closed(self) -> None:
        res = allocate_next_work(
            None,  # type: ignore[arg-type]
            session_id="x",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            candidates=[],
            apply=True,
        )
        self.assertFalse(res["success"])
        self.assertIn("unavailable", res["reasons"][0].lower())
 if __name__ == "__main__":
    unittest.main()
@@ -1,824 +0,0 @@
 """Tests for control-plane DB substrate (#613)."""
 from __future__ import annotations
 import os
 import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import timedelta
 from control_plane_db import (
    ControlPlaneDB,
    InvalidWorkKindError,
    LeaseRequiredError,
    WORK_KINDS,
    _ts,
    _utc_now,
 )
 class ControlPlaneDBTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
        self.db = ControlPlaneDB(self.db_path)
    def tearDown(self) -> None:
        self._tmp.cleanup()
    def test_schema_and_architecture_meta(self) -> None:
        import sqlite3
        conn = sqlite3.connect(self.db_path)
        try:
            rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
        finally:
            conn.close()
        self.assertEqual(rows["schema_version"], "2")
        self.assertIn("DB coordinates", rows["architecture"])
        self.assertIn("bridge", rows["architecture"].lower())
    def test_rejects_raw_incident_as_work_kind(self) -> None:
        with self.assertRaises(InvalidWorkKindError):
            self.db.upsert_work_item(
                remote="prgs",
                org="org",
                repo="repo",
                kind="sentry_incident",
                number=1,
            )
        with self.assertRaises(InvalidWorkKindError):
            self.db.assign_and_lease(
                session_id="s1",
                role="author",
                remote="prgs",
                org="org",
                repo="repo",
                kind="glitchtip_incident",
                number=9,
            )
        self.assertEqual(WORK_KINDS, frozenset({"issue", "pr"}))
    def test_atomic_assign_and_lease_fields(self) -> None:
        self.db.upsert_session(session_id="s-a", role="author", profile="prgs-author")
        result = self.db.assign_and_lease(
            session_id="s-a",
            role="author",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            kind="issue",
            number=613,
            expected_head_sha="abc123",
            allowed_actions=("implement", "comment"),
            forbidden_actions=("approve", "merge"),
        )
        self.assertEqual(result.outcome, "assigned")
        self.assertIsNotNone(result.assignment_id)
        self.assertIsNotNone(result.lease_id)
        self.assertEqual(result.role, "author")
        self.assertEqual(result.work_kind, "issue")
        self.assertEqual(result.work_number, 613)
        self.assertEqual(result.expected_head_sha, "abc123")
        self.assertIn("implement", result.allowed_actions)
        self.assertIn("merge", result.forbidden_actions)
        self.assertIsNotNone(result.expires_at)
    def test_second_session_waits_on_foreign_lease(self) -> None:
        self.db.upsert_session(session_id="s1", role="author")
        self.db.upsert_session(session_id="s2", role="author")
        first = self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=100,
            expected_head_sha="deadbeef",
        )
        self.assertEqual(first.outcome, "assigned")
        second = self.db.assign_and_lease(
            session_id="s2",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=100,
            expected_head_sha="deadbeef",
        )
        self.assertEqual(second.outcome, "wait")
        self.assertEqual(second.owner_session_id, "s1")
    def test_owner_resume_refreshes_lease(self) -> None:
        self.db.upsert_session(session_id="s1", role="reviewer")
        a = self.db.assign_and_lease(
            session_id="s1",
            role="reviewer",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=50,
            expected_head_sha="head-50",
        )
        b = self.db.assign_and_lease(
            session_id="s1",
            role="reviewer",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=50,
            expected_head_sha="head-50",
        )
        self.assertEqual(b.outcome, "assigned")
        self.assertEqual(b.lease_id, a.lease_id)
        self.assertIn("owner-resume", b.reason)
    def test_require_valid_assignment_gates_mutations(self) -> None:
        self.db.upsert_session(session_id="s1", role="author")
        self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=7,
            allowed_actions=("implement",),
            forbidden_actions=("merge",),
        )
        proof = self.db.require_valid_assignment(
            session_id="s1",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=7,
            action="implement",
        )
        self.assertEqual(proof["session_id"], "s1")
        with self.assertRaises(LeaseRequiredError):
            self.db.require_valid_assignment(
                session_id="s1",
                remote="prgs",
                org="o",
                repo="r",
                kind="issue",
                number=7,
                action="merge",
            )
        with self.assertRaises(LeaseRequiredError):
            self.db.require_valid_assignment(
                session_id="s-other",
                remote="prgs",
                org="o",
                repo="r",
                kind="issue",
                number=7,
                action="implement",
            )
    def test_expired_lease_allows_reassign(self) -> None:
        self.db.upsert_session(session_id="s1", role="author")
        self.db.upsert_session(session_id="s2", role="author")
        past = _utc_now() - timedelta(hours=1)
        # Create lease already expired by using negative TTL edge via direct assign then expire
        assigned = self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=3,
            lease_ttl_seconds=1,
        )
        self.assertEqual(assigned.outcome, "assigned")
        # Force expiry in DB
        import sqlite3
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE leases SET expires_at = ? WHERE lease_id = ?",
                (_ts(past), assigned.lease_id),
            )
            conn.commit()
        finally:
            conn.close()
        n = self.db.expire_stale_leases()
        self.assertGreaterEqual(n, 1)
        second = self.db.assign_and_lease(
            session_id="s2",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=3,
        )
        self.assertEqual(second.outcome, "assigned")
        self.assertEqual(second.session_id, "s2")
    def test_merged_work_never_assigned(self) -> None:
        self.db.upsert_session(session_id="s1", role="merger")
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=99,
            state="merged",
        )
        result = self.db.assign_and_lease(
            session_id="s1",
            role="merger",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=99,
            expected_head_sha="merged-head",
        )
        self.assertEqual(result.outcome, "no_safe_work")
    def test_four_concurrent_sessions_unique_assignments(self) -> None:
        """Four concurrent assigners on four different issues — all succeed uniquely.
        Also two concurrent assigners on the *same* issue: at most one assigned.
        """
        for i in range(4):
            self.db.upsert_session(session_id=f"sess-{i}", role="author")
        def claim_unique(i: int):
            return self.db.assign_and_lease(
                session_id=f"sess-{i}",
                role="author",
                remote="prgs",
                org="o",
                repo="r",
                kind="issue",
                number=1000 + i,
            )
        with ThreadPoolExecutor(max_workers=4) as pool:
            results = [f.result() for f in as_completed([pool.submit(claim_unique, i) for i in range(4)])]
        self.assertEqual({r.outcome for r in results}, {"assigned"})
        numbers = sorted(r.work_number for r in results)
        self.assertEqual(numbers, [1000, 1001, 1002, 1003])
        # Contention on one item
        self.db.upsert_session(session_id="c1", role="author")
        self.db.upsert_session(session_id="c2", role="author")
        self.db.upsert_session(session_id="c3", role="author")
        self.db.upsert_session(session_id="c4", role="author")
        barrier = threading.Barrier(4)
        outcomes: list[str] = []
        lock = threading.Lock()
        def contend(sid: str) -> None:
            barrier.wait()
            res = self.db.assign_and_lease(
                session_id=sid,
                role="author",
                remote="prgs",
                org="o",
                repo="r",
                kind="issue",
                number=7777,
            )
            with lock:
                outcomes.append(res.outcome)
        threads = [threading.Thread(target=contend, args=(f"c{i}",)) for i in range(1, 5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        self.assertEqual(outcomes.count("assigned"), 1)
        self.assertEqual(outcomes.count("wait"), 3)
    def test_terminal_lock_index(self) -> None:
        self.db.set_terminal_lock(
            remote="prgs",
            org="o",
            repo="r",
            terminal_pr=332,
            review_id="rev-1",
            decision="approve",
        )
        row = self.db.get_active_terminal_lock(remote="prgs", org="o", repo="r")
        self.assertIsNotNone(row)
        assert row is not None
        self.assertEqual(row["terminal_pr"], 332)
        self.assertEqual(row["status"], "active")
    def test_incident_links_not_work_items(self) -> None:
        link = self.db.upsert_incident_link(
            provider="sentry",
            provider_base_url="https://sentry.prgs.cc",
            provider_org="prgs",
            provider_project="gitea-tools-mcp",
            provider_issue_id="12345",
            gitea_org="Scaled-Tech-Consulting",
            gitea_repo="Gitea-Tools",
            gitea_issue_number=9001,
            fingerprint="fp-1",
        )
        self.assertEqual(link["gitea_issue_number"], 9001)
        found = self.db.get_incident_link_for_gitea_issue(
            gitea_org="Scaled-Tech-Consulting",
            gitea_repo="Gitea-Tools",
            gitea_issue_number=9001,
        )
        self.assertIsNotNone(found)
        # Linking does not create a work_item of incident kind
        with self.assertRaises(InvalidWorkKindError):
            self.db.upsert_work_item(
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
                kind="sentry",
                number=12345,
            )
    def test_heartbeat_and_release(self) -> None:
        self.db.upsert_session(session_id="s1", role="author")
        a = self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=1,
        )
        hb = self.db.heartbeat_lease(a.lease_id, session_id="s1")
        self.assertEqual(hb["lease_id"], a.lease_id)
        self.db.release_lease(a.lease_id, session_id="s1")
        # After release another session can claim
        self.db.upsert_session(session_id="s2", role="author")
        b = self.db.assign_and_lease(
            session_id="s2",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=1,
        )
        self.assertEqual(b.outcome, "assigned")
        self.assertEqual(b.session_id, "s2")
    def test_require_valid_assignment_rejects_stale_head(self) -> None:
        """Assignment must not authorize mutations after work-item head drifts."""
        self.db.upsert_session(session_id="s1", role="author")
        self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=42,
            expected_head_sha="head-v1",
            allowed_actions=("implement",),
        )
        # Head drifts after assignment
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=42,
            current_head_sha="head-v2",
        )
        with self.assertRaises(LeaseRequiredError) as ctx:
            self.db.require_valid_assignment(
                session_id="s1",
                remote="prgs",
                org="o",
                repo="r",
                kind="pr",
                number=42,
                action="implement",
            )
        self.assertIn("stale head", str(ctx.exception).lower())
    def test_require_valid_assignment_rejects_terminal_state(self) -> None:
        """Assignment must not authorize mutations after work item is merged/closed."""
        self.db.upsert_session(session_id="s1", role="author")
        self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=55,
            expected_head_sha="abc",
            allowed_actions=("implement",),
        )
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=55,
            state="merged",
            current_head_sha="abc",
        )
        with self.assertRaises(LeaseRequiredError) as ctx:
            self.db.require_valid_assignment(
                session_id="s1",
                remote="prgs",
                org="o",
                repo="r",
                kind="pr",
                number=55,
                action="implement",
            )
        self.assertIn("terminal", str(ctx.exception).lower())
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=56,
            state="open",
        )
        self.db.assign_and_lease(
            session_id="s1",
            role="author",
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=56,
            allowed_actions=("implement",),
        )
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="issue",
            number=56,
            state="closed",
        )
        with self.assertRaises(LeaseRequiredError):
            self.db.require_valid_assignment(
                session_id="s1",
                remote="prgs",
                org="o",
                repo="r",
                kind="issue",
                number=56,
                action="implement",
            )
    def test_pr_assignment_requires_expected_head_sha(self) -> None:
        """PR assign and mutation must fail closed without a head pin."""
        self.db.upsert_session(session_id="s1", role="author")
        with self.assertRaises(LeaseRequiredError) as ctx:
            self.db.assign_and_lease(
                session_id="s1",
                role="author",
                remote="prgs",
                org="o",
                repo="r",
                kind="pr",
                number=88,
                expected_head_sha=None,
                allowed_actions=("implement",),
            )
        self.assertIn("expected_head_sha", str(ctx.exception))
        # Legacy path: force an unpinned PR assignment into the DB, then
        # populate head and prove mutation is still rejected.
        import sqlite3
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=89,
            current_head_sha=None,
        )
        conn = sqlite3.connect(self.db_path)
        try:
            wid = conn.execute(
                "SELECT work_item_id FROM work_items WHERE kind='pr' AND number=89"
            ).fetchone()[0]
            conn.execute(
                """
                INSERT INTO sessions(session_id, role, started_at, last_heartbeat_at, status)
                VALUES ('legacy', 'author', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active')
                """
            )
            conn.execute(
                """
                INSERT INTO leases(
                    lease_id, work_item_id, session_id, role, phase,
                    expires_at, heartbeat_at, status
                ) VALUES (
                    'lease-legacy', ?, 'legacy', 'author', 'claimed',
                    '2099-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active'
                )
                """,
                (wid,),
            )
            conn.execute(
                """
                INSERT INTO assignments(
                    assignment_id, work_item_id, session_id, lease_id,
                    allowed_actions, forbidden_actions, expected_head_sha,
                    role, status, created_at
                ) VALUES (
                    'asn-legacy', ?, 'legacy', 'lease-legacy',
                    '["implement"]', '["merge"]', NULL,
                    'author', 'active', '2020-01-01T00:00:00Z'
                )
                """,
                (wid,),
            )
            conn.commit()
        finally:
            conn.close()
        self.db.upsert_work_item(
            remote="prgs",
            org="o",
            repo="r",
            kind="pr",
            number=89,
            current_head_sha="populated-head",
        )
        with self.assertRaises(LeaseRequiredError) as ctx2:
            self.db.require_valid_assignment(
                session_id="legacy",
                remote="prgs",
                org="o",
                repo="r",
                kind="pr",
                number=89,
                action="implement",
            )
        self.assertIn("expected_head_sha pin", str(ctx2.exception))
    def test_migrate_duplicate_null_scope_incident_links(self) -> None:
        """Legacy NULL-scope duplicates must migrate without UNIQUE crash."""
        import sqlite3
        from control_plane_db import ControlPlaneDB, ControlPlaneError
        # Build a v1-like table with nullable scope columns and insert dups
        # that collapse under normalization, then open ControlPlaneDB on it.
        path = os.path.join(self._tmp.name, "legacy_dups.sqlite3")
        conn = sqlite3.connect(path)
        try:
            conn.executescript(
                """
                CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
                CREATE TABLE incident_links (
                    link_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    provider TEXT NOT NULL,
                    provider_base_url TEXT,
                    provider_org TEXT,
                    provider_project TEXT,
                    provider_issue_id TEXT NOT NULL,
                    provider_short_id TEXT,
                    provider_permalink TEXT,
                    fingerprint TEXT,
                    gitea_org TEXT NOT NULL,
                    gitea_repo TEXT NOT NULL,
                    gitea_issue_number INTEGER NOT NULL,
                    linked_pr_numbers TEXT,
                    first_seen TEXT,
                    last_seen TEXT,
                    event_count INTEGER,
                    status TEXT NOT NULL DEFAULT 'open',
                    release_resolved_at TEXT,
                    last_sync_at TEXT,
                    UNIQUE (
                        provider, provider_base_url, provider_org,
                        provider_project, provider_issue_id
                    )
                );
                INSERT INTO incident_links(
                    provider, provider_base_url, provider_org, provider_project,
                    provider_issue_id, gitea_org, gitea_repo, gitea_issue_number, status
                ) VALUES
                    ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'),
                    ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open');
                """
            )
            # SQLite allows two NULL-scope rows with same provider/issue under UNIQUE.
            n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
            self.assertEqual(n, 2)
            conn.commit()
        finally:
            conn.close()
        db = ControlPlaneDB(path)
        conn2 = sqlite3.connect(path)
        try:
            n2 = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
            rows = conn2.execute(
                "SELECT provider_base_url, provider_org, provider_project, gitea_issue_number "
                "FROM incident_links"
            ).fetchall()
        finally:
            conn2.close()
        self.assertEqual(n2, 1)
        self.assertEqual(rows[0][0], "")
        self.assertEqual(rows[0][1], "")
        self.assertEqual(rows[0][2], "")
        self.assertEqual(rows[0][3], 10)
        # Touch to silence unused import in type checkers if needed
        self.assertTrue(issubclass(ControlPlaneError, Exception))
        del db
    def test_migrate_conflicting_duplicate_incident_links_fails_closed(self) -> None:
        """Conflicting Gitea targets for the same provider key must fail closed."""
        import sqlite3
        from control_plane_db import ControlPlaneDB, ControlPlaneError
        path = os.path.join(self._tmp.name, "legacy_conflict.sqlite3")
        conn = sqlite3.connect(path)
        try:
            conn.executescript(
                """
                CREATE TABLE incident_links (
                    link_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    provider TEXT NOT NULL,
                    provider_base_url TEXT,
                    provider_org TEXT,
                    provider_project TEXT,
                    provider_issue_id TEXT NOT NULL,
                    gitea_org TEXT NOT NULL,
                    gitea_repo TEXT NOT NULL,
                    gitea_issue_number INTEGER NOT NULL,
                    status TEXT NOT NULL DEFAULT 'open',
                    UNIQUE (
                        provider, provider_base_url, provider_org,
                        provider_project, provider_issue_id
                    )
                );
                INSERT INTO incident_links(
                    provider, provider_base_url, provider_org, provider_project,
                    provider_issue_id, gitea_org, gitea_repo, gitea_issue_number
                ) VALUES
                    ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 1),
                    ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 2);
                """
            )
            conn.commit()
        finally:
            conn.close()
        with self.assertRaises(ControlPlaneError) as ctx:
            ControlPlaneDB(path)
        self.assertIn("conflicting", str(ctx.exception).lower())
    def test_migrate_conflicting_observation_metadata_fails_closed(self) -> None:
        """Same provider key + same Gitea target but differing obs metadata must fail closed.
        Regression for silent data loss: migration used to keep lowest link_id and
        delete peers after comparing only Gitea targets (#619 RC3).
        """
        import sqlite3
        from control_plane_db import ControlPlaneDB, ControlPlaneError
        path = os.path.join(self._tmp.name, "legacy_meta_conflict.sqlite3")
        conn = sqlite3.connect(path)
        try:
            conn.executescript(
                """
                CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
                CREATE TABLE incident_links (
                    link_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    provider TEXT NOT NULL,
                    provider_base_url TEXT,
                    provider_org TEXT,
                    provider_project TEXT,
                    provider_issue_id TEXT NOT NULL,
                    provider_short_id TEXT,
                    provider_permalink TEXT,
                    fingerprint TEXT,
                    gitea_org TEXT NOT NULL,
                    gitea_repo TEXT NOT NULL,
                    gitea_issue_number INTEGER NOT NULL,
                    linked_pr_numbers TEXT,
                    first_seen TEXT,
                    last_seen TEXT,
                    event_count INTEGER,
                    status TEXT NOT NULL DEFAULT 'open',
                    release_resolved_at TEXT,
                    last_sync_at TEXT,
                    UNIQUE (
                        provider, provider_base_url, provider_org,
                        provider_project, provider_issue_id
                    )
                );
                INSERT INTO incident_links(
                    provider, provider_base_url, provider_org, provider_project,
                    provider_issue_id, provider_permalink, fingerprint,
                    gitea_org, gitea_repo, gitea_issue_number,
                    event_count, status, first_seen, last_seen
                ) VALUES
                    (
                        'sentry', NULL, NULL, NULL, 'dup-meta',
                        'https://sentry.example/issues/1', 'fingerprint-A',
                        'org', 'repo', 42,
                        1, 'open', '2026-01-01T00:00:00Z', '2026-01-01T01:00:00Z'
                    ),
                    (
                        'sentry', NULL, NULL, NULL, 'dup-meta',
                        'https://sentry.example/issues/1', 'fingerprint-B',
                        'org', 'repo', 42,
                        99, 'resolved', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z'
                    );
                """
            )
            n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
            self.assertEqual(n, 2)
            conn.commit()
        finally:
            conn.close()
        with self.assertRaises(ControlPlaneError) as ctx:
            ControlPlaneDB(path)
        msg = str(ctx.exception).lower()
        self.assertIn("conflicting", msg)
        self.assertIn("observation metadata", msg)
        # Rows must still be present — migration must not delete before failing.
        conn2 = sqlite3.connect(path)
        try:
            remaining = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
            fps = {
                r[0]
                for r in conn2.execute(
                    "SELECT fingerprint FROM incident_links ORDER BY link_id"
                ).fetchall()
            }
        finally:
            conn2.close()
        self.assertEqual(remaining, 2)
        self.assertEqual(fps, {"fingerprint-A", "fingerprint-B"})
    def test_incident_links_minimal_upsert_is_canonical(self) -> None:
        """Repeated minimal upserts must update one row (NULL-safe uniqueness)."""
        a = self.db.upsert_incident_link(
            provider="sentry",
            provider_issue_id="inc-1",
            gitea_org="org",
            gitea_repo="repo",
            gitea_issue_number=1,
        )
        b = self.db.upsert_incident_link(
            provider="sentry",
            provider_issue_id="inc-1",
            gitea_org="org",
            gitea_repo="repo",
            gitea_issue_number=2,
            # omit optional scope fields again
        )
        c = self.db.upsert_incident_link(
            provider="sentry",
            provider_issue_id="inc-1",
            gitea_org="org",
            gitea_repo="repo",
            gitea_issue_number=3,
            provider_base_url=None,
            provider_org="",
            provider_project="  ",
        )
        self.assertEqual(a["link_id"], b["link_id"])
        self.assertEqual(b["link_id"], c["link_id"])
        self.assertEqual(c["gitea_issue_number"], 3)
        self.assertEqual(c["provider_base_url"], "")
        self.assertEqual(c["provider_org"], "")
        self.assertEqual(c["provider_project"], "")
        import sqlite3
        conn = sqlite3.connect(self.db_path)
        try:
            n = conn.execute(
                "SELECT COUNT(*) FROM incident_links WHERE provider_issue_id = ?",
                ("inc-1",),
            ).fetchone()[0]
        finally:
            conn.close()
        self.assertEqual(n, 1)
 if __name__ == "__main__":
    unittest.main()
@@ -1,397 +0,0 @@
 """Head-scoped #332 review-decision locks (#620).
 Proves:
  * REQUEST_CHANGES on head A does not block mark_ready/submit on head B
  * APPROVE on head B is allowed after RC on head A (same open PR)
  * same-head duplicate terminal remains blocked
  * open-PR cleanup remains forbidden; assessment reports stale_by_head
  * historical mutations keep their head_sha (preserved ledger)
  * PR #619-style: old RC at 036b78e…, current head 2429d9d…, approve allowed
 """
 from __future__ import annotations
 import os
 import unittest
 from unittest.mock import patch
 import mcp_server
 import stale_review_decision_lock as srdl
 HEAD_A = "036b78e31ea5d036452b3a52e0e086b01e0c763f"
 HEAD_B = "2429d9d2e826e519eb8eb4ade45987c00838aefb"
 HEAD_C = "c" * 40
 def _lock(mutations=None, correction=False, ready_pr=None, ready_head=None, ready_action=None):
    return {
        "task": "review_pr",
        "remote": "prgs",
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "session_pid": os.getpid(),
        "session_profile": "prgs-reviewer",
        "session_profile_lock": "prgs-reviewer",
        "profile_identity": "prgs-reviewer",
        "final_review_decision_ready": False,
        "ready_pr_number": ready_pr,
        "ready_action": ready_action,
        "ready_expected_head_sha": ready_head,
        "ready_remote": "prgs" if ready_pr else None,
        "ready_org": "Scaled-Tech-Consulting" if ready_pr else None,
        "ready_repo": "Gitea-Tools" if ready_pr else None,
        "live_mutations": list(mutations or []),
        "correction_authorized": correction,
        "correction_reason": None,
    }
 RC_619_LEGACY = {
    "pr_number": 619,
    "action": "request_changes",
    "review_id": 410,
    "review_state": "request_changes",
    # no head_sha — pre-#620 durable ledger shape
 }
 RC_619_HEAD_A = {
    "pr_number": 619,
    "action": "request_changes",
    "review_id": 410,
    "review_state": "request_changes",
    "head_sha": HEAD_A,
 }
 APPROVE_619_HEAD_B = {
    "pr_number": 619,
    "action": "approve",
    "review_id": 411,
    "review_state": "approve",
    "head_sha": HEAD_B,
 }
 def _seed(mutations=None, **kwargs):
    import review_workflow_load
    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
    mcp_server._save_review_decision_lock(_lock(mutations, **kwargs))
    mcp_server.gitea_load_review_workflow()
 def _open_pr(pr_number=619, head=HEAD_B):
    return {
        "number": pr_number,
        "state": "open",
        "merged": False,
        "merged_at": None,
        "head": {"sha": head},
    }
 def _no_lease():
    return {"block": False, "reasons": [], "mutation_allowed": True}
 def _feedback(blocking=False, stale=False, success=True):
    return {
        "success": success,
        "has_blocking_change_requests": blocking,
        "review_feedback_stale": stale,
        "current_head_sha": HEAD_B,
    }
 class TestPureHeadScopeHelpers(unittest.TestCase):
    def test_mutation_head_prefers_recorded_sha(self):
        m = {"pr_number": 1, "head_sha": HEAD_A}
        self.assertEqual(srdl.mutation_head_sha(m), HEAD_A)
    def test_legacy_mutation_uses_ready_expected_for_same_pr(self):
        lock = _lock(
            [RC_619_LEGACY],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        self.assertEqual(srdl.mutation_head_sha(RC_619_LEGACY, lock), HEAD_A)
    def test_legacy_mutation_ignores_ready_for_other_pr(self):
        lock = _lock([RC_619_LEGACY], ready_pr=1, ready_head=HEAD_A)
        self.assertIsNone(srdl.mutation_head_sha(RC_619_LEGACY, lock))
    def test_prior_blocks_same_head_not_other_head(self):
        lock = _lock([RC_619_HEAD_A])
        self.assertTrue(
            srdl.prior_live_mutations_block_boundary(
                lock, pr_number=619, expected_head_sha=HEAD_A
            )
        )
        self.assertFalse(
            srdl.prior_live_mutations_block_boundary(
                lock, pr_number=619, expected_head_sha=HEAD_B
            )
        )
    def test_backfill_stamps_legacy_terminals(self):
        lock = _lock(
            [dict(RC_619_LEGACY)],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        srdl.backfill_terminal_heads_from_ready(lock)
        self.assertEqual(lock["live_mutations"][0]["head_sha"], HEAD_A)
 class TestHardStopHeadScope(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_rc_head_a_allows_mark_ready_head_b(self):
        _seed(
            [RC_619_HEAD_A],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(
                619, "mark_ready", expected_head_sha=HEAD_B
            ),
            [],
        )
    def test_rc_head_a_blocks_mark_ready_same_head(self):
        _seed(
            [RC_619_HEAD_A],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        reasons = mcp_server.terminal_review_hard_stop_reasons(
            619, "mark_ready", expected_head_sha=HEAD_A
        )
        self.assertTrue(reasons)
        self.assertIn("#332", reasons[0])
    def test_rc_head_a_blocks_other_pr(self):
        _seed([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A)
        reasons = mcp_server.terminal_review_hard_stop_reasons(
            700, "mark_ready", expected_head_sha=HEAD_B
        )
        self.assertTrue(reasons)
    def test_legacy_619_ledger_allows_new_head(self):
        """PR #619-style durable lock without mutation head_sha."""
        _seed(
            [RC_619_LEGACY],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(
                619, "mark_ready", expected_head_sha=HEAD_B
            ),
            [],
        )
 class TestMarkFinalAndRecord(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def _mark(self, pr, action, head, feedback=None):
        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
        ), patch.object(
            mcp_server,
            "gitea_get_pr_review_feedback",
            return_value=feedback or _feedback(blocking=False, stale=True),
        ), patch.object(
            mcp_server,
            "gitea_check_pr_eligibility",
            return_value={"eligible": True, "head_sha": head},
        ):
            return mcp_server.gitea_mark_final_review_decision(
                pr_number=pr,
                action=action,
                expected_head_sha=head,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
    def test_rc_then_rc_on_new_head(self):
        _seed(
            [RC_619_HEAD_A],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        # Prior RC at head A is unresolved but head moved → feedback stale.
        res = self._mark(
            619,
            "request_changes",
            HEAD_B,
            feedback=_feedback(blocking=True, stale=True),
        )
        self.assertTrue(res.get("marked_ready"), res)
        lock = mcp_server._load_review_decision_lock()
        # Historical head A preserved on mutation after backfill.
        heads = {
            m.get("head_sha")
            for m in lock["live_mutations"]
            if m.get("action") == "request_changes"
        }
        self.assertIn(HEAD_A, heads)
        self.assertEqual(lock["ready_expected_head_sha"], HEAD_B)
    def test_rc_then_approve_on_new_head(self):
        _seed(
            [RC_619_HEAD_A],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        res = self._mark(619, "approve", HEAD_B)
        self.assertTrue(res.get("marked_ready"), res)
        self.assertTrue(res.get("head_scoped"))
    def test_same_head_rc_still_blocked_by_hard_stop(self):
        _seed(
            [RC_619_HEAD_A],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        res = self._mark(619, "approve", HEAD_A)
        self.assertFalse(res.get("marked_ready"))
        self.assertTrue(any("#332" in r for r in res.get("reasons") or []))
    def test_pr619_style_legacy_lock_approve_on_current_head(self):
        _seed(
            [RC_619_LEGACY],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        res = self._mark(619, "approve", HEAD_B)
        self.assertTrue(res.get("marked_ready"), res)
        lock = mcp_server._load_review_decision_lock()
        # Backfill should have stamped HEAD_A onto the legacy mutation.
        self.assertEqual(lock["live_mutations"][0].get("head_sha"), HEAD_A)
        self.assertEqual(lock["ready_expected_head_sha"], HEAD_B)
    def test_record_live_mutation_stores_head(self):
        _seed(ready_pr=619, ready_head=HEAD_B, ready_action="approve")
        lock = mcp_server._load_review_decision_lock()
        lock["final_review_decision_ready"] = True
        lock["ready_pr_number"] = 619
        lock["ready_expected_head_sha"] = HEAD_B
        mcp_server._save_review_decision_lock(lock)
        mcp_server.record_live_review_mutation(619, "approve", review_id=99)
        stored = mcp_server._load_review_decision_lock()["live_mutations"][-1]
        self.assertEqual(stored["head_sha"], HEAD_B)
        self.assertEqual(stored["pr_number"], 619)
    def test_submit_gate_allows_new_head_after_prior_terminal(self):
        """check_review_decision_gate must not block different-head submit."""
        mutations = [RC_619_HEAD_A]
        _seed(
            mutations,
            ready_pr=619,
            ready_head=HEAD_B,
            ready_action="approve",
        )
        lock = mcp_server._load_review_decision_lock()
        lock["final_review_decision_ready"] = True
        lock["ready_pr_number"] = 619
        lock["ready_action"] = "approve"
        lock["ready_expected_head_sha"] = HEAD_B
        lock["ready_remote"] = "prgs"
        lock["ready_org"] = "Scaled-Tech-Consulting"
        lock["ready_repo"] = "Gitea-Tools"
        mcp_server._save_review_decision_lock(lock)
        reasons = mcp_server.check_review_decision_gate(
            619,
            "approve",
            final_review_decision_ready=True,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertEqual(reasons, [], reasons)
    def test_submit_gate_blocks_same_head_duplicate(self):
        mutations = [RC_619_HEAD_A]
        _seed(
            mutations,
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        lock = mcp_server._load_review_decision_lock()
        lock["final_review_decision_ready"] = True
        lock["ready_pr_number"] = 619
        lock["ready_action"] = "request_changes"
        lock["ready_expected_head_sha"] = HEAD_A
        lock["ready_remote"] = "prgs"
        lock["ready_org"] = "Scaled-Tech-Consulting"
        lock["ready_repo"] = "Gitea-Tools"
        mcp_server._save_review_decision_lock(lock)
        reasons = mcp_server.check_review_decision_gate(
            619,
            "request_changes",
            final_review_decision_ready=True,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertTrue(reasons)
 class TestAssessmentOpenPrHead(unittest.TestCase):
    def test_open_pr_stale_by_head_no_cleanup(self):
        lock = _lock(
            [RC_619_HEAD_A],
            ready_pr=619,
            ready_head=HEAD_A,
            ready_action="request_changes",
        )
        a = srdl.assess_stale_review_decision_lock(
            lock, pr_live=_open_pr(619, HEAD_B)
        )
        self.assertTrue(a["has_lock"])
        self.assertFalse(a["is_moot"])
        self.assertFalse(a["cleanup_allowed"])
        self.assertTrue(a["stale_by_head"])
        self.assertTrue(a["fresh_review_on_current_head_allowed"])
        self.assertEqual(a["locked_head_sha"], HEAD_A)
        self.assertEqual(a["current_pr_head_sha"], HEAD_B)
    def test_open_pr_same_head_no_fresh(self):
        lock = _lock([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A)
        a = srdl.assess_stale_review_decision_lock(
            lock, pr_live=_open_pr(619, HEAD_A)
        )
        self.assertFalse(a["stale_by_head"])
        self.assertFalse(a["fresh_review_on_current_head_allowed"])
        self.assertFalse(a["cleanup_allowed"])
    def test_historical_mutation_preserved_in_summary(self):
        lock = _lock(
            [RC_619_HEAD_A, APPROVE_619_HEAD_B],
            ready_pr=619,
            ready_head=HEAD_B,
            ready_action="approve",
        )
        summary = srdl.lock_summary(lock)
        self.assertEqual(summary["live_mutations_count"], 2)
        self.assertEqual(summary["last_terminal"]["head_sha"], HEAD_B)
        self.assertEqual(summary["locked_head_sha"], HEAD_B)
 if __name__ == "__main__":
    unittest.main()
@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
        self.assertTrue(report["recovery"])
 class TestLiveRemoteParity(unittest.TestCase):
    """#610: parity must account for the live remote master, not just local.
    The daemon can be stale relative to the live remote target while the local
    checkout HEAD still matches the daemon's startup commit, so local parity
    reports green even though a mutation would run against outdated code.
    """
    SHA_C = "c" * 40
    def test_distinguishes_three_shas(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertEqual(res["daemon_start_head"], SHA_A)
        self.assertEqual(res["local_head"], SHA_A)
        self.assertEqual(res["live_remote_head"], SHA_B)
    def test_mutation_safe_only_when_all_three_match(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
        self.assertTrue(res["mutation_safe"])
        self.assertTrue(res["live_known"])
        self.assertFalse(res["live_stale"])
    def test_live_stale_when_remote_advanced_past_daemon(self):
        # Local checkout still matches the daemon start (local parity green),
        # but the live remote master has advanced -> daemon is live-stale.
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertTrue(res["in_parity"])  # local parity still green
        self.assertTrue(res["live_stale"])
        self.assertFalse(res["mutation_safe"])
        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
        # Non-goal: unfetchable live remote must not be treated as stale for
        # read-only, but a mutation-safe claim fails closed.
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
        self.assertFalse(res["live_known"])
        self.assertFalse(res["mutation_safe"])
        self.assertFalse(res["live_stale"])
        self.assertTrue(res["in_parity"])
    def test_default_live_remote_preserves_legacy_shape(self):
        # Callers that do not supply a live head keep the pre-#610 behavior:
        # in-parity, not live-stale, no live-derived block.
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        self.assertFalse(res["live_stale"])
        self.assertEqual(mp.parity_block_reasons(res), [])
 class TestLiveStaleBlockAndReport(unittest.TestCase):
    """#610: live-staleness must block mutations and surface a typed blocker."""
    def test_live_stale_produces_block_reasons(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertTrue(mp.parity_block_reasons(res))
    def test_disable_env_suppresses_live_stale_block(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
            self.assertEqual(mp.parity_block_reasons(res), [])
    def test_resolver_disagreement_returns_typed_blocker(self):
        # Parity says local-green, resolver says restart required -> disagreement
        # is a typed, fail-closed blocker naming the resolver as authoritative.
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
        self.assertIsNotNone(blocker)
        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
        self.assertTrue(blocker["restart_required"])
        self.assertTrue(blocker["resolver_authoritative"])
    def test_no_disagreement_when_resolver_agrees(self):
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        self.assertIsNone(
            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
    def test_live_stale_report_names_live_remote(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        report = mp.parity_report(res)
        self.assertEqual(report["live_remote_head"], SHA_B)
        self.assertTrue(report["restart_required"])
 class TestReadGitHead(unittest.TestCase):
    def test_test_override_takes_precedence(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,78 @@ class TestReadGitHead(unittest.TestCase):
            self.assertIsNone(mp.read_git_head(""))
 class TestReadRemoteMasterHead(unittest.TestCase):
    """#610: live remote master head reader (env-overridable, fails to None)."""
    def test_test_override_takes_precedence(self):
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
    def test_blank_override_is_none(self):
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
    def test_unfetchable_remote_is_none(self):
        # No override; a bogus root/remote must fail closed to None, never raise.
        env = {k: v for k, v in os.environ.items()
               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
        with patch.dict(os.environ, env, clear=True):
            self.assertIsNone(
                mp.read_remote_master_head("/nonexistent", remote="nope"))
 class TestRemoteHeadCache(unittest.TestCase):
    """#610: live remote reads are cached with a TTL to stay off the network.
    The parity gate runs on every mutation and every runtime-context read, so an
    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
    """
    def setUp(self):
        mp._clear_remote_head_cache()
        env = {k: v for k, v in os.environ.items()
               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
        self._env = patch.dict(os.environ, env, clear=True)
        self._env.start()
        self.addCleanup(self._env.stop)
        self.addCleanup(mp._clear_remote_head_cache)
    def _fake_run(self, sha):
        class _R:
            returncode = 0
            stdout = f"{sha}\trefs/heads/master\n"
        calls = {"n": 0}
        def run(*args, **kwargs):
            calls["n"] += 1
            return _R()
        return run, calls
    def test_second_call_within_ttl_uses_cache(self):
        run, calls = self._fake_run(SHA_B)
        with patch.object(mp.subprocess, "run", run):
            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
        self.assertEqual(a, SHA_B)
        self.assertEqual(b, SHA_B)
        self.assertEqual(calls["n"], 1)
    def test_zero_ttl_bypasses_cache(self):
        run, calls = self._fake_run(SHA_B)
        with patch.object(mp.subprocess, "run", run):
            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
        self.assertEqual(calls["n"], 2)
    def test_env_override_never_touches_subprocess(self):
        run, calls = self._fake_run(SHA_B)
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
            with patch.object(mp.subprocess, "run", run):
                self.assertEqual(
                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
        self.assertEqual(calls["n"], 0)
 class TestServerWiring(unittest.TestCase):
    """Integration with the gate choke point in the server namespace."""
@@ -105,6 +266,13 @@ class TestServerWiring(unittest.TestCase):
        self._saved = self.srv._STARTUP_PARITY
        self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                    "startup_head": SHA_A}
        # Keep the live-remote read hermetic (no real ls-remote network call):
        # default the live master to the daemon start so parity is fully green
        # unless a test overrides the live head explicitly (#610).
        self._live_patch = patch.dict(
            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
        self._live_patch.start()
        self.addCleanup(self._live_patch.stop)
    def tearDown(self):
        self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +315,36 @@ class TestServerWiring(unittest.TestCase):
            self.assertTrue(out["in_parity"])
            self.assertNotIn("report", out)
    # --- #610: live-remote wiring -------------------------------------------
    def test_live_stale_blocks_mutation_though_local_green(self):
        # Local checkout matches the daemon start (local parity green) but the
        # live remote master has advanced -> mutations must fail closed.
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
            self.assertTrue(
                self.srv._master_parity_block("gitea.pr.create"))
    def test_assess_tool_exposes_three_distinct_shas(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            out = self.srv.gitea_assess_master_parity(remote="prgs")
            self.assertEqual(out["daemon_start_head"], SHA_A)
            self.assertEqual(out["local_head"], SHA_A)
            self.assertEqual(out["live_remote_head"], SHA_B)
            self.assertTrue(out["live_stale"])
            self.assertFalse(out["mutation_safe"])
            self.assertIn("report", out)
    def test_assess_tool_mutation_safe_when_all_three_match(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
            out = self.srv.gitea_assess_master_parity(remote="prgs")
            self.assertTrue(out["mutation_safe"])
            self.assertFalse(out["live_stale"])
            self.assertNotIn("report", out)
 if __name__ == "__main__":
    unittest.main()
@@ -1,208 +0,0 @@
 import unittest
 import gitea_mcp_server
 import mcp_namespace_health
 import review_merge_state_machine as rmsm
 class TestMcpNamespaceHealth(unittest.TestCase):
    def setUp(self):
        gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
    def test_registered_tool_but_live_eof_blocks_merge(self):
        result = mcp_namespace_health.classify_namespace_probe(
            "gitea-reviewer",
            required_tool="gitea_whoami",
            registered_tools=["gitea_whoami", "gitea_list_profiles"],
            probe_result={
                "success": False,
                "error": "client is closing: EOF",
            },
            process={
                "pid": 4321,
                "profile": "prgs-reviewer",
                "env": {
                    "GITEA_MCP_PROFILE": "prgs-reviewer",
                    "GITEA_TOKEN": "must-not-leak",
                },
            },
            config_path="/Users/jasonwalker/.gemini/config/mcp_config.json",
            probe_source="client_namespace",
        )
        self.assertFalse(result["healthy"])
        self.assertEqual(result["error_type"], "namespace_eof")
        self.assertTrue(result["required_tool_registered"])
        self.assertFalse(result["required_tool_callable"])
        self.assertTrue(result["blocks_merge_workflow"])
        self.assertFalse(result["ide_namespace_proven"])
        self.assertEqual(result["probe_source"], "client_namespace")
        self.assertEqual(result["diagnostics"]["process_pid"], 4321)
        self.assertEqual(result["diagnostics"]["profile"], "prgs-reviewer")
        self.assertEqual(
            result["diagnostics"]["env"]["GITEA_MCP_PROFILE"],
            "prgs-reviewer",
        )
        self.assertNotIn("GITEA_TOKEN", result["diagnostics"]["env"])
        self.assertIn("gitea-reviewer", result["reasons"][0])
        self.assertIn("gitea_whoami", result["reasons"][0])
    def test_successful_client_namespace_invocation_is_ide_proven(self):
        result = mcp_namespace_health.classify_namespace_probe(
            "gitea-author",
            registered_tools=["gitea_whoami"],
            probe_result={"success": True, "result": {"authenticated": True}},
            process={"pid": 1234, "profile": "prgs-author"},
            config_path="/tmp/mcp_config.json",
            probe_source="client_namespace",
        )
        self.assertTrue(result["healthy"])
        self.assertTrue(result["required_tool_registered"])
        self.assertTrue(result["required_tool_callable"])
        self.assertTrue(result["ide_namespace_proven"])
        self.assertFalse(result["blocks_merge_workflow"])
        self.assertIsNone(result["error_type"])
    def test_offline_spawn_success_is_not_ide_proof(self):
        result = mcp_namespace_health.classify_namespace_probe(
            "gitea-merger",
            registered_tools=["gitea_whoami"],
            probe_result={"success": True, "result": {}},
            process={"pid": 99, "profile": "prgs-merger"},
            probe_source="offline_spawn",
        )
        self.assertTrue(result["healthy"])
        self.assertFalse(result["ide_namespace_proven"])
        self.assertFalse(result["blocks_merge_workflow"])
        self.assertTrue(
            any("offline_spawn" in r for r in result["reasons"])
        )
    def test_registered_missing_required_tool_fails_before_probe_success(self):
        result = mcp_namespace_health.classify_namespace_probe(
            "gitea-tools",
            registered_tools=["gitea_whoami"],
            probe_result={"success": True, "result": {}},
            probe_source="client_namespace",
        )
        self.assertFalse(result["healthy"])
        self.assertEqual(result["required_tool"], "gitea_list_profiles")
        self.assertFalse(result["required_tool_registered"])
        self.assertEqual(result["error_type"], "tool_missing")
    def test_server_tool_exposes_same_assessment_and_records_session(self):
        result = gitea_mcp_server.gitea_assess_mcp_namespace_health(
            "gitea-merger",
            registered_tools=["gitea_whoami"],
            probe_result={"success": False, "error": "transport closed"},
            process={"pid": 9876, "profile": "prgs-merger"},
            probe_source="client_namespace",
        )
        self.assertFalse(result["success"])
        self.assertEqual(result["error_type"], "namespace_eof")
        self.assertEqual(result["namespace"], "gitea-merger")
        self.assertEqual(result["diagnostics"]["process_pid"], 9876)
        self.assertIn("gitea-merger", gitea_mcp_server._LIVE_NAMESPACE_HEALTH)
        gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
        self.assertTrue(gate)
        self.assertTrue(any("gitea-merger" in r for r in gate))
 class TestLiveNamespaceBlocksMerge(unittest.TestCase):
    """AC5: a broken live namespace must hard-block the review/merge state machine."""
    def setUp(self):
        gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
    def _merge_ready_completion(self):
        # Completion through PRE_MERGE_RECHECK is the state where a clean merge
        # is otherwise allowed (see test_merge_allowed_with_pre_merge_gates).
        idx = rmsm.REVIEW_MERGE_STATES.index("PRE_MERGE_RECHECK")
        return {state: True for state in rmsm.REVIEW_MERGE_STATES[: idx + 1]}
    def _all_gates(self):
        return {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
    def test_assess_workflow_blockers_flags_live_namespace_broken(self):
        clean = rmsm.assess_workflow_blockers()
        self.assertFalse(clean["block"])
        broken = rmsm.assess_workflow_blockers(live_namespace_broken=True)
        self.assertTrue(broken["block"])
        self.assertTrue(any("namespace" in r.lower() for r in broken["reasons"]))
    def test_can_merge_blocks_even_when_all_gates_pass(self):
        # Without the namespace blocker a fully-complete workflow can merge...
        allowed = rmsm.can_merge(
            self._merge_ready_completion(), pre_merge_gates=self._all_gates()
        )
        self.assertTrue(allowed["allowed"])
        # ...but a broken live namespace overrides every satisfied gate.
        blocked = rmsm.can_merge(
            self._merge_ready_completion(),
            pre_merge_gates=self._all_gates(),
            live_namespace_broken=True,
        )
        self.assertTrue(blocked["block"])
        self.assertFalse(blocked["allowed"])
    def test_workflow_status_forwards_namespace_blocker(self):
        status = rmsm.workflow_status(
            self._merge_ready_completion(), live_namespace_broken=True
        )
        self.assertFalse(status["merge_allowed"])
        self.assertFalse(status["approve_allowed"])
    def test_server_tool_blocks_merge_on_live_namespace_broken(self):
        result = gitea_mcp_server.gitea_assess_review_merge_state_machine(
            state_completion=self._merge_ready_completion(),
            pre_merge_gates=self._all_gates(),
            live_namespace_broken=True,
        )
        self.assertTrue(result["blockers"]["block"])
        self.assertFalse(result["merge"]["allowed"])
    def test_classify_verdict_bridges_into_merge_block(self):
        # The classify verdict is the intended feed for live_namespace_broken.
        verdict = mcp_namespace_health.classify_namespace_probe(
            "gitea-merger",
            registered_tools=["gitea_whoami", "gitea_adopt_merger_pr_lease"],
            probe_result={"success": False, "error": "client is closing: EOF"},
            process={"pid": 555, "profile": "prgs-merger"},
            probe_source="client_namespace",
        )
        self.assertTrue(verdict["blocks_merge_workflow"])
        blocked = rmsm.can_merge(
            self._merge_ready_completion(),
            pre_merge_gates=self._all_gates(),
            live_namespace_broken=verdict["blocks_merge_workflow"],
        )
        self.assertFalse(blocked["allowed"])
    def test_offline_spawn_does_not_authorize_mutation_gate(self):
        gitea_mcp_server.gitea_assess_mcp_namespace_health(
            "gitea-merger",
            registered_tools=["gitea_whoami"],
            probe_result={"success": True, "result": {}},
            probe_source="offline_spawn",
        )
        gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
        self.assertTrue(gate)
        self.assertTrue(any("offline_spawn" in r or "client_namespace" in r for r in gate))
    def test_client_namespace_healthy_clears_mutation_gate(self):
        gitea_mcp_server.gitea_assess_mcp_namespace_health(
            "gitea-merger",
            registered_tools=["gitea_whoami"],
            probe_result={"success": True, "result": {}},
            probe_source="client_namespace",
        )
        self.assertEqual(gitea_mcp_server._live_namespace_health_gate("merge_pr"), [])
 if __name__ == "__main__":
    unittest.main()
@@ -1,287 +0,0 @@
 """Tests for preserving and resuming prepared Gitea review drafts (#609)."""
 from __future__ import annotations
 import os
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import patch, MagicMock
 sys_path_root = str(Path(__file__).resolve().parent.parent)
 if sys_path_root not in sys.path:
    sys.path.insert(0, sys_path_root)
 import mcp_session_state
 import gitea_mcp_server
 import reviewer_pr_lease
 class TestReviewDrafts(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self.state_dir = self._tmpdir.name
        self._env = patch.dict(
            os.environ,
            {
                mcp_session_state.STATE_DIR_ENV: self.state_dir,
                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
                "GITEA_MCP_PROFILE": "prgs-reviewer",
                "GITEA_PROFILE_NAME": "prgs-reviewer",
                "PYTEST_CURRENT_TEST": "1",  # skip runtime stale checks
            },
            clear=False,
        )
        self._env.start()
        # Mock workspace binding to pass in test context
        self._nwb_patch = patch(
            "gitea_mcp_server.nwb.assess_namespace_mutation_workspace",
            return_value={"block": False, "mutation_workspace": "/tmp/test-worktree"},
        )
        self._nwb_patch.start()
        # Mock authentication headers
        self._auth_patch = patch(
            "gitea_mcp_server.get_auth_header",
            return_value="Bearer mock-token",
        )
        self._auth_patch.start()
        self._username_patch = patch(
            "gitea_mcp_server._authenticated_username",
            return_value="sysadmin",
        )
        self._username_patch.start()
        # Clear/Mock local session lease in-memory state
        reviewer_pr_lease.clear_session_lease()
    def tearDown(self):
        self._username_patch.stop()
        self._auth_patch.stop()
        self._nwb_patch.stop()
        self._env.stop()
        self._tmpdir.cleanup()
        reviewer_pr_lease.clear_session_lease()
    @patch("gitea_mcp_server.api_request")
    def test_save_draft_success(self, mock_api):
        print("GITEA_MCP_SERVER FILE:", gitea_mcp_server.__file__)
        print("DIR OF GITEA_MCP_SERVER:", dir(gitea_mcp_server))
        mock_api.return_value = {
            "head": {"sha": "headsha1234567890"},
            "base": {"ref": "master", "sha": "basesha0987654321"},
        }
        res = gitea_mcp_server.gitea_save_review_draft(
            pr_number=587,
            action="approve",
            body="Good changes",
            expected_head_sha="headsha1234567890",
            validation_commands="pytest tests/",
            validation_results="all pass",
            blocker_reason="stale daemon",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        self.assertTrue(res.get("success"))
        self.assertEqual(res.get("head_sha"), "headsha1234567890")
        # Load draft and verify fields
        draft = mcp_session_state.load_state(
            kind=mcp_session_state.KIND_REVIEW_DRAFT,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            profile_identity="prgs-reviewer",
        )
        self.assertIsNotNone(draft)
        self.assertEqual(draft.get("pr_number"), 587)
        self.assertEqual(draft.get("action"), "approve")
        self.assertEqual(draft.get("body"), "Good changes")
        self.assertEqual(draft.get("validation_commands"), "pytest tests/")
        self.assertEqual(draft.get("validation_results"), "all pass")
        self.assertEqual(draft.get("blocker_reason"), "stale daemon")
        self.assertEqual(os.path.realpath(draft.get("worktree_path")), os.path.realpath("/tmp/test-worktree"))
    @patch("gitea_mcp_server.api_request")
    @patch("gitea_mcp_server._fetch_pr_comments")
    @patch("gitea_mcp_server.gitea_submit_pr_review")
    def test_resume_draft_and_submit_success(self, mock_submit, mock_comments, mock_api):
        # 1. Save draft
        mock_api.return_value = {
            "state": "open",
            "head": {"sha": "headsha1234567890"},
            "base": {"ref": "master", "sha": "basesha0987654321"},
        }
        gitea_mcp_server.gitea_save_review_draft(
            pr_number=587,
            action="approve",
            body="Good changes",
            expected_head_sha="headsha1234567890",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        # Seed local session lease
        reviewer_pr_lease.record_session_lease({
            "pr_number": 587,
            "session_id": "session-1234",
        })
        # Mock Gitea comments to return active reviewer lease owned by us
        mock_comments.return_value = [
            {
                "id": 1,
                "user": {"username": "sysadmin"},
                "body": (
                    "<!-- mcp-review-lease:v1 -->\n"
                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
                    "pr: #587\n"
                    "reviewer_identity: sysadmin\n"
                    "profile: prgs-reviewer\n"
                    "session_id: session-1234\n"
                    "phase: claimed\n"
                ),
            }
        ]
        mock_submit.return_value = {"performed": True, "success": True}
        # 2. Resume draft
        res = gitea_mcp_server.gitea_resume_review_draft(
            pr_number=587,
            submit=True,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        self.assertTrue(res.get("success"))
        self.assertTrue(res.get("marked_ready"))
        self.assertTrue(res.get("submitted"))
        mock_submit.assert_called_once()
        # Verify draft was deleted after successful submit
        draft = mcp_session_state.load_state(
            kind=mcp_session_state.KIND_REVIEW_DRAFT,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            profile_identity="prgs-reviewer",
        )
        self.assertIsNone(draft)
    @patch("gitea_mcp_server.api_request")
    @patch("gitea_mcp_server._fetch_pr_comments")
    def test_resume_draft_fails_checks(self, mock_comments, mock_api):
        # Save a draft
        mock_api.return_value = {
            "state": "open",
            "head": {"sha": "headsha1234567890"},
            "base": {"ref": "master", "sha": "basesha0987654321"},
        }
        gitea_mcp_server.gitea_save_review_draft(
            pr_number=587,
            action="approve",
            body="Good changes",
            expected_head_sha="headsha1234567890",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        # Seed local session lease
        reviewer_pr_lease.record_session_lease({
            "pr_number": 587,
            "session_id": "session-1234",
        })
        # Mock Gitea comments to return active reviewer lease owned by us
        mock_comments.return_value = [
            {
                "id": 1,
                "user": {"username": "sysadmin"},
                "body": (
                    "<!-- mcp-review-lease:v1 -->\n"
                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
                    "pr: #587\n"
                    "reviewer_identity: sysadmin\n"
                    "profile: prgs-reviewer\n"
                    "session_id: session-1234\n"
                    "phase: claimed\n"
                ),
            }
        ]
        # Case A: PR closed
        mock_api.return_value = {
            "state": "closed",
            "head": {"sha": "headsha1234567890"},
            "base": {"ref": "master", "sha": "basesha0987654321"},
        }
        res = gitea_mcp_server.gitea_resume_review_draft(
            pr_number=587,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        self.assertFalse(res.get("success"))
        self.assertIn("PR #587 is not open", res.get("reasons")[0])
        # Case B: Head changed
        mock_api.return_value = {
            "state": "open",
            "head": {"sha": "headsha_NEW"},
            "base": {"ref": "master", "sha": "basesha0987654321"},
        }
        res = gitea_mcp_server.gitea_resume_review_draft(
            pr_number=587,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        self.assertFalse(res.get("success"))
        self.assertIn("PR head SHA changed", res.get("reasons")[0])
        # Case C: Target branch changed
        mock_api.return_value = {
            "state": "open",
            "head": {"sha": "headsha1234567890"},
            "base": {"ref": "master", "sha": "basesha_NEW"},
        }
        res = gitea_mcp_server.gitea_resume_review_draft(
            pr_number=587,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/test-worktree",
        )
        self.assertFalse(res.get("success"))
        self.assertIn("target branch SHA changed", res.get("reasons")[0])
        # Case D: Worktree mismatch
        mock_api.return_value = {
            "state": "open",
            "head": {"sha": "headsha1234567890"},
            "base": {"ref": "master", "sha": "basesha0987654321"},
        }
        res = gitea_mcp_server.gitea_resume_review_draft(
            pr_number=587,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path="/tmp/different-worktree",
        )
        self.assertFalse(res.get("success"))
        self.assertIn("worktree binding mismatch", res.get("reasons")[0])