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

mcp_session_state.py

@@ -36,7 +36,24 @@ KIND_REVIEW_DRAFT = "review_draft"
 # other session proofs; a contaminated session fails closed on gated mutations
 # until a reconciler audits and clears it.
 KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
+# #709: archive prior terminal decision ledgers instead of silent overwrite.
+KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
+# #709: post-merge cleanup/audit reconciliation-required durable record.
+KIND_POST_MERGE_DECISION_RECOVERY = "post_merge_decision_recovery"
+# #709: truthful record when historical terminal evidence is irrecoverably gone.
+KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
+# #709 F1: server-side non-forgeable authorization artifact for recovery.
+KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 
+# Kinds that must survive the default session-state TTL (forensic / recovery).
+RECOVERY_CRITICAL_KINDS = frozenset(
+    {
+        KIND_DECISION_LOCK_ARCHIVE,
+        KIND_POST_MERGE_DECISION_RECOVERY,
+        KIND_IRRECOVERABLE_DECISION_PROVENANCE,
+        KIND_IRRECOVERABLE_PROVENANCE_AUTH,
+    }
+)
 
 
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
@@ -270,7 +287,11 @@ def identity_match_reasons(
         reasons.append("session state missing recorded_at timestamp (fail closed)")
     else:
         age = _now_utc() - recorded_at
-        if age > timedelta(hours=ttl_hours()):
+        kind = (record.get("kind") or "").strip()
+        ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
+            record.get("recovery_critical")
+        )
+        if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
             reasons.append(
                 f"session state expired after {ttl_hours():g}h (fail closed)"
             )
@@ -447,3 +468,177 @@ def clear_state(
         profile_identity=profile_identity,
         state_dir=state_dir,
     )
+
+def list_decision_lock_profile_identities(
+    state_dir: str | None = None,
+) -> list[str]:
+    """Return profile identities that have a durable review_decision_lock file (#709).
+
+    Filename form: ``review_decision_lock-<profile>.json`` (see ``state_key``).
+    Does not validate TTL or identity — callers must load via ``load_state``.
+    """
+    root = (state_dir or default_state_dir()).strip()
+    if not root or not os.path.isdir(root):
+        return []
+    prefix = f"{_sanitize_segment(KIND_DECISION_LOCK)}-"
+    suffix = ".json"
+    found: list[str] = []
+    try:
+        names = os.listdir(root)
+    except OSError:
+        return []
+    for name in names:
+        if not name.startswith(prefix) or not name.endswith(suffix):
+            continue
+        if name.endswith(".lock"):
+            continue
+        mid = name[len(prefix) : -len(suffix)]
+        if mid:
+            found.append(mid)
+    return sorted(set(found))
+
+
+def load_state_for_profile(
+    *,
+    kind: str,
+    profile_identity: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    state_dir: str | None = None,
+    skip_identity_match: bool = False,
+    enforce_repo_scope: bool = True,
+) -> dict[str, Any] | None:
+    """Load durable state for an explicit profile identity (#709 cross-profile).
+
+    When *skip_identity_match* is True, still requires the file's recorded
+    profile_identity to equal the requested profile (anti-stomp), but does not
+    require the *active* session identity to match — needed so a merger can
+    inspect a reviewer lock after merge.
+
+    #709 F3: remote/org/repo filter reasons are **enforced** (not merely
+    computed). When *enforce_repo_scope* is True (default) and the caller
+    supplies remote/org/repo, mismatches fail closed with ``None``.
+    """
+    # #709 F3: refuse traversal / malformed profile identities.
+    try:
+        from irrecoverable_provenance import assess_profile_path_identity
+
+        path_gate = assess_profile_path_identity(profile_identity)
+        if not path_gate.get("valid"):
+            return None
+    except Exception:
+        # Module may be mid-import in edge bootstraps; fall through to
+        # conservative checks below.
+        raw = str(profile_identity or "")
+        if ".." in raw or "/" in raw or "\\" in raw or "\x00" in raw:
+            return None
+
+    profile = current_profile_identity(profile_identity=profile_identity)
+    root = _ensure_state_dir(state_dir)
+    # Refuse symlink escape of the state root (#709 F3).
+    try:
+        real_root = os.path.realpath(root)
+        path = state_file_path(
+            kind=kind,
+            remote=remote,
+            org=org,
+            repo=repo,
+            profile_identity=profile,
+            state_dir=root,
+        )
+        real_path = os.path.realpath(path) if os.path.exists(path) else path
+        if os.path.exists(path) and not str(real_path).startswith(
+            str(real_root) + os.sep
+        ) and str(real_path) != str(real_root):
+            return None
+    except OSError:
+        return None
+
+    path = state_file_path(
+        kind=kind,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+        state_dir=root,
+    )
+    lock_path = f"{path}.lock"
+    with _exclusive_file_lock(lock_path):
+        envelope = _read_json(path)
+    if not envelope:
+        return None
+    payload = envelope.get("payload")
+    if not isinstance(payload, dict):
+        return None
+    merged = dict(payload)
+    for key in (
+        "kind",
+        "remote",
+        "org",
+        "repo",
+        "profile_identity",
+        "session_profile_lock",
+        "recorded_at",
+        "updated_at",
+        "writer_pid",
+    ):
+        if key in envelope and key not in merged:
+            merged[key] = envelope[key]
+    stored = (merged.get("profile_identity") or "").strip()
+    if stored and stored != profile:
+        return None
+    if skip_identity_match:
+        # Enforce remote/org/repo when caller provides them (#709 F3).
+        # Drop only *active session* profile-identity mismatches; keep
+        # expiry, spoof, and repository-scope reasons.
+        scope_remote = remote if enforce_repo_scope else None
+        scope_org = org if enforce_repo_scope else None
+        scope_repo = repo if enforce_repo_scope else None
+        reasons = identity_match_reasons(
+            merged,
+            remote=scope_remote,
+            org=scope_org,
+            repo=scope_repo,
+            profile_identity=stored or profile,
+        )
+        filtered = [
+            r
+            for r in reasons
+            if "profile identity mismatch" not in r
+            or (stored and stored != profile)
+        ]
+        # When caller requested a specific remote/org/repo, also fail if the
+        # stored record lacks those identity fields entirely (legacy incomplete).
+        if enforce_repo_scope:
+            for field, want in (
+                ("remote", remote),
+                ("org", org),
+                ("repo", repo),
+            ):
+                want_s = (want or "").strip()
+                if not want_s:
+                    continue
+                have = (str(merged.get(field) or "")).strip()
+                if not have:
+                    filtered.append(
+                        f"session state {field} missing on durable record "
+                        f"(expected={want_s!r}; fail closed, #709 F3)"
+                    )
+                elif have != want_s:
+                    # identity_match_reasons already adds mismatch; ensure kept
+                    pass
+        if filtered:
+            return None
+        return merged
+    reasons = identity_match_reasons(
+        merged,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+    )
+    if reasons:
+        return None
+    return merged
+

stale_review_decision_lock.py

@@ -438,3 +438,252 @@ def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
         "This path only clears a lock when the referenced PR is merged/closed.",
     ]
     return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# #709 cross-profile cleanup, overwrite protection, recovery provenance
