Merge branch 'master' into fix/issue-700-durable-walls

docs/mcp-restart-coordinator.md

@@ -144,19 +144,6 @@ tool argument expresses caller intent and cannot be self-asserted by a worker
 session. `break_glass_requested` and `break_glass_authorized` are both reported,
 so a bypass is never silent.
 
-### Break-glass Restart Workflow (`gitea_break_glass_restart`, #664)
-
-The dedicated MCP tool `gitea_break_glass_restart` provides the privileged emergency break-glass restart workflow when graceful drain cannot complete:
-
-- **Authorization (#664 AC1 / B1 / B13 / B15)**: Requires the exact trusted profile `prgs-controller` **and** explicit `runtime.break_glass_restart` and `gitea.issue.create` grants enforced by the real production operation gate (no `gitea.read` fallback). Incident creation is mandatory prior to execution (`gitea.issue.create`), so the deployable production policy for `prgs-controller` includes `allowed_operations`: `["gitea.read", "gitea.pr.close", "gitea.pr.comment", "gitea.issue.comment", "gitea.issue.create", "runtime.break_glass_restart", "gitea.branch.delete", "gitea.decision_lock.irrecoverable_recovery"]`. Ordinary roles, non-controller reconcilers, lookalike profile names (`fake-controller`, …), and env vars cannot authorize. A narrow break-glass capability does **not** redefine the profile's declared global role. Updating a live running `prgs-controller` profile in production requires an operator configuration update and daemon reload post-merge.
-- **Required Parameters (#664 AC2)**:
-  - `reason`: Mandatory non-empty string (min 10 characters).
-  - `confirmation`: Must equal exactly `"I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION"`.
-  - `impact_ack`: Must be `True`.
-- **Automatic Incident Creation (#664 AC3)**: Creates a Gitea incident issue (`[INCIDENT] [REQUESTED] Break-glass MCP restart invoked by ...`) detailing the reason, timestamp, disrupted sessions, and linking `#652 #653 #655 #630 #658 #662 #664`.
-- **Immutable Append-Only Audit Entry**: Records immutable pre-execution (REQUESTED) and post-execution (SUCCEEDED/FAILED) audit log entries with correlation identifiers.
-- **Mandatory Reconciliation (#664 AC4)**: Sets `reconciliation_required=True` requiring post-restart reconciliation via `gitea_reconcile_after_restart` (#662).
-
 ### Fail closed on apply
 
 A missing, malformed, expired, unclean, tampered, or fingerprint-stale drain
@@ -173,4 +160,3 @@ profiles are operational metadata only.
 
 A representative dry-run report is in
 [`mcp-restart-impact-sample.json`](./mcp-restart-impact-sample.json).
-

gitea_audit.py

@@ -27,73 +27,13 @@ ALLOWED = "allowed"
 BLOCKED = "blocked"
 FAILED = "failed"
 SUCCEEDED = "succeeded"
-REQUESTED = "requested"
-ACCEPTED = "accepted"
-PENDING = "pending"
 
 REDACTED = "[REDACTED]"
 
 # A dict key containing any of these (case-insensitive) has its value redacted.
-_SECRET_KEY_HINTS = (
-    "token",
-    "password",
-    "passwd",
-    "pwd",
-    "secret",
-    "authorization",
-    "auth",
-    "api_key",
-    "apikey",
-    "access_key",
-    "private_key",
-    "client_secret",
-    "credential",
-)
+_SECRET_KEY_HINTS = ("token", "password", "secret", "authorization", "auth")
 # A string value starting with one of these has the following run redacted.
-# Space-terminated scheme prefixes only. Colon forms (``token:`` / ``api_key:``)
-# and ``Authorization: Bearer …`` are handled by ``_ASSIGNMENT_SECRET_PATTERN``
-# so policy/docs text that merely *names* a scheme is not itself flagged as a
-# live secret by console detectors.
-_SECRET_VALUE_PREFIXES = (
-    "token ",
-    "Basic ",
-    "Bearer ",
-)
-
-# Bare token-shaped values only — never a broad ``sec-`` prefix that erases
-# ordinary words (#664 B8 over-redaction).
-_BARE_SECRET_PATTERN = re.compile(
-    r'(?i)\b(?:'
-    r'ghp_[A-Za-z0-9_]{16,}'
-    r'|gho_[A-Za-z0-9_]{16,}'
-    r'|ghu_[A-Za-z0-9_]{16,}'
-    r'|ghs_[A-Za-z0-9_]{16,}'
-    r'|ghr_[A-Za-z0-9_]{16,}'
-    r'|sk-live-[A-Za-z0-9_-]{16,}'
-    r'|sk-proj-[A-Za-z0-9_-]{16,}'
-    r'|sk-[A-Za-z0-9_-]{20,}'
-    r'|glpat-[A-Za-z0-9_-]{16,}'
-    r')\b'
-)
-
-# Key/value credentials embedded in free text (password=..., api_key: ..., GITEA_TOKEN=..., etc.).
-# Group 1 captures the key name (e.g. GITEA_TOKEN, password, api_key).
-# Group 2 captures delimiter/whitespace (=, : ).
-# Group 3 captures the secret value, stopping at whitespace or non-secret delimiters (&, ;, ,, quotes, closing brackets).
-_ASSIGNMENT_SECRET_PATTERN = re.compile(
-    r'(?i)\b([A-Za-z0-9_]*?(?:token|password|passwd|pwd|secret|api[_-]?key|access[_-]?key|'
-    r'client[_-]?secret|private[_-]?key|authorization|credential))\b(\s*[:=]\s*)('
-    r'"[^"]*"|\'[^\']*\'|'
-    r'(?:Bearer|Basic|Token)\s+[^\s;&,"\'\)\}\]\>]+|'
-    r'[^\s;&,"\'\)\}\]\>]+'
-    r')'
-)
-
-# Connection-string style credentials: Password=...; User ID=...; etc.
-_CONN_STRING_SECRET_PATTERN = re.compile(
-    r'(?i)\b((?:password|pwd|user\s*id|uid|username|account)\s*=\s*)([^\s;\'"]+)'
-)
-
+_SECRET_VALUE_PREFIXES = ("token ", "Basic ", "Bearer ")
 
 # Known synthetic test-only domains/hostnames to preserve
 _SYNTHETIC_HOSTS = {
@@ -119,8 +59,7 @@ def redact_urls(text: str) -> str:
     if not isinstance(text, str) or not text:
         return text
 
-    # Match any URI scheme (http, https, postgres, mysql, mongodb, redis, etc.)
-    url_pattern = re.compile(r'([a-z0-9\+\.\-]+://[^\s)>\]}]+)', re.IGNORECASE)
+    url_pattern = re.compile(r'(https?://[^\s)>\]}]+)', re.IGNORECASE)
 
     def replace_url(match):
         url_str = match.group(1)
@@ -134,11 +73,11 @@ def redact_urls(text: str) -> str:
                     is_synthetic = True
                     break
 
-            if is_synthetic or (parsed.username or parsed.password) or parsed.scheme.lower() not in ("http", "https"):
-                # Rebuild URL to redact any credentials or query secrets
+            if is_synthetic:
+                # Rebuild synthetic URL to redact any credentials or query secrets
                 new_netloc = parsed.netloc
                 if parsed.username or parsed.password:
-                    netloc_clean = parsed.hostname or ""
+                    netloc_clean = parsed.hostname
                     if parsed.port:
                         netloc_clean = f"{netloc_clean}:{parsed.port}"
                     new_netloc = f"[REDACTED_USER]:[REDACTED_PASS]@{netloc_clean}"
@@ -175,51 +114,23 @@ def redact_urls(text: str) -> str:
     return out
 
 
-def _mask_assignment(match: re.Match) -> str:
-    """Keep the key and separator; replace only the secret value."""
-    val = match.group(3)
-    if val.startswith(REDACTED) or val.startswith("%5BREDACTED") or val.startswith("[REDACTED"):
-        return f"{match.group(1)}{match.group(2)}{val}"
-    return f"{match.group(1)}{match.group(2)}{REDACTED}"
-
-
-def _mask_conn_secret(match: re.Match) -> str:
-    """Keep the connection-string key; replace only the credential value."""
-    val = match.group(2)
-    if val.startswith(REDACTED) or val.startswith("%5BREDACTED") or val.startswith("[REDACTED"):
-        return f"{match.group(1)}{val}"
-    return f"{match.group(1)}{REDACTED}"
-
-
 def _redact_str(text):
-    """Redact credentials, bare token shapes, and raw URLs in *text* (#664 B8).
-
-    Covers key/value credentials, authorization/bearer material, bare
-    token-shaped values, connection-string credentials, and secrets embedded
-    in larger sentences. Deliberately does **not** erase ordinary words that
-    merely begin with a broad ``sec-`` prefix.
-    """
+    """Redact anything that looks like an Authorization credential or raw URL in *text*."""
     if not isinstance(text, str) or not text:
         return text
-    out = redact_urls(text)
-    out = _BARE_SECRET_PATTERN.sub(REDACTED, out)
-    out = _ASSIGNMENT_SECRET_PATTERN.sub(_mask_assignment, out)
-    out = _CONN_STRING_SECRET_PATTERN.sub(_mask_conn_secret, out)
-    out_lower = out.lower()
+    out = text
     for prefix in _SECRET_VALUE_PREFIXES:
-        prefix_lower = prefix.lower()
         idx = 0
         while True:
-            i = out_lower.find(prefix_lower, idx)
+            i = out.find(prefix, idx)
             if i == -1:
                 break
             j = i + len(prefix)
             while j < len(out) and not out[j].isspace():
                 j += 1
             out = out[:i] + prefix + REDACTED + out[j:]
-            out_lower = out.lower()
             idx = i + len(prefix) + len(REDACTED)
-    return out
+    return redact_urls(out)
 
 
 def redact(value):

gitea_config.py

@@ -97,26 +97,6 @@ GITEA_OPERATION_ALIASES = {
 _REVIEW_MERGE_OPS = frozenset({"gitea.pr.approve", "gitea.pr.merge"})
 _AUTHOR_ONLY_OPS = frozenset({"gitea.pr.create", "gitea.branch.push"})
 
-# First-class operation services that may appear in multi-service profile
-# allowlists. ``runtime.*`` is the control-plane capability namespace used by
-# non-Gitea MCP tools such as ``runtime.break_glass_restart`` (#664 B13).
-# Unknown foreign prefixes (e.g. ``jenkins.*`` under service=gitea) still fail
-# closed — they are not registered here.
-KNOWN_OPERATION_SERVICES = frozenset({"gitea", "runtime"})
-
-
-def service_for_operation(op, default="gitea"):
-    """Return the registered service prefix for a fully-qualified *op*.
-
-    Unqualified names and unknown prefixes fall back to *default* so callers
-    keep the historical Gitea-centric gate behaviour.
-    """
-    if isinstance(op, str) and "." in op:
-        prefix = op.split(".", 1)[0]
-        if prefix in KNOWN_OPERATION_SERVICES:
-            return prefix
-    return default
-
 
 def normalize_operation(op, service="gitea"):
     """Return the canonical namespaced name for *op*, or fail closed (#106).
@@ -153,12 +133,6 @@ def check_operation(op, allowed, forbidden=(), service="gitea"):
     Reasons: ``allowed``, ``invalid-operation``, ``invalid-forbidden-entry``,
     ``forbidden``, ``no-allowed-operations``, ``not-allowed``.
 
-    Multi-service profile allowlists (#664 B13): each allow/forbid entry is
-    normalized with its own registered service prefix (``gitea.*`` or
-    ``runtime.*``) so a gate defaulting to service=gitea can still enforce an
-    exact ``runtime.break_glass_restart`` grant. Unknown / misspelled
-    operations and foreign service prefixes remain fail-closed.
-
     Fail-closed rules:
     - an *op* that cannot be normalized is denied (``invalid-operation``)
     - a forbidden entry that cannot be normalized denies the request
@@ -169,16 +143,14 @@ def check_operation(op, allowed, forbidden=(), service="gitea"):
     - ``forbidden`` always overrides ``allowed``
     - an empty or missing allowed list denies everything
     """
-    op_service = service_for_operation(op, default=service)
     try:
-        op_n = normalize_operation(op, op_service)
+        op_n = normalize_operation(op, service)
     except ConfigError:
         return (False, "invalid-operation")
     forbidden_n = set()
     for entry in (forbidden or ()):
         try:
-            entry_service = service_for_operation(entry, default=service)
-            forbidden_n.add(normalize_operation(entry, entry_service))
+            forbidden_n.add(normalize_operation(entry, service))
         except ConfigError:
             return (False, "invalid-forbidden-entry")
     if op_n in forbidden_n:
@@ -188,8 +160,7 @@ def check_operation(op, allowed, forbidden=(), service="gitea"):
     allowed_n = set()
     for entry in allowed:
         try:
-            entry_service = service_for_operation(entry, default=service)
-            allowed_n.add(normalize_operation(entry, entry_service))
+            allowed_n.add(normalize_operation(entry, service))
         except ConfigError:
             continue
     if op_n in allowed_n:

gitea_mcp_server.py

@@ -233,50 +233,28 @@ def _effective_workspace_role() -> str:
     )
 
 
-# Exact production profile → role mapping used only when no declared role is
-# present. Substring lookalikes (fake-controller, not-controller, …) never
-# match (#664 B1/B14).
-_EXACT_PROFILE_ROLE_NAMES = {
-    "prgs-controller": "controller",
-    "prgs-reconciler": "reconciler",
-    "prgs-author": "author",
-    "prgs-reviewer": "reviewer",
-    "prgs-merger": "merger",
-    "mdcps-author": "author",
-    "mdcps-reviewer": "reviewer",
-    "mdcps-merger": "merger",
-    "controller": "controller",
-    "reconciler": "reconciler",
-    "author": "author",
-    "reviewer": "reviewer",
-    "merger": "merger",
-}
-
-# Exact trusted profile that may hold break-glass (#664 B1). Capability
-# ``runtime.break_glass_restart`` is still required and enforced by the real
-# operation gate; this set only rejects lookalike profile names.
-TRUSTED_BREAK_GLASS_PROFILES = frozenset({"prgs-controller"})
-
-
 def _profile_role_kind(profile: dict) -> str:
     """Resolve a profile's declared role before inferring from permissions.
 
     Declared ``role`` / ``role_kind`` always wins so a controller profile is
-    never reclassified as reconciler from permission inference (#840). Exact
-    match only — profile-name / role *substrings* never grant controller (or
-    any) authority (#664 B1/B14). Lookalikes such as ``fake-controller``,
-    ``not-controller``, or ``xcontrollerx`` do not become controller.
+    never reclassified as reconciler from permission inference (#840).
     """
     role = (profile.get("role") or profile.get("role_kind") or "").strip().lower()
     if role:
-        # Exact aliases only; never ``"control" in role`` (matches
-        # "not-controller" / "control-plane").
-        if role in ("controller", "control"):
+        # Normalize aliases / case.
+        if "control" in role:
             return "controller"
         return role
     profile_name = (profile.get("profile_name") or "").strip().lower()
-    if profile_name in _EXACT_PROFILE_ROLE_NAMES:
-        return _EXACT_PROFILE_ROLE_NAMES[profile_name]
+    for candidate in (
+        "controller",
+        "reconciler",
+        "merger",
+        "reviewer",
+        "author",
+    ):
+        if candidate in profile_name:
+            return candidate
     return _role_kind(
         profile.get("allowed_operations") or [],
         profile.get("forbidden_operations") or [],
@@ -599,10 +577,10 @@ def _get_workspace_porcelain(worktree_path: str | None = None) -> str:
 
 
 def _parse_porcelain_entries(porcelain: str) -> dict[str, str]:
-    """Map tracked path -> full porcelain line (untracked ``??`` ignored)."""
+    """Map path -> full porcelain line (including untracked ``??``)."""
     entries: dict[str, str] = {}
     for line in (porcelain or "").splitlines():
-        if not line or len(line) < 4 or line.startswith("??"):
+        if not line or len(line) < 4:
             continue
         path = line[3:].strip()
         if " -> " in path:
@@ -7143,17 +7121,35 @@ def terminal_review_hard_stop_reasons(
     ]
 
 
-def _redact(text: str) -> str:
-    """Strip credentials, bare token shapes, and raw URLs from *text* (#664 B8).
+# Patterns scrubbed from any surfaced error text so a credential can never leak.
+_SECRET_PREFIXES = ("token ", "Basic ")
 
-    Defers to the canonical :mod:`gitea_audit` redactor so MCP tool results,
-    incidents, audits, and delegated error surfaces share one boundary.
+
+def _redact(text: str) -> str:
+    """Strip anything that looks like an Authorization credential or raw URL from *text*.
+
+    Errors raised by ``api_request`` echo the server response body, not the
+    request headers, so a token should never appear — this is defence in depth
+    so a future change can't leak ``token …`` / ``Basic …`` material into a
+    tool result or log line.
     """
     if not text:
         return text
+    out = text
+    for prefix in _SECRET_PREFIXES:
+        idx = 0
+        while True:
+            i = out.find(prefix, idx)
+            if i == -1:
+                break
+            j = i + len(prefix)
+            while j < len(out) and not out[j].isspace():
+                j += 1
+            out = out[:i] + prefix + "[REDACTED]" + out[j:]
+            idx = i + len(prefix) + len("[REDACTED]")
+    # Redact raw URLs, query secrets, hostnames, etc.
     import gitea_audit
-    redacted = gitea_audit._redact_str(str(text))
-    return redacted if isinstance(redacted, str) else str(redacted)
+    return gitea_audit.redact_urls(out)
 
 
 # Review states that carry a submitted verdict. Gitea also emits PENDING
@@ -23926,657 +23922,6 @@ def gitea_request_mcp_restart(
     return payload
 
 
-BREAK_GLASS_CONFIRMATION_PHRASE = "I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION"
-
-# Test-injectable canonical break-glass executor/delegate (#664 B6/B11).
-# Production default never signals the live MCP cohort.
-_break_glass_restart_executor = None
-
-
-def _default_break_glass_restart_executor(request: dict) -> dict:
-    """Canonical non-dry-run break-glass executor/delegate (#664 B6/B11).
-
-    Never kills, signals, or restarts the live MCP cohort in-process.
-    An environment string alone (``GITEA_SANCTIONED_RESTART_HOOK``) does not
-    prove restart execution without an active confirmed delegate handoff.
-    """
-    hook = (os.environ.get("GITEA_SANCTIONED_RESTART_HOOK") or "").strip()
-    if hook:
-        return {
-            "success": False,
-            "apply_supported": True,
-            "apply_authorized": True,
-            "restart_performed": False,
-            "break_glass_executed": False,
-            "execution_mode": "accepted_not_executed",
-            "host_hook_configured": True,
-            "reasons": [
-                "sanctioned host restart hook reference is configured, but environment text "
-                "cannot prove execution without a confirmed delegate handoff (fail closed) (#664 B6/B11)"
-            ],
-        }
-    return {
-        "success": False,
-        "apply_supported": False,
-        "apply_authorized": False,
-        "restart_performed": False,
-        "break_glass_executed": False,
-        "execution_mode": "unsupported",
-        "reasons": [
-            "break-glass apply is unsupported: no sanctioned host restart "
-            "executor is configured (#664)"
-        ],
-    }
-
-
-def _run_break_glass_restart_executor(request: dict) -> dict:
-    """Invoke the installed or default break-glass executor (test-injectable)."""
-    executor = _break_glass_restart_executor or _default_break_glass_restart_executor
-    result = executor(request)
-    if not isinstance(result, dict):
-        return {
-            "success": False,
-            "apply_supported": True,
-            "apply_authorized": False,
-            "restart_performed": False,
-            "break_glass_executed": False,
-            "reasons": ["break-glass executor returned a non-dict result (fail closed)"],
-        }
-    return result
-
-
-@mcp.tool()
-def gitea_break_glass_restart(
-    reason: str,
-    confirmation: str,
-    impact_ack: bool = False,
-    restart_class: str = "full_mcp_restart",
-    create_incident_issue: bool = True,
-    dry_run: bool = False,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    worktree_path: str | None = None,
-) -> dict:
-    """Privileged emergency break-glass MCP restart workflow (#664).
-
-    Break-glass restart permits emergency recovery when graceful drain cannot
-    complete. It requires:
-    1. Exact ``runtime.break_glass_restart`` capability on the trusted
-       ``prgs-controller`` profile (ordinary roles, lookalike names, env vars,
-       and non-controller reconcilers fail closed).
-    2. Explicit non-empty reason (minimum 10 characters).
-    3. Exact confirmation string matching 'I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION'.
-    4. Mandatory impact acknowledgement (impact_ack=True).
-    5. Immutable append-only audit entry recorded prior to execution and after terminal completion.
-    6. Automatic incident record created on Gitea prior to execution.
-    7. Truthful execution reporting via the canonical executor/delegate and
-       mandatory post-restart reconciliation (#662).
-    """
-    # B13: real production gate for the exact canonical operation — never
-    # fall back to gitea.read and never stub this gate in production.
-    capability_block = _profile_operation_gate("runtime.break_glass_restart")
-    if capability_block:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": capability_block,
-            "permission_report": _permission_block_report("runtime.break_glass_restart"),
-            "blocker_kind": "permission_denied",
-        })
-
-    h, o, r = _resolve(remote, host, org, repo)
-    profile = get_profile()
-    active_role = _profile_role_kind(profile)
-    profile_name = (
-        (profile.get("profile_name") or profile.get("execution_profile") or "")
-        .strip()
-    )
-    break_glass_env_auth = bool(
-        (os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or "").strip()
-    )
-    # Env is never an authorization channel for this endpoint (B2/B1).
-    _ = break_glass_env_auth
-
-    # B1: exact trusted profile only — not role substring, not lookalike names.
-    if profile_name not in TRUSTED_BREAK_GLASS_PROFILES:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "active_role": active_role,
-            "profile_name": profile_name or None,
-            "reasons": [
-                f"profile '{profile_name or '(unset)'}' is not the trusted "
-                f"break-glass profile {sorted(TRUSTED_BREAK_GLASS_PROFILES)}; "
-                "lookalike names, ordinary roles, env vars, and non-controller "
-                "reconcilers cannot authorize break-glass (#664 AC1/B1)"
-            ],
-            "blocker_kind": "role_authorization",
-        })
-
-    # 2. Required fields and redaction (B8)
-    raw_reason = (reason or "").strip()
-    clean_reason = _redact(raw_reason)
-    if not raw_reason or len(raw_reason) < 10:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                "reason is required and must be at least 10 characters long (#664 AC2)"
-            ],
-            "blocker_kind": "missing_required_fields",
-        })
-
-    clean_confirmation = (confirmation or "").strip()
-    if clean_confirmation != BREAK_GLASS_CONFIRMATION_PHRASE:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                f"confirmation string mismatch; must equal exactly '{BREAK_GLASS_CONFIRMATION_PHRASE}' (#664 AC2)"
-            ],
-            "blocker_kind": "confirmation_mismatch",
-        })
-
-    if not impact_ack:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                "impact_ack must be True to acknowledge disruption of in-flight sessions (#664 AC2)"
-            ],
-            "blocker_kind": "impact_ack_required",
-        })
-
-    # Gate incident issue creation on gitea.issue.create permission (B3)
-    if create_incident_issue:
-        create_block = _profile_operation_gate("gitea.issue.create")
-        if create_block:
-            return gitea_audit.redact({
-                "success": False,
-                "performed": False,
-                "break_glass_executed": False,
-                "reasons": create_block,
-                "permission_report": _permission_block_report("gitea.issue.create"),
-                "blocker_kind": "permission_denied",
-            })
-
-    # Evaluate impact / disrupted sessions
-    impact_result = gitea_request_mcp_restart(
-        remote=remote,
-        host=host,
-        org=org,
-        repo=repo,
-        dry_run=True,
-        restart_class=restart_class,
-        request_break_glass=True,
-    )
-    disrupted_sessions = list(impact_result.get("affected_sessions") or [])
-    disrupted_count = len(disrupted_sessions)
-
-    identity = _authenticated_username(h) or profile.get("username") or "unknown"
-    now_iso = datetime.now(timezone.utc).isoformat()
-    ns_ctx = _resolve_namespace_mutation_context(worktree_path)
-    mcp_namespace = ns_ctx.get("mcp_namespace") or profile.get("profile_name") or "gitea-controller"
-    correlation_id = f"bg-{uuid.uuid4().hex[:12]}"
-
-    audit_payload = gitea_audit.redact({
-        "event": "break_glass_mcp_restart",
-        "correlation_id": correlation_id,
-        "actor": identity,
-        "role": active_role,
-        "mcp_namespace": mcp_namespace,
-        "timestamp": now_iso,
-        "reason": clean_reason,
-        "confirmation": clean_confirmation,
-        "restart_class": restart_class,
-        "disrupted_sessions_count": disrupted_count,
-        "disrupted_sessions": [
-            s.get("session_id") if isinstance(s, dict) else str(s)
-            for s in disrupted_sessions
-        ],
-        "dry_run": dry_run,
-        "remote": remote,
-        "org": o,
-        "repo": r,
-        "env_auth_present": break_glass_env_auth,
-    })
-
-    # Dry-run handling (B7: no durable mutation)
-    if dry_run:
-        return gitea_audit.redact({
-            "success": True,
-            "performed": False,
-            "dry_run": True,
-            "break_glass_executed": False,
-            "would_execute": True,
-            "correlation_id": correlation_id,
-            "actor": identity,
-            "role": active_role,
-            "mcp_namespace": mcp_namespace,
-            "restart_class": restart_class,
-            "reason": clean_reason,
-            "confirmation": clean_confirmation,
-            "disrupted_sessions_count": disrupted_count,
-            "disrupted_sessions": disrupted_sessions,
-            "audit_record": audit_payload,
-            "saved_audit": None,
-            "incident_issue": None,
-            "reconciliation_required": True,
-            "reconciliation_tool": "gitea_reconcile_after_restart",
-            "follow_up_issue_required": True,
-            "cross_references": ["#652", "#653", "#655", "#630", "#658", "#662", "#664"],
-            "reasons": ["break-glass restart dry-run evaluated successfully"],
-        })
-
-    # Fail closed if create_incident_issue is False on real execution (B5)
-    if not create_incident_issue:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                "create_incident_issue=False is forbidden on real break-glass execution; "
-                "pre-execution incident creation is mandatory (#664 AC3)"
-            ],
-            "blocker_kind": "incident_creation_required",
-        })
-
-    # B9: Fail closed if audit backend is disabled
-    if not gitea_audit.audit_enabled():
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                "audit recording is disabled or unavailable; break-glass restart requires an enabled audit backend (#664 AC3)"
-            ],
-            "blocker_kind": "audit_recording_failed",
-        })
-
-    # Pre-execution recording: Audit record in REQUESTED state (B4, B10)
-    pre_audit_event = gitea_audit.build_event(
-        action="break_glass_mcp_restart_requested",
-        result=gitea_audit.REQUESTED,
-        remote=remote,
-        server=(gitea_url(h, "").rstrip("/") if h else None),
-        repository=r,
-        profile_name=profile.get("profile_name", "unknown"),
-        audit_label=profile.get("audit_label", "unknown"),
-        authenticated_username=identity,
-        reason=clean_reason,
-        mcp_namespace=mcp_namespace,
-        task_role=active_role,
-        operation="break_glass_mcp_restart",
-        now=now_iso,
-        request_metadata={
-            "correlation_id": correlation_id,
-            "confirmation": clean_confirmation,
-            "restart_class": restart_class,
-            "disrupted_sessions_count": disrupted_count,
-            "disrupted_sessions": [
-                s.get("session_id") if isinstance(s, dict) else str(s)
-                for s in disrupted_sessions
-            ],
-        },
-    )
-    audit_write_success = gitea_audit.write_event(pre_audit_event)
-
-    if not audit_write_success:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                "failed to persist required pre-execution audit event (#664 AC3)"
-            ],
-            "blocker_kind": "audit_recording_failed",
-        })
-
-    # Pre-execution recording: Gitea Incident Issue (B5, B8)
-    issue_title = _redact(f"[INCIDENT] [REQUESTED] Break-glass MCP restart invoked by {identity} ({correlation_id})")
-    issue_body = _redact(
-        f"## Break-glass MCP restart incident report (#664)\n\n"
-        f"- **Correlation ID**: `{correlation_id}`\n"
-        f"- **Invoked by**: `{identity}` (role: `{active_role}`, namespace: `{mcp_namespace}`)\n"
-        f"- **Timestamp**: `{now_iso}`\n"
-        f"- **Reason**: {clean_reason}\n"
-        f"- **Confirmation**: `{clean_confirmation}`\n"
-        f"- **Disrupted Sessions Count**: `{disrupted_count}`\n\n"
-        f"### Mandatory Post-Restart Reconciliation (#662)\n"
-        f"Post-restart reconciliation must be executed via `gitea_reconcile_after_restart` "
-        f"to clean up orphaned leases, inspect worktree integrity, and handle disrupted work.\n\n"
-        f"### Cross-references\n"
-        f"Ref #652 #653 #655 #630 #658 #662 #664\n"
-    )
-    incident_issue_result = None
-    try:
-        incident_issue_result = api_request(
-            "POST",
-            f"{repo_api_url(h, o, r)}/issues",
-            _auth(h),
-            {
-                "title": issue_title,
-                "body": issue_body,
-                "labels": ["incident", "mcp-health", "break-glass"],
-            },
-        )
-        if not isinstance(incident_issue_result, dict) or "number" not in incident_issue_result:
-            return gitea_audit.redact({
-                "success": False,
-                "performed": False,
-                "break_glass_executed": False,
-                "reasons": [
-                    f"incident issue creation failed (#664 AC3): {_redact(str(incident_issue_result))}"
-                ],
-                "blocker_kind": "incident_creation_failed",
-                "incident_issue": incident_issue_result,
-            })
-    except Exception as exc:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "reasons": [
-                f"incident issue creation failed with exception (#664 AC3): {_redact(str(exc))}"
-            ],
-            "blocker_kind": "incident_creation_failed",
-            "incident_issue": {"error": _redact(str(exc))},
-        })
-
-    incident_number = incident_issue_result.get("number")
-
-    # B6/B11: reach the canonical non-dry-run executor/delegate.
-    # Authorization and apply_authorized do not mean execution occurred.
-    # Dry-run never reaches this path (returned earlier).
-    restart_exec_result = _run_break_glass_restart_executor({
-        "remote": remote,
-        "host": host,
-        "org": o,
-        "repo": r,
-        "restart_class": restart_class,
-        "correlation_id": correlation_id,
-        "reason": clean_reason,
-        "confirmation": clean_confirmation,
-        "profile_name": profile_name,
-        "active_role": active_role,
-        "incident_number": incident_number,
-        "disrupted_sessions_count": disrupted_count,
-        "request_break_glass": True,
-        "dry_run": False,
-    })
-
-    apply_supported = bool(restart_exec_result.get("apply_supported", False))
-    apply_authorized = bool(restart_exec_result.get("apply_authorized", False))
-    exec_success = bool(restart_exec_result.get("success", False))
-    # break_glass_executed is true only when the executor contract proves
-    # execution (or accepted host delegation) occurred.
-    restart_performed = bool(
-        restart_exec_result.get("restart_performed", False)
-        and restart_exec_result.get("break_glass_executed", False)
-    )
-
-    if not apply_supported:
-        term_audit = gitea_audit.build_event(
-            action="break_glass_mcp_restart_terminal",
-            result=gitea_audit.FAILED,
-            remote=remote,
-            server=(gitea_url(h, "").rstrip("/") if h else None),
-            repository=r,
-            profile_name=profile.get("profile_name", "unknown"),
-            audit_label=profile.get("audit_label", "unknown"),
-            authenticated_username=identity,
-            reason="apply_unsupported",
-            mcp_namespace=mcp_namespace,
-            task_role=active_role,
-            operation="break_glass_mcp_restart",
-            now=datetime.now(timezone.utc).isoformat(),
-            request_metadata={
-                "correlation_id": correlation_id,
-                "incident_number": incident_number,
-                "break_glass_executed": False,
-                "blocker_kind": "apply_unsupported",
-            },
-        )
-        gitea_audit.write_event(term_audit)
-        if incident_number:
-            try:
-                api_request(
-                    "POST",
-                    f"{repo_api_url(h, o, r)}/issues/{incident_number}/comments",
-                    _auth(h),
-                    {"body": _redact(f"**Terminal Status**: `apply_unsupported` - Restart coordinator does not support apply execution. No restart was performed. ({correlation_id})")},
-                )
-            except Exception:
-                pass
-
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "actor": identity,
-            "role": active_role,
-            "mcp_namespace": mcp_namespace,
-            "restart_class": restart_class,
-            "reasons": [
-                "break-glass apply is unsupported by restart coordinator (apply_supported=False) (#664)",
-                *(restart_exec_result.get("reasons") or []),
-            ],
-            "blocker_kind": "apply_unsupported",
-            "restart_result": restart_exec_result,
-            "incident_issue": incident_issue_result,
-            "audit_record": pre_audit_event,
-        })
-
-    if not apply_authorized or not exec_success or not restart_performed:
-        term_audit = gitea_audit.build_event(
-            action="break_glass_mcp_restart_terminal",
-            result=gitea_audit.FAILED,
-            remote=remote,
-            server=(gitea_url(h, "").rstrip("/") if h else None),
-            repository=r,
-            profile_name=profile.get("profile_name", "unknown"),
-            audit_label=profile.get("audit_label", "unknown"),
-            authenticated_username=identity,
-            reason="restart_delegation_failed",
-            mcp_namespace=mcp_namespace,
-            task_role=active_role,
-            operation="break_glass_mcp_restart",
-            now=datetime.now(timezone.utc).isoformat(),
-            request_metadata={
-                "correlation_id": correlation_id,
-                "incident_number": incident_number,
-                "break_glass_executed": False,
-                "blocker_kind": "restart_delegation_failed",
-            },
-        )
-        gitea_audit.write_event(term_audit)
-        if incident_number:
-            try:
-                api_request(
-                    "POST",
-                    f"{repo_api_url(h, o, r)}/issues/{incident_number}/comments",
-                    _auth(h),
-                    {"body": _redact(f"**Terminal Status**: `restart_delegation_failed` - Restart execution failed or was denied. No restart was performed. ({correlation_id})")},
-                )
-            except Exception:
-                pass
-
-        return gitea_audit.redact({
-            "success": False,
-            "performed": False,
-            "break_glass_executed": False,
-            "actor": identity,
-            "role": active_role,
-            "mcp_namespace": mcp_namespace,
-            "restart_class": restart_class,
-            "reasons": [
-                "delegated restart execution failed or was denied by coordinator (#664)",
-                *(restart_exec_result.get("reasons") or []),
-            ],
-            "blocker_kind": "restart_delegation_failed",
-            "restart_result": restart_exec_result,
-            "incident_issue": incident_issue_result,
-            "audit_record": pre_audit_event,
-        })
-
-    # Restart occurred! Perform mandatory post-restart reconciliation (B10)
-    recon_result = None
-    try:
-        recon_result = gitea_reconcile_after_restart(
-            remote=remote,
-            host=host,
-            org=org,
-            repo=repo,
-        )
-    except Exception as exc:
-        recon_result = {"success": False, "error": _redact(str(exc))}
-
-    recon_success = bool(recon_result and isinstance(recon_result, dict) and recon_result.get("success", False))
-
-    if not recon_success:
-        term_audit = gitea_audit.build_event(
-            action="break_glass_mcp_restart_terminal",
-            result=gitea_audit.FAILED,
-            remote=remote,
-            server=(gitea_url(h, "").rstrip("/") if h else None),
-            repository=r,
-            profile_name=profile.get("profile_name", "unknown"),
-            audit_label=profile.get("audit_label", "unknown"),
-            authenticated_username=identity,
-            reason="reconciliation_failed",
-            mcp_namespace=mcp_namespace,
-            task_role=active_role,
-            operation="break_glass_mcp_restart",
-            now=datetime.now(timezone.utc).isoformat(),
-            request_metadata={
-                "correlation_id": correlation_id,
-                "incident_number": incident_number,
-                "break_glass_executed": True,
-                "reconciliation_success": False,
-                "blocker_kind": "reconciliation_failed",
-            },
-        )
-        gitea_audit.write_event(term_audit)
-        if incident_number:
-            try:
-                api_request(
-                    "POST",
-                    f"{repo_api_url(h, o, r)}/issues/{incident_number}/comments",
-                    _auth(h),
-                    {"body": _redact(f"**Terminal Status**: `reconciliation_failed` - Restart was executed but post-restart reconciliation failed. ({correlation_id})")},
-                )
-            except Exception:
-                pass
-
-        return gitea_audit.redact({
-            "success": False,
-            "performed": True,
-            "break_glass_executed": True,
-            "actor": identity,
-            "role": active_role,
-            "mcp_namespace": mcp_namespace,
-            "restart_class": restart_class,
-            "reasons": [
-                "break-glass restart executed but post-restart reconciliation failed (#664/#662)"
-            ],
-            "blocker_kind": "reconciliation_failed",
-            "restart_result": restart_exec_result,
-            "reconciliation_result": recon_result,
-            "incident_issue": incident_issue_result,
-            "audit_record": pre_audit_event,
-        })
-
-    # Terminal audit append for successful execution + reconciliation
-    term_audit = gitea_audit.build_event(
-        action="break_glass_mcp_restart_terminal",
-        result=gitea_audit.SUCCEEDED,
-        remote=remote,
-        server=(gitea_url(h, "").rstrip("/") if h else None),
-        repository=r,
-        profile_name=profile.get("profile_name", "unknown"),
-        audit_label=profile.get("audit_label", "unknown"),
-        authenticated_username=identity,
-        reason=clean_reason,
-        mcp_namespace=mcp_namespace,
-        task_role=active_role,
-        operation="break_glass_mcp_restart",
-        now=datetime.now(timezone.utc).isoformat(),
-        request_metadata={
-            "correlation_id": correlation_id,
-            "incident_number": incident_number,
-            "break_glass_executed": True,
-            "reconciliation_success": True,
-        },
-    )
-    term_write_success = gitea_audit.write_event(term_audit)
-
-    if incident_number:
-        try:
-            api_request(
-                "POST",
-                f"{repo_api_url(h, o, r)}/issues/{incident_number}/comments",
-                _auth(h),
-                {"body": _redact(f"**Terminal Status**: `succeeded` - Break-glass restart executed and reconciled successfully ({correlation_id}).")},
-            )
-        except Exception:
-            pass
-
-    if not term_write_success:
-        return gitea_audit.redact({
-            "success": False,
-            "performed": True,
-            "break_glass_executed": True,
-            "actor": identity,
-            "role": active_role,
-            "mcp_namespace": mcp_namespace,
-            "restart_class": restart_class,
-            "reasons": [
-                "break-glass restart executed and reconciled but terminal audit recording failed (#664)"
-            ],
-            "blocker_kind": "terminal_audit_failed",
-            "restart_result": restart_exec_result,
-            "reconciliation_result": recon_result,
-            "incident_issue": incident_issue_result,
-            "audit_record": pre_audit_event,
-        })
-
-    return gitea_audit.redact({
-        "success": True,
-        "performed": True,
-        "dry_run": False,
-        "break_glass_executed": True,
-        "would_execute": True,
-        "correlation_id": correlation_id,
-        "actor": identity,
-        "role": active_role,
-        "mcp_namespace": mcp_namespace,
-        "restart_class": restart_class,
-        "reason": clean_reason,
-        "confirmation": clean_confirmation,
-        "disrupted_sessions_count": disrupted_count,
-        "disrupted_sessions": disrupted_sessions,
-        "audit_record": term_audit,
-        "saved_audit": term_audit,
-        "incident_issue": incident_issue_result,
-        "reconciliation_result": recon_result,
-        "reconciliation_required": True,
-        "reconciliation_tool": "gitea_reconcile_after_restart",
-        "follow_up_issue_required": True,
-        "cross_references": ["#652", "#653", "#655", "#630", "#658", "#662", "#664"],
-        "reasons": [
-            "break-glass restart executed with incident creation and mandatory reconciliation"
-        ],
-    })
-
-
-
 # --- #662 post-restart reconciliation ---------------------------------------
 
 _POST_RESTART_LAST_PROOF: dict | None = None

namespace_workspace_binding.py

@@ -37,17 +37,12 @@ def normalize_role_kind(
     *,
     profile_name: str | None = None,
 ) -> str:
-    """Map profile/task role to a workspace namespace key.
-
-    Exact profile-name matches only for controller routing (#840 / #664 B1):
-    substring lookalikes such as ``fake-controller`` must not become
-    controller.
-    """
+    """Map profile/task role to a workspace namespace key."""
     role = (role_kind or "author").strip().lower()
     profile = (profile_name or "").strip().lower()
-    if role == "reviewer" and profile in ("prgs-merger", "mdcps-merger", "merger"):
+    if role == "reviewer" and "merger" in profile:
         return "merger"
-    if role == "controller" or profile in ("prgs-controller", "controller"):
+    if "controller" in profile or role == "controller":
         return "controller"
     if role in ROLE_WORKTREE_ENVS:
         return role

task_capability_map.py

@@ -576,15 +576,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "runtime.record_analytics_usage",
         "role": "author",
     },
-    # #664: emergency break-glass MCP restart workflow (privileged controller role).
-    "break_glass_restart": {
-        "permission": "runtime.break_glass_restart",
-        "role": "controller",
-    },
-    "gitea_break_glass_restart": {
-        "permission": "runtime.break_glass_restart",
-        "role": "controller",
-    },
 }
 
 

tests/test_issue_664_break_glass_restart.py

@@ -1,982 +0,0 @@
-"""Tests for emergency break-glass MCP restart workflow (#664).
-
-Regression suite for review #641 remediation: B13, B1, B14, B6/B11, B8, and
-preservation of previously accepted B2/B3/B5/B7/B9/B10 corrections.
-"""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import MagicMock, patch
-
-import gitea_audit
-import gitea_config
-import gitea_mcp_server
-
-
-def _controller_profile(**extra) -> dict:
-    base = {
-        "profile_name": "prgs-controller",
-        "execution_profile": "prgs-controller",
-        "role": "reconciler",  # declared role must not be redefined by capability
-        "allowed_operations": [
-            "gitea.read",
-            "gitea.issue.create",
-            "gitea.branch.delete",
-            "gitea.pr.close",
-            "gitea.pr.comment",
-            "gitea.issue.comment",
-            "runtime.break_glass_restart",
-        ],
-        "forbidden_operations": [
-            "gitea.pr.approve",
-            "gitea.pr.merge",
-            "gitea.pr.create",
-            "gitea.branch.push",
-        ],
-    }
-    base.update(extra)
-    return base
-
-
-def _gate_open_patches():
-    """Keep master-parity / runtime-mode blocks out of unit tests."""
-    return (
-        patch.object(gitea_mcp_server, "_master_parity_block", return_value=[]),
-        patch.object(gitea_mcp_server, "_runtime_mode_block", return_value=[]),
-        patch.object(gitea_mcp_server, "_try_auto_switch_for_operation", return_value=False),
-    )
-
-
-class TestBreakGlassRestart(unittest.TestCase):
-    """Test suite for gitea_break_glass_restart tool and guardrails (#664)."""
-
-    def setUp(self) -> None:
-        self.env_patcher = patch.dict(os.environ, {}, clear=False)
-        self.env_patcher.start()
-        os.environ.pop("GITEA_BREAKGLASS_RESTART_AUTHORIZATION", None)
-        os.environ.pop("GITEA_SANCTIONED_RESTART_HOOK", None)
-        os.environ.pop("GITEA_AUDIT_LOG", None)
-        # Reset injectable executor between tests.
-        gitea_mcp_server._break_glass_restart_executor = None
-
-    def tearDown(self) -> None:
-        gitea_mcp_server._break_glass_restart_executor = None
-        self.env_patcher.stop()
-
-    # ── B13: operation registration / real gate ───────────────────────────
-
-    def test_normalize_operation_accepts_canonical_break_glass_op(self) -> None:
-        """B13: production normalizer accepts exact runtime.break_glass_restart."""
-        self.assertEqual(
-            gitea_config.normalize_operation(
-                "runtime.break_glass_restart", service="runtime"
-            ),
-            "runtime.break_glass_restart",
-        )
-        ok, reason = gitea_config.check_operation(
-            "runtime.break_glass_restart",
-            ["gitea.read", "runtime.break_glass_restart"],
-        )
-        self.assertTrue(ok, reason)
-        self.assertEqual(reason, "allowed")
-
-    def test_normalize_unknown_and_misspelled_ops_fail_closed(self) -> None:
-        """B13: unknown / misspelled operations fail closed (no gitea.read fallback)."""
-        # Well-formed but misspelled runtime op normalizes, then is not allowed.
-        ok, reason = gitea_config.check_operation(
-            "runtime.break_glass_restar",  # misspelled
-            ["gitea.read", "runtime.break_glass_restart"],
-        )
-        self.assertFalse(ok)
-        self.assertEqual(reason, "not-allowed")
-
-        ok2, reason2 = gitea_config.check_operation(
-            "runtime.break_glass_restart",
-            ["gitea.read"],  # capability not granted
-        )
-        self.assertFalse(ok2)
-        self.assertEqual(reason2, "not-allowed")
-
-        ok3, reason3 = gitea_config.check_operation(
-            "frobnicate",
-            ["gitea.read", "runtime.break_glass_restart"],
-        )
-        self.assertFalse(ok3)
-        self.assertEqual(reason3, "invalid-operation")
-
-        # gitea.read grant alone never authorizes break-glass.
-        ok4, reason4 = gitea_config.check_operation(
-            "runtime.break_glass_restart",
-            ["gitea.read"],
-        )
-        self.assertFalse(ok4)
-        self.assertNotEqual(reason4, "allowed")
-
-    def test_real_profile_operation_gate_without_stubbing(self) -> None:
-        """B13: exercise real _profile_operation_gate (not stubbed)."""
-        allowed = _controller_profile()
-        denied = _controller_profile(
-            allowed_operations=["gitea.read", "gitea.issue.create"]
-        )
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=allowed):
-                self.assertEqual(
-                    gitea_mcp_server._profile_operation_gate(
-                        "runtime.break_glass_restart"
-                    ),
-                    [],
-                )
-            with patch.object(gitea_mcp_server, "get_profile", return_value=denied):
-                reasons = gitea_mcp_server._profile_operation_gate(
-                    "runtime.break_glass_restart"
-                )
-                self.assertTrue(reasons)
-                self.assertTrue(
-                    any("runtime.break_glass_restart" in r or "not allowed" in r
-                        for r in reasons)
-                )
-
-    def test_entry_point_uses_same_operation_as_gate(self) -> None:
-        """B13: entry point enforces runtime.break_glass_restart, not gitea.read."""
-        # Profile has gitea.read but not the break-glass capability.
-        prof = _controller_profile(
-            allowed_operations=["gitea.read", "gitea.issue.create"]
-        )
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart required due to deadlock in worker pool",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "permission_denied")
-        self.assertNotIn("gitea.read", " ".join(res.get("reasons") or []))
-
-    # ── B1 / B14: exact profile auth, no substring, role preservation ─────
-
-    def test_trusted_prgs_controller_authorized(self) -> None:
-        """B1: exact trusted prgs-controller with capability is authorized."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart required due to deadlock in worker pool",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=True,
-                    remote="prgs",
-                )
-        self.assertTrue(res["success"], res)
-        self.assertFalse(res["break_glass_executed"])
-
-    def test_fabricated_controller_like_profile_names_denied(self) -> None:
-        """B1: lookalike profile names never become authorized."""
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        for name in (
-            "fake-controller",
-            "controller-copy",
-            "not-controller",
-            "xcontrollerx",
-            "CONTROLLER",
-            "prgs-controller-copy",
-        ):
-            prof = _controller_profile(profile_name=name, execution_profile=name)
-            with p_parity, p_runtime, p_switch:
-                with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                    res = gitea_mcp_server.gitea_break_glass_restart(
-                        reason="Emergency restart required due to deadlock in worker pool",
-                        confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                        impact_ack=True,
-                        dry_run=True,
-                        remote="prgs",
-                    )
-            self.assertFalse(res["success"], name)
-            self.assertEqual(res["blocker_kind"], "role_authorization", name)
-            self.assertFalse(res["break_glass_executed"])
-
-    def test_ordinary_and_non_controller_reconciler_denied(self) -> None:
-        """B1: ordinary roles and non-controller reconcilers are denied."""
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        denied = [
-            {"profile_name": "prgs-author", "role": "author",
-             "allowed_operations": ["gitea.read", "runtime.break_glass_restart"],
-             "forbidden_operations": []},
-            {"profile_name": "prgs-reviewer", "role": "reviewer",
-             "allowed_operations": ["gitea.read", "runtime.break_glass_restart"],
-             "forbidden_operations": []},
-            {"profile_name": "prgs-merger", "role": "merger",
-             "allowed_operations": ["gitea.read", "runtime.break_glass_restart"],
-             "forbidden_operations": []},
-            {"profile_name": "prgs-reconciler", "role": "reconciler",
-             "allowed_operations": [
-                 "gitea.read", "gitea.branch.delete", "runtime.break_glass_restart"
-             ],
-             "forbidden_operations": []},
-            {"profile_name": "prgs-controller", "role": "reconciler",
-             "allowed_operations": ["gitea.read"],
-             "forbidden_operations": []},  # no capability
-        ]
-        for prof in denied:
-            with p_parity, p_runtime, p_switch:
-                with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                    res = gitea_mcp_server.gitea_break_glass_restart(
-                        reason="Emergency restart required due to deadlock in worker pool",
-                        confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                        impact_ack=True,
-                        dry_run=True,
-                        remote="prgs",
-                    )
-            self.assertFalse(res["success"], prof)
-            self.assertFalse(res["break_glass_executed"], prof)
-            self.assertIn(
-                res["blocker_kind"],
-                ("role_authorization", "permission_denied"),
-                prof,
-            )
-
-    def test_env_var_cannot_grant_authorization_or_bypass_denial(self) -> None:
-        """B2 preserved: env var cannot grant break-glass authorization."""
-        os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "secret-bypass-token"
-        prof = {
-            "profile_name": "prgs-author",
-            "role": "author",
-            "allowed_operations": [
-                "gitea.read", "gitea.issue.create", "runtime.break_glass_restart"
-            ],
-            "forbidden_operations": [],
-        }
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart attempting env var bypass",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "role_authorization")
-
-    def test_profile_role_kind_no_substring_authority(self) -> None:
-        """B1/B14: substring lookalikes do not become controller."""
-        for name in (
-            "fake-controller",
-            "controller-copy",
-            "not-controller",
-            "xcontrollerx",
-            "myCONTROLLER",
-        ):
-            prof = {
-                "profile_name": name,
-                "role": "author",
-                "allowed_operations": ["gitea.read"],
-            }
-            self.assertEqual(
-                gitea_mcp_server._profile_role_kind(prof),
-                "author",
-                name,
-            )
-        # Role substrings must not promote.
-        for role in ("not-controller", "control-plane", "xcontrollerx"):
-            prof = {"profile_name": "other", "role": role, "allowed_operations": ["gitea.read"]}
-            self.assertEqual(
-                gitea_mcp_server._profile_role_kind(prof),
-                role,
-                role,
-            )
-
-    def test_prgs_controller_retains_declared_reconciler_role(self) -> None:
-        """B14: break-glass capability does not redefine global role."""
-        prof = _controller_profile(role="reconciler")
-        self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler")
-        # Declared controller still wins when declared.
-        prof2 = _controller_profile(role="controller")
-        self.assertEqual(gitea_mcp_server._profile_role_kind(prof2), "controller")
-
-    def test_cleanup_merged_pr_branch_role_resolution_unchanged(self) -> None:
-        """B14: prgs-controller with reconciler role still resolves for cleanup."""
-        # When declared reconciler, cleanup gate's role check should see reconciler.
-        prof = _controller_profile(role="reconciler")
-        self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler")
-        # Fabricated controller-like names with reconciler ops do not become controller.
-        fake = {
-            "profile_name": "fake-controller",
-            "role": "reconciler",
-            "allowed_operations": [
-                "gitea.read", "gitea.branch.delete", "gitea.pr.close"
-            ],
-        }
-        self.assertEqual(gitea_mcp_server._profile_role_kind(fake), "reconciler")
-
-    def test_narrow_break_glass_capability_does_not_redefine_role(self) -> None:
-        """B14: granting runtime.break_glass_restart does not invent controller role."""
-        prof = {
-            "profile_name": "prgs-reconciler",
-            "role": "reconciler",
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.branch.delete",
-                "runtime.break_glass_restart",
-            ],
-        }
-        self.assertEqual(gitea_mcp_server._profile_role_kind(prof), "reconciler")
-
-    # ── B2-style input validation (preserved) ─────────────────────────────
-
-    def test_reason_validation(self) -> None:
-        """AC2: Reason is required and must be at least 10 characters long."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        for invalid_reason in ["", "   ", "too short", "123456789"]:
-            with p_parity, p_runtime, p_switch:
-                with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                    res = gitea_mcp_server.gitea_break_glass_restart(
-                        reason=invalid_reason,
-                        confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                        impact_ack=True,
-                        remote="prgs",
-                    )
-            self.assertFalse(res["success"])
-            self.assertEqual(res["blocker_kind"], "missing_required_fields")
-
-    def test_confirmation_validation(self) -> None:
-        """AC2: Confirmation phrase must match exact required string."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart needed due to stuck daemon processes",
-                    confirmation="wrong_confirmation_phrase",
-                    impact_ack=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertEqual(res["blocker_kind"], "confirmation_mismatch")
-
-    def test_impact_ack_validation(self) -> None:
-        """AC2: impact_ack=True is mandatory."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart needed due to stuck daemon processes",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=False,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertEqual(res["blocker_kind"], "impact_ack_required")
-
-    def test_incident_permission_gate(self) -> None:
-        """B3 preserved: Gate incident creation on gitea.issue.create permission."""
-        prof = _controller_profile(
-            allowed_operations=["gitea.read", "runtime.break_glass_restart"]
-        )
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart needed due to hung worker process cohort",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    create_incident_issue=True,
-                    dry_run=False,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertEqual(res["blocker_kind"], "permission_denied")
-
-    # ── B8: redaction ─────────────────────────────────────────────────────
-
-    def test_redaction_key_value_and_connection_strings(self) -> None:
-        """B8: key/value, bearer, connection-string, and embedded secrets."""
-        cases = [
-            ("password=hunter2supersecret", ["hunter2supersecret"], "password"),
-            ("api_key: sk-live-abcdef1234567890ab", ["sk-live-abcdef1234567890ab"], "api_key"),
-            ("Server=db;Password=s3cretValue;Uid=sa", ["s3cretValue"], "password"),
-            (
-                "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.abc.def",
-                ["eyJhbGciOiJIUzI1NiJ9.abc.def", "Bearer eyJ"],
-                "authorization",
-            ),
-            (
-                "token ghp_1234567890abcdef12345678 embedded",
-                ["ghp_1234567890abcdef12345678"],
-                "token",
-            ),
-            ("nested note password=letmein12345 end", ["letmein12345"], "password"),
-        ]
-        for raw, secrets, key in cases:
-            out = gitea_audit._redact_str(raw)
-            self.assertIn("[REDACTED]", out, raw)
-            for secret in secrets:
-                self.assertNotIn(secret, out, raw)
-            self.assertIn(key, out.lower(), raw)
-
-        nested = gitea_audit.redact({
-            "reason": "password=supersecret99",
-            "items": [{"api_key": "abc123xyz"}, "token ghp_abcdefghijklmnop1234"],
-            "error": RuntimeError("pwd=nestedSecret99"),
-        })
-        # Exception objects pass through redact as non-str/non-container; ensure
-        # string forms are covered via str conversion in _redact_str usage.
-        self.assertEqual(nested["items"][0]["api_key"], gitea_audit.REDACTED)
-        self.assertNotIn("supersecret99", nested["reason"])
-        self.assertNotIn("ghp_abcdefghijklmnop1234", nested["items"][1])
-
-    def test_redaction_preserves_benign_sec_prefix_text(self) -> None:
-        """B8: ordinary text beginning with sec- must not be erased."""
-        benign = (
-            "Emergency restart in sec-primary-region for sector-planning "
-            "and secondary-health checks"
-        )
-        out = gitea_audit._redact_str(benign)
-        self.assertIn("sec-primary-region", out)
-        self.assertIn("sector-planning", out)
-        self.assertIn("secondary-health", out)
-        self.assertNotIn("[REDACTED]", out)
-
-    def test_redaction_across_break_glass_surfaces(self) -> None:
-        """B8: operator reason is redacted on result, incident, and audit surfaces."""
-        prof = _controller_profile()
-        raw_reason = (
-            "Emergency restart: password=supersecret99 api_key: "
-            "sk-live-abcdef1234567890ab and url https://user:[email protected]/api"
-        )
-        mock_api_request = MagicMock(return_value={"number": 101, "title": "[INCIDENT]"})
-        fake_exec = {
-            "success": True,
-            "apply_supported": True,
-            "apply_authorized": True,
-            "restart_performed": True,
-            "break_glass_executed": True,
-        }
-        gitea_mcp_server._break_glass_restart_executor = lambda req: fake_exec
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "api_request", mock_api_request
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_mcp_server, "gitea_reconcile_after_restart",
-                return_value={"success": True},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=True
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason=raw_reason,
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertTrue(res["success"], res)
-        self.assertNotIn("supersecret99", res["reason"])
-        self.assertNotIn("sk-live-abcdef1234567890ab", res["reason"])
-        self.assertNotIn("user:[email protected]", res["reason"])
-        posted_body = mock_api_request.call_args[0][3]["body"]
-        self.assertNotIn("supersecret99", posted_body)
-        self.assertNotIn("sk-live-abcdef1234567890ab", posted_body)
-
-    # ── B6/B11: reachable executor / truthful flags ───────────────────────
-
-    def test_dry_run_never_executes_and_reports_false(self) -> None:
-        """B7/B6: dry-run never executes; break_glass_executed always false."""
-        prof = _controller_profile()
-        called = {"n": 0}
-
-        def _should_not_run(_req):
-            called["n"] += 1
-            return {"restart_performed": True, "break_glass_executed": True}
-
-        gitea_mcp_server._break_glass_restart_executor = _should_not_run
-        mock_audit = MagicMock()
-        mock_api = MagicMock()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": [{"session_id": "s1"}]},
-            ), patch.object(
-                gitea_audit, "write_event", mock_audit
-            ), patch.object(
-                gitea_mcp_server, "api_request", mock_api
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart preview in dry-run mode",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=True,
-                    remote="prgs",
-                )
-        self.assertTrue(res["success"])
-        self.assertTrue(res["dry_run"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertTrue(res["would_execute"])
-        self.assertEqual(called["n"], 0)
-        mock_audit.assert_not_called()
-        mock_api.assert_not_called()
-
-    def test_unsupported_apply_truthful(self) -> None:
-        """B6/B11: unsupported apply returns blocked result, execution false."""
-        prof = _controller_profile()
-        gitea_mcp_server._break_glass_restart_executor = lambda req: {
-            "success": False,
-            "apply_supported": False,
-            "apply_authorized": False,
-            "restart_performed": False,
-            "break_glass_executed": False,
-            "reasons": ["no hook"],
-        }
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=True
-            ), patch.object(
-                gitea_mcp_server, "api_request",
-                return_value={"number": 555, "title": "[INCIDENT]"},
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Privileged restart request with unsupported apply",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["performed"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "apply_unsupported")
-
-    def test_delegation_success_rejection_and_failure(self) -> None:
-        """B6/B11: distinct terminal states for success / rejection / failure."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-
-        def _run(exec_result, recon=None):
-            gitea_mcp_server._break_glass_restart_executor = lambda req: exec_result
-            with p_parity, p_runtime, p_switch:
-                with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                    gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-                ), patch.object(
-                    gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-                ), patch.object(
-                    gitea_mcp_server, "gitea_request_mcp_restart",
-                    return_value={"affected_sessions": []},
-                ), patch.object(
-                    gitea_mcp_server, "gitea_reconcile_after_restart",
-                    return_value=recon or {"success": True},
-                ), patch.object(
-                    gitea_audit, "audit_enabled", return_value=True
-                ), patch.object(
-                    gitea_audit, "write_event", return_value=True
-                ), patch.object(
-                    gitea_mcp_server, "api_request",
-                    return_value={"number": 555, "title": "[INCIDENT]"},
-                ):
-                    return gitea_mcp_server.gitea_break_glass_restart(
-                        reason="Privileged break-glass restart delegation path",
-                        confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                        impact_ack=True,
-                        dry_run=False,
-                        create_incident_issue=True,
-                        remote="prgs",
-                    )
-
-        ok = _run({
-            "success": True,
-            "apply_supported": True,
-            "apply_authorized": True,
-            "restart_performed": True,
-            "break_glass_executed": True,
-        })
-        self.assertTrue(ok["success"], ok)
-        self.assertTrue(ok["break_glass_executed"])
-        self.assertTrue(ok["performed"])
-
-        rejected = _run({
-            "success": False,
-            "apply_supported": True,
-            "apply_authorized": False,
-            "restart_performed": False,
-            "break_glass_executed": False,
-            "reasons": ["class denied"],
-        })
-        self.assertFalse(rejected["success"])
-        self.assertFalse(rejected["break_glass_executed"])
-        self.assertEqual(rejected["blocker_kind"], "restart_delegation_failed")
-
-        failed = _run({
-            "success": False,
-            "apply_supported": True,
-            "apply_authorized": True,
-            "restart_performed": False,
-            "break_glass_executed": False,
-            "reasons": ["executor error"],
-        })
-        self.assertFalse(failed["success"])
-        self.assertFalse(failed["break_glass_executed"])
-        self.assertEqual(failed["blocker_kind"], "restart_delegation_failed")
-
-    def test_reconciliation_success_and_failure_truthful_flags(self) -> None:
-        """B10 preserved: recon failure keeps break_glass_executed=true."""
-        prof = _controller_profile()
-        gitea_mcp_server._break_glass_restart_executor = lambda req: {
-            "success": True,
-            "apply_supported": True,
-            "apply_authorized": True,
-            "restart_performed": True,
-            "break_glass_executed": True,
-        }
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_mcp_server, "gitea_reconcile_after_restart",
-                return_value={"success": False, "error": "lease cleanup failed"},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=True
-            ), patch.object(
-                gitea_mcp_server, "api_request",
-                return_value={"number": 555, "title": "[INCIDENT]"},
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Privileged restart request with failing reconciliation",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertTrue(res["performed"])
-        self.assertTrue(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "reconciliation_failed")
-
-    def test_authorization_is_not_execution(self) -> None:
-        """B6: apply_authorized alone never sets break_glass_executed."""
-        # Default executor without hook → unsupported, not executed.
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=True
-            ), patch.object(
-                gitea_mcp_server, "api_request",
-                return_value={"number": 555, "title": "[INCIDENT]"},
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Privileged restart without host hook configured",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "apply_unsupported")
-
-    # ── preserved fail-closed pre-exec (B5/B9) ────────────────────────────
-
-    def test_audit_failure_before_execution_fails_closed(self) -> None:
-        """B9 preserved: Audit recording failure stops execution fail-closed."""
-        prof = _controller_profile()
-        called = {"n": 0}
-        gitea_mcp_server._break_glass_restart_executor = lambda req: called.__setitem__("n", called["n"] + 1) or {}
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=False
-            ), patch.object(
-                gitea_mcp_server, "api_request", MagicMock()
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart with failing audit sink",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "audit_recording_failed")
-        self.assertEqual(called["n"], 0)
-
-    def test_incident_creation_failure_before_execution_fails_closed(self) -> None:
-        """B5 preserved: Incident creation failure stops execution fail-closed."""
-        prof = _controller_profile()
-        called = {"n": 0}
-        gitea_mcp_server._break_glass_restart_executor = lambda req: called.__setitem__("n", called["n"] + 1) or {}
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=True
-            ), patch.object(
-                gitea_mcp_server, "api_request",
-                side_effect=RuntimeError("Gitea 500 API Error"),
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart with failing incident POST",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "incident_creation_failed")
-        self.assertEqual(called["n"], 0)
-
-    def test_incident_opt_out_on_real_execution_fails_closed(self) -> None:
-        """B5 preserved: create_incident_issue=False fails closed on real execution."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart trying to skip incident creation",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=False,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "incident_creation_required")
-
-    def test_audit_disabled_fails_closed(self) -> None:
-        """B9 preserved: Disabling audit recording blocks execution fail-closed."""
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_audit, "audit_enabled", return_value=False
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Emergency restart with disabled audit logging",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "audit_recording_failed")
-
-    def test_capability_map_registration(self) -> None:
-        """B12/B13: capability map registers exact runtime.break_glass_restart."""
-        from task_capability_map import TASK_CAPABILITY_MAP
-        entry = TASK_CAPABILITY_MAP.get("gitea_break_glass_restart")
-        self.assertIsNotNone(entry)
-        self.assertEqual(entry["permission"], "runtime.break_glass_restart")
-        self.assertEqual(entry["role"], "controller")
-        entry2 = TASK_CAPABILITY_MAP.get("break_glass_restart")
-        self.assertEqual(entry2["permission"], "runtime.break_glass_restart")
-
-    # ── B8 / B6 / B15 remediation tests ─────────────────────────────────────
-
-    def test_b8_redaction_gitea_token_and_uri_credentials(self) -> None:
-        """B8: GITEA_TOKEN= and URI userinfo credentials redacted without erasing neighbours."""
-        # GITEA_TOKEN= with underscore key
-        out1 = gitea_audit._redact_str("failed with GITEA_TOKEN=synthetic_tok_123456789")
-        self.assertIn("GITEA_TOKEN=[REDACTED]", out1)
-        self.assertNotIn("synthetic_tok_123456789", out1)
-
-        # Connection string with URI userinfo
-        out2 = gitea_audit._redact_str("conn postgres://user:[email protected]:5432/app")
-        self.assertIn("postgres://[REDACTED_USER]:[REDACTED_PASS]@db.internal:5432/app", out2)
-        self.assertNotIn("s3cr3tpw", out2)
-
-        # Value boundary preserving adjacent audit evidence (correlation_id, incident_number)
-        raw_audit = "password=secret123;correlation_id=bg-7f2a1c;incident_number=4242"
-        out3 = gitea_audit._redact_str(raw_audit)
-        self.assertIn("password=[REDACTED]", out3)
-        self.assertIn("correlation_id=bg-7f2a1c", out3)
-        self.assertIn("incident_number=4242", out3)
-        self.assertNotIn("secret123", out3)
-
-        # Query param boundary in URL preserving adjacent parameters
-        raw_url = "token=abc-123&pr=908&issue=664&head=c67f39b4"
-        out4 = gitea_audit._redact_str(raw_url)
-        self.assertIn("token=[REDACTED]", out4)
-        self.assertIn("pr=908", out4)
-        self.assertIn("issue=664", out4)
-        self.assertIn("head=c67f39b4", out4)
-
-    def test_b6_default_executor_environment_text_cannot_imply_execution(self) -> None:
-        """B6/B11: GITEA_SANCTIONED_RESTART_HOOK string alone returns break_glass_executed=False."""
-        os.environ["GITEA_SANCTIONED_RESTART_HOOK"] = "this-string-is-never-invoked"
-        prof = _controller_profile()
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prof), patch.object(
-                gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
-            ), patch.object(
-                gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
-            ), patch.object(
-                gitea_mcp_server, "gitea_request_mcp_restart",
-                return_value={"affected_sessions": []},
-            ), patch.object(
-                gitea_audit, "audit_enabled", return_value=True
-            ), patch.object(
-                gitea_audit, "write_event", return_value=True
-            ), patch.object(
-                gitea_mcp_server, "api_request",
-                return_value={"number": 555, "title": "[INCIDENT]"},
-            ):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Privileged restart request with non-empty hook env var",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertFalse(res["performed"])
-        self.assertFalse(res["break_glass_executed"])
-        self.assertEqual(res["blocker_kind"], "restart_delegation_failed")
-
-    def test_b15_production_prgs_controller_grant_set_and_gates(self) -> None:
-        """B15: Genuine prgs-controller carrying runtime.break_glass_restart and gitea.issue.create passes real gates."""
-        # Full production-shaped prgs-controller profile
-        prod_profile = {
-            "profile_name": "prgs-controller",
-            "execution_profile": "prgs-controller",
-            "role": "reconciler",
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.pr.close",
-                "gitea.pr.comment",
-                "gitea.issue.comment",
-                "gitea.issue.create",
-                "runtime.break_glass_restart",
-                "gitea.branch.delete",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.merge",
-                "gitea.pr.create",
-                "gitea.branch.push",
-            ],
-        }
-
-        # 1. Real _profile_operation_gate checks
-        p_parity, p_runtime, p_switch = _gate_open_patches()
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=prod_profile):
-                # Both required operations pass the real operation gate (no stubs)
-                self.assertEqual(
-                    gitea_mcp_server._profile_operation_gate("runtime.break_glass_restart"),
-                    [],
-                )
-                self.assertEqual(
-                    gitea_mcp_server._profile_operation_gate("gitea.issue.create"),
-                    [],
-                )
-
-        # 2. Missing gitea.issue.create fails incident creation gate
-        no_issue_create = dict(prod_profile)
-        no_issue_create["allowed_operations"] = [
-            "gitea.read", "gitea.pr.close", "runtime.break_glass_restart"
-        ]
-        with p_parity, p_runtime, p_switch:
-            with patch.object(gitea_mcp_server, "get_profile", return_value=no_issue_create):
-                res = gitea_mcp_server.gitea_break_glass_restart(
-                    reason="Restart testing missing gitea.issue.create permission",
-                    confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
-                    impact_ack=True,
-                    dry_run=False,
-                    create_incident_issue=True,
-                    remote="prgs",
-                )
-        self.assertFalse(res["success"])
-        self.assertEqual(res["blocker_kind"], "permission_denied")
-        self.assertIn("gitea.issue.create", " ".join(res.get("reasons") or []))
-
-
-if __name__ == "__main__":
-    unittest.main()
-

tests/test_issue_697_untracked_preflight.py

@@ -0,0 +1,32 @@
+"""Regression test for Issue #697: Native workspace preflight explicitly includes untracked files.
+
+Verifies that `_parse_porcelain_entries` includes untracked files (lines starting with '??')
+so workspace preflight fails closed on untracked or dirty state.
+"""
+
+from __future__ import annotations
+
+import unittest
+import gitea_mcp_server as mcp_server
+
+
+class TestIssue697UntrackedPreflight(unittest.TestCase):
+    def test_parse_porcelain_entries_includes_untracked_files(self):
+        porcelain = " M gitea_mcp_server.py\n?? untracked_file.txt\n?? new_dir/test.py\n"
+        entries = mcp_server._parse_porcelain_entries(porcelain)
+        self.assertIn("untracked_file.txt", entries)
+        self.assertIn("new_dir/test.py", entries)
+        self.assertEqual(entries["untracked_file.txt"], "?? untracked_file.txt")
+        self.assertEqual(entries["new_dir/test.py"], "?? new_dir/test.py")
+
+    def test_parse_porcelain_entries_includes_tracked_and_untracked(self):
+        porcelain = " M tracked.py\n?? untracked.py\n D deleted.py\n"
+        entries = mcp_server._parse_porcelain_entries(porcelain)
+        self.assertEqual(len(entries), 3)
+        self.assertIn("tracked.py", entries)
+        self.assertIn("untracked.py", entries)
+        self.assertIn("deleted.py", entries)
+
+
+if __name__ == "__main__":
+    unittest.main()