fix(workflow): durable HMAC, dedicated mint capability, head-exact approve match (#709)

Address formal review 435 REQUEST_CHANGES on PR #710:
- F4: require durable GITEA_IRRECOVERABLE_AUTH_HMAC_KEY (fail closed; no ephemeral
  per-process secret); bind key_version into HMAC; cross-process verify works
- F5: dedicated gitea.decision_lock.irrecoverable_recovery only; reject reconciler
  equivalence; authoritative incident body + author + content_digest; reject
  self-authored incident evidence
- F3 residual: lock_targets_merged_pr_approval requires recorded-head match when
  expected_head_sha is provided (legacy no-head approve no longer primary-clears)

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

.env.example

@@ -55,3 +55,12 @@ GITEA_MCP_PROFILE=prgs
 # GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
 # GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
 # GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override
+
+# Durable HMAC key for irrecoverable decision-lock provenance artifacts (#709 F4).
+# REQUIRED in production native MCP processes that mint or verify recovery
+# authorization (reconciler + merger must share the same key). Hex (preferred),
+# base64, or utf-8 literal (>=16 bytes). Never generate an ephemeral per-process
+# key — cross-process / post-restart verification would fail closed.
+# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
+# Optional key version string bound into the signature (for rotation).
+# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION=v1

gitea_mcp_server.py

@@ -5140,12 +5140,14 @@ def gitea_issue_irrecoverable_provenance_authorization(
     org: str | None = None,
     repo: str | None = None,
 ) -> dict:
-    """Mint a server-side authorization artifact for irrecoverable recovery (#709 F1).
+    """Mint a server-side authorization artifact for irrecoverable recovery (#709 F1/F5).
 
-    Non-forgeable: requires production native MCP transport (or pytest), a
+    Non-forgeable: requires production native MCP transport (or pytest), the
-    dedicated/reconciler mutation capability, live head equality, and validated
+    dedicated ``gitea.decision_lock.irrecoverable_recovery`` capability (no
-    incident evidence. Confirmation is human intent only — never authorization.
+    reconciler equivalence), live head equality, durable HMAC key, and
-    Caller Booleans are not accepted.
+    authoritative incident evidence (author + canonical content_digest).
+    Confirmation is human intent only — never authorization. Caller Booleans
+    are not accepted.
     """
     import irrecoverable_provenance as irp
 
@@ -5224,7 +5226,7 @@ def gitea_issue_irrecoverable_provenance_authorization(
         report["reasons"].extend(head_gate.get("reasons") or [])
         return report
 
-    # Incident evidence live validation.
+    # Incident evidence live validation (author + canonical digest, #709 F5).
     comment_payload = None
     comment_err = None
     try:
@@ -5243,6 +5245,10 @@ def gitea_issue_irrecoverable_provenance_authorization(
         expected_remote=remote,
         expected_org=o,
         expected_repo=r,
+        expected_pr_number=pr_number,
+        expected_head_sha=str(expected_head_sha),
+        mint_actor_username=actor,
+        reject_self_authored=True,
     )
     if not incident_gate.get("valid"):
         report["reasons"].extend(incident_gate.get("reasons") or [])
@@ -5285,6 +5291,7 @@ def gitea_issue_irrecoverable_provenance_authorization(
             )
             return report
 
+    try:
         artifact = irp.build_authorization_artifact(
             remote=remote,
             org=o,
@@ -5298,6 +5305,9 @@ def gitea_issue_irrecoverable_provenance_authorization(
             issuer_profile=profile_name or "unknown",
             native_provenance=mcp_daemon_guard.mutation_provenance_fields(),
         )
+    except irp.AuthSecretError as exc:
+        report["reasons"].append(str(exc))
+        return report
     artifact["kind"] = mcp_session_state.KIND_IRRECOVERABLE_PROVENANCE_AUTH
     saved = mcp_session_state.save_state(
         kind=mcp_session_state.KIND_IRRECOVERABLE_PROVENANCE_AUTH,
@@ -5313,7 +5323,8 @@ def gitea_issue_irrecoverable_provenance_authorization(
     report["success"] = True
     report["reasons"].append(
         "issued server-side irrecoverable provenance authorization "
-        "(non-forgeable; bound to remote/org/repo/PR/head/incident)"
+        "(non-forgeable; bound to remote/org/repo/PR/head/incident; "
+        f"key_version={artifact.get('key_version')})"
     )
     return report
 
@@ -5471,8 +5482,13 @@ def gitea_record_irrecoverable_decision_lock_provenance(
         incident_comment_id=incident_comment_id,
         comment_payload=comment_payload if isinstance(comment_payload, dict) else None,
         comment_lookup_error=comment_err,
+        expected_remote=remote,
         expected_org=o,
         expected_repo=r,
+        expected_pr_number=pr_number,
+        expected_head_sha=str(expected_head_sha),
+        mint_actor_username=actor,
+        reject_self_authored=True,
     )
     if not incident_gate.get("valid"):
         report["reasons"].extend(incident_gate.get("reasons") or [])

irrecoverable_provenance.py

@@ -14,7 +14,6 @@ import hashlib
 import hmac
 import json
 import os
-import secrets
 import uuid
 from datetime import datetime, timedelta, timezone
 from typing import Any
@@ -30,14 +29,28 @@ KIND_AUTH = "irrecoverable_provenance_authorization"
 KIND_RECOVERY = mcp_session_state.KIND_IRRECOVERABLE_DECISION_PROVENANCE
 
 CONFIRMATION_PREFIX = "IRRECOVERABLE DECISION PROVENANCE PR"
+# Canonical incident-comment marker (#709 review 435 F5).
+INCIDENT_MARKER = "IRRECOVERABLE DECISION PROVENANCE INCIDENT"
 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
+# Durable HMAC key origin (#709 review 435 F4). Never generate an ephemeral
-# gets a deterministic salt so hermetic tests are stable; production uses
+# production key — mint/verify across reconciler/merger processes and restarts
-# transport fingerprint + random secret minted at process start.
+# requires a shared secret from env (or pytest constant / explicit env override).
+ENV_AUTH_HMAC_KEY = "GITEA_IRRECOVERABLE_AUTH_HMAC_KEY"
+ENV_AUTH_HMAC_KEY_VERSION = "GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION"
+DEFAULT_KEY_VERSION = "v1"
+_PYTEST_AUTH_SECRET = b"pytest-irrecoverable-auth-v1"
+_PYTEST_KEY_VERSION = "pytest-v1"
+
 _PROCESS_AUTH_SECRET: bytes | None = None
+_PROCESS_AUTH_KEY_VERSION: str | None = None
+_PROCESS_AUTH_SECRET_SOURCE: str | None = None
+
+
+class AuthSecretError(RuntimeError):
+    """Raised when the durable HMAC signing key cannot be resolved (fail closed)."""
 
 
 def _now() -> datetime:
@@ -48,15 +61,92 @@ def _now_iso() -> str:
     return _now().isoformat()
 
 
+def _decode_key_material(raw: str) -> bytes:
+    """Decode operator-supplied key material (hex, base64, or utf-8 literal)."""
+    text = (raw or "").strip()
+    if not text:
+        raise AuthSecretError("empty HMAC key material (fail closed, #709 F4)")
+    # Prefer hex (64 chars = 32 bytes).
+    if len(text) >= 32 and all(c in "0123456789abcdefABCDEF" for c in text):
+        try:
+            if len(text) % 2 == 0:
+                decoded = bytes.fromhex(text)
+                if len(decoded) >= 16:
+                    return decoded
+        except ValueError:
+            pass
+    # base64
+    try:
+        import base64
+
+        decoded = base64.b64decode(text, validate=True)
+        if len(decoded) >= 16:
+            return decoded
+    except Exception:
+        pass
+    # utf-8 literal (min 16 chars after strip)
+    encoded = text.encode("utf-8")
+    if len(encoded) < 16:
+        raise AuthSecretError(
+            "HMAC key material too short (need >=16 bytes; fail closed, #709 F4)"
+        )
+    return encoded
+
+
+def reset_process_auth_secret_for_tests() -> None:
+    """Clear cached HMAC key (tests only)."""
+    global _PROCESS_AUTH_SECRET, _PROCESS_AUTH_KEY_VERSION, _PROCESS_AUTH_SECRET_SOURCE
+    _PROCESS_AUTH_SECRET = None
+    _PROCESS_AUTH_KEY_VERSION = None
+    _PROCESS_AUTH_SECRET_SOURCE = None
+
+
+def auth_key_version() -> str:
+    """Return the active signing-key version string (never secret material)."""
+    _process_secret()  # ensure version is resolved with the secret
+    return _PROCESS_AUTH_KEY_VERSION or DEFAULT_KEY_VERSION
+
+
 def _process_secret() -> bytes:
-    global _PROCESS_AUTH_SECRET
+    """Resolve durable HMAC key for irrecoverable auth artifacts (#709 F4).
-    if _PROCESS_AUTH_SECRET is None:
+
-        if mcp_daemon_guard.is_pytest_runtime():
+    Production (non-pytest): **requires** ``GITEA_IRRECOVERABLE_AUTH_HMAC_KEY``.
-            _PROCESS_AUTH_SECRET = b"pytest-irrecoverable-auth-v1"
+    Silently generating an ephemeral per-process key is forbidden — that made
-        else:
+    mint+verify fail across merger/reconciler processes and restarts.
-            _PROCESS_AUTH_SECRET = secrets.token_bytes(32)
+
+    Pytest: uses a fixed constant unless the env key is set (so cross-process
+    regression tests can inject a shared durable key).
+    """
+    global _PROCESS_AUTH_SECRET, _PROCESS_AUTH_KEY_VERSION, _PROCESS_AUTH_SECRET_SOURCE
+    if _PROCESS_AUTH_SECRET is not None:
         return _PROCESS_AUTH_SECRET
 
+    env_raw = (os.environ.get(ENV_AUTH_HMAC_KEY) or "").strip()
+    env_ver = (os.environ.get(ENV_AUTH_HMAC_KEY_VERSION) or "").strip()
+    pytest = mcp_daemon_guard.is_pytest_runtime()
+
+    if env_raw:
+        _PROCESS_AUTH_SECRET = _decode_key_material(env_raw)
+        _PROCESS_AUTH_KEY_VERSION = env_ver or (
+            _PYTEST_KEY_VERSION if pytest else DEFAULT_KEY_VERSION
+        )
+        _PROCESS_AUTH_SECRET_SOURCE = "env"
+        return _PROCESS_AUTH_SECRET
+
+    if pytest:
+        _PROCESS_AUTH_SECRET = _PYTEST_AUTH_SECRET
+        _PROCESS_AUTH_KEY_VERSION = env_ver or _PYTEST_KEY_VERSION
+        _PROCESS_AUTH_SECRET_SOURCE = "pytest_constant"
+        return _PROCESS_AUTH_SECRET
+
+    # Production fail-closed: do NOT call secrets.token_bytes.
+    raise AuthSecretError(
+        f"{ENV_AUTH_HMAC_KEY} is required for production irrecoverable auth HMAC; "
+        "ephemeral per-process key generation is forbidden (fail closed, #709 F4 "
+        "review 435). Configure a durable shared secret for mint+verify across "
+        "processes and restarts."
+    )
+
 
 def expected_confirmation(pr_number: int) -> str:
     """Human intent confirmation text (not an authorization credential)."""
@@ -137,9 +227,19 @@ def _scope_payload(
     }
 
 
-def _sign_scope(scope: dict[str, Any], native_provenance: dict[str, Any]) -> str:
+def _sign_scope(
-    """HMAC over canonical scope + native transport fingerprint (non-caller)."""
+    scope: dict[str, Any],
+    native_provenance: dict[str, Any],
+    *,
+    key_version: str | None = None,
+) -> str:
+    """HMAC over key_version + canonical scope + native transport fingerprint.
+
+    The signing key itself is never serialized. Key *version* is bound into the
+    MAC so operators can rotate durable keys without ambiguous verification.
+    """
     material = {
+        "key_version": (key_version or auth_key_version()),
         "scope": scope,
         "native": {
             "native_mcp_transport": bool(
@@ -166,15 +266,14 @@ def assess_capability_for_irrecoverable_recovery(
     role_kind: str | None = None,
     profile_name: str | None = None,
 ) -> dict[str, Any]:
-    """Whether the active profile may mint/use irrecoverable recovery (#709 F1).
+    """Whether the active profile may mint/use irrecoverable recovery (#709 F5).
 
-    Accepts the dedicated capability, or a reconciler-shaped profile that
+    **Dedicated capability only.** Reconciler role / ``gitea.issue.comment``
-    already holds issue-comment mutation rights (interim equivalence until
+    equivalence is intentionally rejected so a reconciler cannot self-mint
-    operators grant the dedicated op). Never treats bare ``gitea.read`` as
+    recovery authority by authoring its own incident comment (#709 review 435
-    sufficient.
+    F5). Bare ``gitea.read`` is never sufficient.
     """
     import gitea_config
-    import reconciler_profile
 
     allowed = list(allowed_operations or [])
     forbidden = list(forbidden_operations or [])
@@ -191,35 +290,14 @@ def assess_capability_for_irrecoverable_recovery(
             "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
+    role_hint = role or name or "unknown"
-    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 "
+        f"(profile={role_hint!r}; reconciler/issue.comment equivalence is "
-            "profile or explicit grant, #709 F1)"
+        "not accepted — dedicated grant required, #709 F5 review 435; "
+        "gitea.read alone is insufficient)"
     )
     return {
         "allowed": False,
@@ -250,6 +328,108 @@ def assess_transport_for_auth_mint() -> dict[str, Any]:
     }
 
 
+def _incident_author(comment_payload: dict[str, Any]) -> str | None:
+    user = comment_payload.get("user")
+    if isinstance(user, dict):
+        login = (user.get("login") or user.get("username") or "").strip()
+        return login or None
+    if isinstance(user, str) and user.strip():
+        return user.strip()
+    # Some payloads flatten author.
+    for key in ("login", "author", "username"):
+        val = comment_payload.get(key)
+        if isinstance(val, str) and val.strip():
+            return val.strip()
+    return None
+
+
+def canonical_incident_fields(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+    incident_issue: int,
+) -> dict[str, str]:
+    """Ordered fields that form the authoritative incident content digest."""
+    head = normalize_head_sha(expected_head_sha) or ""
+    return {
+        "marker": INCIDENT_MARKER,
+        "remote": str(remote or "").strip(),
+        "org": str(org or "").strip(),
+        "repo": str(repo or "").strip(),
+        "pr_number": str(int(pr_number)),
+        "expected_head_sha": head,
+        "incident_issue": str(int(incident_issue)),
+    }
+
+
+def incident_content_digest(fields: dict[str, str]) -> str:
+    """SHA-256 hex digest over canonical field lines (stable, sort-free order)."""
+    # Fixed key order — do not sort; operator body must match this order.
+    order = (
+        "marker",
+        "remote",
+        "org",
+        "repo",
+        "pr_number",
+        "expected_head_sha",
+        "incident_issue",
+    )
+    lines = [f"{k}={fields[k]}" for k in order]
+    blob = "\n".join(lines).encode("utf-8")
+    return hashlib.sha256(blob).hexdigest()
+
+
+def build_canonical_incident_body(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    pr_number: int,
+    expected_head_sha: str,
+    incident_issue: int,
+    narrative: str | None = None,
+) -> str:
+    """Build an operator-postable canonical incident comment body (#709 F5)."""
+    fields = canonical_incident_fields(
+        remote=remote,
+        org=org,
+        repo=repo,
+        pr_number=pr_number,
+        expected_head_sha=expected_head_sha,
+        incident_issue=incident_issue,
+    )
+    digest = incident_content_digest(fields)
+    lines = [
+        fields["marker"],
+        f"remote: {fields['remote']}",
+        f"org: {fields['org']}",
+        f"repo: {fields['repo']}",
+        f"pr_number: {fields['pr_number']}",
+        f"expected_head_sha: {fields['expected_head_sha']}",
+        f"incident_issue: {fields['incident_issue']}",
+        f"content_digest: {digest}",
+    ]
+    if narrative and narrative.strip():
+        lines.extend(["", narrative.strip()])
+    return "\n".join(lines)
+
+
+def _parse_incident_field(body: str, name: str) -> str | None:
+    """Extract ``name: value`` or ``name=value`` from incident body lines."""
+    for line in (body or "").splitlines():
+        text = line.strip()
+        if not text or text.startswith("#"):
+            continue
+        for sep in (":", "="):
+            prefix = f"{name}{sep}"
+            if text.lower().startswith(prefix.lower()):
+                return text[len(prefix) :].strip()
+    return None
+
+
 def assess_incident_evidence(
     *,
     incident_issue: int | None,
@@ -259,8 +439,19 @@ def assess_incident_evidence(
     expected_remote: str | None = None,
     expected_org: str | None = None,
     expected_repo: str | None = None,
+    expected_pr_number: int | None = None,
+    expected_head_sha: str | None = None,
+    mint_actor_username: str | None = None,
+    reject_self_authored: bool = True,
 ) -> dict[str, Any]:
-    """Validate canonical incident evidence is present and live-fetched."""
+    """Validate authoritative incident evidence (live-fetched, #709 F5).
+
+    Requires:
+    - live comment id match
+    - non-empty author identity
+    - canonical marker + scope fields + content_digest integrity
+    - optional mint-actor independence (mint actor ≠ comment author)
+    """
     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")
@@ -293,16 +484,124 @@ def assess_incident_evidence(
     if not body:
         reasons.append("incident comment body is empty (fail closed)")
 
-    # Soft scope hints when URLs are present (never hard-code issue numbers).
+    author = _incident_author(comment_payload)
+    if not author:
+        reasons.append(
+            "incident comment missing author identity (fail closed, #709 F5)"
+        )
+
+    # Self-mint prevention: minting actor must not be the sole incident author.
+    if (
+        reject_self_authored
+        and author
+        and mint_actor_username
+        and author.strip().lower() == str(mint_actor_username).strip().lower()
+    ):
+        reasons.append(
+            "incident comment author matches mint actor; self-authored incident "
+            "evidence is not accepted (fail closed, #709 F5 review 435)"
+        )
+
+    # Canonical content + digest (not any non-empty body).
+    if body and INCIDENT_MARKER not in body:
+        reasons.append(
+            f"incident comment missing canonical marker {INCIDENT_MARKER!r} "
+            "(fail closed, #709 F5)"
+        )
+    elif body:
+        remote_f = _parse_incident_field(body, "remote")
+        org_f = _parse_incident_field(body, "org")
+        repo_f = _parse_incident_field(body, "repo")
+        pr_f = _parse_incident_field(body, "pr_number")
+        head_f = _parse_incident_field(body, "expected_head_sha")
+        issue_f = _parse_incident_field(body, "incident_issue")
+        digest_f = _parse_incident_field(body, "content_digest")
+
+        if expected_remote and remote_f and remote_f != str(expected_remote).strip():
+            reasons.append(
+                "incident body remote does not match expected remote (fail closed)"
+            )
+        if expected_org and org_f and org_f != str(expected_org).strip():
+            reasons.append(
+                "incident body org does not match expected org (fail closed)"
+            )
+        if expected_repo and repo_f and repo_f != str(expected_repo).strip():
+            reasons.append(
+                "incident body repo does not match expected repo (fail closed)"
+            )
+        if expected_pr_number is not None:
+            try:
+                if pr_f is None or int(pr_f) != int(expected_pr_number):
+                    reasons.append(
+                        "incident body pr_number does not match mint PR "
+                        "(fail closed, #709 F5)"
+                    )
+            except (TypeError, ValueError):
+                reasons.append(
+                    "incident body pr_number missing or invalid (fail closed)"
+                )
+        if expected_head_sha:
+            if not head_f or not heads_equal(head_f, expected_head_sha):
+                reasons.append(
+                    "incident body expected_head_sha does not match live mint "
+                    "head (fail closed, #709 F5)"
+                )
+        try:
+            if issue_f is None or int(issue_f) != int(incident_issue):  # type: ignore[arg-type]
+                reasons.append(
+                    "incident body incident_issue does not match provided "
+                    "incident_issue (fail closed, #709 F5)"
+                )
+        except (TypeError, ValueError):
+            reasons.append(
+                "incident body incident_issue missing or invalid (fail closed)"
+            )
+
+        # Content digest integrity when we have enough scope to recompute.
+        if (
+            remote_f
+            and org_f
+            and repo_f
+            and pr_f
+            and head_f
+            and issue_f
+            and digest_f
+        ):
+            try:
+                fields = canonical_incident_fields(
+                    remote=remote_f,
+                    org=org_f,
+                    repo=repo_f,
+                    pr_number=int(pr_f),
+                    expected_head_sha=head_f,
+                    incident_issue=int(issue_f),
+                )
+                expected_digest = incident_content_digest(fields)
+                if not hmac.compare_digest(
+                    expected_digest.lower(), str(digest_f).strip().lower()
+                ):
+                    reasons.append(
+                        "incident content_digest mismatch (forged or incomplete "
+                        "canonical body; fail closed, #709 F5)"
+                    )
+            except (TypeError, ValueError) as exc:
+                reasons.append(
+                    f"incident content_digest recompute failed: {exc} (fail closed)"
+                )
+        else:
+            reasons.append(
+                "incident body missing required canonical fields "
+                "(remote/org/repo/pr_number/expected_head_sha/incident_issue/"
+                "content_digest; fail closed, #709 F5)"
+            )
+
+    # Hard scope check when URLs are present.
     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:
+    if expected_org and issue_url and f"/{expected_org}/" not in 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 "
@@ -314,13 +613,10 @@ def assess_incident_evidence(
         "reasons": reasons,
         "comment": {
             "id": comment_payload.get("id"),
-            "author": (
+            "author": 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),
+            "has_canonical_marker": INCIDENT_MARKER in body if body else False,
         },
     }
 
@@ -396,7 +692,12 @@ def build_authorization_artifact(
         expires_at=expires.isoformat(),
         authorization_id=authorization_id,
     )
-    signature = _sign_scope(scope, provenance)
+    try:
+        kv = auth_key_version()
+        signature = _sign_scope(scope, provenance, key_version=kv)
+    except AuthSecretError as exc:
+        # Surface fail-closed mint: caller must not get a forgeable blank sig.
+        raise AuthSecretError(str(exc)) from exc
     return {
         "kind": KIND_AUTH,
         "auth_type": AUTH_TYPE,
@@ -407,6 +708,8 @@ def build_authorization_artifact(
         "recovery_critical": True,
         "issue_ref": "#709",
         "server_signature": signature,
+        "key_version": kv,
+        # Never serialize the secret; only the version id.
         "native_provenance": provenance,
         **scope,
         "timestamp": created.isoformat(),
@@ -487,7 +790,7 @@ def verify_authorization_artifact(
     if not (auth.get("server_signature") or "").strip():
         reasons.append("authorization missing server_signature (fail closed)")
 
-    # Recompute signature over stored scope fields.
+    # Recompute signature over stored scope fields + key_version.
     try:
         scope = _scope_payload(
             remote=str(auth.get("remote") or ""),
@@ -507,14 +810,18 @@ def verify_authorization_artifact(
         native = auth.get("native_provenance") or {}
         if not isinstance(native, dict):
             native = {}
-        expected_sig = _sign_scope(scope, native)
+        stored_kv = str(auth.get("key_version") or auth_key_version())
+        expected_sig = _sign_scope(scope, native, key_version=stored_kv)
         if not hmac.compare_digest(
             expected_sig, str(auth.get("server_signature") or "")
         ):
             reasons.append(
-                "authorization server_signature invalid (forged or corrupt; "
+                "authorization server_signature invalid (forged, corrupt, or "
-                "fail closed, #709 F1)"
+                "HMAC key mismatch across process/restart; fail closed, "
+                "#709 F4/F1)"
             )
+    except AuthSecretError as exc:
+        reasons.append(f"authorization HMAC key unavailable: {exc} (fail closed)")
     except (TypeError, ValueError) as exc:
         reasons.append(f"authorization scope incomplete: {exc} (fail closed)")
 

stale_review_decision_lock.py

@@ -504,7 +504,13 @@ def lock_targets_merged_pr_approval(
     pr_number: int,
     expected_head_sha: str | None = None,
 ) -> bool:
-    """True when *lock*'s last terminal mutation is approve of *pr_number*."""
+    """True when *lock*'s last terminal mutation is approve of *pr_number*.
+
+    When *expected_head_sha* is provided, a **recorded** terminal head must
+    match. Legacy same-repo approve locks with no recorded head do **not**
+    match (fail closed, #709 F3 residual / review 435) — callers must use the
+    strict secondary path that refuses no-head clears.
+    """
     last = last_terminal_mutation(lock)
     if last is None:
         return False
@@ -513,8 +519,12 @@ def lock_targets_merged_pr_approval(
     if last.get("pr_number") != pr_number:
         return False
     if expected_head_sha:
+        want = normalize_head_sha(expected_head_sha)
+        if not want:
+            return False
         locked = mutation_head_sha(last, lock)
-        if locked and not heads_equal(locked, expected_head_sha):
+        # Require recorded-head match; unrecorded head is not a match (#709 F3).
+        if not locked or not heads_equal(locked, want):
             return False
     return True
 

tests/test_issue_709_decision_lock_cross_profile.py

@@ -1,7 +1,8 @@
 """#709: cross-profile decision-lock cleanup, overwrite protection, recovery.
 
-Covers AC1–AC8 plus review-434 F1/F2/F3 remediations without fabricating
+Covers AC1–AC8 plus review-434 F1/F2/F3 and review-435 F3-residual/F4/F5
-historical PR provenance or special-casing live PR numbers in production code.
+remediations without fabricating historical PR provenance or special-casing
+live PR numbers in production code.
 """
 
 from __future__ import annotations
@@ -63,6 +64,58 @@ RECONCILER_OPS = [
     "gitea.pr.comment",
     "gitea.issue.comment",
 ]
+DEDICATED_RECOVERY_OPS = RECONCILER_OPS + [
+    irp.CAPABILITY_IRRECOVERABLE_RECOVERY,
+]
+DURABLE_TEST_HMAC_KEY = "0" * 64  # 32-byte hex durable key for F4 tests
+
+
+def _canonical_incident_body(
+    *,
+    pr_number=42,
+    head=HEAD_A,
+    remote="prgs",
+    org="Scaled-Tech-Consulting",
+    repo="Gitea-Tools",
+    incident_issue=700,
+    narrative="forensic diagnosis",
+):
+    return irp.build_canonical_incident_body(
+        remote=remote,
+        org=org,
+        repo=repo,
+        pr_number=pr_number,
+        expected_head_sha=head,
+        incident_issue=incident_issue,
+        narrative=narrative,
+    )
+
+
+def _incident_comment_payload(
+    *,
+    comment_id=11489,
+    author="controller-ops",
+    pr_number=42,
+    head=HEAD_A,
+    remote="prgs",
+    org="Scaled-Tech-Consulting",
+    repo="Gitea-Tools",
+    incident_issue=700,
+):
+    return {
+        "id": comment_id,
+        "body": _canonical_incident_body(
+            pr_number=pr_number,
+            head=head,
+            remote=remote,
+            org=org,
+            repo=repo,
+            incident_issue=incident_issue,
+        ),
+        "user": {"login": author},
+        "created_at": "2026-07-13T00:00:00Z",
+        "html_url": f"https://gitea.example/{org}/{repo}/issues/{incident_issue}#issuecomment-{comment_id}",
+    }
 
 
 def _mint_auth(
@@ -142,6 +195,24 @@ class TestAC1TargetApproval(unittest.TestCase):
             )
         )
 
+    def test_f3_residual_rejects_legacy_no_head_when_expected_head_given(self):
+        """Primary approve-match requires recorded-head (#709 F3 residual / 435)."""
+        # APPROVE without head fields — legacy ledger.
+        legacy = _lock([APPROVE])  # no head=
+        self.assertIsNone(srdl.mutation_head_sha(APPROVE, legacy))
+        self.assertFalse(
+            srdl.lock_targets_merged_pr_approval(
+                legacy,
+                pr_number=100,
+                expected_head_sha=HEAD_A,
+            )
+        )
+        # Without expected_head_sha, PR-number-only match still works for
+        # non-destructive callers that do not pass a head pin.
+        self.assertTrue(
+            srdl.lock_targets_merged_pr_approval(legacy, pr_number=100)
+        )
+
 
 class TestF1AuthorizationNotSelfAssertable(unittest.TestCase):
     def test_operator_authorized_true_cannot_authorize_via_build(self):
@@ -336,7 +407,8 @@ class TestF1AuthorizationNotSelfAssertable(unittest.TestCase):
         )
         self.assertFalse(a["allowed"])
 
-    def test_reconciler_capability_allowed(self):
+    def test_f5_reconciler_equivalence_rejected(self):
+        """Reconciler profile without dedicated capability cannot mint (#709 F5)."""
         a = irp.assess_capability_for_irrecoverable_recovery(
             allowed_operations=RECONCILER_OPS,
             forbidden_operations=[
@@ -347,7 +419,241 @@ class TestF1AuthorizationNotSelfAssertable(unittest.TestCase):
             role_kind="reconciler",
             profile_name="prgs-reconciler",
         )
+        self.assertFalse(a["allowed"], a)
+        self.assertTrue(
+            any("dedicated" in r for r in a["reasons"]),
+            msg=a["reasons"],
+        )
+
+    def test_f5_dedicated_capability_allowed(self):
+        a = irp.assess_capability_for_irrecoverable_recovery(
+            allowed_operations=DEDICATED_RECOVERY_OPS,
+            forbidden_operations=[
+                "gitea.pr.approve",
+                "gitea.pr.merge",
+                "gitea.pr.review",
+            ],
+            role_kind="reconciler",
+            profile_name="prgs-reconciler",
+        )
         self.assertTrue(a["allowed"], a)
+        self.assertEqual(a["via"], "dedicated_capability")
+
+
+class TestF5AuthoritativeIncidentEvidence(unittest.TestCase):
+    def test_any_nonempty_body_rejected(self):
+        g = irp.assess_incident_evidence(
+            incident_issue=700,
+            incident_comment_id=11489,
+            comment_payload={
+                "id": 11489,
+                "body": "random forensic note without canonical fields",
+                "user": {"login": "controller-ops"},
+            },
+            expected_remote="prgs",
+            expected_org="Scaled-Tech-Consulting",
+            expected_repo="Gitea-Tools",
+            expected_pr_number=42,
+            expected_head_sha=HEAD_A,
+        )
+        self.assertFalse(g["valid"], g)
+
+    def test_missing_author_rejected(self):
+        body = _canonical_incident_body()
+        g = irp.assess_incident_evidence(
+            incident_issue=700,
+            incident_comment_id=11489,
+            comment_payload={"id": 11489, "body": body, "user": {}},
+            expected_remote="prgs",
+            expected_org="Scaled-Tech-Consulting",
+            expected_repo="Gitea-Tools",
+            expected_pr_number=42,
+            expected_head_sha=HEAD_A,
+        )
+        self.assertFalse(g["valid"], g)
+        self.assertTrue(any("author" in r for r in g["reasons"]), g["reasons"])
+
+    def test_self_authored_rejected(self):
+        g = irp.assess_incident_evidence(
+            incident_issue=700,
+            incident_comment_id=11489,
+            comment_payload=_incident_comment_payload(author="sysadmin"),
+            expected_remote="prgs",
+            expected_org="Scaled-Tech-Consulting",
+            expected_repo="Gitea-Tools",
+            expected_pr_number=42,
+            expected_head_sha=HEAD_A,
+            mint_actor_username="sysadmin",
+            reject_self_authored=True,
+        )
+        self.assertFalse(g["valid"], g)
+        self.assertTrue(
+            any("self-authored" in r for r in g["reasons"]), g["reasons"]
+        )
+
+    def test_canonical_body_with_independent_author_accepted(self):
+        g = irp.assess_incident_evidence(
+            incident_issue=700,
+            incident_comment_id=11489,
+            comment_payload=_incident_comment_payload(author="controller-ops"),
+            expected_remote="prgs",
+            expected_org="Scaled-Tech-Consulting",
+            expected_repo="Gitea-Tools",
+            expected_pr_number=42,
+            expected_head_sha=HEAD_A,
+            mint_actor_username="sysadmin",
+            reject_self_authored=True,
+        )
+        self.assertTrue(g["valid"], g)
+
+    def test_tampered_content_digest_rejected(self):
+        payload = _incident_comment_payload()
+        payload["body"] = payload["body"].replace(
+            "content_digest: ", "content_digest: " + "f" * 64 + "x"
+        )
+        # Force bad digest line
+        lines = []
+        for line in payload["body"].splitlines():
+            if line.startswith("content_digest:"):
+                lines.append("content_digest: " + "0" * 64)
+            else:
+                lines.append(line)
+        payload["body"] = "\n".join(lines)
+        g = irp.assess_incident_evidence(
+            incident_issue=700,
+            incident_comment_id=11489,
+            comment_payload=payload,
+            expected_remote="prgs",
+            expected_org="Scaled-Tech-Consulting",
+            expected_repo="Gitea-Tools",
+            expected_pr_number=42,
+            expected_head_sha=HEAD_A,
+            mint_actor_username="sysadmin",
+        )
+        self.assertFalse(g["valid"], g)
+        self.assertTrue(
+            any("content_digest" in r for r in g["reasons"]), g["reasons"]
+        )
+
+
+class TestF4DurableHmacKey(unittest.TestCase):
+    def tearDown(self):
+        os.environ.pop(irp.ENV_AUTH_HMAC_KEY, None)
+        os.environ.pop(irp.ENV_AUTH_HMAC_KEY_VERSION, None)
+        irp.reset_process_auth_secret_for_tests()
+
+    def test_key_version_bound_into_artifact(self):
+        irp.reset_process_auth_secret_for_tests()
+        auth = _mint_auth()
+        self.assertIn("key_version", auth)
+        self.assertTrue(auth["key_version"])
+        self.assertNotIn("server_secret", auth)
+        self.assertNotIn("hmac_key", auth)
+
+    def test_production_fails_closed_without_durable_key(self):
+        """Non-pytest process without env key must not generate ephemeral secret."""
+        script = (
+            "import os, sys\n"
+            "os.environ.pop('PYTEST_CURRENT_TEST', None)\n"
+            "os.environ.pop('GITEA_IRRECOVERABLE_AUTH_HMAC_KEY', None)\n"
+            # Force non-pytest path by patching guard after import.
+            "import mcp_daemon_guard as g\n"
+            "g.is_pytest_runtime = lambda: False\n"
+            "import irrecoverable_provenance as irp\n"
+            "irp.reset_process_auth_secret_for_tests()\n"
+            "try:\n"
+            "    irp._process_secret()\n"
+            "    print('UNEXPECTED_OK')\n"
+            "except irp.AuthSecretError as e:\n"
+            "    print('FAIL_CLOSED', 'ephemeral' in str(e).lower() or 'required' in str(e).lower())\n"
+        )
+        env = {k: v for k, v in os.environ.items() if not k.startswith("PYTEST")}
+        env.pop("PYTEST_CURRENT_TEST", None)
+        env.pop(irp.ENV_AUTH_HMAC_KEY, None)
+        proc = subprocess.run(
+            [sys.executable, "-c", script],
+            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            capture_output=True,
+            text=True,
+            env=env,
+            timeout=30,
+        )
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        out = (proc.stdout or "").strip()
+        self.assertTrue(out.startswith("FAIL_CLOSED"), msg=out)
+
+    def test_cross_process_verify_with_durable_key(self):
+        """Mint in process A, verify in process B with same durable key (#709 F4)."""
+        import json as _json
+
+        mint_script = (
+            "import json, os, irrecoverable_provenance as irp\n"
+            "irp.reset_process_auth_secret_for_tests()\n"
+            "auth = irp.build_authorization_artifact(\n"
+            " remote='prgs', org='o', repo='r', pr_number=10,\n"
+            f" expected_head_sha={HEAD_A!r}, incident_issue=1, incident_comment_id=2,\n"
+            " destroyed_subject=None, issuer_username='sysadmin',\n"
+            " issuer_profile='prgs-reconciler',\n"
+            " native_provenance={'native_mcp_transport': True, 'pytest': True,\n"
+            "  'token_fingerprint': 'fp', 'entrypoint': 'pytest', 'pid': 1})\n"
+            "print(json.dumps(auth))\n"
+        )
+        env = dict(os.environ)
+        env[irp.ENV_AUTH_HMAC_KEY] = DURABLE_TEST_HMAC_KEY
+        env[irp.ENV_AUTH_HMAC_KEY_VERSION] = "test-v1"
+        mint = subprocess.run(
+            [sys.executable, "-c", mint_script],
+            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            capture_output=True,
+            text=True,
+            env=env,
+            timeout=30,
+        )
+        self.assertEqual(mint.returncode, 0, mint.stderr)
+        auth = _json.loads(mint.stdout.strip())
+        self.assertEqual(auth.get("key_version"), "test-v1")
+
+        verify_script = (
+            "import json, sys, irrecoverable_provenance as irp\n"
+            "irp.reset_process_auth_secret_for_tests()\n"
+            "auth = json.loads(sys.stdin.read())\n"
+            "v = irp.verify_authorization_artifact(\n"
+            " auth, remote='prgs', org='o', repo='r', pr_number=10,\n"
+            f" expected_head_sha={HEAD_A!r}, incident_issue=1, incident_comment_id=2)\n"
+            "print(v.get('valid'), v.get('reasons'))\n"
+        )
+        verify = subprocess.run(
+            [sys.executable, "-c", verify_script],
+            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            input=_json.dumps(auth),
+            capture_output=True,
+            text=True,
+            env=env,
+            timeout=30,
+        )
+        self.assertEqual(verify.returncode, 0, verify.stderr)
+        self.assertTrue(
+            (verify.stdout or "").strip().startswith("True"),
+            msg=verify.stdout + verify.stderr,
+        )
+
+        # Different durable key must fail verification (cross-process mismatch).
+        env_bad = dict(env)
+        env_bad[irp.ENV_AUTH_HMAC_KEY] = "1" * 64
+        verify_bad = subprocess.run(
+            [sys.executable, "-c", verify_script],
+            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            input=_json.dumps(auth),
+            capture_output=True,
+            text=True,
+            env=env_bad,
+            timeout=30,
+        )
+        self.assertEqual(verify_bad.returncode, 0, verify_bad.stderr)
+        self.assertTrue(
+            (verify_bad.stdout or "").strip().startswith("False"),
+            msg=verify_bad.stdout,
+        )
 
 
 class TestF2MergerConsumer(unittest.TestCase):
@@ -807,7 +1113,7 @@ class TestIrrecoverableToolF1(unittest.TestCase):
                 "GITEA_MCP_SESSION_STATE_DIR": self._tmp.name,
                 "GITEA_SESSION_PROFILE_LOCK": "prgs-reconciler",
                 "GITEA_PROFILE_NAME": "prgs-reconciler",
-                "GITEA_ALLOWED_OPERATIONS": ",".join(RECONCILER_OPS),
+                "GITEA_ALLOWED_OPERATIONS": ",".join(DEDICATED_RECOVERY_OPS),
             },
             clear=False,
         )
@@ -824,7 +1130,7 @@ class TestIrrecoverableToolF1(unittest.TestCase):
         return {
             "profile_name": "prgs-reconciler",
             "role": "reconciler",
-            "allowed_operations": RECONCILER_OPS,
+            "allowed_operations": DEDICATED_RECOVERY_OPS,
             "forbidden_operations": [
                 "gitea.pr.approve",
                 "gitea.pr.merge",
@@ -924,17 +1230,18 @@ class TestIrrecoverableToolF1(unittest.TestCase):
             if "/pulls/" in str(url):
                 return {"head": {"sha": HEAD_A}, "state": "open"}
             if "/issues/comments/" in str(url):
-                return {
+                return _incident_comment_payload(
-                    "id": 11489,
+                    comment_id=11489,
-                    "body": "forensic diagnosis",
+                    author="controller-ops",
-                    "user": {"login": "sysadmin"},
+                    pr_number=50,
-                    "created_at": "2026-07-13T00:00:00Z",
+                    head=HEAD_A,
-                }
+                    remote="prgs",
+                    org="Scaled-Tech-Consulting",
+                    repo="Gitea-Tools",
+                    incident_issue=700,
+                )
             raise AssertionError(f"unexpected API {method} {url}")
 
-        common = dict(
-            get_profile=self._profile(),
-        )
         with patch.object(self.mcp, "get_profile", return_value=self._profile()), patch.object(
             self.mcp, "_authenticated_username", return_value="sysadmin"
         ), patch.object(