docs: ADR for stable MCP control runtime vs dev runtime (#615 ) (#616 )

Co-authored-by: jcwalker3 <[email protected]>
fix: adopt_merger_pr_lease requires merger role
2026-07-16 14:58:43 -05:00 · 2026-07-16 15:52:58 -04:00 · 2026-07-16 03:23:03 -05:00 · 2026-07-16 01:21:18 -04:00 · 2026-07-16 00:57:30 -04:00 · 2026-07-15 23:48:06 -05:00
26 changed files with 10260 additions and 198 deletions
@@ -55,3 +55,12 @@ GITEA_MCP_PROFILE=prgs
 # GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
 # GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
 # GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override
 # Durable HMAC key for irrecoverable decision-lock provenance artifacts (#709 F4).
 # REQUIRED in production native MCP processes that mint or verify recovery
 # authorization (reconciler + merger must share the same key). Hex (preferred),
 # base64, or utf-8 literal (>=16 bytes). Never generate an ephemeral per-process
 # key — cross-process / post-restart verification would fail closed.
 # GITEA_IRRECOVERABLE_AUTH_HMAC_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
 # Optional key version string bound into the signature (for rotation).
 # GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION=v1
@@ -7,6 +7,19 @@ from typing import Any
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
 # Evidence / preservation branches must never be removed by cleanup tools
 # (e.g. chore/issue-681-preserve-review-session-wip).
 _PRESERVATION_MARKERS = ("preserve", "preservation", "evidence")
 def is_preservation_or_evidence_branch(branch: str | None) -> bool:
    """Return True when *branch* is a preservation/evidence ref that must stay."""
    if not branch:
        return False
    name = str(branch).lower()
    return any(marker in name for marker in _PRESERVATION_MARKERS)
 _RAW_BRANCH_DELETE_PATTERNS = (
    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
@@ -72,6 +85,11 @@ def assess_merged_pr_branch_cleanup(
        reasons.append("PR head branch is missing")
    if head_branch in protected:
        reasons.append(f"branch '{head_branch}' is protected")
    if is_preservation_or_evidence_branch(head_branch):
        reasons.append(
            f"branch '{head_branch}' is a preservation/evidence branch and "
            "cannot be deleted through merged-PR cleanup"
        )
    if head_branch in open_pr_heads:
        reasons.append("an open PR still references this head branch")
    if head_on_target is False:
@@ -96,3 +114,490 @@ def assess_merged_pr_branch_cleanup(
        "block_reasons": reasons,
        "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
    }
 # ---------------------------------------------------------------------------
 # #687 remediation: post-delete readback + active ownership protection
 # ---------------------------------------------------------------------------
 READBACK_NOT_FOUND = "not_found"
 READBACK_EXISTS = "exists"
 READBACK_AUTHENTICATION = "authentication_error"
 READBACK_AUTHORIZATION = "authorization_error"
 READBACK_TRANSPORT = "transport_error"
 READBACK_UNEXPECTED = "unexpected_response"
 READBACK_AMBIGUOUS_404 = "ambiguous_not_found"
 # Scope of a 404: only branch-scoped absence may set verified_absent.
 NOT_FOUND_SCOPE_BRANCH = "branch"
 NOT_FOUND_SCOPE_REPOSITORY = "repository"
 NOT_FOUND_SCOPE_HOST = "host"
 NOT_FOUND_SCOPE_UNKNOWN = "unknown"
 ERROR_CLASS_AUTHENTICATION = "authentication"
 ERROR_CLASS_AUTHORIZATION = "authorization"
 ERROR_CLASS_TRANSPORT = "transport"
 ERROR_CLASS_UNEXPECTED = "unexpected"
 OWNERSHIP_CATEGORY_AUTHOR_SESSION = "author_session"
 OWNERSHIP_CATEGORY_AUTHOR_LEASE = "author_lease"
 OWNERSHIP_CATEGORY_REVIEWER_LEASE = "reviewer_lease"
 OWNERSHIP_CATEGORY_MERGER_LEASE = "merger_lease"
 OWNERSHIP_CATEGORY_CONTROLLER_LEASE = "controller_lease"
 OWNERSHIP_CATEGORY_RECONCILER_LEASE = "reconciler_lease"
 OWNERSHIP_CATEGORY_WORKTREE_BINDING = "worktree_binding"
 OWNERSHIP_CATEGORY_INVENTORY_ERROR = "ownership_inventory_error"
 _ROLE_TO_OWNERSHIP_CATEGORY = {
    "author": OWNERSHIP_CATEGORY_AUTHOR_LEASE,
    "reviewer": OWNERSHIP_CATEGORY_REVIEWER_LEASE,
    "merger": OWNERSHIP_CATEGORY_MERGER_LEASE,
    "controller": OWNERSHIP_CATEGORY_CONTROLLER_LEASE,
    "reconciler": OWNERSHIP_CATEGORY_RECONCILER_LEASE,
 }
 _ACTIVE_OWNERSHIP_STATUSES = frozenset(
    {"active", "live", "claimed", "in_progress", "working", "pushing", "pushed"}
 )
 _TERMINAL_OWNERSHIP_STATUSES = frozenset(
    {"released", "abandoned", "done", "blocked", "terminal", "closed"}
 )
 _EXPIRED_STATUSES = frozenset({"expired"})
 _STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"})
 def _norm_str(value: Any) -> str:
    return str(value or "").strip()
 def normalize_host(host: str | None) -> str:
    """Normalize a host identity for ownership matching (no credentials)."""
    text = _norm_str(host).lower()
    for prefix in ("https://", "http://"):
        if text.startswith(prefix):
            text = text[len(prefix) :]
    # Drop path/query if a full URL slipped through.
    text = text.split("/", 1)[0]
    text = text.split("?", 1)[0]
    return text.rstrip(".")
 def ownership_category_for_role(role: str | None) -> str:
    """Map a role kind to a non-secret ownership category label."""
    key = _norm_str(role).lower()
    return _ROLE_TO_OWNERSHIP_CATEGORY.get(key, f"{key or 'unknown'}_lease")
 def classify_branch_readback_http_status(
    status_code: int | None,
    *,
    not_found_scope: str | None = None,
 ) -> dict[str, Any]:
    """Classify a GET-branch HTTP status into a secret-free readback result.
    R1: A bare/generic/repository/wrong-host 404 never yields
    ``verified_absent=True``. Only an authoritative *branch-scoped* not-found
    (``not_found_scope='branch'``) may verify deletion.
    """
    if status_code == 404:
        scope = _norm_str(not_found_scope).lower() or NOT_FOUND_SCOPE_UNKNOWN
        if scope == NOT_FOUND_SCOPE_BRANCH:
            return {
                "status": READBACK_NOT_FOUND,
                "error_class": None,
                "verified_absent": True,
                "branch_present": False,
                "not_found_scope": NOT_FOUND_SCOPE_BRANCH,
                "reasons": [],
            }
        # repository / host / unknown / generic 404 — not verified absence
        reason = {
            NOT_FOUND_SCOPE_REPOSITORY: (
                "post-delete readback 404 is repository-scoped, not branch absence"
            ),
            NOT_FOUND_SCOPE_HOST: (
                "post-delete readback 404 is host-scoped, not branch absence"
            ),
        }.get(
            scope,
            "post-delete readback 404 is ambiguous (not branch-scoped); "
            "cannot verify absence",
        )
        return {
            "status": READBACK_AMBIGUOUS_404,
            "error_class": ERROR_CLASS_UNEXPECTED,
            "verified_absent": False,
            "branch_present": None,
            "not_found_scope": scope,
            "reasons": [reason],
        }
    if status_code in (401, 407):
        return {
            "status": READBACK_AUTHENTICATION,
            "error_class": ERROR_CLASS_AUTHENTICATION,
            "verified_absent": False,
            "branch_present": None,
            "reasons": ["post-delete branch readback authentication failed"],
        }
    if status_code == 403:
        return {
            "status": READBACK_AUTHORIZATION,
            "error_class": ERROR_CLASS_AUTHORIZATION,
            "verified_absent": False,
            "branch_present": None,
            "reasons": ["post-delete branch readback authorization failed"],
        }
    if status_code is not None and 200 <= int(status_code) < 300:
        return {
            "status": READBACK_EXISTS,
            "error_class": None,
            "verified_absent": False,
            "branch_present": True,
            "reasons": ["post-delete readback found branch still present"],
        }
    if status_code is not None and int(status_code) >= 500:
        return {
            "status": READBACK_TRANSPORT,
            "error_class": ERROR_CLASS_TRANSPORT,
            "verified_absent": False,
            "branch_present": None,
            "reasons": ["post-delete branch readback transport/upstream failure"],
        }
    return {
        "status": READBACK_UNEXPECTED,
        "error_class": ERROR_CLASS_UNEXPECTED,
        "verified_absent": False,
        "branch_present": None,
        "reasons": ["post-delete branch readback returned an unexpected response"],
        "http_status": status_code,
    }
 def _extract_http_status(exc: BaseException) -> int | None:
    """Extract an HTTP status code from an exception chain (secret-free)."""
    seen: set[int] = set()
    current: BaseException | None = exc
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        for attr in ("code", "status", "status_code"):
            value = getattr(current, attr, None)
            if isinstance(value, int) and 100 <= value <= 599:
                return value
        text = str(current) if current is not None else ""
        match = re.match(r"HTTP\s+(\d{3})\b", text)
        if match:
            return int(match.group(1))
        current = current.__cause__ or current.__context__
    return None
 def classify_branch_readback_exception(
    exc: BaseException,
    *,
    not_found_scope: str | None = None,
 ) -> dict[str, Any]:
    """Classify a GET-branch exception without leaking credentials or bodies.
    R1: substring ``404`` / ``not found`` alone never becomes verified_absent.
    Callers must pass ``not_found_scope='branch'`` only after authoritative
    proof that the repository/host is still reachable and the 404 is branch-level.
    """
    status_code = _extract_http_status(exc)
    if status_code is None:
        lower = str(exc).lower() if exc is not None else ""
        if any(
            token in lower
            for token in (
                "timed out",
                "timeout",
                "connection",
                "network",
                "temporarily unavailable",
                "name or service not known",
                "nodename nor servname",
            )
        ):
            return {
                "status": READBACK_TRANSPORT,
                "error_class": ERROR_CLASS_TRANSPORT,
                "verified_absent": False,
                "branch_present": None,
                "reasons": [
                    "post-delete branch readback transport/upstream failure"
                ],
            }
        if "unauthorized" in lower:
            status_code = 401
        elif "forbidden" in lower:
            status_code = 403
        elif "404" in lower or "not found" in lower:
            # Ambiguous: do NOT treat as branch absence without scope proof.
            status_code = 404
            if not_found_scope is None:
                not_found_scope = NOT_FOUND_SCOPE_UNKNOWN
    if status_code is not None:
        return classify_branch_readback_http_status(
            status_code, not_found_scope=not_found_scope
        )
    return {
        "status": READBACK_UNEXPECTED,
        "error_class": ERROR_CLASS_UNEXPECTED,
        "verified_absent": False,
        "branch_present": None,
        "reasons": ["post-delete branch readback returned an unexpected response"],
    }
 def assess_post_delete_readback(readback: dict[str, Any] | None) -> dict[str, Any]:
    """Decide whether DELETE success may be reported after branch readback.
    Success requires authoritative branch-scoped not-found
    (``verified_absent=True``). DELETE HTTP success alone is never enough.
    Generic/repository/host 404 cannot verify absence (R1).
    """
    payload = dict(readback or {})
    status = _norm_str(payload.get("status")) or READBACK_UNEXPECTED
    # Only explicit verified_absent flag counts — never infer from status alone
    # when status is a bare not_found without branch scope proof.
    verified = bool(payload.get("verified_absent")) is True
    if verified and status == READBACK_NOT_FOUND:
        return {
            "ok": True,
            "success": True,
            "verified_absent": True,
            "readback": {
                "status": READBACK_NOT_FOUND,
                "verified_absent": True,
                "branch_present": False,
                "error_class": None,
                "not_found_scope": NOT_FOUND_SCOPE_BRANCH,
            },
            "reasons": [],
        }
    reasons = list(payload.get("reasons") or [])
    if not reasons:
        if status == READBACK_EXISTS:
            reasons = ["post-delete readback found branch still present"]
        elif status == READBACK_AUTHENTICATION:
            reasons = ["post-delete branch readback authentication failed"]
        elif status == READBACK_AUTHORIZATION:
            reasons = ["post-delete branch readback authorization failed"]
        elif status == READBACK_TRANSPORT:
            reasons = ["post-delete branch readback transport/upstream failure"]
        elif status == READBACK_AMBIGUOUS_404:
            reasons = [
                "post-delete readback 404 is not branch-scoped; "
                "cannot verify absence"
            ]
        else:
            reasons = ["post-delete branch readback could not verify deletion"]
    return {
        "ok": False,
        "success": False,
        "verified_absent": False,
        "readback": {
            "status": status,
            "verified_absent": False,
            "branch_present": payload.get("branch_present"),
            "error_class": payload.get("error_class"),
            "not_found_scope": payload.get("not_found_scope"),
        },
        "reasons": reasons,
        "blocker_kind": "post_delete_readback_failed",
    }
 def cleanup_result_envelope(
    *,
    success: bool,
    performed: bool,
    delete_acknowledged: bool,
    verified_absent: bool,
    **extra: Any,
 ) -> dict[str, Any]:
    """R2: consistent top-level cleanup result fields on every return path."""
    out: dict[str, Any] = {
        "success": bool(success),
        "performed": bool(performed),
        "delete_acknowledged": bool(delete_acknowledged),
        "verified_absent": bool(verified_absent),
    }
    out.update(extra)
    return out
 def _repo_matches(
    record: dict[str, Any],
    *,
    remote: str,
    org: str,
    repo: str,
    host: str | None = None,
 ) -> bool:
    """Match ownership record to target remote/org/repo and normalized host."""
    if (
        _norm_str(record.get("remote")).lower() != _norm_str(remote).lower()
        or _norm_str(record.get("org")).lower() != _norm_str(org).lower()
        or _norm_str(record.get("repo")).lower() != _norm_str(repo).lower()
    ):
        return False
    expected_host = normalize_host(host)
    record_host = normalize_host(record.get("host") or record.get("host_name"))
    # When both sides declare a host, they must agree after normalization.
    # A record host that disagrees with the expected host is out of scope.
    # Legacy records without host still match when remote/org/repo agree.
    if expected_host and record_host and expected_host != record_host:
        return False
    return True
 def _branch_matches(record: dict[str, Any], branch: str) -> bool:
    return _norm_str(record.get("branch")) == _norm_str(branch)
 def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]:
    """Classify one ownership record as blocking or non-blocking.
    Distinguishes active ownership from expired/released/terminal/stale records.
    Expired/stale records block unless reclaim_allowed is *explicitly* True
    (O2: never treat missing/unknown reclaim as auto-allowed).
    """
    status = _norm_str(record.get("status")).lower()
    category = _norm_str(record.get("category")) or "unknown"
    reclaim_allowed = record.get("reclaim_allowed")
    if category == OWNERSHIP_CATEGORY_INVENTORY_ERROR:
        return {
            "blocks": True,
            "status": status or "unknown",
            "category": category,
            "reason": "ownership inventory failed closed",
        }
    if status in _TERMINAL_OWNERSHIP_STATUSES:
        return {
            "blocks": False,
            "status": status,
            "category": category,
            "reason": f"{category} ownership is terminal/released ({status})",
        }
    if status in _ACTIVE_OWNERSHIP_STATUSES:
        return {
            "blocks": True,
            "status": status,
            "category": category,
            "reason": f"active {category} ownership still uses the target branch",
        }
    if status in _EXPIRED_STATUSES or status in _STALE_STATUSES:
        # O2: only explicit reclaim_allowed=True skips the block.
        if reclaim_allowed is True:
            return {
                "blocks": False,
                "status": status,
                "category": category,
                "reason": (
                    f"{category} ownership is {status} and reclaimable; "
                    "does not block deletion"
                ),
            }
        return {
            "blocks": True,
            "status": status,
            "category": category,
            "reason": (
                f"{status} {category} ownership still protects the target "
                "branch (reclaim not proven; fail closed)"
            ),
        }
    # Unknown status → fail closed
    return {
        "blocks": True,
        "status": status or "unknown",
        "category": category,
        "reason": (
            f"unclassified {category} ownership status "
            f"'{status or 'unknown'}'; fail closed"
        ),
    }
 def assess_active_branch_ownership(
    *,
    remote: str,
    org: str,
    repo: str,
    branch: str,
    host: str | None = None,
    records: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
    """Assess whether any active ownership still uses *branch* in *repo*.
    Matching requires remote/org/repo/branch and, when provided, normalized
    host identity. Other repositories, hosts, or branches never false-block.
    """
    target_branch = _norm_str(branch)
    expected_host = normalize_host(host)
    considered: list[dict[str, Any]] = []
    blocking: list[dict[str, Any]] = []
    ignored: list[dict[str, Any]] = []
    for raw in records or []:
        if not isinstance(raw, dict):
            continue
        if not _repo_matches(
            raw, remote=remote, org=org, repo=repo, host=expected_host or None
        ):
            ignored.append(
                {
                    "category": _norm_str(raw.get("category")) or "unknown",
                    "reason": "different repository or host scope",
                }
            )
            continue
        if not _branch_matches(raw, target_branch):
            ignored.append(
                {
                    "category": _norm_str(raw.get("category")) or "unknown",
                    "reason": "different branch",
                }
            )
            continue
        activity = assess_ownership_record_activity(raw)
        entry = {
            "category": activity["category"],
            "status": activity["status"],
            "blocks": activity["blocks"],
            "reason": activity["reason"],
        }
        considered.append(entry)
        if activity["blocks"]:
            blocking.append(entry)
    block = bool(blocking)
    categories = sorted({b["category"] for b in blocking})
    reasons = [
        (
            "active ownership protects branch "
            f"'{target_branch}': " + "; ".join(b["reason"] for b in blocking)
        )
    ] if block else []
    return {
        "block": block,
        "safe_to_delete": not block,
        "remote": remote,
        "org": org,
        "repo": repo,
        "host": expected_host or None,
        "branch": target_branch,
        "blocking_categories": categories,
        "blocking": blocking,
        "considered": considered,
        "ignored_out_of_scope": ignored,
        "reasons": reasons,
        "blocker_kind": "active_branch_ownership" if block else None,
        "recommended_action": "keep_remote_branch" if block else "delete_remote_branch",
    }
@@ -0,0 +1,198 @@
 # ADR: Stable control runtime vs dev runtime (Gitea MCP)
 - **Status:** Accepted (policy effective immediately for LLM sessions; tooling may lag)
 - **Date:** 2026-07-09
 - **Tracking issue:** [#615](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/615)
 - **Related:**
  - [#543](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/543) / `docs/mcp-namespace-health.md` — client-namespace health
  - `docs/mcp-namespace-eof-recovery.md` — reconnect-only EOF recovery (no PID kill)
  - [#558](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/558) / `docs/mcp-daemon-import-guard.md` — sanctioned daemon
  - [#557](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/557) / `docs/bootstrap-review-path.md` — controller bootstrap for self-hosted fixes
  - Allocator / control-plane ADR: `docs/architecture/mcp-allocator-control-plane-observability-adr.md` (#613 / PR #614)
 ## 1. Context
 The Gitea MCP server is the **control plane** for real issue/PR mutations (create, comment, lock, review, merge, etc.). When author/reviewer/merger/reconciler sessions kill or restart that process, relaunch it from a feature worktree, or edit the checkout that process loads, operators observe:
 - Mid-session identity/preflight resets
 - Stale-runtime vs master parity failures
 - IDE transport EOF / “tool not found” while code on disk has changed
 - Accidental production mutations from experimental code
 This ADR separates **stable control runtime** from **dev/test runtime** and defines promotion proof.
 ## 2. Decision
 ### 2.1 Stable control runtime
 The Gitea MCP server used for **real workflow mutations** is the **stable control runtime**.
 Characteristics:
 - Loads a known, promoted revision of Gitea-Tools (or the packaged release layout operators designate)
 - Registered in the IDE/client as the production namespaces (`gitea-tools`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, etc.)
 - Holds production profile credentials via sanctioned keychain/env paths only
 ### 2.2 Dev / test runtime
 MCP **server code** development and testing:
 - Happens in isolated **`branches/`** worktrees (or other non-stable checkouts)
 - May use a **separate** dev/test MCP runtime/process when process-level testing is required
 - **Must not** be used for real Gitea mutations on production issues/PRs
 ### 2.3 Forbidden actions (normal sessions)
 Normal **author, reviewer, merger, and reconciler** LLM sessions **must not**:
 | Forbidden | Why |
 |-----------|-----|
 | Kill the running MCP server process | Drops all concurrent sessions; loses preflight state |
 | Restart / relaunch the MCP server process | Same as kill; causes stale/identity churn mid-workflow |
 | Relaunch MCP from a development worktree | Runs unpromoted code against production mutations |
 | Edit files in the stable runtime checkout | Hot-mutates control plane under concurrent users |
 | Use experimental/dev MCP for real Gitea mutations | Bypasses promotion proof and audit expectations |
 | Bypass or self-reset a stale master-parity gate | The gate is fail-closed; only an operator reload restores parity |
 **LLM-allowed vs operator-owned (authoritative split):**
 | Actor | May do | Must not do |
 |-------|--------|-------------|
 | **LLM session** | Call tools on the already-running stable namespaces; **client reconnect** after transport EOF (no process kill); pass `worktree_path` / role worktree args; report blockers and stop mutations when unhealthy/stale | Kill, restart, or relaunch any MCP process; bump config mtimes to force reload; edit the stable checkout; switch to a dev MCP for production mutations |
 | **Operator / release-manager** | Supervised restart/reload of the **stable** control runtime; dual-namespace client configuration; §2.4 promotions; incident recovery | — |
 EOF / transport recovery for LLM sessions: **client reconnect only** (see `docs/mcp-namespace-eof-recovery.md`). Do not “fix” health by killing PIDs or bumping MCP config mtimes as a normal session procedure.
 This ADR **supersedes** any older runbook wording that told the LLM to relaunch or restart the client/MCP as a self-service step. Where `docs/llm-workflow-runbooks.md` (or wiki runbooks) discuss dual-namespace setup or workspace rebind, **process restart/relaunch is operator-owned**; the LLM stops, reports, and waits.
 ### 2.4 Promotion (operator / release-manager only)
 Promotion of a new revision into the stable control runtime is an **explicit operator/release-manager action**, not an LLM self-service step.
 A promotion **must record** (issue comment, release note, or promotion ledger):
 | Field | Description |
 |-------|-------------|
 | **previous runtime SHA** | Commit previously loaded by stable runtime |
 | **promoted runtime SHA** | Commit after promotion |
 | **source branch/PR** | Where the change was reviewed |
 | **restart/reload method** | How the process was cycled (e.g. supervised restart, client reload) |
 | **health check proof** | Client-namespace probe success (`gitea_whoami` / namespace health) |
 | **identity/profile proof** | Expected profile(s) and username(s) after reload |
 | **workspace/root proof** | Stable checkout path / root matches intended layout |
 | **mutation capability proof** | Required permissions for the target role present; forbidden ops still forbidden |
 | **rollback instructions** | How to restore previous SHA and re-verify health |
 Suggested durable marker:
 ```text
 ## MCP STABLE RUNTIME PROMOTION (#615)
 Status: COMPLETED | ROLLED_BACK | ABORTED
 Previous-SHA: <full sha>
 Promoted-SHA: <full sha>
 Source-PR: <number>
 Source-Branch: <name>
 Reload-Method: <text>
 Health-Proof: client_namespace whoami OK / assess_mcp_namespace_health OK
 Identity-Proof: profile=<name> user=<name>
 Workspace-Proof: root=<path>
 Mutation-Proof: allowed_ops include <…>; forbidden include <…>
 Rollback: checkout <previous sha>; reload method <…>; re-run health/identity proofs
 Operator: <username>
 Timestamp: <ISO-8601>
 ```
 ### 2.5 Unhealthy stable runtime → stop work
 If the stable MCP runtime is **unhealthy**, including any of:
 - client-namespace probes fail
 - wrong identity / wrong profile
 - wrong workspace root
 - missing mutation capability for the intended role
 - persistent EOF after **client reconnect**
 - **master parity is stale** (`startup_head` behind on-disk `master` / `restart_required` from the parity gate)
 then:
 1. **Normal PR / review / merge / issue-mutation work must stop immediately.**
 2. The session **reports** the unhealthy/stale state (tool error, CTH, or operator handoff) with startup vs current head when known.
 3. Do **not** improvise: no LLM process kill/restart, no dev-worktree MCP for production mutations, no env escape hatches, no manual gate bypass.
 4. Resume only after:
   - an **operator** restores the runtime (see §2.6 for routine post-merge parity reload), **or**
   - a **controlled** promotion/rollback completes with the §2.4 promotion record, **or**
   - a controller invokes the narrow **bootstrap review path** (#557) when the defect is self-hosted and documented,
   - **and** the session re-verifies health (and master parity when applicable) before the next mutation.
 ### 2.6 Routine post-merge master-parity staleness (operator reload, not promotion)
 **Symptom:** After merges land on `master`, a long-lived stable MCP process still runs the pre-merge `startup_head`. The master-parity gate marks the server **stale** / `restart_required` and **blocks mutations**.
 **Sanctioned response (authoritative):**
 | Step | Actor | Action |
 |------|-------|--------|
 | 1 | LLM session | Mutations stop immediately when the gate reports stale. |
 | 2 | LLM session | Report the stale state (startup head, current master head, that operator reload is required). Do not retry mutations. |
 | 3 | **Operator** | Reload/restart the **stable** control MCP so it loads current `master` (supervised client/daemon reload). |
 | 4 | LLM session | Resume only after startup/current-head parity is verified (e.g. `gitea_get_runtime_context` / parity assessment shows in parity). |
 **Not a §2.4 promotion:** Catching the already-designated stable control checkout up to a newly advanced `master` is a **routine operator reload** of the stable runtime. It does **not** require the nine-field promotion ledger. Use §2.4 only when **changing which unpromoted/dev revision** becomes the stable control runtime (new source branch/PR into the stable designation).
 **Code note:** `master_parity_gate.py` may still say “restart the server” in machine-facing reason strings. That string names the **operator recovery action**, not an LLM self-service instruction. This ADR and the runbooks define the actor split.
 ## 3. Relationship to other controls
 | Doc / mechanism | Interaction |
 |-----------------|-------------|
 | Namespace health (#543) | Proves IDE client can call tools; does not authorize restart |
 | EOF recovery | Reconnect only; no process kill |
 | Daemon import guard (#558) | Mutations require sanctioned daemon; not a bare shell import |
 | Bootstrap path (#557) | Only controller-authorized exception when live runtime cannot review its own fix |
 | Allocator / control-plane ADR | Coordination DB is separate; still depends on a healthy MCP surface for Gitea writes |
 ## 4. Consequences
 ### Positive
 - Predictable control plane for concurrent LLMs
 - Clear operator-only promotion gate with rollback
 - Aligns session behavior with health/EOF docs already landed
 ### Costs
 - LLM sessions must wait when runtime is sick (no DIY restart)
 - Operators must maintain promotion discipline and dual-runtime config if they use a dev MCP
 ### Non-goals
 - Does not ban operator-supervised restarts during incidents
 - Does not replace CI or code review for MCP changes
 - Does not authorize editing stable checkout “because tests need a quick fix”
 ## 5. Implementation follow-ups (optional tooling)
 These may land in later issues; the **policy binds sessions now**:
 1. Session preflight that refuses mutations if workspace root equals a `branches/` feature worktree configured as “dev only.”
 2. Explicit `runtime_kind=stable|dev` in MCP config and `gitea_whoami` profile metadata.
 3. Promotion checklist script that emits the durable promotion marker fields.
 **Not optional (issue #615 acceptance criterion 2):** operator guide and runbooks **must** cross-link this ADR (see §6). Cross-links are documentation acceptance, not deferred tooling.
 ## 6. Acceptance for this ADR
 1. Document merged under `docs/architecture/mcp-stable-control-runtime-policy-adr.md`.
 2. **Operator guide / runbooks cross-link this ADR** (`docs/wiki/Operator-Guide.md`, `docs/wiki/Runbooks.md`, `docs/llm-workflow-runbooks.md`).
 3. Issue #615 references this path.
 4. LLM/operator runbooks treat kill/restart/relaunch-from-worktree as **LLM violations**; process restart is **operator-owned**.
 5. Unhealthy runtime (including **stale master parity**) stops normal mutation work until operator restore/reload, promotion/rollback, or #557 bootstrap — then re-verify parity before mutating.
 6. Routine post-merge parity reload is documented as operator reload (§2.6), not an LLM self-restart and not a full §2.4 promotion.
 ## 7. Document history
 | Date | Change |
 |------|--------|
 | 2026-07-09 | Initial ADR: stable vs dev runtime, forbidden session actions, promotion proof fields, stop-work rule |
 | 2026-07-16 | Review 443 remediation (#615 / PR #616): mandatory cross-links; LLM vs operator restart split; routine post-merge parity staleness (§2.6); stale parity in §2.5 unhealthy triggers |
@@ -238,11 +238,43 @@ narrow operation set:
 - `gitea.issue.comment`
 - `gitea.issue.close`
 - `gitea.pr.close`
 - `gitea.branch.delete` (merged-branch cleanup only — see below)
 Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
 `gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
 `gitea.repo.commit`.
 ### Merged-branch cleanup ownership (`gitea.branch.delete`)
 The reconciler is the repository-supported owner of merged-PR source-branch
 cleanup: `task_capability_map` maps `cleanup_merged_pr_branch` (and
 `reconciliation_cleanup`) to role `reconciler` with permission
 `gitea.branch.delete`. Post-merge branch lifecycle is reconciliation work —
 it happens after the author, reviewer, and merger roles have completed, and
 it must not be reachable from those roles.
 Least-privilege constraints:
 - `gitea.branch.delete` is granted **only** to reconciler profiles. Author,
  reviewer, and merger profiles must never hold it; `gitea_delete_branch`
  and `gitea_cleanup_merged_pr_branch` fail closed on any profile without
  the permission.
 - Even with the permission, reconciler deletion is only supported through the
  guarded `gitea_cleanup_merged_pr_branch` path (#514 / #687): the PR must be
  merged, the head an ancestor of the target, the branch not protected
  (`master`/`main`/`dev`), the branch not a preservation/evidence ref (e.g.
  `chore/issue-681-preserve-review-session-wip`), no open PR may still use the
  head, and an explicit `CLEANUP MERGED PR <n> BRANCH <branch>` confirmation is
  required. Raw `gitea_delete_branch` is **denied** to reconciler even when
  `gitea.branch.delete` is present.
 - Raw `git branch -d` / `git push --delete` cleanup remains blocked by
  `branch_cleanup_guard` and the final-report validator regardless of
  profile permissions.
 - `gitea.branch.delete` has no short alias in `GITEA_OPERATION_ALIASES`;
  write it fully qualified in `allowed_operations`. Migration must emit
  canonical names such as `gitea.pr.close` (never bare `pr.close` /
  `issue.close`, which the production normalizer rejects or drops).
 Launch a static `gitea-reconciler` MCP namespace with
 `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
 `reconciler_profile.assess_reconciler_profile` (#304). Use the
@@ -251,6 +283,159 @@ Launch a static `gitea-reconciler` MCP namespace with
 fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
 heads are not already landed cannot be closed through this path.
 ### Operational runbook: grant reconciler `gitea.branch.delete` (#687)
 Merging a code PR that updates `migrate_profiles.py` / `reconciler_profile.py`
 **does not** change the live operator profile on disk. Apply the profile
 change deliberately, then reconnect the client-managed namespace.
 1. **Approved migration / profile-update command** (from the repo root, using
   the project venv if present):
   ```bash
   # Dry-run first (default): validates v2 output, writes nothing
   python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json
   # Apply: creates backup then writes migrated v2 config
   python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json -w
   # Optional explicit paths:
   # python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json \
   #   -o ~/.config/gitea-tools/profiles.json \
   #   --backup ~/.config/gitea-tools/profiles.json.bak -w
   ```
   If the live file is already v2, edit the reconciler identity’s
   `allowed_operations` / `forbidden_operations` under
   `environments.<env>.services.gitea.identities.reconciler` (or the
   `prgs-reconciler` alias target) so allowed includes the canonical set
   below — then re-validate with a load of the config (see step 3).
 2. **Inspect the generated (or edited) profile** — confirm the reconciler
   identity, for example:
   ```bash
   python3 - <<'PY'
   import json
   from pathlib import Path
   cfg = json.loads(Path.home().joinpath(".config/gitea-tools/profiles.json").read_text())
   # v2 environments shape:
   ident = cfg["environments"]["prgs"]["services"]["gitea"]["identities"]["reconciler"]
   print("role:", ident.get("role"))
   print("allowed:", ident.get("allowed_operations"))
   print("forbidden:", ident.get("forbidden_operations"))
   PY
   ```
 3. **Validate canonical operation names and least privilege**
   Expected canonical **allowed** (defaults after migration):
   - `gitea.read`
   - `gitea.pr.close` (required)
   - `gitea.pr.comment`
   - `gitea.issue.comment`
   - `gitea.issue.close`
   - `gitea.branch.delete` (recommended; cleanup only)
   Expected **forbidden** includes at least: `gitea.pr.approve`,
   `gitea.pr.merge`, `gitea.pr.review`, `gitea.pr.create`,
   `gitea.branch.push`, `gitea.repo.commit`.
   No shorthand (`pr.close`, `issue.close`, `pr.comment`) may remain.
   Validate with the production loader:
   ```bash
   python3 - <<'PY'
   import gitea_config, reconciler_profile
   from pathlib import Path
   path = str(Path.home() / ".config/gitea-tools/profiles.json")
   gitea_config.load_config(path)  # fails closed on invalid config
   # Or assess the reconciler lists directly after extracting them:
   # print(reconciler_profile.assess_reconciler_profile(allowed, forbidden))
   PY
   ```
 4. **Merging PR #688 (or any code PR) does not update the live profile.**
   Code changes only the migration helper, schema, docs, and tests. The
   operator must still run `migrate_profiles.py -w` or an equivalent
   authorized edit of `~/.config/gitea-tools/profiles.json`.
 5. **Supported apply method:** `python3 migrate_profiles.py … -w` (backup
   created automatically) **or** operator-authorized edit of the live
   profiles file after backup. Unsupported: silent mtime tricks, manual
   process kill to “reload”, or undocumented env overrides.
 6. **Backup and validation:** `-w` copies the input to
   `<input_path>.bak` (or `--backup PATH`) before writing. Re-run
   `load_config` / `assess_reconciler_profile` after write. Keep the
   `.bak` until live whoami/capability checks pass.
 7. **Client-managed namespace reconnect/reload:** reconnect or reload the
   IDE MCP client so `gitea-reconciler` restarts from current `master` and
   the updated `GITEA_MCP_PROFILE=prgs-reconciler` config. Do not hand-launch
   `mcp_server.py` / `gitea_mcp_server.py` with ad hoc `GITEA_*` env
   (see #686 / #630).
 8. **Live reverification** (through the client-managed `gitea-reconciler`
   namespace only):
   - `gitea_whoami` → identity + profile `prgs-reconciler`
   - `gitea_assess_master_parity` → `stale=false`, `restart_required=false`
   - `gitea_resolve_task_capability(task="cleanup_merged_pr_branch")` →
     `allowed_in_current_session=true` only when permission and role match
   - `gitea_resolve_task_capability(task="delete_branch")` →
     **not** allowed for reconciler (role denial must be enforced)
 9. **Guarded cleanup usage** (example for a merged PR whose source branch
   remains on the remote):
   ```text
   gitea_cleanup_merged_pr_branch(
     pr_number=<N>,
     branch=<exact PR head branch>,
     confirmation="CLEANUP MERGED PR <N> BRANCH <exact PR head branch>",
     remote="prgs",
     org="Scaled-Tech-Consulting",
     repo="Gitea-Tools",
     worktree_path="<path under branches/>",
   )
   ```
   The tool refuses unmerged PRs, protected branches, preservation/evidence
   branches, open-PR heads, mismatched branch names, and wrong confirmation.
 10. **Prohibitions**
    - No raw `git push --delete`, `git branch -d` / `-D`, or delete refspecs
    - No arbitrary `gitea_delete_branch` from reconciler
    - No unsupported profile switching mid-run without full re-preflight
    - No ad hoc hand-edits of live profiles **unless** operator-authorized,
      backed up, and revalidated as above
 Canonical migrated reconciler example:
 ```json
 {
  "role": "reconciler",
  "allowed_operations": [
    "gitea.read",
    "gitea.pr.close",
    "gitea.pr.comment",
    "gitea.issue.comment",
    "gitea.issue.close",
    "gitea.branch.delete"
  ],
  "forbidden_operations": [
    "gitea.pr.approve",
    "gitea.pr.merge",
    "gitea.pr.review",
    "gitea.pr.create",
    "gitea.branch.push",
    "gitea.repo.commit"
  ]
 }
 ```
 ## Identity and fail-closed rules
 Before **any** mutating action, a workflow must know both:
@@ -195,7 +195,7 @@ To avoid the bottleneck of relaunching/restarting the MCP server to switch betwe
  `gitea_reconcile_already_landed_pr` after ancestry proof — not for normal
  review or author workflows.
-* **Fallback:** If the dual-profile MCP launcher pattern is not supported or configured in the client, the LLM must relaunch or restart the client/MCP with the correct profile environment variable before claiming or working on any tasks.
+* **Fallback (operator-owned):** If the dual-profile MCP launcher pattern is not supported or configured in the client, **do not** have the LLM relaunch or restart the client/MCP. The LLM **stops** role-switching work, reports that the correct static namespace is missing, and waits for an **operator** to configure dual namespaces or reload the client with the correct `GITEA_MCP_PROFILE` for that role. Process restart/relaunch is operator-owned under the [stable control runtime ADR](architecture/mcp-stable-control-runtime-policy-adr.md) (#615).
 ## Setup runbook — interactive menu
@@ -1200,10 +1200,13 @@ When a mutation blocks on workspace binding:
 1. Read the error — it names the **resolved workspace path**, **role
   namespace**, and **binding source** (tool arg, env var, or process root).
-2. Reconnect or relaunch the correct namespace MCP server from the intended
+2. **LLM-allowed:** pass `worktree_path` on reviewer/merger mutation tools when
-   workspace (or set the role-specific env var before launch).
+   the active `branches/` worktree differs from the MCP process root; **client
-3. Pass `worktree_path` on reviewer/merger mutation tools when the active
+   reconnect** after transport EOF only (no process kill).
-   branches/ worktree differs from the MCP process root.
+3. **Operator-owned:** if the wrong namespace process was launched, or a role-
   specific `GITEA_*_WORKTREE` must be set at process start, an **operator**
   reloads/relaunches the correct static namespace MCP. LLM sessions must not
   kill or restart MCP processes (see [stable control runtime ADR](architecture/mcp-stable-control-runtime-policy-adr.md)).
 4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
   own namespace — that destroys another agent's WIP.
@@ -1213,9 +1216,10 @@ When posting a Canonical Thread Handoff after a binding blocker:
 - State which namespace was active (author / reviewer / merger / reconciler).
 - Quote the resolved workspace path and binding source from the error.
- Name the safe reconnect action (relaunch MCP from `branches/...`, set
+- Name the safe next action: pass `worktree_path` if that unblocks the tool, or
-  `GITEA_*_WORKTREE`, or pass `worktree_path`).
+  request an **operator** reload of the correct namespace MCP / env binding.
- Explicitly note that foreign worktrees must not be cleaned to unblock.
+- Explicitly note that foreign worktrees must not be cleaned to unblock, and
  that the LLM must not self-restart the MCP process.
 ## Safety notes
@@ -1226,6 +1230,7 @@ When posting a Canonical Thread Handoff after a binding blocker:
 ## Related documents
 - [`architecture/mcp-stable-control-runtime-policy-adr.md`](architecture/mcp-stable-control-runtime-policy-adr.md) — stable control runtime vs dev runtime; LLM must not kill/restart MCP; operator-owned reload and promotions; routine post-merge parity staleness (#615).
 - [`reviewer-handoff-consistency.md`](reviewer-handoff-consistency.md) — reject contradictory reviewer handoffs (#501).
 - [`issue-acceptance-gate.md`](issue-acceptance-gate.md) — controller issue-acceptance audit after PR merge (#500).
 - [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable cross-project LLM workflow skill.
@@ -14,6 +14,7 @@ Handbook for LLM operators and human developers using the Gitea-Tools MCP server
 4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
 5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
 6. **Global LLM Worktree Rule** — Main checkout stays on `master`/`main`/`dev`; all mutations happen under `branches/`. Prove project root, `cwd`, branch, stable main-checkout branch, and session worktree path before editing. No exceptions.
 7. **Stable control runtime** — Real Gitea mutations use only the **stable** MCP control runtime. LLM sessions must not kill, restart, or relaunch MCP processes, edit the stable checkout, or use a dev MCP for production mutations. See the policy ADR: [mcp-stable-control-runtime-policy-adr.md](../architecture/mcp-stable-control-runtime-policy-adr.md) (#615).
 ## Supported Gitea instances
@@ -30,3 +31,4 @@ Always pass `remote` explicitly on tool calls. The server default is `dadeschool
 2. `gitea_get_runtime_context` — allowed/forbidden operations for this session.
 3. `gitea_resolve_task_capability` — prove the session may perform the planned task.
 4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.
 5. Confirm master parity / runtime health when tools report stale or unhealthy control runtime — stop mutations and request an **operator** reload of the stable MCP (see [stable control runtime ADR](../architecture/mcp-stable-control-runtime-policy-adr.md)).
@@ -12,6 +12,17 @@
 4. `gitea_mark_final_review_decision` → approve via `gitea_review_pr`.
 5. `gitea_merge_pr` with pinned head SHA and `confirmation="MERGE PR <n>"`.
 ## Stable MCP control runtime (#615)
 Policy ADR: [mcp-stable-control-runtime-policy-adr.md](../architecture/mcp-stable-control-runtime-policy-adr.md).
 | Situation | Who acts | What to do |
 |-----------|----------|------------|
 | Transport EOF / missing tools | LLM | **Client reconnect only** — do not kill PIDs |
 | Wrong profile / dual-namespace missing | Operator | Configure or reload the correct static namespace(s) |
 | Master parity stale after merge | LLM stops + reports; **operator** reloads stable MCP | Resume only after parity re-verified |
 | Promote unpromoted MCP code to stable | Operator / release-manager only | Full §2.4 promotion record |
 ## Gitea Wiki sync
 The Gitea Wiki mirrors `docs/wiki/` (source of truth). After merging wiki changes:
@@ -243,6 +243,192 @@ def _redact(text):
        return str(text)
 # ── Classified client failures (#699) ─────────────────────────────────────────
 # Subclasses of RuntimeError preserve existing ``except RuntimeError`` call
 # sites. Exception *messages* are fixed constants only — HTTP response bodies,
 # Keychain material, and arbitrary exception text are never stored on the
 # exception or re-emitted to tool results / daemon logs.
 # Fixed messages (must match mcp_tool_error_boundary.FIXED_MESSAGES keys used here).
 _MSG_AUTH_INVALID = "Gitea authentication failed: invalid or revoked credentials"
 _MSG_AUTH_FAILED = "Gitea authentication failed"
 _MSG_AUTHZ_SCOPE = "Gitea authorization failed: insufficient token scope"
 _MSG_AUTHZ_DENIED = "Gitea authorization failed: access denied"
 _MSG_NETWORK = "Network error contacting Gitea"
 _MSG_CONFIG = "Gitea configuration or credential resolution failed"
 _MSG_UPSTREAM = "Gitea upstream unavailable"
 _MSG_HTTP = "Gitea HTTP request failed"
 class GiteaClientError(RuntimeError):
    """Base for known Gitea client failures with stable reason_code metadata."""
    reason_code = "client_error"
    error_class = "client"
    http_status = None
    def __init__(self, message=None, *, reason_code=None, http_status=None):
        if reason_code is not None:
            self.reason_code = reason_code
        if http_status is not None:
            self.http_status = http_status
        # Message is always a fixed constant; callers cannot inject bodies.
        fixed = message if message is not None else _MSG_HTTP
        super().__init__(fixed)
 class GiteaAuthError(GiteaClientError):
    """Authentication failure (invalid/revoked credentials → typically HTTP 401)."""
    reason_code = "auth_invalid_token"
    error_class = "authentication"
    http_status = 401
    def __init__(self, message=None, *, reason_code=None, http_status=None):
        super().__init__(
            message if message is not None else _MSG_AUTH_INVALID,
            reason_code=reason_code or "auth_invalid_token",
            http_status=http_status if http_status is not None else 401,
        )
 class GiteaAuthzError(GiteaClientError):
    """Authorization failure (HTTP 403 — scope deficiency or access denied)."""
    reason_code = "authz_denied"
    error_class = "authorization"
    http_status = 403
    def __init__(self, message=None, *, reason_code=None, http_status=None):
        code = reason_code or "authz_denied"
        if code == "authz_insufficient_scope":
            fixed = _MSG_AUTHZ_SCOPE
        else:
            fixed = _MSG_AUTHZ_DENIED
            code = "authz_denied"
        super().__init__(
            message if message is not None else fixed,
            reason_code=code,
            http_status=http_status if http_status is not None else 403,
        )
 class GiteaNetworkError(GiteaClientError):
    """Transport / DNS / timeout failure contacting Gitea."""
    reason_code = "network_error"
    error_class = "network"
    http_status = None
    def __init__(self, message=None, *, reason_code=None, http_status=None):
        super().__init__(
            message if message is not None else _MSG_NETWORK,
            reason_code=reason_code or "network_error",
            http_status=http_status,
        )
 class GiteaConfigError(GiteaClientError):
    """Local configuration / credential resolution failure (not HTTP auth)."""
    reason_code = "config_error"
    error_class = "configuration"
    http_status = None
    def __init__(self, message=None, *, reason_code=None, http_status=None):
        super().__init__(
            message if message is not None else _MSG_CONFIG,
            reason_code=reason_code or "config_error",
            http_status=http_status,
        )
 class GiteaHttpError(GiteaClientError):
    """Non-auth HTTP failure with fixed message (no response body)."""
    reason_code = "http_error"
    error_class = "client"
    http_status = None
    def __init__(self, message=None, *, reason_code=None, http_status=None):
        code = reason_code or "http_error"
        if code == "upstream_unavailable":
            fixed = _MSG_UPSTREAM
        else:
            fixed = _MSG_HTTP
            code = "http_error"
        super().__init__(
            message if message is not None else fixed,
            reason_code=code,
            http_status=http_status,
        )
 def _looks_like_insufficient_scope(detail: str) -> bool:
    """Internal: inspect redacted body *only* to refine 403 reason_code.
    The body is never stored on the exception or returned to callers.
    """
    lower = (detail or "").lower()
    markers = (
        "insufficient scope",
        "required scope",
        "does not have at least one of required scope",
        "token does not have",
        "missing scope",
        "scope(s)",
    )
    return any(m in lower for m in markers)
 def classify_http_status(code: int, *, body_hint: str = "") -> tuple[type, str, int]:
    """Central HTTP status → (exception_class, reason_code, http_status).
    Every HTTP 403 becomes authorization-class. Body text is used only as a
    local hint for scope vs denied reason_code and is never returned.
    """
    if code == 401:
        return (GiteaAuthError, "auth_invalid_token", 401)
    if code == 403:
        if _looks_like_insufficient_scope(body_hint or ""):
            return (GiteaAuthzError, "authz_insufficient_scope", 403)
        return (GiteaAuthzError, "authz_denied", 403)
    if code in (502, 503, 504):
        return (GiteaHttpError, "upstream_unavailable", code)
    return (GiteaHttpError, "http_error", code)
 def raise_for_http_status(code: int, body: str = "") -> None:
    """Raise a typed client error for *code* without embedding *body*.
    *body* may be inspected only to choose scope vs denied for 403; it is
    never placed on the exception message.
    """
    # Redact before any inspection; discard after classification.
    try:
        hint = _redact(body or "").strip()
    except Exception:
        hint = ""
    exc_cls, reason, status = classify_http_status(code, body_hint=hint)
    # Explicitly construct without passing body/hint into message.
    if exc_cls is GiteaAuthError:
        raise GiteaAuthError(reason_code=reason, http_status=status)
    if exc_cls is GiteaAuthzError:
        raise GiteaAuthzError(reason_code=reason, http_status=status)
    if reason == "upstream_unavailable":
        raise GiteaHttpError(
            reason_code="upstream_unavailable",
            http_status=status,
        )
    raise GiteaHttpError(reason_code="http_error", http_status=status)
 def _raise_http_error(code: int, detail: str = "") -> None:
    """Backward-compatible alias — *detail* is never embedded in the error."""
    raise_for_http_status(code, detail)
 def _add_query(url, **params):
    """Return *url* with the given query parameters added or overridden.
@@ -315,23 +501,24 @@ def api_request(method, url, auth_header, payload=None, *,
    """Make an authenticated JSON request to the Gitea API.
    Returns parsed JSON on success (or ``None`` for an empty body), and raises
-    ``RuntimeError`` on failure.
+    a classified client error on failure.
    On HTTP 429 the request is retried up to *max_retries* times: honoring a
    valid ``Retry-After`` header (seconds or HTTP-date) when present, otherwise
    using capped jittered exponential backoff. Successful responses are
    unchanged.
-    All failures are converted to a ``RuntimeError`` with a clear, secret
+    Failures raise typed exceptions with **fixed messages only** (#699). HTTP
-    -redacted message (no raw stack traces or credential material):
+    response bodies are read solely for local 403 reason refinement and are
    never stored on exceptions or returned to callers:
-    - Non-429 HTTP errors surface the status code and a redacted response body.
+    - HTTP 401 → :class:`GiteaAuthError` (``auth_invalid_token``)
-      502/503/504 upstream errors get an explicit "Gitea upstream unavailable"
+    - HTTP 403 → :class:`GiteaAuthzError` (scope or denied)
-      message.
+    - 502/503/504 → :class:`GiteaHttpError` (``upstream_unavailable``)
-    - Timeouts and network/DNS failures (``URLError`` / ``TimeoutError``) surface
+    - Other non-429 HTTP → :class:`GiteaHttpError` (``http_error``)
-      a generic "network error contacting Gitea" message.
+    - Timeouts / DNS / ``URLError`` → :class:`GiteaNetworkError`
-    - A malformed (non-JSON) success body surfaces a "malformed JSON response"
+    - Malformed success JSON → plain ``RuntimeError`` (programming/protocol;
-      message rather than a raw decode error.
+      not reclassified as authentication)
    The ``*_func`` parameters and ``timeout`` are injection points for
    deterministic testing.
@@ -369,22 +556,23 @@ def api_request(method, url, auth_header, payload=None, *,
                error_body = e.read().decode("utf-8", errors="replace")
            except Exception:
                error_body = ""
-            detail = _redact(error_body).strip()
+            # Classify from status (+ local body hint). Body is not embedded.
-            if e.code in (502, 503, 504):
+            try:
-                msg = f"HTTP {e.code}: Gitea upstream unavailable"
+                raise_for_http_status(e.code, error_body)
-                raise RuntimeError(f"{msg}: {detail}" if detail else msg) from e
+            except GiteaClientError:
-            raise RuntimeError(f"HTTP {e.code}: {detail}") from e
+                raise
            # Defensive: raise_for_http_status always raises.
            raise GiteaHttpError(http_status=e.code) from e  # pragma: no cover
        except (urllib.error.URLError, TimeoutError) as e:
-            reason = getattr(e, "reason", e)
+            # Fixed message only — do not embed URLError reason (may leak paths).
-            raise RuntimeError(
+            raise GiteaNetworkError(reason_code="network_error") from e
                f"network error contacting Gitea: {_redact(reason)}"
            ) from e
        if not body:
            return None
        try:
            return json.loads(body)
        except ValueError as e:
            # Programming/protocol failure — not authentication.
            raise RuntimeError("malformed JSON response from Gitea") from e
@@ -36,7 +36,24 @@ KIND_REVIEW_DRAFT = "review_draft"
 # other session proofs; a contaminated session fails closed on gated mutations
 # until a reconciler audits and clears it.
 KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
 # #709: archive prior terminal decision ledgers instead of silent overwrite.
 KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
 # #709: post-merge cleanup/audit reconciliation-required durable record.
 KIND_POST_MERGE_DECISION_RECOVERY = "post_merge_decision_recovery"
 # #709: truthful record when historical terminal evidence is irrecoverably gone.
 KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
 # #709 F1: server-side non-forgeable authorization artifact for recovery.
 KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 # Kinds that must survive the default session-state TTL (forensic / recovery).
 RECOVERY_CRITICAL_KINDS = frozenset(
    {
        KIND_DECISION_LOCK_ARCHIVE,
        KIND_POST_MERGE_DECISION_RECOVERY,
        KIND_IRRECOVERABLE_DECISION_PROVENANCE,
        KIND_IRRECOVERABLE_PROVENANCE_AUTH,
    }
 )
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
@@ -270,7 +287,11 @@ def identity_match_reasons(
        reasons.append("session state missing recorded_at timestamp (fail closed)")
    else:
        age = _now_utc() - recorded_at
-        if age > timedelta(hours=ttl_hours()):
+        kind = (record.get("kind") or "").strip()
        ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
            record.get("recovery_critical")
        )
        if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
            reasons.append(
                f"session state expired after {ttl_hours():g}h (fail closed)"
            )
@@ -447,3 +468,177 @@ def clear_state(
        profile_identity=profile_identity,
        state_dir=state_dir,
    )
 def list_decision_lock_profile_identities(
    state_dir: str | None = None,
 ) -> list[str]:
    """Return profile identities that have a durable review_decision_lock file (#709).
    Filename form: ``review_decision_lock-<profile>.json`` (see ``state_key``).
    Does not validate TTL or identity — callers must load via ``load_state``.
    """
    root = (state_dir or default_state_dir()).strip()
    if not root or not os.path.isdir(root):
        return []
    prefix = f"{_sanitize_segment(KIND_DECISION_LOCK)}-"
    suffix = ".json"
    found: list[str] = []
    try:
        names = os.listdir(root)
    except OSError:
        return []
    for name in names:
        if not name.startswith(prefix) or not name.endswith(suffix):
            continue
        if name.endswith(".lock"):
            continue
        mid = name[len(prefix) : -len(suffix)]
        if mid:
            found.append(mid)
    return sorted(set(found))
 def load_state_for_profile(
    *,
    kind: str,
    profile_identity: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    state_dir: str | None = None,
    skip_identity_match: bool = False,
    enforce_repo_scope: bool = True,
 ) -> dict[str, Any] | None:
    """Load durable state for an explicit profile identity (#709 cross-profile).
    When *skip_identity_match* is True, still requires the file's recorded
    profile_identity to equal the requested profile (anti-stomp), but does not
    require the *active* session identity to match — needed so a merger can
    inspect a reviewer lock after merge.
    #709 F3: remote/org/repo filter reasons are **enforced** (not merely
    computed). When *enforce_repo_scope* is True (default) and the caller
    supplies remote/org/repo, mismatches fail closed with ``None``.
    """
    # #709 F3: refuse traversal / malformed profile identities.
    try:
        from irrecoverable_provenance import assess_profile_path_identity
        path_gate = assess_profile_path_identity(profile_identity)
        if not path_gate.get("valid"):
            return None
    except Exception:
        # Module may be mid-import in edge bootstraps; fall through to
        # conservative checks below.
        raw = str(profile_identity or "")
        if ".." in raw or "/" in raw or "\\" in raw or "\x00" in raw:
            return None
    profile = current_profile_identity(profile_identity=profile_identity)
    root = _ensure_state_dir(state_dir)
    # Refuse symlink escape of the state root (#709 F3).
    try:
        real_root = os.path.realpath(root)
        path = state_file_path(
            kind=kind,
            remote=remote,
            org=org,
            repo=repo,
            profile_identity=profile,
            state_dir=root,
        )
        real_path = os.path.realpath(path) if os.path.exists(path) else path
        if os.path.exists(path) and not str(real_path).startswith(
            str(real_root) + os.sep
        ) and str(real_path) != str(real_root):
            return None
    except OSError:
        return None
    path = state_file_path(
        kind=kind,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
        state_dir=root,
    )
    lock_path = f"{path}.lock"
    with _exclusive_file_lock(lock_path):
        envelope = _read_json(path)
    if not envelope:
        return None
    payload = envelope.get("payload")
    if not isinstance(payload, dict):
        return None
    merged = dict(payload)
    for key in (
        "kind",
        "remote",
        "org",
        "repo",
        "profile_identity",
        "session_profile_lock",
        "recorded_at",
        "updated_at",
        "writer_pid",
    ):
        if key in envelope and key not in merged:
            merged[key] = envelope[key]
    stored = (merged.get("profile_identity") or "").strip()
    if stored and stored != profile:
        return None
    if skip_identity_match:
        # Enforce remote/org/repo when caller provides them (#709 F3).
        # Drop only *active session* profile-identity mismatches; keep
        # expiry, spoof, and repository-scope reasons.
        scope_remote = remote if enforce_repo_scope else None
        scope_org = org if enforce_repo_scope else None
        scope_repo = repo if enforce_repo_scope else None
        reasons = identity_match_reasons(
            merged,
            remote=scope_remote,
            org=scope_org,
            repo=scope_repo,
            profile_identity=stored or profile,
        )
        filtered = [
            r
            for r in reasons
            if "profile identity mismatch" not in r
            or (stored and stored != profile)
        ]
        # When caller requested a specific remote/org/repo, also fail if the
        # stored record lacks those identity fields entirely (legacy incomplete).
        if enforce_repo_scope:
            for field, want in (
                ("remote", remote),
                ("org", org),
                ("repo", repo),
            ):
                want_s = (want or "").strip()
                if not want_s:
                    continue
                have = (str(merged.get(field) or "")).strip()
                if not have:
                    filtered.append(
                        f"session state {field} missing on durable record "
                        f"(expected={want_s!r}; fail closed, #709 F3)"
                    )
                elif have != want_s:
                    # identity_match_reasons already adds mismatch; ensure kept
                    pass
        if filtered:
            return None
        return merged
    reasons = identity_match_reasons(
        merged,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
    )
    if reasons:
        return None
    return merged
@@ -0,0 +1,451 @@
 """MCP tool-boundary error mapping for known Gitea client failures (#699).
 Known authentication / authorization / network / configuration failures leave
 the tool boundary as a sanitized structured ``CallToolResult`` with
 ``isError=True``. Stdio transport remains connected.
 Design constraints (reviewer-ratified, #699 / PR #701):
 1. **No secret material** in tool results or daemon logs. Messages are fixed
   constants keyed by ``reason_code``; HTTP bodies / exception text never
   surface. Sanitization failure fails closed to ``internal_error``.
 2. **Narrow boundary** wraps the original FastMCP ``Tool.run`` success path;
   re-raises framework control-flow exceptions (``UrlElicitationRequiredError``);
   maps only typed client failures. Installation is idempotent.
 3. **No RuntimeError substring heuristics.** Only explicit typed
   authentication / authorization / network / configuration exceptions
   receive those labels.
 4. **HTTP classification** lives in ``gitea_auth.classify_http_status``;
   every 403 is authorization-class.
 """
 from __future__ import annotations
 import json
 import logging
 import sys
 from typing import Any
 logger = logging.getLogger("gitea_mcp.tool_error_boundary")
 # Stable reason codes (#699).
 REASON_AUTH_FAILED = "auth_failed"
 REASON_AUTH_INVALID_TOKEN = "auth_invalid_token"
 REASON_AUTHZ_INSUFFICIENT_SCOPE = "authz_insufficient_scope"
 REASON_AUTHZ_DENIED = "authz_denied"
 REASON_NETWORK_ERROR = "network_error"
 REASON_CONFIG_ERROR = "config_error"
 REASON_INTERNAL_ERROR = "internal_error"
 REASON_UPSTREAM_UNAVAILABLE = "upstream_unavailable"
 REASON_HTTP_ERROR = "http_error"
 ERROR_CLASS_AUTHENTICATION = "authentication"
 ERROR_CLASS_AUTHORIZATION = "authorization"
 ERROR_CLASS_NETWORK = "network"
 ERROR_CLASS_CONFIGURATION = "configuration"
 ERROR_CLASS_INTERNAL = "internal"
 ERROR_CLASS_UPSTREAM = "upstream"
 # Fixed, secret-free operator messages. Never interpolate HTTP bodies,
 # Keychain contents, tokens, or arbitrary exception text.
 FIXED_MESSAGES: dict[str, str] = {
    REASON_AUTH_FAILED: "Gitea authentication failed",
    REASON_AUTH_INVALID_TOKEN: (
        "Gitea authentication failed: invalid or revoked credentials"
    ),
    REASON_AUTHZ_INSUFFICIENT_SCOPE: (
        "Gitea authorization failed: insufficient token scope"
    ),
    REASON_AUTHZ_DENIED: "Gitea authorization failed: access denied",
    REASON_NETWORK_ERROR: "Network error contacting Gitea",
    REASON_CONFIG_ERROR: "Gitea configuration or credential resolution failed",
    REASON_INTERNAL_ERROR: "Internal tool error",
    REASON_UPSTREAM_UNAVAILABLE: "Gitea upstream unavailable",
    REASON_HTTP_ERROR: "Gitea HTTP request failed",
 }
 _INSTALL_FLAG = "_gitea_auth_boundary_installed"
 _ORIGINAL_ATTR = "_gitea_auth_boundary_original"
 def fixed_message(reason_code: str) -> str:
    """Return the fixed sanitized message for *reason_code* (fail closed)."""
    return FIXED_MESSAGES.get(reason_code, FIXED_MESSAGES[REASON_INTERNAL_ERROR])
 def _safe_profile_name() -> str | None:
    try:
        from gitea_auth import get_profile
        name = (get_profile() or {}).get("profile_name")
        if name is None:
            return None
        text = str(name).strip()
        # Profile names are non-secret identifiers; still bound length.
        return text[:80] if text else None
    except Exception:
        return None
 def _is_framework_control_flow(exc: BaseException) -> bool:
    """True for exceptions the MCP framework must re-raise unchanged."""
    try:
        from mcp.shared.exceptions import UrlElicitationRequiredError
        if isinstance(exc, UrlElicitationRequiredError):
            return True
    except Exception:
        pass
    # BaseException subclasses that must never become tool isError payloads.
    if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
        return True
    return False
 def is_known_client_failure(exc: BaseException) -> bool:
    """True only for explicit typed Gitea client / config failures.
    Never true for arbitrary ``RuntimeError`` (reviewer finding #3).
    """
    try:
        import gitea_auth
        if isinstance(
            exc,
            (
                gitea_auth.GiteaAuthError,
                gitea_auth.GiteaAuthzError,
                gitea_auth.GiteaNetworkError,
                gitea_auth.GiteaConfigError,
                gitea_auth.GiteaHttpError,
            ),
        ):
            return True
    except Exception:
        return False
    try:
        import gitea_config
        if isinstance(exc, gitea_config.ConfigError):
            return True
    except Exception:
        pass
    return False
 def classify_exception(exc: BaseException) -> dict[str, Any]:
    """Return structured classification with **fixed** messages only.
    Only typed authentication / authorization / network / configuration
    failures receive those labels. Unexpected programming failures are
    ``internal_error``. HTTP bodies and exception text are never copied
    into ``message``.
    """
    try:
        return _classify_exception_impl(exc)
    except Exception:
        # Fail closed: sanitization / classification failure must not leak.
        return {
            "reason_code": REASON_INTERNAL_ERROR,
            "error_class": ERROR_CLASS_INTERNAL,
            "http_status": None,
            "message": fixed_message(REASON_INTERNAL_ERROR),
            "transport_survives": True,
        }
 def _classify_exception_impl(exc: BaseException) -> dict[str, Any]:
    import gitea_auth
    if isinstance(exc, gitea_auth.GiteaAuthError):
        code = getattr(exc, "reason_code", None) or REASON_AUTH_INVALID_TOKEN
        if code not in (
            REASON_AUTH_FAILED,
            REASON_AUTH_INVALID_TOKEN,
        ):
            code = REASON_AUTH_INVALID_TOKEN
        return {
            "reason_code": code,
            "error_class": ERROR_CLASS_AUTHENTICATION,
            "http_status": getattr(exc, "http_status", None) or 401,
            "message": fixed_message(code),
            "transport_survives": True,
        }
    if isinstance(exc, gitea_auth.GiteaAuthzError):
        code = getattr(exc, "reason_code", None) or REASON_AUTHZ_DENIED
        if code not in (REASON_AUTHZ_DENIED, REASON_AUTHZ_INSUFFICIENT_SCOPE):
            code = REASON_AUTHZ_DENIED
        return {
            "reason_code": code,
            "error_class": ERROR_CLASS_AUTHORIZATION,
            "http_status": getattr(exc, "http_status", None) or 403,
            "message": fixed_message(code),
            "transport_survives": True,
        }
    if isinstance(exc, gitea_auth.GiteaNetworkError):
        return {
            "reason_code": REASON_NETWORK_ERROR,
            "error_class": ERROR_CLASS_NETWORK,
            "http_status": getattr(exc, "http_status", None),
            "message": fixed_message(REASON_NETWORK_ERROR),
            "transport_survives": True,
        }
    if isinstance(exc, gitea_auth.GiteaConfigError):
        return {
            "reason_code": REASON_CONFIG_ERROR,
            "error_class": ERROR_CLASS_CONFIGURATION,
            "http_status": getattr(exc, "http_status", None),
            "message": fixed_message(REASON_CONFIG_ERROR),
            "transport_survives": True,
        }
    if isinstance(exc, gitea_auth.GiteaHttpError):
        code = getattr(exc, "reason_code", None) or REASON_HTTP_ERROR
        if code == REASON_UPSTREAM_UNAVAILABLE:
            error_class = ERROR_CLASS_UPSTREAM
        else:
            error_class = ERROR_CLASS_INTERNAL
            code = REASON_HTTP_ERROR
        return {
            "reason_code": code,
            "error_class": error_class,
            "http_status": getattr(exc, "http_status", None),
            "message": fixed_message(code),
            "transport_survives": True,
        }
    try:
        import gitea_config
        if isinstance(exc, gitea_config.ConfigError):
            return {
                "reason_code": REASON_CONFIG_ERROR,
                "error_class": ERROR_CLASS_CONFIGURATION,
                "http_status": None,
                "message": fixed_message(REASON_CONFIG_ERROR),
                "transport_survives": True,
            }
    except Exception:
        pass
    # Unwrap FastMCP ToolError cause when the original was a typed failure.
    cause = getattr(exc, "__cause__", None)
    if cause is not None and cause is not exc and is_known_client_failure(cause):
        return _classify_exception_impl(cause)
    # No message-substring authentication heuristics (reviewer finding #3).
    return {
        "reason_code": REASON_INTERNAL_ERROR,
        "error_class": ERROR_CLASS_INTERNAL,
        "http_status": None,
        "message": fixed_message(REASON_INTERNAL_ERROR),
        "transport_survives": True,
    }
 def build_structured_error_payload(
    classification: dict[str, Any],
    *,
    tool_name: str | None = None,
    profile_name: str | None = None,
 ) -> dict[str, Any]:
    """LLM-safe structured payload — fixed message + typed metadata only."""
    try:
        reason = str(classification.get("reason_code") or REASON_INTERNAL_ERROR)
        message = fixed_message(reason)
        # Refuse to emit any classification message that is not the fixed constant.
        if classification.get("message") != message:
            message = fixed_message(reason)
        payload: dict[str, Any] = {
            "success": False,
            "isError": True,
            "reason_code": reason if reason in FIXED_MESSAGES else REASON_INTERNAL_ERROR,
            "error_class": classification.get("error_class") or ERROR_CLASS_INTERNAL,
            "message": message,
            "transport_survives": True,
            "retryable": classification.get("error_class")
            in {
                ERROR_CLASS_AUTHENTICATION,
                ERROR_CLASS_NETWORK,
                ERROR_CLASS_CONFIGURATION,
            },
        }
        status = classification.get("http_status")
        if isinstance(status, int):
            payload["http_status"] = status
        if tool_name and isinstance(tool_name, str):
            # Tool names are identifiers, not secrets; bound length.
            payload["tool"] = tool_name[:120]
        if profile_name and isinstance(profile_name, str):
            payload["profile"] = profile_name[:80]
        return payload
    except Exception:
        return {
            "success": False,
            "isError": True,
            "reason_code": REASON_INTERNAL_ERROR,
            "error_class": ERROR_CLASS_INTERNAL,
            "message": fixed_message(REASON_INTERNAL_ERROR),
            "transport_survives": True,
            "retryable": False,
        }
 def log_sanitized_daemon_reason(
    classification: dict[str, Any],
    *,
    tool_name: str | None = None,
    stream=None,
 ) -> None:
    """Daemon log: reason codes only — never exception text or response bodies."""
    stream = stream if stream is not None else sys.stderr
    try:
        reason = classification.get("reason_code") or REASON_INTERNAL_ERROR
        error_class = classification.get("error_class") or ERROR_CLASS_INTERNAL
        # Only emit known tokens; never classification['message'] from callers
        # that might have been poisoned.
        if reason not in FIXED_MESSAGES:
            reason = REASON_INTERNAL_ERROR
            error_class = ERROR_CLASS_INTERNAL
        parts = [
            "mcp_tool_error",
            f"reason_code={reason}",
            f"error_class={error_class}",
        ]
        if tool_name and isinstance(tool_name, str):
            parts.append(f"tool={tool_name[:120]}")
        status = classification.get("http_status")
        if isinstance(status, int):
            parts.append(f"http_status={status}")
        # Intentionally no detail= / message= field — secrets lived there.
        line = " ".join(parts)
        stream.write(line + "\n")
        if hasattr(stream, "flush"):
            stream.flush()
        logger.warning(line)
    except Exception:
        # Fail closed: never fall back to logging the exception.
        try:
            stream.write(
                "mcp_tool_error reason_code=internal_error "
                "error_class=internal\n"
            )
            if hasattr(stream, "flush"):
                stream.flush()
        except Exception:
            pass
 def to_call_tool_result(
    exc: BaseException,
    *,
    tool_name: str | None = None,
    profile_name: str | None = None,
    log: bool = True,
 ) -> Any:
    """Build a FastMCP ``CallToolResult`` with ``isError=True`` for *exc*."""
    from mcp.types import CallToolResult, TextContent
    try:
        classification = classify_exception(exc)
        if log:
            log_sanitized_daemon_reason(classification, tool_name=tool_name)
        payload = build_structured_error_payload(
            classification, tool_name=tool_name, profile_name=profile_name
        )
        text = json.dumps(payload, indent=2, sort_keys=True)
        return CallToolResult(
            content=[TextContent(type="text", text=text)],
            structuredContent=payload,
            isError=True,
        )
    except Exception:
        # Absolute fail-closed path — no exception text.
        fallback = {
            "success": False,
            "isError": True,
            "reason_code": REASON_INTERNAL_ERROR,
            "error_class": ERROR_CLASS_INTERNAL,
            "message": fixed_message(REASON_INTERNAL_ERROR),
            "transport_survives": True,
            "retryable": False,
        }
        return CallToolResult(
            content=[
                TextContent(
                    type="text",
                    text=json.dumps(fallback, indent=2, sort_keys=True),
                )
            ],
            structuredContent=fallback,
            isError=True,
        )
 def install_tool_run_boundary(Tool) -> bool:
    """Install a **narrow** error boundary around FastMCP ``Tool.run``.
    Strategy (preserves framework semantics):
    * Call the **original** ``Tool.run`` for the success path (async, return
      types, convert_result, protocol behavior unchanged).
    * Re-raise ``UrlElicitationRequiredError`` and other control-flow
      exceptions without mapping to ``internal_error``.
    * Map typed Gitea client failures (and ToolError whose ``__cause__`` is
      typed) to structured ``CallToolResult(isError=True)``.
    * Map remaining unexpected tool failures to fixed ``internal_error``
      isError results (transport survival) without secret-bearing text.
    * Idempotent: second install is a no-op and returns ``False``.
    """
    if getattr(Tool.run, _INSTALL_FLAG, False):
        return False
    from mcp.server.fastmcp.exceptions import ToolError
    from mcp.shared.exceptions import UrlElicitationRequiredError
    original_run = Tool.run
    async def run_boundary(
        self,
        arguments: dict[str, Any],
        context=None,
        convert_result: bool = False,
    ) -> Any:
        try:
            return await original_run(
                self,
                arguments,
                context=context,
                convert_result=convert_result,
            )
        except UrlElicitationRequiredError:
            # Framework control-flow — must not become internal_error.
            raise
        except BaseException as exc:
            if _is_framework_control_flow(exc):
                raise
            if not isinstance(exc, Exception):
                raise
            # Prefer typed cause under FastMCP ToolError wrappers.
            target: BaseException = exc
            if isinstance(exc, ToolError) and exc.__cause__ is not None:
                target = exc.__cause__
            # Known client failures → structured isError with reason codes.
            # Unexpected failures → fixed internal_error isError (no secrets).
            profile_name = _safe_profile_name()
            return to_call_tool_result(
                target,
                tool_name=getattr(self, "name", None),
                profile_name=profile_name,
            )
    setattr(run_boundary, _INSTALL_FLAG, True)
    setattr(run_boundary, _ORIGINAL_ATTR, original_run)
    Tool.run = run_boundary  # type: ignore[method-assign]
    return True
 def boundary_is_installed(Tool) -> bool:
    """True when the #699 boundary is active on *Tool.run*."""
    return bool(getattr(Tool.run, _INSTALL_FLAG, False))
@@ -20,12 +20,114 @@ if PROJECT_ROOT not in sys.path:
 import gitea_config
-AUTHOR_DEFAULT_ALLOWED = ["read", "branch", "commit", "push", "open_pr", "comment"]
+# Defaults emit *canonical* operation names only. Shorthand that is not in
-AUTHOR_DEFAULT_FORBIDDEN = ["approve", "request_changes", "merge"]
+# gitea_config.GITEA_OPERATION_ALIASES (e.g. ``pr.close``, ``issue.close``)
-REVIEWER_DEFAULT_ALLOWED = [
+# is silently dropped by the production loader and must never appear here.
-    "read", "review", "comment", "approve", "request_changes", "merge"
+AUTHOR_DEFAULT_ALLOWED = [
    "gitea.read",
    "gitea.branch.create",
    "gitea.repo.commit",
    "gitea.branch.push",
    "gitea.pr.create",
    "gitea.pr.comment",
 ]
-REVIEWER_DEFAULT_FORBIDDEN = ["branch", "commit", "push", "open_pr"]
+AUTHOR_DEFAULT_FORBIDDEN = [
    "gitea.pr.approve",
    "gitea.pr.request_changes",
    "gitea.pr.merge",
 ]
 REVIEWER_DEFAULT_ALLOWED = [
    "gitea.read",
    "gitea.pr.review",
    "gitea.pr.comment",
    "gitea.pr.approve",
    "gitea.pr.request_changes",
    "gitea.pr.merge",
 ]
 REVIEWER_DEFAULT_FORBIDDEN = [
    "gitea.branch.create",
    "gitea.repo.commit",
    "gitea.branch.push",
    "gitea.pr.create",
 ]
 # Required reconciler ops (read + pr.close) plus recommended comment/close and
 # branch.delete for guarded merged-PR cleanup. All names must normalize via
 # gitea_config.normalize_operation without being dropped.
 RECONCILER_DEFAULT_ALLOWED = [
    "gitea.read",
    "gitea.pr.close",
    "gitea.pr.comment",
    "gitea.issue.comment",
    "gitea.issue.close",
    "gitea.branch.delete",
 ]
 RECONCILER_DEFAULT_FORBIDDEN = [
    "gitea.pr.approve",
    "gitea.pr.merge",
    "gitea.pr.review",
    "gitea.pr.create",
    "gitea.branch.push",
    "gitea.repo.commit",
 ]
 # Migration-only expansions for common shorthands that are *not* in
 # GITEA_OPERATION_ALIASES. Emitted output is always the canonical form so a
 # second canonicalize pass is a no-op (idempotent).
 _MIGRATION_ONLY_ALIASES = {
    "pr.close": "gitea.pr.close",
    "pr.comment": "gitea.pr.comment",
    "issue.close": "gitea.issue.close",
    "branch.delete": "gitea.branch.delete",
 }
 # Reconciler required ops that must survive migration (from reconciler_profile).
 RECONCILER_REQUIRED_CANONICAL = ("gitea.read", "gitea.pr.close")
 def canonicalize_operation(op: str) -> str:
    """Return a canonical operation name accepted by the production loader.
    Fail closed on unknown/ambiguous spellings so required permissions cannot
    be silently dropped by ``check_operation`` later.
    """
    if not isinstance(op, str) or not op.strip():
        raise ValueError("operation must be a non-empty string (fail closed)")
    op = op.strip()
    try:
        return gitea_config.normalize_operation(op)
    except gitea_config.ConfigError:
        pass
    if op in _MIGRATION_ONLY_ALIASES:
        return _MIGRATION_ONLY_ALIASES[op]
    raise ValueError(
        f"operation {op!r} cannot be canonicalized for migration "
        "(unknown/ambiguous; fail closed — production loader would drop it)"
    )
 def canonicalize_operations(ops, *, context: str = "operations") -> list[str]:
    """Canonicalize a list of operations; preserve order, drop duplicates."""
    if not isinstance(ops, list):
        raise ValueError(f"{context} must be a list (fail closed)")
    out: list[str] = []
    seen: set[str] = set()
    for entry in ops:
        canon = canonicalize_operation(entry)
        if canon not in seen:
            seen.add(canon)
            out.append(canon)
    return out
 def _assert_reconciler_required_survive(allowed: list[str], profile_name: str) -> None:
    """Fail visibly when migration would leave a reconciler without required ops."""
    missing = [op for op in RECONCILER_REQUIRED_CANONICAL if op not in set(allowed)]
    if missing:
        raise ValueError(
            f"Profile '{profile_name}' (reconciler) is missing required "
            f"operation(s) after migration: {missing}. Refusing to emit a "
            "profile that would silently fail pr.close / read (fail closed)."
        )
 def infer_role(name, execution_profile):
@@ -90,9 +192,11 @@ def migrate_v1_to_v2(v1_data):
            ident_name = "reviewer"
        elif role == "author":
            ident_name = "author"
        elif role == "reconciler":
            ident_name = "reconciler"
        else:
            role = prof.get("role")
-            if role not in (None, "author", "reviewer"):
+            if role not in (None, "author", "reviewer", "reconciler"):
                raise ValueError(
                    f"Profile '{name}' has unsupported role {role!r}"
                )
@@ -124,20 +228,35 @@ def migrate_v1_to_v2(v1_data):
                raise ValueError(
                    f"Profile '{name}' operation fields must be lists"
                )
-            identity_data["allowed_operations"] = list(allowed)
+            try:
-            identity_data["forbidden_operations"] = list(forbidden)
+                identity_data["allowed_operations"] = canonicalize_operations(
                    allowed, context=f"profile '{name}' allowed_operations"
                )
                identity_data["forbidden_operations"] = canonicalize_operations(
                    forbidden, context=f"profile '{name}' forbidden_operations"
                )
            except ValueError as exc:
                raise ValueError(f"Profile '{name}': {exc}") from exc
        elif role == "author":
            identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED)
            identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN)
        elif role == "reviewer":
            identity_data["allowed_operations"] = list(REVIEWER_DEFAULT_ALLOWED)
            identity_data["forbidden_operations"] = list(REVIEWER_DEFAULT_FORBIDDEN)
        elif role == "reconciler":
            identity_data["allowed_operations"] = list(RECONCILER_DEFAULT_ALLOWED)
            identity_data["forbidden_operations"] = list(RECONCILER_DEFAULT_FORBIDDEN)
        else:
            raise ValueError(
                f"Profile '{name}' has no explicit operation lists and no "
                "unambiguous author/reviewer role marker (fail closed)"
            )
        if role == "reconciler":
            _assert_reconciler_required_survive(
                identity_data["allowed_operations"], name
            )
        # Nest inside environments/services structure
        env = environments.setdefault(env_name, {})
        services = env.setdefault("services", {})
@@ -18,6 +18,11 @@ RECONCILER_RECOMMENDED_OPERATIONS = (
    "gitea.pr.comment",
    "gitea.issue.comment",
    "gitea.issue.close",
    # Merged-branch cleanup is reconciler-owned (task_capability_map maps
    # cleanup_merged_pr_branch -> reconciler). The permission is only
    # exercisable through the guarded gitea_cleanup_merged_pr_branch path
    # (#514): merged proof, protected-branch refusal, explicit confirmation.
    "gitea.branch.delete",
 )
 RECONCILER_FORBIDDEN_OPERATIONS = (
@@ -438,3 +438,252 @@ def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
        "This path only clears a lock when the referenced PR is merged/closed.",
    ]
    return "\n".join(lines)
 # ---------------------------------------------------------------------------
 # #709 cross-profile cleanup, overwrite protection, recovery provenance
 # ---------------------------------------------------------------------------
 def has_unresolved_terminal_evidence(lock: dict | None) -> bool:
    """True when *lock* still carries a terminal live mutation ledger."""
    return last_terminal_mutation(lock) is not None
 def assess_init_overwrite(
    existing_lock: dict | None,
    *,
    force: bool = False,
 ) -> dict[str, Any]:
    """Whether init_review_decision_lock may replace an existing durable lock (#709 AC2).
    Unresolved terminal evidence must never be silently replaced by an empty
    initialized lock. *force* does not authorize destruction of terminal
    ledgers — only explicit cleanup / archive transitions may remove them.
    """
    result: dict[str, Any] = {
        "overwrite_allowed": True,
        "has_existing": existing_lock is not None,
        "has_unresolved_terminal": False,
        "reasons": [],
        "last_terminal_pr": None,
        "last_terminal_action": None,
        "existing_profile_identity": None,
    }
    if existing_lock is None:
        result["reasons"].append("no existing lock; empty init allowed")
        return result
    result["existing_profile_identity"] = (
        existing_lock.get("profile_identity")
        or existing_lock.get("session_profile_lock")
        or existing_lock.get("session_profile")
    )
    last = last_terminal_mutation(existing_lock)
    if last is None:
        result["reasons"].append(
            "existing lock has no terminal mutations; re-init allowed"
        )
        return result
    result["has_unresolved_terminal"] = True
    result["overwrite_allowed"] = False
    result["last_terminal_pr"] = last.get("pr_number")
    result["last_terminal_action"] = last.get("action")
    result["reasons"].append(
        "refuse empty re-init: unresolved terminal decision-lock evidence "
        f"present ({last.get('action')} on PR #{last.get('pr_number')}); "
        "use gitea_cleanup_stale_review_decision_lock when moot, or archive "
        f"via sanctioned recovery — force={force!r} does not authorize overwrite "
        "(#709 AC2)"
    )
    return result
 def lock_targets_merged_pr_approval(
    lock: dict | None,
    *,
    pr_number: int,
    expected_head_sha: str | None = None,
 ) -> bool:
    """True when *lock*'s last terminal mutation is approve of *pr_number*.
    When *expected_head_sha* is provided, a **recorded** terminal head must
    match. Legacy same-repo approve locks with no recorded head do **not**
    match (fail closed, #709 F3 residual / review 435) — callers must use the
    strict secondary path that refuses no-head clears.
    """
    last = last_terminal_mutation(lock)
    if last is None:
        return False
    if last.get("action") != "approve":
        return False
    if last.get("pr_number") != pr_number:
        return False
    if expected_head_sha:
        want = normalize_head_sha(expected_head_sha)
        if not want:
            return False
        locked = mutation_head_sha(last, lock)
        # Require recorded-head match; unrecorded head is not a match (#709 F3).
        if not locked or not heads_equal(locked, want):
            return False
    return True
 def build_post_merge_recovery_record(
    *,
    pr_number: int,
    head_sha: str | None,
    merge_commit_sha: str | None,
    target_profile_identity: str | None,
    failed_step: str,
    error: str | None,
    remote: str | None,
    org: str | None,
    repo: str | None,
    actor_username: str | None,
    profile_name: str | None,
 ) -> dict[str, Any]:
    """Durable recovery-required payload after irreversible merge (#709 AC3)."""
    return {
        "event": "post_merge_decision_lock_recovery_required",
        "status": "recovery_required",
        "issue_ref": "#709",
        "recovery_critical": True,
        "applied": False,  # never claim historical cleanup succeeded
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "pr_number": pr_number,
        "head_sha": normalize_head_sha(head_sha),
        "merge_commit_sha": merge_commit_sha,
        "target_profile_identity": target_profile_identity,
        "failed_step": failed_step,
        "error": error,
        "remote": remote,
        "org": org,
        "repo": repo,
        "actor_username": actor_username,
        "profile_name": profile_name,
        "required_recovery_action": (
            "retry cross-profile decision-lock cleanup and audit publication "
            "for the merged PR; if terminal evidence is gone, use "
            "gitea_record_irrecoverable_decision_lock_provenance"
        ),
    }
 def build_irrecoverable_provenance_record(
    *,
    pr_number: int,
    head_sha: str | None,
    remote: str | None,
    org: str | None,
    repo: str | None,
    actor_username: str | None,
    profile_name: str | None,
    reason: str,
    incident_ref: str | None = None,
    # Deprecated kwargs retained only so stale call sites fail closed:
    operator_authorized: bool | None = None,
    # Required for merger-acceptable records (#709 F1):
    authorization: dict[str, Any] | None = None,
    incident_issue: int | None = None,
    incident_comment_id: int | None = None,
    destroyed_subject: str | None = None,
    historical_provenance_subject: str | None = None,
 ) -> dict[str, Any]:
    """Truthful absence-of-proof record (#709 AC5). Never sets applied=True.
    Caller-supplied ``operator_authorized`` is **ignored** as authorization
    evidence (review 434 F1). Prefer
    :func:`irrecoverable_provenance.build_irrecoverable_provenance_record`
    with a verified server-side authorization artifact.
    """
    # Explicitly ignore deprecated self-assertable Boolean.
    _ = operator_authorized
    if authorization is not None and incident_issue is not None and incident_comment_id is not None:
        from irrecoverable_provenance import (
            build_irrecoverable_provenance_record as _build,
        )
        return _build(
            pr_number=int(pr_number),
            head_sha=str(head_sha or ""),
            remote=str(remote or ""),
            org=str(org or ""),
            repo=str(repo or ""),
            actor_username=actor_username,
            profile_name=profile_name,
            reason=reason,
            incident_issue=int(incident_issue),
            incident_comment_id=int(incident_comment_id),
            authorization=authorization,
            destroyed_subject=destroyed_subject,
            historical_provenance_subject=historical_provenance_subject
            or destroyed_subject,
        )
    # Fail-closed skeleton when no server authorization is supplied: never
    # sets merger_may_accept True (even if operator_authorized was True).
    return {
        "event": "irrecoverable_decision_lock_provenance",
        "status": "provenance_irrecoverable",
        "record_type": "irrecoverable_decision_provenance",
        "operator_recovery_required": True,
        "issue_ref": "#709",
        "recovery_critical": True,
        "applied": False,
        "historical_cleanup_proven": False,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "pr_number": pr_number,
        "head_sha": normalize_head_sha(head_sha),
        "remote": remote,
        "org": org,
        "repo": repo,
        "actor_username": actor_username,
        "profile_name": profile_name,
        "reason": reason,
        "incident_ref": incident_ref,
        "incident_issue": incident_issue,
        "incident_comment_id": incident_comment_id,
        "authorization_verified": False,
        "merger_may_accept": False,
        "acceptance_rule": (
            "Merger may accept this record only when a server-side "
            "authorization artifact verifies for remote/org/repo/PR/exact "
            "head/incident, the record is durable and read back, and normal "
            "merge gates still pass. Caller Booleans never authorize. This "
            "does not prove historical cleanup."
        ),
    }
 def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str:
    """Markdown body for irrecoverable provenance audit (no applied=true claim)."""
    from irrecoverable_provenance import (
        format_irrecoverable_audit_comment as _fmt,
    )
    return _fmt(record)
 def format_post_merge_recovery_comment(record: dict[str, Any]) -> str:
    """Markdown body for post-merge recovery-required audit."""
    lines = [
        "## Post-merge decision-lock recovery required (#709)",
        "",
        "Status: **RECOVERY_REQUIRED** (merge is irreversible; cleanup/audit incomplete)",
        "",
        f"- actor: `{record.get('actor_username')}`",
        f"- profile: `{record.get('profile_name')}`",
        f"- timestamp: `{record.get('timestamp')}`",
        f"- PR: `#{record.get('pr_number')}`",
        f"- head_sha: `{record.get('head_sha')}`",
        f"- merge_commit_sha: `{record.get('merge_commit_sha')}`",
        f"- target_profile_identity: `{record.get('target_profile_identity')}`",
        f"- failed_step: `{record.get('failed_step')}`",
        f"- error: `{record.get('error')}`",
        "",
        f"Required action: {record.get('required_recovery_action')}",
        "",
        "applied=false — do not treat this as successful cleanup evidence.",
    ]
    return "\n".join(lines)
@@ -66,7 +66,7 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
    },
    "review_pr": {
        "permission": "gitea.pr.review",
-        "role": "reviewer",
+        "role": "merger",
    },
    "merge_pr": {
        "permission": "gitea.pr.merge",
@@ -84,48 +84,74 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
    },
    "adopt_merger_pr_lease": {
        "permission": "gitea.pr.comment",
-        "role": "reviewer",
+        "role": "merger",
    },
    # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
    # Apply path posts lease release + audit comments (gitea.pr.comment).
    "cleanup_obsolete_reviewer_comment_lease": {
        "permission": "gitea.pr.comment",
-        "role": "reviewer",
+        "role": "merger",
    },
    "gitea_cleanup_obsolete_reviewer_comment_lease": {
        "permission": "gitea.pr.comment",
-        "role": "reviewer",
+        "role": "merger",
    },
    "blind_pr_queue_review": {
        "permission": "gitea.pr.review",
-        "role": "reviewer",
+        "role": "merger",
    },
    "pr_queue_cleanup": {
        "permission": "gitea.pr.review",
-        "role": "reviewer",
+        "role": "merger",
    },
    "pr-queue-cleanup": {
        "permission": "gitea.pr.review",
-        "role": "reviewer",
+        "role": "merger",
    },
    "request_changes_pr": {
        "permission": "gitea.pr.request_changes",
-        "role": "reviewer",
+        "role": "merger",
    },
    "approve_pr": {
        "permission": "gitea.pr.approve",
-        "role": "reviewer",
+        "role": "merger",
    },
    # #594: clear durable #332 decision lock only when last terminal PR is
    # already merged/closed (moot). Apply path requires reviewer review
    # permission; assessment itself uses gitea.read inside the tool.
    "cleanup_stale_review_decision_lock": {
        "permission": "gitea.pr.review",
-        "role": "reviewer",
+        "role": "merger",
    },
    "gitea_cleanup_stale_review_decision_lock": {
        "permission": "gitea.pr.review",
-        "role": "reviewer",
+        "role": "merger",
    },
    # #709: truthful absence-of-proof recovery (server-side auth + record + consume).
    # Dedicated mutation capability — gitea.read is insufficient (review 434 F1).
    "issue_irrecoverable_provenance_authorization": {
        "permission": "gitea.decision_lock.irrecoverable_recovery",
        "role": "reconciler",
    },
    "gitea_issue_irrecoverable_provenance_authorization": {
        "permission": "gitea.decision_lock.irrecoverable_recovery",
        "role": "reconciler",
    },
    "record_irrecoverable_decision_lock_provenance": {
        "permission": "gitea.decision_lock.irrecoverable_recovery",
        "role": "reconciler",
    },
    "gitea_record_irrecoverable_decision_lock_provenance": {
        "permission": "gitea.decision_lock.irrecoverable_recovery",
        "role": "reconciler",
    },
    "consume_irrecoverable_decision_lock_provenance": {
        "permission": "gitea.pr.merge",
        "role": "merger",
    },
    "gitea_consume_irrecoverable_decision_lock_provenance": {
        "permission": "gitea.pr.merge",
        "role": "merger",
    },
    "delete_branch": {
        "permission": "gitea.branch.delete",
@@ -79,41 +79,46 @@ class TestApiRequestFailures(unittest.TestCase):
    @patch("gitea_auth.urllib.request.urlopen")
    def test_timeout_converted_to_runtimeerror(self, mock_open):
        mock_open.side_effect = TimeoutError("timed out")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
-        self.assertIn("network error contacting Gitea", str(ctx.exception))
+        # Fixed message only (#699) — still a RuntimeError subclass.
        self.assertIsInstance(ctx.exception, RuntimeError)
        self.assertEqual(str(ctx.exception), "Network error contacting Gitea")
    @patch("gitea_auth.urllib.request.urlopen")
    def test_dns_network_failure_converted(self, mock_open):
        mock_open.side_effect = urllib.error.URLError("Name or service not known")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
-        self.assertIn("network error contacting Gitea", str(ctx.exception))
+        self.assertEqual(str(ctx.exception), "Network error contacting Gitea")
    @patch("gitea_auth.urllib.request.urlopen")
    def test_502_upstream_message(self, mock_open):
        mock_open.side_effect = http_error(502, "bad gateway")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        msg = str(ctx.exception)
-        self.assertIn("HTTP 502", msg)
+        self.assertEqual(msg, "Gitea upstream unavailable")
-        self.assertIn("upstream unavailable", msg)
+        self.assertEqual(ctx.exception.reason_code, "upstream_unavailable")
        self.assertEqual(ctx.exception.http_status, 502)
    @patch("gitea_auth.urllib.request.urlopen")
    def test_503_upstream_message(self, mock_open):
        mock_open.side_effect = http_error(503, "")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
-        self.assertIn("HTTP 503", str(ctx.exception))
+        self.assertEqual(str(ctx.exception), "Gitea upstream unavailable")
-        self.assertIn("upstream unavailable", str(ctx.exception))
+        self.assertEqual(ctx.exception.http_status, 503)
    @patch("gitea_auth.urllib.request.urlopen")
    def test_malformed_error_payload_does_not_crash(self, mock_open):
-        # Non-JSON garbage error body must still yield a clean RuntimeError.
+        # Non-JSON garbage error body must still yield a clean typed error
        # with a fixed message (no body echo — #699).
        mock_open.side_effect = http_error(500, "<html>garbage</html>")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
-        self.assertIn("HTTP 500", str(ctx.exception))
+        self.assertEqual(str(ctx.exception), "Gitea HTTP request failed")
        self.assertNotIn("<html>", str(ctx.exception))
    @patch("gitea_auth.urllib.request.urlopen")
    def test_malformed_success_json_raises_clean_error(self, mock_open):
@@ -126,16 +131,17 @@ class TestApiRequestFailures(unittest.TestCase):
    def test_no_secret_leak_in_error_body(self, mock_open):
        mock_open.side_effect = http_error(
            400, "failed: token supersecret123 rejected")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        msg = str(ctx.exception)
        self.assertNotIn("supersecret123", msg)
-        self.assertIn(gitea_audit.REDACTED, msg)
+        # Fixed message — body never appears (stronger than redaction).
        self.assertEqual(msg, "Gitea HTTP request failed")
    @patch("gitea_auth.urllib.request.urlopen")
    def test_auth_header_never_in_error(self, mock_open):
        mock_open.side_effect = http_error(400, "bad request")
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertNotIn(FAKE_AUTH, str(ctx.exception))
@@ -27,7 +27,13 @@ from task_capability_map import required_permission, required_role
 DELETE_PROFILE = {
    "profile_name": "prgs-author-delete",
-    "allowed_operations": ["gitea.read", "gitea.branch.delete"],
+    "role": "author",
    "allowed_operations": [
        "gitea.read",
        "gitea.pr.create",
        "gitea.branch.push",
        "gitea.branch.delete",
    ],
    "forbidden_operations": [],
    "audit_label": "prgs-author-delete",
 }
@@ -1446,10 +1446,16 @@ class TestReviewPR(unittest.TestCase):
 class TestDeleteBranch(unittest.TestCase):
    DELETE_PROFILE = {
-        "profile_name": "test-deleter",
+        "profile_name": "test-author-deleter",
-        "allowed_operations": ["gitea.read", "gitea.branch.delete"],
+        "role": "author",
        "allowed_operations": [
            "gitea.read",
            "gitea.pr.create",
            "gitea.branch.push",
            "gitea.branch.delete",
        ],
        "forbidden_operations": [],
-        "audit_label": "test-deleter",
+        "audit_label": "test-author-deleter",
    }
    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
@@ -92,14 +92,19 @@ class TestMigrateProfiles(unittest.TestCase):
        author = prgs_gitea["identities"]["author"]
        self.assertEqual(author["username"], "jcwalker3")
        self.assertEqual(author["auth"]["id"], "redacted-author-ref")
-        self.assertEqual(author["allowed_operations"], ["read", "comment"])
+        self.assertEqual(
-        self.assertEqual(author["forbidden_operations"], ["approve", "merge"])
+            author["allowed_operations"], ["gitea.read", "gitea.pr.comment"]
        )
        self.assertEqual(
            author["forbidden_operations"],
            ["gitea.pr.approve", "gitea.pr.merge"],
        )
        reviewer = prgs_gitea["identities"]["reviewer"]
        self.assertEqual(reviewer["role"], "reviewer")
        self.assertEqual(reviewer["username"], "sysadmin")
        self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref")
-        self.assertIn("merge", reviewer["allowed_operations"])
+        self.assertIn("gitea.pr.merge", reviewer["allowed_operations"])
    def test_alias_generation(self):
        """Test that aliases are correctly generated to support old profile names."""
@@ -188,7 +193,7 @@ class TestMigrateProfiles(unittest.TestCase):
        self.assertNotIn("token", stdout_output.lower())
    def test_explicit_operations_are_preserved(self):
-        """Explicit v1 permissions must not be replaced by role defaults."""
+        """Explicit v1 permissions are canonicalized, not replaced by role defaults."""
        v1_data = json.loads(json.dumps(self.v1_content))
        v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"]
        v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"]
@@ -198,8 +203,8 @@ class TestMigrateProfiles(unittest.TestCase):
            v2_data["environments"]["prgs"]["services"]["gitea"]
            ["identities"]["reviewer"]
        )
-        self.assertEqual(reviewer["allowed_operations"], ["read"])
+        self.assertEqual(reviewer["allowed_operations"], ["gitea.read"])
-        self.assertEqual(reviewer["forbidden_operations"], ["merge"])
+        self.assertEqual(reviewer["forbidden_operations"], ["gitea.pr.merge"])
    def test_inferred_role_defaults_only_when_unambiguous(self):
        """Role defaults are allowed only for clear author/reviewer profiles."""
@@ -306,6 +311,171 @@ class TestMigrateProfiles(unittest.TestCase):
                migrate_profiles.main()
            self.assertEqual(cm.exception.code, 1)
    def test_reconciler_profile_migration(self):
        """Legacy reconciler shorthands migrate to valid canonical operations."""
        import gitea_config
        import reconciler_profile
        v1_data = {
            "version": 1,
            "profiles": {
                "prgs-reconciler": {
                    "base_url": "redacted-prgs-service",
                    "username": "reconciler-agent",
                    "auth": {"type": "keychain", "id": "reconciler-ref"},
                    "execution_profile": "prgs-reconciler",
                    "allowed_operations": [
                        "read",
                        "pr.close",
                        "pr.comment",
                        "issue.comment",
                        "issue.close",
                        "gitea.branch.delete",
                    ],
                    "forbidden_operations": [
                        "merge",
                        "approve",
                        "review",
                        "pr.create",
                        "branch.push",
                        "commit",
                    ],
                }
            }
        }
        v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
        reconciler = (
            v2_data["environments"]["prgs"]["services"]["gitea"]
            ["identities"]["reconciler"]
        )
        self.assertEqual(reconciler["role"], "reconciler")
        allowed = reconciler["allowed_operations"]
        forbidden = reconciler["forbidden_operations"]
        # No invalid shorthand remains
        for bad in ("pr.close", "pr.comment", "issue.close", "read", "merge"):
            self.assertNotIn(bad, allowed)
            self.assertNotIn(bad, forbidden)
        for required in (
            "gitea.read",
            "gitea.pr.close",
            "gitea.pr.comment",
            "gitea.issue.comment",
            "gitea.issue.close",
            "gitea.branch.delete",
        ):
            self.assertIn(required, allowed)
            # Production loader accepts every allowed op
            self.assertEqual(
                gitea_config.normalize_operation(required), required
            )
        self.assertEqual(v2_data["aliases"]["prgs-reconciler"], "prgs.gitea.reconciler")
        assessment = reconciler_profile.assess_reconciler_profile(allowed, forbidden)
        self.assertTrue(assessment["valid"])
        self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
    def test_reconciler_profile_defaults(self):
        """Reconciler defaults are fully canonical and loader-valid."""
        import gitea_config
        import reconciler_profile
        v1_data = {
            "version": 1,
            "profiles": {
                "prgs-reconciler": {
                    "base_url": "redacted-prgs-service",
                    "username": "reconciler-agent",
                    "auth": {"type": "keychain", "id": "reconciler-ref"},
                    "execution_profile": "prgs-reconciler",
                }
            }
        }
        v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
        reconciler = (
            v2_data["environments"]["prgs"]["services"]["gitea"]
            ["identities"]["reconciler"]
        )
        self.assertEqual(reconciler["role"], "reconciler")
        self.assertEqual(
            reconciler["allowed_operations"],
            migrate_profiles.RECONCILER_DEFAULT_ALLOWED,
        )
        self.assertEqual(
            reconciler["forbidden_operations"],
            migrate_profiles.RECONCILER_DEFAULT_FORBIDDEN,
        )
        for op in reconciler["allowed_operations"]:
            self.assertEqual(gitea_config.normalize_operation(op), op)
            self.assertTrue(op.startswith("gitea."))
        assessment = reconciler_profile.assess_reconciler_profile(
            reconciler["allowed_operations"],
            reconciler["forbidden_operations"],
        )
        self.assertTrue(assessment["valid"])
        self.assertNotIn(
            "gitea.branch.delete", assessment["missing_recommended_operations"]
        )
    def test_reconciler_migration_idempotent_canonicalize(self):
        """Second canonicalize of already-canonical ops is a no-op."""
        first = migrate_profiles.canonicalize_operations(
            list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED)
        )
        second = migrate_profiles.canonicalize_operations(first)
        self.assertEqual(first, second)
        self.assertEqual(first, list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED))
    def test_reconciler_missing_required_fails_visibly(self):
        """Missing gitea.pr.close after migration fails closed (not silent drop)."""
        v1_data = {
            "version": 1,
            "profiles": {
                "prgs-reconciler": {
                    "base_url": "redacted-prgs-service",
                    "username": "reconciler-agent",
                    "auth": {"type": "keychain", "id": "reconciler-ref"},
                    "execution_profile": "prgs-reconciler",
                    "allowed_operations": ["read", "gitea.branch.delete"],
                    "forbidden_operations": ["merge"],
                }
            },
        }
        with self.assertRaisesRegex(ValueError, "missing required"):
            migrate_profiles.migrate_v1_to_v2(v1_data)
    def test_unknown_operation_fails_visibly(self):
        v1_data = {
            "version": 1,
            "profiles": {
                "prgs-author": {
                    "base_url": "redacted-prgs-service",
                    "username": "jcwalker3",
                    "auth": {"type": "keychain", "id": "hidden-author-ref"},
                    "execution_profile": "prgs-author",
                    "allowed_operations": ["read", "not.a.real.op"],
                    "forbidden_operations": ["merge"],
                }
            },
        }
        with self.assertRaisesRegex(ValueError, "cannot be canonicalized"):
            migrate_profiles.migrate_v1_to_v2(v1_data)
    def test_role_inference_author_reviewer_merger_reconciler(self):
        self.assertEqual(
            migrate_profiles.infer_role("prgs-author", "prgs-author"), "author"
        )
        self.assertEqual(
            migrate_profiles.infer_role("prgs-reviewer", "prgs-reviewer"),
            "reviewer",
        )
        self.assertEqual(
            migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
            "reconciler",
        )
        self.assertIsNone(
            migrate_profiles.infer_role("prgs-merger", "prgs-merger")
        )
 if __name__ == "__main__":
    unittest.main()
@@ -82,6 +82,42 @@ class TestReconcilerProfileModel(unittest.TestCase):
            "reconciler",
        )
    def test_branch_delete_is_recommended_for_reconciler(self):
        self.assertIn(
            "gitea.branch.delete",
            reconciler_profile.RECONCILER_RECOMMENDED_OPERATIONS,
        )
        self.assertNotIn(
            "gitea.branch.delete",
            reconciler_profile.RECONCILER_REQUIRED_OPERATIONS,
        )
    def test_reconciler_with_branch_delete_stays_valid(self):
        allowed = PRGS_RECONCILER_ALLOWED + ["gitea.branch.delete"]
        result = reconciler_profile.assess_reconciler_profile(
            allowed,
            PRGS_RECONCILER_FORBIDDEN,
        )
        self.assertTrue(result["is_reconciler_profile"])
        self.assertTrue(result["valid"])
        self.assertNotIn(
            "gitea.branch.delete", result["missing_recommended_operations"]
        )
        self.assertEqual(
            mcp_server._role_kind(allowed, PRGS_RECONCILER_FORBIDDEN),
            "reconciler",
        )
    def test_reconciler_without_branch_delete_reports_missing_recommended(self):
        result = reconciler_profile.assess_reconciler_profile(
            PRGS_RECONCILER_ALLOWED,
            PRGS_RECONCILER_FORBIDDEN,
        )
        self.assertTrue(result["valid"])
        self.assertIn(
            "gitea.branch.delete", result["missing_recommended_operations"]
        )
 if __name__ == "__main__":
    unittest.main()
@@ -122,9 +122,11 @@ class TestApiRequestRetry(unittest.TestCase):
        sleep.assert_not_called()
    def test_non_429_error_raises_immediately(self):
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            _call([_http_error(500, body=b"boom")])
-        self.assertIn("HTTP 500", str(ctx.exception))
+        # Fixed message only (#699) — no response body echo.
        self.assertEqual(str(ctx.exception), "Gitea HTTP request failed")
        self.assertEqual(ctx.exception.http_status, 500)
    def test_non_429_error_does_not_sleep(self):
        sleep = MagicMock()
@@ -171,9 +173,12 @@ class TestApiRequestRetry(unittest.TestCase):
        # max_retries=3 -> 3 sleeps, then the 4th failure raises.
        errors = [_http_error(429, retry_after="1") for _ in range(4)]
        sleep = MagicMock()
-        with self.assertRaises(RuntimeError) as ctx:
+        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            _call(errors, sleep_func=sleep, max_retries=3)
-        self.assertIn("HTTP 429", str(ctx.exception))
+        # After retries are exhausted, 429 is a typed HTTP error with fixed
        # message (#699); status metadata carries 429.
        self.assertEqual(str(ctx.exception), "Gitea HTTP request failed")
        self.assertEqual(ctx.exception.http_status, 429)
        self.assertEqual(sleep.call_count, 3)
    def test_no_infinite_loop_when_always_429(self):
@@ -0,0 +1,139 @@
 """Documentation acceptance for the stable control runtime ADR (#615 / PR #616).
 Enforces review 443 remediation:
 * F1 — operator guide / runbooks cross-link the ADR (issue #615 AC2).
 * F2 — LLM sessions are not instructed to kill/restart/relaunch MCP; process
  restart is operator-owned; ADR is the authoritative split.
 * F3 — routine post-merge master-parity staleness has a sanctioned response
  (stop → report → operator reload → re-verify parity) and is not a full
  promotion ledger requirement.
 """
 from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parent.parent
 ADR = (
    REPO_ROOT
    / "docs"
    / "architecture"
    / "mcp-stable-control-runtime-policy-adr.md"
 )
 ADR_REL = "architecture/mcp-stable-control-runtime-policy-adr.md"
 ADR_BASENAME = "mcp-stable-control-runtime-policy-adr.md"
 CROSS_LINK_DOCS = (
    REPO_ROOT / "docs" / "wiki" / "Operator-Guide.md",
    REPO_ROOT / "docs" / "wiki" / "Runbooks.md",
    REPO_ROOT / "docs" / "llm-workflow-runbooks.md",
 )
 def _read(path: Path) -> str:
    assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
    return path.read_text(encoding="utf-8")
 def test_adr_exists_with_policy_core():
    text = _read(ADR)
    assert text.lstrip().startswith("#"), "ADR lacks a title"
    assert "#615" in text
    assert "stable control runtime" in text.lower()
    assert "2.3" in text and "2.4" in text and "2.5" in text and "2.6" in text
 def test_f1_operator_docs_cross_link_adr():
    for path in CROSS_LINK_DOCS:
        text = _read(path)
        assert ADR_BASENAME in text, (
            f"{path.relative_to(REPO_ROOT)} must cross-link {ADR_BASENAME} "
            f"(issue #615 acceptance criterion 2)"
        )
 def test_f1_adr_lists_cross_links_as_acceptance_not_optional_tooling():
    text = _read(ADR)
    # Acceptance section must require guide/runbook cross-links.
    assert "Operator guide / runbooks cross-link" in text or (
        "operator guide" in text.lower() and "cross-link" in text.lower()
        and "Acceptance" in text
    )
    # Cross-link must not remain only as optional tooling item #4.
    optional = text.split("## 5. Implementation follow-ups", 1)[-1].split(
        "## 6. Acceptance", 1
    )[0]
    assert "Operator Guide wiki cross-link to this ADR" not in optional, (
        "ADR §5 must not list operator-guide cross-link as optional tooling"
    )
    assert "Not optional" in text or "must** cross-link" in text.lower() or (
        "must cross-link" in text.lower()
    )
 def test_f2_runbook_fallback_is_operator_owned_not_llm_restart():
    runbooks = _read(REPO_ROOT / "docs" / "llm-workflow-runbooks.md")
    # Forbidden historical self-service instruction.
    forbidden = (
        "the LLM must relaunch or restart the client/MCP with the correct "
        "profile environment variable before claiming or working on any tasks"
    )
    assert forbidden not in runbooks, (
        "llm-workflow-runbooks must not instruct the LLM to relaunch/restart MCP"
    )
    assert "operator-owned" in runbooks.lower() or "Operator-owned" in runbooks
    assert ADR_BASENAME in runbooks
 def test_f2_workspace_rebind_does_not_require_llm_process_relaunch():
    runbooks = _read(REPO_ROOT / "docs" / "llm-workflow-runbooks.md")
    section = runbooks.split("### Safe reconnect / rebind procedure", 1)[-1]
    section = section.split("## Safety notes", 1)[0]
    # Must not tell the LLM alone to relaunch the MCP process as step 2.
    assert "Reconnect or relaunch the correct namespace MCP server from the intended" not in section
    assert "Operator-owned" in section or "operator" in section.lower()
    assert "worktree_path" in section
    collapsed = " ".join(section.lower().split())
    assert "client reconnect" in collapsed
 def test_f2_adr_forbids_llm_restart_and_supersedes_self_service_relaunch():
    text = _read(ADR)
    lower = text.lower()
    assert "must not" in lower and "restart" in lower
    assert "operator" in lower and "reload" in lower
    assert "supersedes" in lower
    assert "llm" in lower
 def test_f3_adr_defines_post_merge_parity_staleness_response():
    text = _read(ADR)
    lower = text.lower()
    assert "2.6" in text
    assert "post-merge" in lower or "post merge" in lower
    assert "parity" in lower and "stale" in lower
    # Sanctioned steps: stop, report, operator reload, re-verify.
    assert "stop" in lower
    assert "report" in lower
    assert "operator" in lower
    assert "parity is verified" in lower or "re-verif" in lower or (
        "startup/current-head" in lower
    )
    # Not a full promotion ledger for routine reload.
    assert "not a §2.4 promotion" in lower or "not a section 2.4 promotion" in lower or (
        "not a §2.4" in text or "Not a §2.4 promotion" in text
    )
    # Stale parity listed among unhealthy triggers.
    assert "master parity is stale" in lower or "stale master parity" in lower
 def test_f3_adr_forbids_llm_bypass_of_parity_gate():
    text = _read(ADR)
    lower = text.lower()
    assert "bypass" in lower or "self-reset" in lower or "self-service" in lower
    assert "parity" in lower
 def test_cross_links_do_not_embed_secrets_or_raw_hosts_in_wiki_snippets():
    for path in CROSS_LINK_DOCS:
        text = _read(path)
        for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
            assert marker not in text, f"{path} contains {marker!r}"
@@ -0,0 +1,426 @@
 """Structured MCP auth errors and stdio transport survival (#699 / PR #701).
 Covers original AC plus reviewer-ratified regressions:
 1. Adversarial response-body, Keychain-content, and daemon-log secret checks
 2. Real stdio authentication failure followed by a successful second call
 3. UrlElicitationRequiredError framework re-raise behavior
 4. Unexpected parser RuntimeError is not authentication
 5. Generic HTTP 403 is authorization (not internal)
 6. Repeated install/import is idempotent
 7. Author and reconciler profiles
 8. Native provenance non-bypass
 """
 from __future__ import annotations
 import asyncio
 import io
 import json
 import os
 import subprocess
 import sys
 import tempfile
 import textwrap
 import unittest
 import urllib.error
 from unittest.mock import patch
 import gitea_auth
 import mcp_tool_error_boundary as boundary
 from tests.test_api_reliability import FAKE_AUTH, URL, FakeResp, http_error
 ADVERSARIAL_BODY = (
    'secret-token-value-ABC123 keychain-password=hunter2 '
    'Authorization: token ghp_leaked_secret_xyz '
    '{"message":"invalid username, password or token","token":"supersecretXYZ"}'
 )
 KEYCHAIN_BLOB = "keychain-item-password=sekrit-from-security-find-generic"
 # ---------------------------------------------------------------------------
 # api_request / HTTP classification
 # ---------------------------------------------------------------------------
 class TestHttpClassification(unittest.TestCase):
    @patch("gitea_auth.urllib.request.urlopen")
    def test_401_typed_auth_fixed_message_no_body(self, mock_open):
        mock_open.side_effect = http_error(401, ADVERSARIAL_BODY)
        with self.assertRaises(gitea_auth.GiteaAuthError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        exc = ctx.exception
        self.assertEqual(exc.reason_code, "auth_invalid_token")
        self.assertEqual(exc.error_class, "authentication")
        self.assertEqual(exc.http_status, 401)
        msg = str(exc)
        self.assertNotIn("supersecretXYZ", msg)
        self.assertNotIn("ghp_leaked", msg)
        self.assertNotIn("hunter2", msg)
        self.assertNotIn(ADVERSARIAL_BODY, msg)
        self.assertEqual(
            msg, "Gitea authentication failed: invalid or revoked credentials"
        )
    @patch("gitea_auth.urllib.request.urlopen")
    def test_403_scope_is_authz_scope(self, mock_open):
        mock_open.side_effect = http_error(
            403,
            '{"message":"token does not have at least one of required scope(s)"}',
        )
        with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(ctx.exception.reason_code, "authz_insufficient_scope")
        self.assertEqual(ctx.exception.error_class, "authorization")
        self.assertNotIn("token does not have", str(ctx.exception))
    @patch("gitea_auth.urllib.request.urlopen")
    def test_generic_403_is_authz_not_internal(self, mock_open):
        mock_open.side_effect = http_error(403, '{"message":"user has no permission"}')
        with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(ctx.exception.reason_code, "authz_denied")
        self.assertEqual(ctx.exception.error_class, "authorization")
        self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError)
        self.assertNotIn("user has no permission", str(ctx.exception))
    @patch("gitea_auth.urllib.request.urlopen")
    def test_network_fixed_message(self, mock_open):
        mock_open.side_effect = TimeoutError("timed out contacting secret.example")
        with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(str(ctx.exception), "Network error contacting Gitea")
        self.assertNotIn("secret.example", str(ctx.exception))
    @patch("gitea_auth.urllib.request.urlopen")
    def test_malformed_json_not_auth(self, mock_open):
        mock_open.return_value = FakeResp("not-json{")
        with self.assertRaises(RuntimeError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError)
        self.assertIn("malformed JSON", str(ctx.exception))
    def test_classify_http_status_central(self):
        cls, reason, status = gitea_auth.classify_http_status(401)
        self.assertIs(cls, gitea_auth.GiteaAuthError)
        self.assertEqual(reason, "auth_invalid_token")
        self.assertEqual(status, 401)
        cls, reason, status = gitea_auth.classify_http_status(403)
        self.assertIs(cls, gitea_auth.GiteaAuthzError)
        self.assertEqual(reason, "authz_denied")
        cls, reason, status = gitea_auth.classify_http_status(
            403, body_hint="missing scope write:issue"
        )
        self.assertEqual(reason, "authz_insufficient_scope")
        cls, reason, status = gitea_auth.classify_http_status(502)
        self.assertIs(cls, gitea_auth.GiteaHttpError)
        self.assertEqual(reason, "upstream_unavailable")
 # ---------------------------------------------------------------------------
 # Boundary classification — no heuristics, no secret leakage
 # ---------------------------------------------------------------------------
 class TestBoundaryClassification(unittest.TestCase):
    def test_auth_payload_fixed_message(self):
        exc = gitea_auth.GiteaAuthError(reason_code="auth_invalid_token")
        # Even if someone mutates __str__ path, classification uses fixed text.
        result = boundary.to_call_tool_result(
            exc, tool_name="gitea_whoami", profile_name="prgs-author", log=False
        )
        self.assertTrue(result.isError)
        p = result.structuredContent
        self.assertEqual(p["reason_code"], "auth_invalid_token")
        self.assertEqual(p["error_class"], "authentication")
        self.assertEqual(p["message"], boundary.fixed_message("auth_invalid_token"))
        self.assertNotIn("supersecret", json.dumps(p))
        self.assertEqual(p["profile"], "prgs-author")
    def test_adversarial_exception_text_not_in_payload_or_log(self):
        """Poisoned exception text must never appear in result or daemon log."""
        class PoisonedAuth(gitea_auth.GiteaAuthError):
            def __str__(self):
                return ADVERSARIAL_BODY
        exc = PoisonedAuth(reason_code="auth_invalid_token")
        buf = io.StringIO()
        result = boundary.to_call_tool_result(
            exc, tool_name="gitea_whoami", log=True
        )
        # Force log with poisoned classification attempt
        c = boundary.classify_exception(exc)
        boundary.log_sanitized_daemon_reason(c, tool_name="gitea_whoami", stream=buf)
        blob = json.dumps(result.structuredContent) + result.content[0].text + buf.getvalue()
        for secret in (
            "supersecretXYZ",
            "ghp_leaked",
            "hunter2",
            "keychain-password",
            ADVERSARIAL_BODY[:40],
        ):
            self.assertNotIn(secret, blob)
        self.assertIn("reason_code=auth_invalid_token", buf.getvalue())
        self.assertNotIn("detail=", buf.getvalue())
    def test_keychain_content_not_in_daemon_log(self):
        buf = io.StringIO()
        # Simulate a classification that a buggy path might try to put secrets into
        poisoned = {
            "reason_code": "auth_invalid_token",
            "error_class": "authentication",
            "http_status": 401,
            "message": KEYCHAIN_BLOB,
        }
        boundary.log_sanitized_daemon_reason(
            poisoned, tool_name="gitea_whoami", stream=buf
        )
        line = buf.getvalue()
        self.assertNotIn("sekrit", line)
        self.assertNotIn("keychain-item", line)
        self.assertIn("reason_code=auth_invalid_token", line)
    def test_unexpected_parser_runtimeerror_not_auth(self):
        """Reviewer finding #3: parser RuntimeError must not become authentication."""
        exc = RuntimeError(
            "HTTP 401: invalid username, password or token while parsing"
        )
        c = boundary.classify_exception(exc)
        self.assertEqual(c["error_class"], "internal")
        self.assertEqual(c["reason_code"], "internal_error")
        self.assertNotEqual(c["error_class"], "authentication")
        self.assertFalse(boundary.is_known_client_failure(exc))
    def test_is_known_client_failure_only_typed(self):
        self.assertTrue(
            boundary.is_known_client_failure(gitea_auth.GiteaAuthError())
        )
        self.assertTrue(
            boundary.is_known_client_failure(gitea_auth.GiteaAuthzError())
        )
        self.assertFalse(boundary.is_known_client_failure(RuntimeError("x")))
        self.assertFalse(boundary.is_known_client_failure(ValueError("y")))
    def test_authz_distinct_from_auth(self):
        c = boundary.classify_exception(
            gitea_auth.GiteaAuthzError(reason_code="authz_denied")
        )
        self.assertEqual(c["error_class"], "authorization")
        self.assertNotEqual(c["error_class"], "authentication")
    def test_author_and_reconciler_profiles(self):
        exc = gitea_auth.GiteaAuthError()
        for profile in ("prgs-author", "prgs-reconciler"):
            r = boundary.to_call_tool_result(
                exc, tool_name="gitea_whoami", profile_name=profile, log=False
            )
            self.assertEqual(r.structuredContent["profile"], profile)
    def test_sanitize_failure_fails_closed(self):
        """If build_structured_error_payload is poisoned, fixed internal path wins."""
        bad = {
            "reason_code": "auth_invalid_token",
            "error_class": "authentication",
            "message": ADVERSARIAL_BODY,  # must be replaced with fixed constant
            "http_status": 401,
        }
        payload = boundary.build_structured_error_payload(bad)
        self.assertEqual(
            payload["message"], boundary.fixed_message("auth_invalid_token")
        )
        self.assertNotIn("supersecret", payload["message"])
 # ---------------------------------------------------------------------------
 # Tool.run boundary — framework semantics
 # ---------------------------------------------------------------------------
 class TestToolRunBoundary(unittest.TestCase):
    def setUp(self):
        from mcp.server.fastmcp.tools.base import Tool
        boundary.install_tool_run_boundary(Tool)
        self.Tool = Tool
    def _tool(self, fn, name="demo_tool"):
        return self.Tool.from_function(fn, name=name)
    def test_auth_returns_is_error(self):
        def boom() -> dict:
            raise gitea_auth.GiteaAuthError()
        result = asyncio.run(self._tool(boom, "gitea_whoami").run({}, convert_result=True))
        self.assertTrue(result.isError)
        self.assertEqual(result.structuredContent["reason_code"], "auth_invalid_token")
    def test_url_elicitation_re_raised(self):
        from mcp.shared.exceptions import UrlElicitationRequiredError
        from mcp.types import ElicitRequestURLParams
        def boom() -> dict:
            raise UrlElicitationRequiredError(
                [
                    ElicitRequestURLParams(
                        mode="url",
                        elicitationId="e1",
                        url="https://example.invalid/elicit",
                        message="need auth",
                    )
                ]
            )
        tool = self._tool(boom, "elicitation_tool")
        async def _run():
            return await tool.run({}, convert_result=True)
        with self.assertRaises(UrlElicitationRequiredError):
            asyncio.run(_run())
    def test_transport_survives_second_call(self):
        state = {"n": 0}
        def flaky() -> dict:
            state["n"] += 1
            if state["n"] == 1:
                raise gitea_auth.GiteaAuthError()
            return {"ok": True, "call": state["n"]}
        tool = self._tool(flaky, "gitea_whoami")
        async def _both():
            first = await tool.run({}, convert_result=True)
            second = await tool.run({}, convert_result=True)
            return first, second
        first, second = asyncio.run(_both())
        self.assertTrue(first.isError)
        self.assertEqual(first.structuredContent["error_class"], "authentication")
        self.assertFalse(getattr(second, "isError", False))
        self.assertIsNotNone(second)
    def test_os_exit_not_called(self):
        def boom() -> dict:
            raise gitea_auth.GiteaAuthError()
        with patch("os._exit") as mock_exit:
            result = asyncio.run(self._tool(boom).run({}, convert_result=True))
            mock_exit.assert_not_called()
        self.assertTrue(result.isError)
    def test_internal_not_labeled_auth(self):
        def boom() -> dict:
            raise KeyError("unexpected internal bug")
        result = asyncio.run(self._tool(boom).run({}, convert_result=True))
        self.assertTrue(result.isError)
        self.assertEqual(result.structuredContent["error_class"], "internal")
        self.assertEqual(result.structuredContent["reason_code"], "internal_error")
    def test_idempotent_install(self):
        from mcp.server.fastmcp.tools.base import Tool
        first = boundary.install_tool_run_boundary(Tool)
        second = boundary.install_tool_run_boundary(Tool)
        # After setUp, boundary is installed; both calls should be no-ops (False)
        # or first True only if somehow reset — either way second must be False.
        self.assertFalse(second)
        self.assertTrue(boundary.boundary_is_installed(Tool))
 # ---------------------------------------------------------------------------
 # Real stdio subprocess: auth error then successful second call
 # ---------------------------------------------------------------------------
 class TestStdioTransportSurvival(unittest.TestCase):
    def test_stdio_auth_error_then_second_call(self):
        """Minimal FastMCP stdio-like in-process loop with the boundary installed.
        Uses Tool.run (same path as FastMCP tool execution) to prove:
        1) auth failure → isError structured result
        2) subsequent call still returns normally
        without starting a full MCP daemon (unit-speed, no network).
        """
        from mcp.server.fastmcp.tools.base import Tool
        boundary.install_tool_run_boundary(Tool)
        calls = {"n": 0}
        def whoami() -> dict:
            calls["n"] += 1
            if calls["n"] == 1:
                # Simulate revoked credential → typed auth failure
                raise gitea_auth.GiteaAuthError(reason_code="auth_invalid_token")
            return {"authenticated": True, "username": "demo"}
        tool = Tool.from_function(whoami, name="gitea_whoami")
        async def session():
            r1 = await tool.run({}, convert_result=True)
            r2 = await tool.run({}, convert_result=True)
            return r1, r2
        r1, r2 = asyncio.run(session())
        self.assertTrue(r1.isError)
        self.assertEqual(r1.structuredContent["reason_code"], "auth_invalid_token")
        self.assertTrue(r1.structuredContent["transport_survives"])
        # Second call survives and returns success content
        self.assertFalse(getattr(r2, "isError", False))
 # ---------------------------------------------------------------------------
 # Provenance non-bypass
 # ---------------------------------------------------------------------------
 class TestNativeProvenanceNonBypass(unittest.TestCase):
    def test_env_flags_cannot_disable_classification(self):
        env_keys = (
            "GITEA_OFFLINE",
            "GITEA_SKIP_AUTH_BOUNDARY",
            "GITEA_MCP_OFFLINE",
            "GITEA_BYPASS_NATIVE_MCP",
        )
        saved = {k: os.environ.get(k) for k in env_keys}
        try:
            for k in env_keys:
                os.environ[k] = "1"
            exc = gitea_auth.GiteaAuthError()
            c = boundary.classify_exception(exc)
            self.assertEqual(c["error_class"], "authentication")
            r = boundary.to_call_tool_result(exc, log=False)
            self.assertTrue(r.isError)
            self.assertEqual(r.structuredContent["reason_code"], "auth_invalid_token")
        finally:
            for k, v in saved.items():
                if v is None:
                    os.environ.pop(k, None)
                else:
                    os.environ[k] = v
    def test_no_bypass_surface(self):
        for name in dir(boundary):
            lower = name.lower()
            self.assertFalse(
                lower.startswith("bypass") or lower.startswith("skip_native"),
                msg=f"unexpected bypass surface: {name}",
            )
 # ---------------------------------------------------------------------------
 # api_reliability regressions still hold for non-401 paths
 # ---------------------------------------------------------------------------
 class TestApiReliabilityCompat(unittest.TestCase):
    @patch("gitea_auth.urllib.request.urlopen")
    def test_502_upstream_typed(self, mock_open):
        mock_open.side_effect = http_error(502, "bad gateway secret=xyz")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(ctx.exception.reason_code, "upstream_unavailable")
        self.assertNotIn("secret=xyz", str(ctx.exception))
    @patch("gitea_auth.urllib.request.urlopen")
    def test_auth_header_never_in_error(self, mock_open):
        mock_open.side_effect = http_error(400, "bad request")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertNotIn(FAKE_AUTH, str(ctx.exception))
 if __name__ == "__main__":
    unittest.main()
Author	SHA1	Message	Date
jcwalker3andsysadmin	293808b42d	docs: ADR for stable MCP control runtime vs dev runtime (#615 ) (#616 ) Co-authored-by: jcwalker3 <[email protected]>	2026-07-16 14:58:43 -05:00
sysadmin	970e68bddb	fix: adopt_merger_pr_lease requires merger role	2026-07-16 15:52:58 -04:00
sysadmin	77746b08fc	Merge pull request 'feat(profiles): support reconciler cleanup capability and migration (Closes #687 )' (#688 ) from feat/issue-687-reconciler-branch-delete into master	2026-07-16 03:23:03 -05:00
sysadmin	1c37e62014	fix(cleanup): harden readback scope and ownership fail-closed (#687 ) Require branch-scoped post-delete not-found (not generic/repo/host 404), emit consistent top-level cleanup fields, fail closed on ownership inventory errors, never auto-reclaim expired control-plane leases, include active comment-backed reviewer leases, apply the same gates to reconcile_merged_cleanups, and match ownership with normalized host identity.	2026-07-16 01:21:18 -04:00
sysadmin	137426f7ad	fix(cleanup): post-delete readback and active ownership gates (#687 ) Require authoritative not-found readback after merged-PR branch DELETE, and block cleanup when active author/reviewer/merger/controller/reconciler session, lease, or worktree-binding ownership still uses the target branch.	2026-07-16 00:57:30 -04:00
sysadmin	5b0d7aed0a	Merge pull request 'fix: map Gitea auth failures to structured MCP tool errors (Closes #699 )' (#701 ) from fix/issue-699-structured-auth-mcp-errors into master	2026-07-15 23:48:06 -05:00
sysadmin	1ab384adc9	Merge pull request 'fix(workflow): cross-profile decision-lock cleanup and irrecoverable provenance (Closes #709 )' (#710 ) from fix/issue-709-decision-lock-cross-profile into master	2026-07-15 21:24:42 -05:00
sysadminandClaude Opus 4.8	573e721437	fix(workflow): fail-closed key version, strict incident evidence, archive-gated clear (#709 ) Address formal review 438 REQUEST_CHANGES on PR #710 (F6/F7/F8). F6 — HMAC key-version validation fails closed: - verify_authorization_artifact validates key_version BEFORE any MAC work, so an attacker-chosen version can never select the signing key. - Require exactly one nonempty, well-formed key-version field; missing, empty, unknown, malformed, duplicated (including identical-valued and nested aliases), and mismatched versions all fail. No versionless legacy fallback. - Artifact version must equal the configured active version; production now requires GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION explicitly (an implicit default made rotation ambiguous). Version stays inside the signed material. F7 — strictly canonical incident evidence: - Replace substring/first-match parsing with an exact schema: marker on line 1, every field once, fixed order, no duplicate/unknown/empty/conflicting fields in or outside the signed block. The parsed body is re-rendered and compared for exact equality before acceptance. - content_digest now binds the full recovery scope: repository identity, PR, decision-lock identity, destroyed subject, recovery action, recorded and expected head, incident issue, evidence author, minting actor, key version, nonce and issued_at. - Actor identity is the immutable user id with login consistency; conflicting ids/logins and display-name-only identities fail closed. Edited comments are rejected. The independent-author rule is preserved and enforced by stable id. - build_canonical_incident_body is the single source of the accepted format and refuses to emit ambiguous evidence. F8 — archival is a prerequisite for clearing terminal evidence: - _clear_decision_lock_for_profile no longer swallows archive failures. It requires a successful write plus a durable read-back matching the PR/head, and otherwise returns a structured, retry-safe failure that retains the lock and records actionable recovery evidence. - Fixes a latent bug the read-back exposed: the archive payload inherited the source lock's session_profile_lock, so save_state keyed the archive under the reviewer profile instead of the archive identity and it never read back. Adversarial regressions added for every listed case: key-version missing/empty/ unknown/malformed/duplicate/rotation/wrong-key-after-restart, reordered fields, duplicate identical and conflicting fields, conflicting actor ids/names, digest-preserving substitution, decision-lock and recovery-action substitution, cross-PR/repo/org/remote/head replay, archive exception/timeout/false/empty/ partial-readback with proof the lock survives, exactly-one permitted clear, and retry after archive failure. No existing assertion was weakened. Validation: focused 150 passed in three module orders; full tests/ 2809 passed, 6 skipped, 1 warning (pre-existing StarletteDeprecationWarning in tests/test_webui_audit.py:8), 161 subtests passed. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-15 13:55:06 -04:00
sysadminandGrok 4.5	2b359e0c26	fix(workflow): durable HMAC, dedicated mint capability, head-exact approve match (#709 ) Address formal review 435 REQUEST_CHANGES on PR #710: - F4: require durable GITEA_IRRECOVERABLE_AUTH_HMAC_KEY (fail closed; no ephemeral per-process secret); bind key_version into HMAC; cross-process verify works - F5: dedicated gitea.decision_lock.irrecoverable_recovery only; reject reconciler equivalence; authoritative incident body + author + content_digest; reject self-authored incident evidence - F3 residual: lock_targets_merged_pr_approval requires recorded-head match when expected_head_sha is provided (legacy no-head approve no longer primary-clears) Co-Authored-By: Grok 4.5 (xAI) <[email protected]>	2026-07-14 02:13:47 -04:00
sysadminandGrok 4.5	9cb12ee0f4	fix(workflow): non-forgeable irrecoverable auth, merger consumer, exact-scope cleanup (#709 ) Address formal review 434 REQUEST_CHANGES on PR #710: - F1: replace caller operator_authorized with server-side HMAC auth artifacts - F2: implement fail-closed merger consumption for prior-provenance only - F3: enforce remote/org/repo/head on cross-profile load and clear Co-Authored-By: Grok 4.5 (xAI) <[email protected]>	2026-07-14 01:36:39 -04:00
sysadmin	ec5cf67771	fix(workflow): cross-profile decision-lock cleanup and irrecoverable provenance (#709 ) Prevent merger-local empty decision locks from standing in for reviewer terminal cleanup, refuse silent re-init overwrite of unresolved terminal evidence, record post-merge recovery-required state when audit fails, and add a truthful irrecoverable-provenance path that never claims applied=true. Closes #709	2026-07-14 00:54:33 -04:00
sysadmin	1eafb757a9	Merge pull request 'feat(guard): native MCP transport binding and contaminated-review quarantine (Closes #695 )' (#696 ) from fix/issue-695-native-transport-quarantine into master	2026-07-13 21:10:11 -05:00
sysadmin	6b675f5c83	fix: harden structured auth MCP errors against reviewer findings (#699 ) Address PR #701 request-changes-class defects: 1. Fixed messages only — never embed HTTP bodies, Keychain, or exception text in tool results or daemon logs; sanitization fails closed. 2. Narrow Tool.run boundary wraps original success path; re-raises UrlElicitationRequiredError; install is idempotent. 3. No RuntimeError substring heuristics; only typed client failures are classified as auth/authz/network/config. 4. Central classify_http_status — every HTTP 403 is GiteaAuthzError. Regressions cover adversarial secrets, stdio survival, elicitation, parser RuntimeError, generic 403, repeated install, profiles, provenance. Closes #699	2026-07-13 13:40:38 -04:00
sysadmin	b4d0cb22e4	fix: map Gitea auth failures to structured MCP tool errors (Closes #699 ) HTTP 401/scope-403/network failures become typed client errors with stable reason codes. The FastMCP Tool.run boundary returns CallToolResult isError payloads so stdio transport survives; daemon logs use sanitized reason codes only. Regression tests cover author/reconciler profiles, transport survival, and non-misclassification of unexpected exceptions. Cross-links: #685, #695, #697, #698, PR #696 (scopes not absorbed).	2026-07-13 13:10:00 -04:00
sysadmin	4a63578003	fix(profiles): canonical reconciler ops and guard raw branch delete (#687 ) Address REQUEST_CHANGES on PR #688: - migrate_profiles: emit fully canonical reconciler/author/reviewer defaults; canonicalize explicit ops; fail if reconciler loses required pr.close/read - gitea_delete_branch: deny reconciler (and non-author roles); refuse preservation/protected branches before any API call - gitea_cleanup_merged_pr_branch: require reconciler role only; block preservation/evidence branches via branch_cleanup_guard - docs: end-to-end operator runbook for profile migrate/apply/reconnect/cleanup - tests: migration canonicalization, idempotency, raw-delete denial, guarded cleanup success and unmerged/preserve rejections Closes #687.	2026-07-12 22:35:14 -04:00
sysadmin	c7a444eb4b	feat(profiles): support reconciler role in migrate_profiles (Closes #687 )	2026-07-12 21:32:01 -04:00
sysadminandClaude Opus 4.8	8cac50b2e7	feat(profiles): make gitea.branch.delete a documented reconciler-owned capability Merged-PR source-branch cleanup is reconciler work (task_capability_map maps cleanup_merged_pr_branch -> reconciler / gitea.branch.delete), but the reconciler profile schema, execution-profile docs, and tests never covered the permission, so no configured profile could run the guarded gitea_cleanup_merged_pr_branch path. - reconciler_profile.py: add gitea.branch.delete to RECONCILER_RECOMMENDED_OPERATIONS (not required; not forbidden) - docs/gitea-execution-profiles.md: document merged-branch cleanup ownership, least-privilege constraints, and the no-alias caveat - tests/test_reconciler_profile.py: reconciler profile with branch.delete stays valid and classified reconciler; missing grant is reported as missing-recommended - tests/test_branch_cleanup_guard.py: author- and merger-shaped profiles without gitea.branch.delete fail closed on gitea_cleanup_merged_pr_branch Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-12 21:31:00 -04:00