Merge master into feat/issue-663-restart-classes (resolve #886 conflict)

Brings PR #886 up to date with master @ 2f4dec8323
(8 commits behind), resolving the single conflicted file.

Conflict: gitea_mcp_server.py, both hunks inside gitea_request_mcp_restart.
Both sides were purely additive to the same tool, so both are kept in full:

- Branch side (#663, restart classes): parameters restart_class,
  target_session_id, target_role, target_connector; payload keys
  controller_approval_authorized, requester_role, requester_permissions.
- Master side (#661 via PR #882, drain-proof hard gate): parameters
  drain_proof_json, request_break_glass; the explanatory comment describing
  the apply-path hard gate and break-glass authorization.

No behaviour from either side was dropped, reordered, or reimplemented. Every
parameter from both sides is already consumed by the auto-merged function body
(restart_class and the three target_* arguments flow into the coordinator call;
drain_proof_json and request_break_glass drive the dry_run=False hard gate), so
the union is the only resolution that keeps the merged function coherent.

Validation on the merged tree:

  python -m pytest tests/test_drain_proof.py tests/test_restart_classes.py \
    tests/test_restart_coordinator.py tests/test_mcp_restart_paths.py \
    tests/test_mcp_restart_governance_docs.py tests/test_webui_sanctioned_restart.py \
    tests/test_issue_662_post_restart_reconcile.py -q
  # 165 passed, 68 subtests passed

py_compile on gitea_mcp_server.py passes and no conflict markers remain.

Closes #663

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -2068,6 +2068,7 @@ import lease_lifecycle  # noqa: E402
 import lease_policy  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import restart_coordinator  # noqa: E402  # #658 MCP restart coordinator/impact
+import drain_proof  # noqa: E402  # #661 pre-restart drain proof and hard gate
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
 import sentry_incident_bridge  # noqa: E402  (#607 Sentry→Gitea incident bridge)
@@ -22346,6 +22347,8 @@ def gitea_request_mcp_restart(
     target_session_id: str | None = None,
     target_role: str | None = None,
     target_connector: str | None = None,
+    drain_proof_json: str | None = None,
+    request_break_glass: bool = False,
 ) -> dict:
     """Evaluate a proposed MCP restart and return an impact preview (#658).
 
@@ -22356,10 +22359,15 @@ def gitea_request_mcp_restart(
     verdict, so the console (#642/#652) and operators can see what a restart
     would disrupt *before* any concurrent LLM work is destroyed.
 
-    This tool is **dry-run and never restarts anything.** The mutative apply
-    path is a separate child gated by a drain proof (non-goal here); calling
-    with ``dry_run=False`` still performs no restart and reports that apply is
-    not yet available.
+    This tool **never restarts a process.** In dry-run (the default) it returns
+    only the impact preview. With ``dry_run=False`` it enforces the #661 hard
+    gate: the apply request must present a valid, unexpired, clean drain proof
+    (``drain_proof_json``) or it is denied and a durable incident descriptor is
+    returned under ``incident``. Break-glass is the only bypass and is honoured
+    only when ``request_break_glass`` is set *and* the environment carries
+    ``GITEA_BREAKGLASS_RESTART_AUTHORIZATION``. Even an authorized gate performs
+    no restart here; actual execution is a further child. The gate outcome is
+    reported under ``apply_gate`` / ``apply_authorized``.
 
     Operator override authority is read from the process environment
     (``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
@@ -22501,12 +22509,54 @@ def gitea_request_mcp_restart(
     payload["controller_approval_authorized"] = controller_approved
     payload["requester_role"] = requester_role
     payload["requester_permissions"] = list(requester_permissions)
+    # Actual restart execution remains a further child; this tool never restarts
+    # a process. What #661 adds is the *hard gate*: an apply request (dry_run
+    # False) must present a valid, unexpired, clean drain proof, or it is denied
+    # and a durable incident is raised. Break-glass is the only bypass and its
+    # authorization is read from the environment, never self-asserted.
     payload["apply_supported"] = False
     if not dry_run:
-        payload["reasons"] = list(payload.get("reasons") or []) + [
-            "apply requested but not supported: sanctioned restart apply is "
-            "gated by a drain proof (separate child); no restart performed (#658)"
-        ]
+        proof_obj: dict | None = None
+        proof_parse_error: str | None = None
+        if drain_proof_json:
+            try:
+                parsed = json.loads(drain_proof_json)
+                proof_obj = parsed if isinstance(parsed, dict) else None
+                if proof_obj is None:
+                    proof_parse_error = "drain_proof_json is not a JSON object"
+            except (ValueError, TypeError) as exc:
+                proof_parse_error = f"invalid drain_proof_json: {_redact(str(exc))}"
+
+        break_glass_authorized = bool(
+            (
+                os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or ""
+            ).strip()
+        )
+        break_glass = bool(request_break_glass and break_glass_authorized)
+
+        expected_fp = drain_proof.impact_fingerprint(report.as_dict())
+        gate = drain_proof.gate_apply_restart(
+            proof=proof_obj,
+            break_glass=break_glass,
+            expected_impact_fingerprint=expected_fp,
+            requesting_session_id=sid,
+        )
+        gate_payload = gate.as_dict()
+        if proof_parse_error and not break_glass:
+            gate_payload["reasons"] = [proof_parse_error] + list(
+                gate_payload.get("reasons") or []
+            )
+        payload["apply_gate"] = gate_payload
+        payload["apply_authorized"] = gate.allow
+        payload["break_glass_requested"] = bool(request_break_glass)
+        payload["break_glass_authorized"] = break_glass_authorized
+        # Even an authorized gate performs no restart here: execution is a later
+        # child. The gate proves the apply path *would* be permitted.
+        payload["reasons"] = list(payload.get("reasons") or []) + list(
+            gate_payload.get("reasons") or []
+        )
+        if not gate.allow and gate.incident is not None:
+            payload["incident"] = gate.incident
     return payload