feat: guarded merger lease adoption flow (#536)

Replace manual in-process _SESSION_LEASE seeding with an auditable
gitea_adopt_merger_pr_lease tool that posts durable adoption proof and
records sanctioned session provenance. Mutation gates now reject bare
record_session_lease calls without provenance from acquire/adopt/heartbeat.

Closes #536
This commit is contained in:
2026-07-08 11:03:56 -04:00
parent 6b45212659
commit 08ea214526
10 changed files with 905 additions and 10 deletions
+31 -5
View File
@@ -23,6 +23,7 @@ _ACTIVE_PHASES = frozenset({
"approved",
"request-changes",
"merging",
"adopted",
})
DEFAULT_LEASE_TTL_MINUTES = 120
@@ -380,9 +381,22 @@ def assess_post_merge_moot_lease(
}
def record_session_lease(lease: dict[str, Any]) -> dict[str, Any]:
def record_session_lease(
lease: dict[str, Any],
*,
lease_provenance: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Record the in-session lease mirror for mutation gates.
*lease_provenance* must be supplied by sanctioned MCP tools (#536). Bare
manual seeding without provenance cannot satisfy merger/reviewer mutation
gates.
"""
global _SESSION_LEASE
_SESSION_LEASE = dict(lease)
stored = dict(lease)
if lease_provenance:
stored["lease_provenance"] = dict(lease_provenance)
_SESSION_LEASE = stored
return dict(_SESSION_LEASE)
@@ -415,13 +429,25 @@ def assess_mutation_lease_gate(
if not session:
reasons.append(
f"no in-session reviewer lease recorded; acquire via "
f"gitea_acquire_reviewer_pr_lease before {mutation}"
f"gitea_acquire_reviewer_pr_lease or adopt via "
f"gitea_adopt_merger_pr_lease before {mutation}"
)
elif session.get("pr_number") != pr_number:
else:
import merger_lease_adoption as mla
if not mla.is_sanctioned_session_lease(session):
reasons.append(
"in-session lease lacks sanctioned provenance; manual "
"_SESSION_LEASE seeding is not canonical proof — use "
"gitea_acquire_reviewer_pr_lease or gitea_adopt_merger_pr_lease"
)
if session and session.get("pr_number") != pr_number:
reasons.append(
f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}"
)
elif (session.get("session_id") or "") != (session_id or session.get("session_id")):
elif session and (session.get("session_id") or "") != (
session_id or session.get("session_id")
):
reasons.append("session lease session_id mismatch (fail closed)")
if active: