fix(workflow): non-forgeable irrecoverable auth, merger consumer, exact-scope cleanup (#709)

Address formal review 434 REQUEST_CHANGES on PR #710:
- F1: replace caller operator_authorized with server-side HMAC auth artifacts
- F2: implement fail-closed merger consumption for prior-provenance only
- F3: enforce remote/org/repo/head on cross-profile load and clear

Co-Authored-By: Grok 4.5 (xAI) <[email protected]>

irrecoverable_provenance.py

@@ -0,0 +1,902 @@
+"""Server-side irrecoverable decision-lock provenance authorization (#709 AC5).
+
+Authorization is **not** a caller-supplied Boolean. A durable, non-forgeable
+authorization artifact must be minted under production native MCP transport
+(or pytest) with a dedicated mutation capability, live head binding, and
+validated incident evidence. Merger consumption is fail-closed and resolves
+only the historical-provenance blocker — never normal approval, lease,
+mergeability, anti-stomp, or workspace gates.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import mcp_daemon_guard
+import mcp_session_state
+from stale_review_decision_lock import heads_equal, normalize_head_sha
+
+# Dedicated mutation capability (#709 review 434 F1).
+CAPABILITY_IRRECOVERABLE_RECOVERY = "gitea.decision_lock.irrecoverable_recovery"
+
+KIND_AUTH = "irrecoverable_provenance_authorization"
+KIND_RECOVERY = mcp_session_state.KIND_IRRECOVERABLE_DECISION_PROVENANCE
+
+CONFIRMATION_PREFIX = "IRRECOVERABLE DECISION PROVENANCE PR"
+AUTH_TTL_HOURS = 24.0
+RECORD_TYPE = "irrecoverable_decision_provenance"
+AUTH_TYPE = "irrecoverable_provenance_authorization"
+
+# Internal HMAC material is process-local and never caller-supplied. Pytest
+# gets a deterministic salt so hermetic tests are stable; production uses
+# transport fingerprint + random secret minted at process start.
+_PROCESS_AUTH_SECRET: bytes | None = None
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _now_iso() -> str:
+    return _now().isoformat()
+
+
+def _process_secret() -> bytes:
+    global _PROCESS_AUTH_SECRET
+    if _PROCESS_AUTH_SECRET is None:
+        if mcp_daemon_guard.is_pytest_runtime():
+            _PROCESS_AUTH_SECRET = b"pytest-irrecoverable-auth-v1"
+        else:
+            _PROCESS_AUTH_SECRET = secrets.token_bytes(32)
+    return _PROCESS_AUTH_SECRET
+
+
+def expected_confirmation(pr_number: int) -> str:
+    """Human intent confirmation text (not an authorization credential)."""
+    return f"{CONFIRMATION_PREFIX} {int(pr_number)}"
+
+
+def auth_state_profile_identity(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+) -> str:
+    """Stable durable key segment for one exact-scope authorization."""
+    head = normalize_head_sha(expected_head_sha) or "nohead"
+    segs = [
+        KIND_AUTH,
+        mcp_session_state._sanitize_segment(remote),
+        mcp_session_state._sanitize_segment(org),
+        mcp_session_state._sanitize_segment(repo),
+        f"pr{int(pr_number)}",
+        head[:16],
+    ]
+    return "-".join(segs)
+
+
+def recovery_state_profile_identity(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+) -> str:
+    head = normalize_head_sha(expected_head_sha) or "nohead"
+    segs = [
+        KIND_RECOVERY,
+        mcp_session_state._sanitize_segment(remote),
+        mcp_session_state._sanitize_segment(org),
+        mcp_session_state._sanitize_segment(repo),
+        f"pr{int(pr_number)}",
+        head[:16],
+    ]
+    return "-".join(segs)
+
+
+def _scope_payload(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+    incident_issue: int,
+    incident_comment_id: int,
+    destroyed_subject: str | None,
+    issuer_username: str,
+    issuer_profile: str,
+    created_at: str,
+    expires_at: str,
+    authorization_id: str,
+) -> dict[str, Any]:
+    return {
+        "authorization_id": authorization_id,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "blocked_pr_number": int(pr_number),
+        "expected_head_sha": normalize_head_sha(expected_head_sha),
+        "incident_issue": int(incident_issue),
+        "incident_comment_id": int(incident_comment_id),
+        "destroyed_subject": (destroyed_subject or "").strip() or None,
+        "issuer_username": issuer_username,
+        "issuer_profile": issuer_profile,
+        "created_at": created_at,
+        "expires_at": expires_at,
+    }
+
+
+def _sign_scope(scope: dict[str, Any], native_provenance: dict[str, Any]) -> str:
+    """HMAC over canonical scope + native transport fingerprint (non-caller)."""
+    material = {
+        "scope": scope,
+        "native": {
+            "native_mcp_transport": bool(
+                native_provenance.get("native_mcp_transport")
+            ),
+            "production_native_mcp_transport": bool(
+                native_provenance.get("production_native_mcp_transport")
+            ),
+            "token_fingerprint": native_provenance.get("token_fingerprint"),
+            "entrypoint": native_provenance.get("entrypoint"),
+            "pid": native_provenance.get("pid"),
+        },
+    }
+    blob = json.dumps(material, sort_keys=True, separators=(",", ":")).encode(
+        "utf-8"
+    )
+    return hmac.new(_process_secret(), blob, hashlib.sha256).hexdigest()
+
+
+def assess_capability_for_irrecoverable_recovery(
+    *,
+    allowed_operations: list[str] | None,
+    forbidden_operations: list[str] | None = None,
+    role_kind: str | None = None,
+    profile_name: str | None = None,
+) -> dict[str, Any]:
+    """Whether the active profile may mint/use irrecoverable recovery (#709 F1).
+
+    Accepts the dedicated capability, or a reconciler-shaped profile that
+    already holds issue-comment mutation rights (interim equivalence until
+    operators grant the dedicated op). Never treats bare ``gitea.read`` as
+    sufficient.
+    """
+    import gitea_config
+    import reconciler_profile
+
+    allowed = list(allowed_operations or [])
+    forbidden = list(forbidden_operations or [])
+    reasons: list[str] = []
+
+    dedicated_ok, _ = gitea_config.check_operation(
+        CAPABILITY_IRRECOVERABLE_RECOVERY, allowed, forbidden
+    )
+    if dedicated_ok:
+        return {
+            "allowed": True,
+            "capability": CAPABILITY_IRRECOVERABLE_RECOVERY,
+            "via": "dedicated_capability",
+            "reasons": [],
+        }
+
+    # Interim: reconciler profile with issue.comment (mutation, not read-only).
+    is_reconciler = reconciler_profile.is_reconciler_profile(allowed, forbidden)
+    comment_ok, _ = gitea_config.check_operation(
+        "gitea.issue.comment", allowed, forbidden
+    )
+    role = (role_kind or "").strip().lower()
+    name = (profile_name or "").strip().lower()
+    role_looks_reconciler = role == "reconciler" or "reconciler" in name
+    if is_reconciler and comment_ok:
+        return {
+            "allowed": True,
+            "capability": CAPABILITY_IRRECOVERABLE_RECOVERY,
+            "via": "reconciler_profile_equivalence",
+            "reasons": [],
+        }
+    if role_looks_reconciler and comment_ok and not dedicated_ok:
+        # Role metadata says reconciler but ops incomplete — still fail if
+        # is_reconciler_profile is false (missing pr.close).
+        reasons.append(
+            "reconciler role metadata without reconciler-required operations "
+            f"(need {CAPABILITY_IRRECOVERABLE_RECOVERY} or reconciler profile "
+            "with gitea.pr.close + gitea.issue.comment; gitea.read alone is "
+            "insufficient, #709 F1)"
+        )
+    else:
+        reasons.append(
+            f"missing dedicated capability {CAPABILITY_IRRECOVERABLE_RECOVERY} "
+            "(gitea.read is insufficient; require reconciler-capable mutation "
+            "profile or explicit grant, #709 F1)"
+        )
+    return {
+        "allowed": False,
+        "capability": CAPABILITY_IRRECOVERABLE_RECOVERY,
+        "via": None,
+        "reasons": reasons,
+    }
+
+
+def assess_transport_for_auth_mint() -> dict[str, Any]:
+    """Native transport required for minting non-forgeable auth artifacts."""
+    reasons: list[str] = []
+    native = mcp_daemon_guard.is_native_mcp_transport()
+    pytest = mcp_daemon_guard.is_pytest_runtime()
+    production = mcp_daemon_guard.is_production_native_mcp_transport()
+    if not native and not pytest:
+        reasons.append(
+            "irrecoverable provenance authorization requires production native "
+            "MCP transport; ordinary Python processes cannot mint acceptable "
+            "recovery authorization (#709 F1)"
+        )
+    return {
+        "allowed": not reasons,
+        "native_mcp_transport": native,
+        "production_native_mcp_transport": production,
+        "pytest": pytest,
+        "reasons": reasons,
+    }
+
+
+def assess_incident_evidence(
+    *,
+    incident_issue: int | None,
+    incident_comment_id: int | None,
+    comment_payload: dict[str, Any] | None,
+    comment_lookup_error: str | None = None,
+    expected_remote: str | None = None,
+    expected_org: str | None = None,
+    expected_repo: str | None = None,
+) -> dict[str, Any]:
+    """Validate canonical incident evidence is present and live-fetched."""
+    reasons: list[str] = []
+    if incident_issue is None or int(incident_issue) <= 0:
+        reasons.append("incident_issue is required and must be a positive integer")
+    if incident_comment_id is None or int(incident_comment_id) <= 0:
+        reasons.append(
+            "incident_comment_id is required and must be a positive integer"
+        )
+    if comment_lookup_error:
+        reasons.append(
+            f"incident evidence lookup failed: {comment_lookup_error} (fail closed)"
+        )
+    if not isinstance(comment_payload, dict):
+        if not comment_lookup_error:
+            reasons.append(
+                "incident evidence not found or not a comment object (fail closed)"
+            )
+        return {"valid": False, "reasons": reasons, "comment": None}
+
+    # Gitea returns comment with id; optional issue_url / html_url for scope.
+    cid = comment_payload.get("id")
+    try:
+        if int(cid) != int(incident_comment_id):  # type: ignore[arg-type]
+            reasons.append(
+                "incident comment id mismatch against live payload (fail closed)"
+            )
+    except (TypeError, ValueError):
+        reasons.append("incident comment payload missing valid id (fail closed)")
+
+    body = (comment_payload.get("body") or "").strip()
+    if not body:
+        reasons.append("incident comment body is empty (fail closed)")
+
+    # Soft scope hints when URLs are present (never hard-code issue numbers).
+    issue_url = str(
+        comment_payload.get("issue_url")
+        or comment_payload.get("html_url")
+        or ""
+    )
+    if expected_org and expected_org not in issue_url and issue_url:
+        # Only fail when URL is present and clearly wrong-org; missing URL ok.
+        if f"/{expected_org}/" not in issue_url:
+            # html_url may be /user/repo/issues/n — check repo if provided
+            if expected_repo and f"/{expected_repo}/" not in issue_url:
+                reasons.append(
+                    "incident evidence URL does not match expected repository "
+                    "(fail closed)"
+                )
+
+    return {
+        "valid": not reasons,
+        "reasons": reasons,
+        "comment": {
+            "id": comment_payload.get("id"),
+            "author": (
+                (comment_payload.get("user") or {}).get("login")
+                if isinstance(comment_payload.get("user"), dict)
+                else comment_payload.get("user")
+            ),
+            "created_at": comment_payload.get("created_at"),
+            "body_len": len(body),
+        },
+    }
+
+
+def assess_live_head_binding(
+    *,
+    expected_head_sha: str | None,
+    live_head_sha: str | None,
+    pr_lookup_error: str | None = None,
+    pr_state: str | None = None,
+) -> dict[str, Any]:
+    """expected_head_sha mandatory and must equal live PR head."""
+    reasons: list[str] = []
+    want = normalize_head_sha(expected_head_sha)
+    have = normalize_head_sha(live_head_sha)
+    if not want:
+        reasons.append(
+            "expected_head_sha is mandatory and must be a non-empty SHA "
+            "(fail closed, #709 F1)"
+        )
+    if pr_lookup_error:
+        reasons.append(f"live PR head lookup failed: {pr_lookup_error} (fail closed)")
+    if want and not have:
+        reasons.append("live PR head SHA unavailable (fail closed)")
+    if want and have and not heads_equal(want, have):
+        reasons.append(
+            "expected_head_sha does not equal live PR head "
+            f"(expected={want[:12]}… live={have[:12]}…; fail closed, #709 F1)"
+        )
+    return {
+        "valid": not reasons,
+        "expected_head_sha": want,
+        "live_head_sha": have,
+        "pr_state": pr_state,
+        "reasons": reasons,
+    }
+
+
+def build_authorization_artifact(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+    incident_issue: int,
+    incident_comment_id: int,
+    destroyed_subject: str | None,
+    issuer_username: str,
+    issuer_profile: str,
+    native_provenance: dict[str, Any] | None = None,
+    ttl_hours: float = AUTH_TTL_HOURS,
+) -> dict[str, Any]:
+    """Build a server-side authorization artifact (caller cannot forge signature)."""
+    provenance = dict(
+        native_provenance or mcp_daemon_guard.mutation_provenance_fields()
+    )
+    created = _now()
+    expires = created + timedelta(hours=float(ttl_hours))
+    authorization_id = str(uuid.uuid4())
+    scope = _scope_payload(
+        remote=remote,
+        org=org,
+        repo=repo,
+        pr_number=pr_number,
+        expected_head_sha=expected_head_sha,
+        incident_issue=incident_issue,
+        incident_comment_id=incident_comment_id,
+        destroyed_subject=destroyed_subject,
+        issuer_username=issuer_username,
+        issuer_profile=issuer_profile,
+        created_at=created.isoformat(),
+        expires_at=expires.isoformat(),
+        authorization_id=authorization_id,
+    )
+    signature = _sign_scope(scope, provenance)
+    return {
+        "kind": KIND_AUTH,
+        "auth_type": AUTH_TYPE,
+        "record_type": AUTH_TYPE,
+        "status": "issued",
+        "consumption_state": "issued",
+        "consumed_at": None,
+        "recovery_critical": True,
+        "issue_ref": "#709",
+        "server_signature": signature,
+        "native_provenance": provenance,
+        **scope,
+        "timestamp": created.isoformat(),
+        "recorded_at": created.isoformat(),
+        "updated_at": created.isoformat(),
+    }
+
+
+def verify_authorization_artifact(
+    auth: dict[str, Any] | None,
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+    incident_issue: int | None = None,
+    incident_comment_id: int | None = None,
+    require_unconsumed: bool = True,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Fail-closed verification of a server-side authorization artifact."""
+    reasons: list[str] = []
+    if not isinstance(auth, dict):
+        return {
+            "valid": False,
+            "reasons": ["authorization artifact missing (fail closed)"],
+        }
+    if (auth.get("kind") or auth.get("auth_type") or "") not in (
+        KIND_AUTH,
+        AUTH_TYPE,
+    ) and auth.get("record_type") != AUTH_TYPE:
+        if (auth.get("kind") or "") != KIND_AUTH:
+            reasons.append(
+                f"authorization kind mismatch (expected {KIND_AUTH!r}; fail closed)"
+            )
+
+    for field, want in (
+        ("remote", remote),
+        ("org", org),
+        ("repo", repo),
+    ):
+        have = (str(auth.get(field) or "")).strip()
+        if not have or have != (want or "").strip():
+            reasons.append(
+                f"authorization {field} mismatch "
+                f"(stored={have!r}, expected={want!r}; fail closed)"
+            )
+
+    try:
+        if int(auth.get("blocked_pr_number")) != int(pr_number):
+            reasons.append("authorization PR number mismatch (fail closed)")
+    except (TypeError, ValueError):
+        reasons.append("authorization missing blocked_pr_number (fail closed)")
+
+    if not heads_equal(auth.get("expected_head_sha"), expected_head_sha):
+        reasons.append("authorization head SHA mismatch (fail closed)")
+
+    if incident_issue is not None:
+        try:
+            if int(auth.get("incident_issue")) != int(incident_issue):
+                reasons.append("authorization incident_issue mismatch (fail closed)")
+        except (TypeError, ValueError):
+            reasons.append("authorization missing incident_issue (fail closed)")
+    if incident_comment_id is not None:
+        try:
+            if int(auth.get("incident_comment_id")) != int(incident_comment_id):
+                reasons.append(
+                    "authorization incident_comment_id mismatch (fail closed)"
+                )
+        except (TypeError, ValueError):
+            reasons.append("authorization missing incident_comment_id (fail closed)")
+
+    if not (auth.get("issuer_username") or "").strip():
+        reasons.append("authorization missing issuer_username (fail closed)")
+    if not (auth.get("issuer_profile") or "").strip():
+        reasons.append("authorization missing issuer_profile (fail closed)")
+    if not (auth.get("server_signature") or "").strip():
+        reasons.append("authorization missing server_signature (fail closed)")
+
+    # Recompute signature over stored scope fields.
+    try:
+        scope = _scope_payload(
+            remote=str(auth.get("remote") or ""),
+            org=str(auth.get("org") or ""),
+            repo=str(auth.get("repo") or ""),
+            pr_number=int(auth.get("blocked_pr_number")),
+            expected_head_sha=str(auth.get("expected_head_sha") or ""),
+            incident_issue=int(auth.get("incident_issue")),
+            incident_comment_id=int(auth.get("incident_comment_id")),
+            destroyed_subject=auth.get("destroyed_subject"),
+            issuer_username=str(auth.get("issuer_username") or ""),
+            issuer_profile=str(auth.get("issuer_profile") or ""),
+            created_at=str(auth.get("created_at") or ""),
+            expires_at=str(auth.get("expires_at") or ""),
+            authorization_id=str(auth.get("authorization_id") or ""),
+        )
+        native = auth.get("native_provenance") or {}
+        if not isinstance(native, dict):
+            native = {}
+        expected_sig = _sign_scope(scope, native)
+        if not hmac.compare_digest(
+            expected_sig, str(auth.get("server_signature") or "")
+        ):
+            reasons.append(
+                "authorization server_signature invalid (forged or corrupt; "
+                "fail closed, #709 F1)"
+            )
+    except (TypeError, ValueError) as exc:
+        reasons.append(f"authorization scope incomplete: {exc} (fail closed)")
+
+    # Native provenance required on the artifact itself.
+    native = auth.get("native_provenance") or {}
+    if not isinstance(native, dict) or not (
+        native.get("native_mcp_transport") or native.get("pytest")
+    ):
+        # Pytest artifacts stamp pytest=True via mutation_provenance_fields.
+        if not mcp_daemon_guard.is_pytest_runtime():
+            if not (isinstance(native, dict) and native.get("native_mcp_transport")):
+                reasons.append(
+                    "authorization lacks native transport provenance (fail closed)"
+                )
+
+    # Expiry / consumption.
+    now_dt = now or _now()
+    expires_raw = auth.get("expires_at")
+    expires_dt = None
+    if expires_raw:
+        text = str(expires_raw).strip()
+        if text.endswith("Z"):
+            text = text[:-1] + "+00:00"
+        try:
+            expires_dt = datetime.fromisoformat(text)
+            if expires_dt.tzinfo is None:
+                expires_dt = expires_dt.replace(tzinfo=timezone.utc)
+        except ValueError:
+            reasons.append("authorization expires_at unparseable (fail closed)")
+    else:
+        reasons.append("authorization missing expires_at (fail closed)")
+    if expires_dt is not None and now_dt > expires_dt:
+        reasons.append("authorization expired (fail closed)")
+
+    state = (auth.get("consumption_state") or auth.get("status") or "").strip()
+    if require_unconsumed and state in ("consumed", "expired"):
+        reasons.append(
+            f"authorization already {state}; cannot be replayed (fail closed)"
+        )
+    if require_unconsumed and auth.get("consumed_at"):
+        reasons.append("authorization already consumed (fail closed)")
+
+    return {
+        "valid": not reasons,
+        "reasons": reasons,
+        "authorization_id": auth.get("authorization_id"),
+        "consumption_state": state or None,
+    }
+
+
+def build_irrecoverable_provenance_record(
+    *,
+    pr_number: int,
+    head_sha: str,
+    remote: str,
+    org: str,
+    repo: str,
+    actor_username: str | None,
+    profile_name: str | None,
+    reason: str,
+    incident_issue: int,
+    incident_comment_id: int,
+    authorization: dict[str, Any],
+    destroyed_subject: str | None = None,
+    historical_provenance_subject: str | None = None,
+) -> dict[str, Any]:
+    """Truthful absence-of-proof record. Never sets applied=True.
+
+    ``merger_may_accept`` is True only when *authorization* verifies for the
+    exact scope. Caller-supplied Booleans are never consulted.
+    """
+    head = normalize_head_sha(head_sha)
+    auth_check = verify_authorization_artifact(
+        authorization,
+        remote=remote,
+        org=org,
+        repo=repo,
+        pr_number=pr_number,
+        expected_head_sha=head or "",
+        incident_issue=incident_issue,
+        incident_comment_id=incident_comment_id,
+        require_unconsumed=True,
+    )
+    may_accept = bool(auth_check.get("valid"))
+    return {
+        "event": "irrecoverable_decision_lock_provenance",
+        "status": "provenance_irrecoverable",
+        "record_type": RECORD_TYPE,
+        "kind": KIND_RECOVERY,
+        "operator_recovery_required": True,
+        "issue_ref": "#709",
+        "recovery_critical": True,
+        "applied": False,
+        "historical_cleanup_proven": False,
+        "timestamp": _now_iso(),
+        "pr_number": int(pr_number),
+        "blocked_pr_number": int(pr_number),
+        "head_sha": head,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "reason": reason,
+        "incident_issue": int(incident_issue),
+        "incident_comment_id": int(incident_comment_id),
+        # Legacy field for audit readability; not a caller Boolean gate.
+        "incident_ref": f"issue:{int(incident_issue)}/comment:{int(incident_comment_id)}",
+        "authorization_id": authorization.get("authorization_id"),
+        "authorization_issuer": authorization.get("issuer_username"),
+        "authorization_issuer_profile": authorization.get("issuer_profile"),
+        "authorization_verified": may_accept,
+        "authorization_verify_reasons": list(auth_check.get("reasons") or []),
+        "destroyed_subject": (destroyed_subject or "").strip() or None,
+        "historical_provenance_subject": (
+            (historical_provenance_subject or destroyed_subject or "").strip()
+            or None
+        ),
+        "consumption_state": "issued",
+        "consumed_at": None,
+        "merger_may_accept": may_accept,
+        "acceptance_rule": (
+            "Merger may accept this record only when a server-side authorization "
+            "artifact verifies for remote/org/repo/PR/exact-head/incident, the "
+            "record is durable and read back, the auth is unexpired and unconsumed, "
+            "and normal merge gates still pass. Resolves only the historical "
+            "prior-provenance blocker; never proves historical cleanup "
+            "(applied=false, historical_cleanup_proven=false)."
+        ),
+        "native_provenance": mcp_daemon_guard.mutation_provenance_fields(),
+    }
+
+
+def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str:
+    """Markdown body for irrecoverable provenance audit (no applied=true claim)."""
+    lines = [
+        "## Irrecoverable decision-lock provenance (#709)",
+        "",
+        "Status: **PROVENANCE_IRRECOVERABLE** (not applied cleanup)",
+        "",
+        f"- actor: `{record.get('actor_username')}`",
+        f"- profile: `{record.get('profile_name')}`",
+        f"- timestamp: `{record.get('timestamp')}`",
+        f"- PR: `#{record.get('pr_number')}`",
+        f"- head_sha: `{record.get('head_sha')}`",
+        f"- incident_issue: `{record.get('incident_issue')}`",
+        f"- incident_comment_id: `{record.get('incident_comment_id')}`",
+        f"- authorization_id: `{record.get('authorization_id')}`",
+        f"- authorization_issuer: `{record.get('authorization_issuer')}`",
+        f"- authorization_verified: `{record.get('authorization_verified')}`",
+        f"- historical_cleanup_proven: `{record.get('historical_cleanup_proven')}`",
+        f"- applied: `{record.get('applied')}` (must remain false)",
+        f"- merger_may_accept: `{record.get('merger_may_accept')}`",
+        f"- destroyed_subject: `{record.get('destroyed_subject')}`",
+        "",
+        f"Reason: {record.get('reason')}",
+        "",
+        "This record documents **absence of proof**, not successful cleanup.",
+        "It must not be reused for a different PR or head (#709 AC6).",
+        "Authorization is a server-side artifact — not a caller Boolean.",
+    ]
+    return "\n".join(lines)
+
+
+def assess_merger_consumption(
+    recovery: dict[str, Any] | None,
+    authorization: dict[str, Any] | None,
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    live_head_sha: str | None,
+    # Normal merge gate outcomes (must still pass independently).
+    approval_at_current_head: bool | None = None,
+    has_blocking_change_requests: bool | None = None,
+    mergeable: bool | None = None,
+    lease_ok: bool | None = None,
+    runtime_ok: bool | None = None,
+    workspace_ok: bool | None = None,
+    anti_stomp_ok: bool | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Fail-closed merger assessment for consuming an irrecoverable recovery.
+
+    Resolves **only** the historical prior-provenance blocker when all
+    recovery checks pass. Never grants a pass when normal merge gates fail.
+    """
+    reasons: list[str] = []
+    result: dict[str, Any] = {
+        "allowed": False,
+        "resolves_prior_provenance_blocker": False,
+        "historical_cleanup_proven": False,
+        "irrecoverable_recovery_authorized": False,
+        "recovery_record_consumed": False,
+        "reasons": reasons,
+        "normal_gates": {
+            "approval_at_current_head": approval_at_current_head,
+            "has_blocking_change_requests": has_blocking_change_requests,
+            "mergeable": mergeable,
+            "lease_ok": lease_ok,
+            "runtime_ok": runtime_ok,
+            "workspace_ok": workspace_ok,
+            "anti_stomp_ok": anti_stomp_ok,
+        },
+    }
+
+    if not isinstance(recovery, dict):
+        reasons.append("recovery record missing (fail closed)")
+        return result
+    if (recovery.get("record_type") or recovery.get("kind")) not in (
+        RECORD_TYPE,
+        KIND_RECOVERY,
+        "irrecoverable_decision_provenance",
+    ):
+        if recovery.get("status") != "provenance_irrecoverable":
+            reasons.append("recovery record type/status invalid (fail closed)")
+
+    if recovery.get("applied") is True:
+        reasons.append(
+            "recovery record claims applied=true; refuse (fabrication, fail closed)"
+        )
+    if recovery.get("historical_cleanup_proven") is True:
+        reasons.append(
+            "recovery record claims historical_cleanup_proven=true; refuse "
+            "(fail closed)"
+        )
+
+    for field, want in (("remote", remote), ("org", org), ("repo", repo)):
+        have = (str(recovery.get(field) or "")).strip()
+        if have != (want or "").strip():
+            reasons.append(
+                f"recovery {field} mismatch (stored={have!r}, expected={want!r})"
+            )
+
+    try:
+        if int(recovery.get("pr_number") or recovery.get("blocked_pr_number")) != int(
+            pr_number
+        ):
+            reasons.append("recovery PR number mismatch (fail closed)")
+    except (TypeError, ValueError):
+        reasons.append("recovery missing pr_number (fail closed)")
+
+    if not heads_equal(recovery.get("head_sha"), live_head_sha):
+        reasons.append(
+            "recovery head SHA does not match live PR head (fail closed)"
+        )
+
+    if recovery.get("consumption_state") == "consumed" or recovery.get("consumed_at"):
+        # Idempotent: already consumed for this exact scope is OK if head matches.
+        result["recovery_record_consumed"] = True
+        reasons.append("recovery record already consumed (idempotent check)")
+
+    auth_check = verify_authorization_artifact(
+        authorization,
+        remote=remote,
+        org=org,
+        repo=repo,
+        pr_number=pr_number,
+        expected_head_sha=str(live_head_sha or ""),
+        incident_issue=recovery.get("incident_issue"),
+        incident_comment_id=recovery.get("incident_comment_id"),
+        # When recovery already consumed, allow already-consumed auth for
+        # idempotent re-report; otherwise require unconsumed.
+        require_unconsumed=not result["recovery_record_consumed"],
+        now=now,
+    )
+    if not auth_check.get("valid"):
+        reasons.extend(auth_check.get("reasons") or ["authorization invalid"])
+    else:
+        result["irrecoverable_recovery_authorized"] = True
+
+    if not recovery.get("merger_may_accept") and not result["recovery_record_consumed"]:
+        reasons.append(
+            "recovery record merger_may_accept is false (fail closed)"
+        )
+
+    # Auth id binding.
+    if (
+        authorization
+        and recovery.get("authorization_id")
+        and authorization.get("authorization_id")
+        and recovery.get("authorization_id") != authorization.get("authorization_id")
+    ):
+        reasons.append("recovery authorization_id does not match artifact (fail closed)")
+
+    # Normal gates: if explicitly False, refuse consumption as merge-authorizing.
+    normal_blockers: list[str] = []
+    if approval_at_current_head is False:
+        normal_blockers.append("missing/stale approval at current head")
+    if has_blocking_change_requests is True:
+        normal_blockers.append("blocking change requests present")
+    if mergeable is False:
+        normal_blockers.append("PR not mergeable")
+    if lease_ok is False:
+        normal_blockers.append("lease gate failed")
+    if runtime_ok is False:
+        normal_blockers.append("runtime gate failed")
+    if workspace_ok is False:
+        normal_blockers.append("workspace gate failed")
+    if anti_stomp_ok is False:
+        normal_blockers.append("anti-stomp gate failed")
+    if normal_blockers:
+        reasons.append(
+            "recovery cannot bypass normal merge gates: "
+            + "; ".join(normal_blockers)
+            + " (#709 F2)"
+        )
+        result["resolves_prior_provenance_blocker"] = False
+        result["allowed"] = False
+        result["reasons"] = reasons
+        return result
+
+    # Filter pure informational "already consumed" when everything else matches
+    # for idempotent success.
+    hard = [
+        r
+        for r in reasons
+        if "already consumed" not in r
+    ]
+    if not hard and result["irrecoverable_recovery_authorized"]:
+        result["allowed"] = True
+        result["resolves_prior_provenance_blocker"] = True
+        result["historical_cleanup_proven"] = False
+        if result["recovery_record_consumed"]:
+            reasons.append(
+                "idempotent: prior-provenance blocker already resolved for this scope"
+            )
+        else:
+            reasons.append(
+                "prior-provenance blocker may be resolved by consuming this record "
+                "(historical cleanup remains unproven)"
+            )
+    result["reasons"] = reasons
+    return result
+
+
+def mark_consumed(
+    record: dict[str, Any],
+    *,
+    consumer_username: str | None,
+    consumer_profile: str | None,
+) -> dict[str, Any]:
+    """Return a copy of *record* marked consumed (crash-safe write is caller's job)."""
+    out = dict(record)
+    out["consumption_state"] = "consumed"
+    out["status"] = out.get("status") or "issued"
+    if out.get("kind") == KIND_AUTH or out.get("auth_type") == AUTH_TYPE:
+        out["status"] = "consumed"
+    out["consumed_at"] = _now_iso()
+    out["consumed_by"] = consumer_username
+    out["consumed_by_profile"] = consumer_profile
+    out["updated_at"] = out["consumed_at"]
+    return out
+
+
+def assess_profile_path_identity(profile_identity: str | None) -> dict[str, Any]:
+    """Reject traversal / malformed profile identity segments (#709 F3)."""
+    reasons: list[str] = []
+    raw = profile_identity if profile_identity is not None else ""
+    text = str(raw)
+    if not text.strip():
+        reasons.append("profile identity empty (fail closed)")
+        return {"valid": False, "reasons": reasons, "sanitized": None}
+    if text != text.strip():
+        reasons.append("profile identity has surrounding whitespace (fail closed)")
+    if ".." in text or "/" in text or "\\" in text or "\x00" in text:
+        reasons.append(
+            "profile identity contains path traversal or separator characters "
+            "(fail closed, #709 F3)"
+        )
+    if text.startswith("-") or text.startswith("."):
+        reasons.append("profile identity has unsafe leading character (fail closed)")
+    # After sanitize, must not collapse to something that collides emptily.
+    sanitized = mcp_session_state._sanitize_segment(text)
+    if sanitized in ("_", ""):
+        reasons.append("profile identity sanitizes to empty (fail closed)")
+    if sanitized != text and any(c in text for c in ("..", "/", "\\")):
+        # Already covered; keep fail closed.
+        pass
+    return {
+        "valid": not reasons,
+        "reasons": reasons,
+        "sanitized": sanitized if not reasons else None,
+    }

mcp_session_state.py

@@ -42,6 +42,8 @@ KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
 KIND_POST_MERGE_DECISION_RECOVERY = "post_merge_decision_recovery"
 # #709: truthful record when historical terminal evidence is irrecoverably gone.
 KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
+# #709 F1: server-side non-forgeable authorization artifact for recovery.
+KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 
 # Kinds that must survive the default session-state TTL (forensic / recovery).
 RECOVERY_CRITICAL_KINDS = frozenset(
@@ -49,6 +51,7 @@ RECOVERY_CRITICAL_KINDS = frozenset(
         KIND_DECISION_LOCK_ARCHIVE,
         KIND_POST_MERGE_DECISION_RECOVERY,
         KIND_IRRECOVERABLE_DECISION_PROVENANCE,
+        KIND_IRRECOVERABLE_PROVENANCE_AUTH,
     }
 )
 
@@ -504,6 +507,7 @@ def load_state_for_profile(
     repo: str | None = None,
     state_dir: str | None = None,
     skip_identity_match: bool = False,
+    enforce_repo_scope: bool = True,
 ) -> dict[str, Any] | None:
     """Load durable state for an explicit profile identity (#709 cross-profile).
 
@@ -511,9 +515,46 @@ def load_state_for_profile(
     profile_identity to equal the requested profile (anti-stomp), but does not
     require the *active* session identity to match — needed so a merger can
     inspect a reviewer lock after merge.
+
+    #709 F3: remote/org/repo filter reasons are **enforced** (not merely
+    computed). When *enforce_repo_scope* is True (default) and the caller
+    supplies remote/org/repo, mismatches fail closed with ``None``.
     """
+    # #709 F3: refuse traversal / malformed profile identities.
+    try:
+        from irrecoverable_provenance import assess_profile_path_identity
+
+        path_gate = assess_profile_path_identity(profile_identity)
+        if not path_gate.get("valid"):
+            return None
+    except Exception:
+        # Module may be mid-import in edge bootstraps; fall through to
+        # conservative checks below.
+        raw = str(profile_identity or "")
+        if ".." in raw or "/" in raw or "\\" in raw or "\x00" in raw:
+            return None
+
     profile = current_profile_identity(profile_identity=profile_identity)
     root = _ensure_state_dir(state_dir)
+    # Refuse symlink escape of the state root (#709 F3).
+    try:
+        real_root = os.path.realpath(root)
+        path = state_file_path(
+            kind=kind,
+            remote=remote,
+            org=org,
+            repo=repo,
+            profile_identity=profile,
+            state_dir=root,
+        )
+        real_path = os.path.realpath(path) if os.path.exists(path) else path
+        if os.path.exists(path) and not str(real_path).startswith(
+            str(real_root) + os.sep
+        ) and str(real_path) != str(real_root):
+            return None
+    except OSError:
+        return None
+
     path = state_file_path(
         kind=kind,
         remote=remote,
@@ -548,34 +589,46 @@ def load_state_for_profile(
     if stored and stored != profile:
         return None
     if skip_identity_match:
-        # Still enforce TTL / future-dated so dead records do not authorize cleanup.
+        # Enforce remote/org/repo when caller provides them (#709 F3).
+        # Drop only *active session* profile-identity mismatches; keep
+        # expiry, spoof, and repository-scope reasons.
+        scope_remote = remote if enforce_repo_scope else None
+        scope_org = org if enforce_repo_scope else None
+        scope_repo = repo if enforce_repo_scope else None
         reasons = identity_match_reasons(
             merged,
-            remote=remote or merged.get("remote"),
+            remote=scope_remote,
-            org=org or merged.get("org"),
+            org=scope_org,
-            repo=repo or merged.get("repo"),
+            repo=scope_repo,
             profile_identity=stored or profile,
         )
-        # Drop active-session-only mismatches; keep expiry / spoof reasons.
         filtered = [
             r
             for r in reasons
             if "profile identity mismatch" not in r
             or (stored and stored != profile)
         ]
-        # Re-run only expiry/future/missing checks via identity when profile matches
+        # When caller requested a specific remote/org/repo, also fail if the
-        expiry_reasons = [
+        # stored record lacks those identity fields entirely (legacy incomplete).
-            r
+        if enforce_repo_scope:
-            for r in identity_match_reasons(
+            for field, want in (
-                merged,
+                ("remote", remote),
-                remote=None,
+                ("org", org),
-                org=None,
+                ("repo", repo),
-                repo=None,
+            ):
-                profile_identity=stored or profile,
+                want_s = (want or "").strip()
-            )
+                if not want_s:
-            if any(x in r for x in ("expired", "future", "missing recorded_at"))
+                    continue
-        ]
+                have = (str(merged.get(field) or "")).strip()
-        if expiry_reasons:
+                if not have:
+                    filtered.append(
+                        f"session state {field} missing on durable record "
+                        f"(expected={want_s!r}; fail closed, #709 F3)"
+                    )
+                elif have != want_s:
+                    # identity_match_reasons already adds mismatch; ensure kept
+                    pass
+        if filtered:
             return None
         return merged
     reasons = identity_match_reasons(

stale_review_decision_lock.py

@@ -570,13 +570,52 @@ def build_irrecoverable_provenance_record(
     actor_username: str | None,
     profile_name: str | None,
     reason: str,
-    incident_ref: str | None,
+    incident_ref: str | None = None,
-    operator_authorized: bool,
+    # Deprecated kwargs retained only so stale call sites fail closed:
+    operator_authorized: bool | None = None,
+    # Required for merger-acceptable records (#709 F1):
+    authorization: dict[str, Any] | None = None,
+    incident_issue: int | None = None,
+    incident_comment_id: int | None = None,
+    destroyed_subject: str | None = None,
+    historical_provenance_subject: str | None = None,
 ) -> dict[str, Any]:
-    """Truthful absence-of-proof record (#709 AC5). Never sets applied=True."""
+    """Truthful absence-of-proof record (#709 AC5). Never sets applied=True.
+
+    Caller-supplied ``operator_authorized`` is **ignored** as authorization
+    evidence (review 434 F1). Prefer
+    :func:`irrecoverable_provenance.build_irrecoverable_provenance_record`
+    with a verified server-side authorization artifact.
+    """
+    # Explicitly ignore deprecated self-assertable Boolean.
+    _ = operator_authorized
+    if authorization is not None and incident_issue is not None and incident_comment_id is not None:
+        from irrecoverable_provenance import (
+            build_irrecoverable_provenance_record as _build,
+        )
+
+        return _build(
+            pr_number=int(pr_number),
+            head_sha=str(head_sha or ""),
+            remote=str(remote or ""),
+            org=str(org or ""),
+            repo=str(repo or ""),
+            actor_username=actor_username,
+            profile_name=profile_name,
+            reason=reason,
+            incident_issue=int(incident_issue),
+            incident_comment_id=int(incident_comment_id),
+            authorization=authorization,
+            destroyed_subject=destroyed_subject,
+            historical_provenance_subject=historical_provenance_subject
+            or destroyed_subject,
+        )
+    # Fail-closed skeleton when no server authorization is supplied: never
+    # sets merger_may_accept True (even if operator_authorized was True).
     return {
         "event": "irrecoverable_decision_lock_provenance",
         "status": "provenance_irrecoverable",
+        "record_type": "irrecoverable_decision_provenance",
         "operator_recovery_required": True,
         "issue_ref": "#709",
         "recovery_critical": True,
@@ -592,40 +631,27 @@ def build_irrecoverable_provenance_record(
         "profile_name": profile_name,
         "reason": reason,
         "incident_ref": incident_ref,
-        "operator_authorized": bool(operator_authorized),
+        "incident_issue": incident_issue,
-        "merger_may_accept": bool(operator_authorized),
+        "incident_comment_id": incident_comment_id,
+        "authorization_verified": False,
+        "merger_may_accept": False,
         "acceptance_rule": (
-            "Merger may accept this record only when operator_authorized=true, "
+            "Merger may accept this record only when a server-side "
-            "repository/PR/head match the live target, the record is durable "
+            "authorization artifact verifies for remote/org/repo/PR/exact "
-            "and read back, and no conflicting terminal lock remains for a "
+            "head/incident, the record is durable and read back, and normal "
-            "different PR/head. This does not prove historical cleanup."
+            "merge gates still pass. Caller Booleans never authorize. This "
+            "does not prove historical cleanup."
         ),
     }
 
 
 def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str:
     """Markdown body for irrecoverable provenance audit (no applied=true claim)."""
-    lines = [
+    from irrecoverable_provenance import (
-        "## Irrecoverable decision-lock provenance (#709)",
+        format_irrecoverable_audit_comment as _fmt,
-        "",
+    )
-        "Status: **PROVENANCE_IRRECOVERABLE** (not applied cleanup)",
+
-        "",
+    return _fmt(record)
-        f"- actor: `{record.get('actor_username')}`",
-        f"- profile: `{record.get('profile_name')}`",
-        f"- timestamp: `{record.get('timestamp')}`",
-        f"- PR: `#{record.get('pr_number')}`",
-        f"- head_sha: `{record.get('head_sha')}`",
-        f"- incident_ref: `{record.get('incident_ref')}`",
-        f"- operator_authorized: `{record.get('operator_authorized')}`",
-        f"- historical_cleanup_proven: `{record.get('historical_cleanup_proven')}`",
-        f"- applied: `{record.get('applied')}` (must remain false)",
-        "",
-        f"Reason: {record.get('reason')}",
-        "",
-        "This record documents **absence of proof**, not successful cleanup.",
-        "It must not be reused for a different PR or head (#709 AC6).",
-    ]
-    return "\n".join(lines)
 
 
 def format_post_merge_recovery_comment(record: dict[str, Any]) -> str:

task_capability_map.py

@@ -127,15 +127,32 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.review",
         "role": "reviewer",
     },
-    # #709: truthful absence-of-proof recovery record (not applied cleanup).
+    # #709: truthful absence-of-proof recovery (server-side auth + record + consume).
+    # Dedicated mutation capability — gitea.read is insufficient (review 434 F1).
+    "issue_irrecoverable_provenance_authorization": {
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
+        "role": "reconciler",
+    },
+    "gitea_issue_irrecoverable_provenance_authorization": {
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
+        "role": "reconciler",
+    },
     "record_irrecoverable_decision_lock_provenance": {
-        "permission": "gitea.issue.comment",
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
         "role": "reconciler",
     },
     "gitea_record_irrecoverable_decision_lock_provenance": {
-        "permission": "gitea.issue.comment",
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
         "role": "reconciler",
     },
+    "consume_irrecoverable_decision_lock_provenance": {
+        "permission": "gitea.pr.merge",
+        "role": "merger",
+    },
+    "gitea_consume_irrecoverable_decision_lock_provenance": {
+        "permission": "gitea.pr.merge",
+        "role": "merger",
+    },
     "delete_branch": {
         "permission": "gitea.branch.delete",
         "role": "author",