+# ---------------------------------------------------------------------------
+
+
+def has_unresolved_terminal_evidence(lock: dict | None) -> bool:
+    """True when *lock* still carries a terminal live mutation ledger."""
+    return last_terminal_mutation(lock) is not None
+
+
+def assess_init_overwrite(
+    existing_lock: dict | None,
+    *,
+    force: bool = False,
+) -> dict[str, Any]:
+    """Whether init_review_decision_lock may replace an existing durable lock (#709 AC2).
+
+    Unresolved terminal evidence must never be silently replaced by an empty
+    initialized lock. *force* does not authorize destruction of terminal
+    ledgers — only explicit cleanup / archive transitions may remove them.
+    """
+    result: dict[str, Any] = {
+        "overwrite_allowed": True,
+        "has_existing": existing_lock is not None,
+        "has_unresolved_terminal": False,
+        "reasons": [],
+        "last_terminal_pr": None,
+        "last_terminal_action": None,
+        "existing_profile_identity": None,
+    }
+    if existing_lock is None:
+        result["reasons"].append("no existing lock; empty init allowed")
+        return result
+    result["existing_profile_identity"] = (
+        existing_lock.get("profile_identity")
+        or existing_lock.get("session_profile_lock")
+        or existing_lock.get("session_profile")
+    )
+    last = last_terminal_mutation(existing_lock)
+    if last is None:
+        result["reasons"].append(
+            "existing lock has no terminal mutations; re-init allowed"
+        )
+        return result
+    result["has_unresolved_terminal"] = True
+    result["overwrite_allowed"] = False
+    result["last_terminal_pr"] = last.get("pr_number")
+    result["last_terminal_action"] = last.get("action")
+    result["reasons"].append(
+        "refuse empty re-init: unresolved terminal decision-lock evidence "
+        f"present ({last.get('action')} on PR #{last.get('pr_number')}); "
+        "use gitea_cleanup_stale_review_decision_lock when moot, or archive "
+        f"via sanctioned recovery — force={force!r} does not authorize overwrite "
+        "(#709 AC2)"
+    )
+    return result
+
+
+def lock_targets_merged_pr_approval(
+    lock: dict | None,
+    *,
+    pr_number: int,
+    expected_head_sha: str | None = None,
+) -> bool:
+    """True when *lock*'s last terminal mutation is approve of *pr_number*.
+
+    When *expected_head_sha* is provided, a **recorded** terminal head must
+    match. Legacy same-repo approve locks with no recorded head do **not**
+    match (fail closed, #709 F3 residual / review 435) — callers must use the
+    strict secondary path that refuses no-head clears.
+    """
+    last = last_terminal_mutation(lock)
+    if last is None:
+        return False
+    if last.get("action") != "approve":
+        return False
+    if last.get("pr_number") != pr_number:
+        return False
+    if expected_head_sha:
+        want = normalize_head_sha(expected_head_sha)
+        if not want:
+            return False
+        locked = mutation_head_sha(last, lock)
+        # Require recorded-head match; unrecorded head is not a match (#709 F3).
+        if not locked or not heads_equal(locked, want):
+            return False
+    return True
+
+
+def build_post_merge_recovery_record(
+    *,
+    pr_number: int,
+    head_sha: str | None,
+    merge_commit_sha: str | None,
+    target_profile_identity: str | None,
+    failed_step: str,
+    error: str | None,
+    remote: str | None,
+    org: str | None,
+    repo: str | None,
+    actor_username: str | None,
+    profile_name: str | None,
+) -> dict[str, Any]:
+    """Durable recovery-required payload after irreversible merge (#709 AC3)."""
+    return {
+        "event": "post_merge_decision_lock_recovery_required",
+        "status": "recovery_required",
+        "issue_ref": "#709",
+        "recovery_critical": True,
+        "applied": False,  # never claim historical cleanup succeeded
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "pr_number": pr_number,
+        "head_sha": normalize_head_sha(head_sha),
+        "merge_commit_sha": merge_commit_sha,
+        "target_profile_identity": target_profile_identity,
+        "failed_step": failed_step,
+        "error": error,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "required_recovery_action": (
+            "retry cross-profile decision-lock cleanup and audit publication "
+            "for the merged PR; if terminal evidence is gone, use "
+            "gitea_record_irrecoverable_decision_lock_provenance"
+        ),
+    }
+
+
+def build_irrecoverable_provenance_record(
+    *,
+    pr_number: int,
+    head_sha: str | None,
+    remote: str | None,
+    org: str | None,
+    repo: str | None,
+    actor_username: str | None,
+    profile_name: str | None,
+    reason: str,
+    incident_ref: str | None = None,
+    # Deprecated kwargs retained only so stale call sites fail closed:
+    operator_authorized: bool | None = None,
+    # Required for merger-acceptable records (#709 F1):
+    authorization: dict[str, Any] | None = None,
+    incident_issue: int | None = None,
+    incident_comment_id: int | None = None,
+    destroyed_subject: str | None = None,
+    historical_provenance_subject: str | None = None,
+) -> dict[str, Any]:
+    """Truthful absence-of-proof record (#709 AC5). Never sets applied=True.
+
+    Caller-supplied ``operator_authorized`` is **ignored** as authorization
+    evidence (review 434 F1). Prefer
+    :func:`irrecoverable_provenance.build_irrecoverable_provenance_record`
+    with a verified server-side authorization artifact.
+    """
+    # Explicitly ignore deprecated self-assertable Boolean.
+    _ = operator_authorized
+    if authorization is not None and incident_issue is not None and incident_comment_id is not None:
+        from irrecoverable_provenance import (
+            build_irrecoverable_provenance_record as _build,
+        )
+
+        return _build(
+            pr_number=int(pr_number),
+            head_sha=str(head_sha or ""),
+            remote=str(remote or ""),
+            org=str(org or ""),
+            repo=str(repo or ""),
+            actor_username=actor_username,
+            profile_name=profile_name,
+            reason=reason,
+            incident_issue=int(incident_issue),
+            incident_comment_id=int(incident_comment_id),
+            authorization=authorization,
+            destroyed_subject=destroyed_subject,
+            historical_provenance_subject=historical_provenance_subject
+            or destroyed_subject,
+        )
+    # Fail-closed skeleton when no server authorization is supplied: never
+    # sets merger_may_accept True (even if operator_authorized was True).
+    return {
+        "event": "irrecoverable_decision_lock_provenance",
+        "status": "provenance_irrecoverable",
+        "record_type": "irrecoverable_decision_provenance",
+        "operator_recovery_required": True,
+        "issue_ref": "#709",
+        "recovery_critical": True,
+        "applied": False,
+        "historical_cleanup_proven": False,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "pr_number": pr_number,
+        "head_sha": normalize_head_sha(head_sha),
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "reason": reason,
+        "incident_ref": incident_ref,
+        "incident_issue": incident_issue,
+        "incident_comment_id": incident_comment_id,
+        "authorization_verified": False,
+        "merger_may_accept": False,
+        "acceptance_rule": (
+            "Merger may accept this record only when a server-side "
+            "authorization artifact verifies for remote/org/repo/PR/exact "
+            "head/incident, the record is durable and read back, and normal "
+            "merge gates still pass. Caller Booleans never authorize. This "
+            "does not prove historical cleanup."
+        ),
+    }
+
+
+def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str:
+    """Markdown body for irrecoverable provenance audit (no applied=true claim)."""
+    from irrecoverable_provenance import (
+        format_irrecoverable_audit_comment as _fmt,
+    )
+
+    return _fmt(record)
+
+
+def format_post_merge_recovery_comment(record: dict[str, Any]) -> str:
+    """Markdown body for post-merge recovery-required audit."""
+    lines = [
+        "## Post-merge decision-lock recovery required (#709)",
+        "",
+        "Status: **RECOVERY_REQUIRED** (merge is irreversible; cleanup/audit incomplete)",
+        "",
+        f"- actor: `{record.get('actor_username')}`",
+        f"- profile: `{record.get('profile_name')}`",
+        f"- timestamp: `{record.get('timestamp')}`",
+        f"- PR: `#{record.get('pr_number')}`",
+        f"- head_sha: `{record.get('head_sha')}`",
+        f"- merge_commit_sha: `{record.get('merge_commit_sha')}`",
+        f"- target_profile_identity: `{record.get('target_profile_identity')}`",
+        f"- failed_step: `{record.get('failed_step')}`",
+        f"- error: `{record.get('error')}`",
+        "",
+        f"Required action: {record.get('required_recovery_action')}",
+        "",
+        "applied=false — do not treat this as successful cleanup evidence.",
+    ]
+    return "\n".join(lines)
+

task_capability_map.py

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