Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

fix: resolve conflicts for PR #486

Browse Source 
Merge prgs/master into PR branch to restore mergeability.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-08 22:28:12 -04:00
co-authored by Claude Opus 4.8
parent 3b4f222a7d 0e791325f8
commit b12212493c
47 changed files with 6721 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+704 -19
View File
@@ -211,6 +211,28 @@ def _effective_workspace_role() -> str:
)
def _actual_profile_role() -> str:
"""Resolve the workspace role from the ACTIVE PROFILE alone (#540).
Unlike :func:`_effective_workspace_role`, this never consults
``_preflight_resolved_role``. A task capability whose ``required_role_kind``
is ``author`` (e.g. ``comment_issue``) stamps ``_preflight_resolved_role =
"author"``; the #274/#475 role exemptions must key off the real profile
identity so that stamp cannot poison a genuine reviewer/merger/reconciler
session into being treated as an author. Author profiles still classify as
``author`` here, so author blocking is preserved.
"""
profile = get_profile()
role = _role_kind(
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
return nwb.normalize_role_kind(
role,
profile_name=profile.get("profile_name"),
)
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
role = _effective_workspace_role()
@@ -539,11 +561,21 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
"""#274: author file/branch mutations must run from a branches/ worktree.
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
and merger metadata mutations must not require ``GITEA_AUTHOR_WORKTREE``
is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
(#468, #483). Non-author namespaces use dedicated workspace env vars (#510).
The exemption honours BOTH the effective workspace role and the actual
profile role (#540). ``comment_issue`` preflight stamps
``_preflight_resolved_role = "author"`` (its ``required_role_kind``), which
would otherwise poison :func:`_effective_workspace_role` into classifying a
genuine reconciler as an author and defeat this exemption. Keying off the
real profile role as well preserves the exemption without weakening author
blocking — an actual author profile classifies as ``author`` in both.
"""
role = _effective_workspace_role()
if role in nwb.NON_AUTHOR_ROLES:
if (
_effective_workspace_role() in nwb.NON_AUTHOR_ROLES
or _actual_profile_role() in nwb.NON_AUTHOR_ROLES
):
return
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
@@ -651,6 +683,7 @@ def verify_preflight_purity(
f"{_format_preflight_files(reviewer_delta)}"
)
_enforce_root_checkout_guard(worktree_path)
_enforce_branches_only_author_mutation(worktree_path)
_clear_preflight_capability_state()
@@ -693,6 +726,28 @@ def _verify_role_mutation_workspace(
verify_preflight_purity(remote, worktree_path=resolved, task=task)
return resolved
def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
"""#475: fail closed when the stable control checkout is contaminated."""
ctx = _resolve_author_mutation_context(worktree_path)
canonical_root = ctx["canonical_repo_root"]
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path=workspace,
canonical_repo_root=canonical_root,
current_branch=git_state.get("current_branch"),
head_sha=git_state.get("head_sha"),
porcelain_status=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha,
resolved_role=_preflight_resolved_role,
actual_role=_actual_profile_role(),
)
if assessment["block"]:
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
from mcp.server.fastmcp import FastMCP # noqa: E402
from gitea_auth import ( # noqa: E402
@@ -714,6 +769,7 @@ import role_session_router # noqa: E402
import role_namespace_gate # noqa: E402
import task_capability_map # noqa: E402
import review_proofs # noqa: E402
import review_workflow_boundary # noqa: E402
import review_workflow_load # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
@@ -724,16 +780,21 @@ import stacked_pr_support # noqa: E402
import merge_approval_gate # noqa: E402
import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import root_checkout_guard # noqa: E402
import remote_repo_guard # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import reviewer_pr_lease # noqa: E402
import merger_lease_adoption # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import worktree_cleanup_audit # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
import audit_reconciliation_mode # noqa: E402
import review_merge_state_machine # noqa: E402
import pr_work_lease # noqa: E402
import native_mcp_preference # noqa: E402
import worktree_cleanup_audit # noqa: E402
# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -1135,11 +1196,49 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
profile = REMOTES[remote]
return (
host or profile["host"],
org or profile["org"],
repo or profile["repo"],
resolved_host = host or profile["host"]
resolved_org = org or profile["org"]
resolved_repo = repo or profile["repo"]
_enforce_remote_repo_guard(
remote,
resolved_org,
resolved_repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
return (resolved_host, resolved_org, resolved_repo)
def _enforce_remote_repo_guard(
remote: str,
resolved_org: str,
resolved_repo: str,
*,
org_explicit: bool,
repo_explicit: bool,
) -> None:
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530).
Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
set, so the unit suite (which calls tools with bare remotes against mocked APIs)
is unaffected. In production it protects every read/lookup/mutation tool because
they all resolve targets through :func:`_resolve`.
"""
if "pytest" in sys.modules and not os.environ.get(
"GITEA_FORCE_REMOTE_REPO_CHECK"
):
return
local_remote_url = _local_git_remote_url(remote)
assessment = remote_repo_guard.assess_remote_repo_match(
remote=remote,
resolved_org=resolved_org,
resolved_repo=resolved_repo,
local_remote_url=local_remote_url,
org_explicit=org_explicit,
repo_explicit=repo_explicit,
)
if assessment["block"]:
raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
def _auth(host: str) -> str:
@@ -2394,10 +2493,18 @@ def _review_decision_session_reasons(lock: dict | None) -> list[str]:
return reasons
def init_review_decision_lock(remote: str | None, task: str | None):
def init_review_decision_lock(remote: str | None, task: str | None, force: bool = True):
"""Seed read-only-until-ready state for reviewer PR review tasks."""
if task != "review_pr":
return
if not force:
lock = _load_review_decision_lock()
if lock is not None:
if lock.get("remote") == remote and lock.get("session_pid") == os.getpid():
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = (lock.get("session_profile_lock") or "").strip()
if not env_lock or not stored_lock or env_lock == stored_lock:
return
review_workflow_load.clear_review_workflow_load()
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip()
@@ -2790,6 +2897,85 @@ def gitea_get_pr_review_feedback(
}
def _fetch_pr_lease_comments_safe(
pr_number: int,
*,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
limit: int = 100,
require_open: bool = False,
) -> dict:
"""Fetch PR thread comments with structured fail-closed errors (#519)."""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
resolved_repo = f"{o}/{r}"
pr_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
try:
pr = api_request("GET", pr_url, auth)
except RuntimeError as exc:
return {
"success": False,
"comments": [],
"reasons": [
"PR lookup failed before conflict-fix push assessment "
f"(pr_number={pr_number}, repo={resolved_repo}, remote={remote}): "
f"{_redact(str(exc))}"
],
"pr_lookup": "failed",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
}
pr_state = (pr.get("state") or "").strip().lower()
if require_open and pr_state != "open":
return {
"success": False,
"comments": [],
"reasons": [
f"PR #{pr_number} on {resolved_repo} is not open "
f"(state={pr_state or 'unknown'})"
],
"pr_lookup": "not_open",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
"head_sha": (pr.get("head") or {}).get("sha"),
}
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
try:
comments = api_request("GET", api, auth)
except RuntimeError as exc:
return {
"success": False,
"comments": [],
"reasons": [
"PR comment fetch failed during conflict-fix push assessment "
f"(pr_number={pr_number}, issue_index={pr_number}, "
f"repo={resolved_repo}, remote={remote}): "
f"{_redact(str(exc))}"
],
"pr_lookup": "ok",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
"head_sha": (pr.get("head") or {}).get("sha"),
}
if not isinstance(comments, list):
comments = []
return {
"success": True,
"comments": list(comments[:limit]),
"reasons": [],
"pr_lookup": "ok",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
"head_sha": (pr.get("head") or {}).get("sha"),
}
def _list_pr_lease_comments(
pr_number: int,
*,
@@ -2799,7 +2985,15 @@ def _list_pr_lease_comments(
repo: str | None,
limit: int = 100,
) -> list[dict]:
"""Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
"""Fetch PR/issue thread comments used for reviewer/conflict-fix leases.
Intentionally does **not** pre-fetch the PR via GET /pulls/{n}. Shared
callers (merge approval feedback, reviewer lease gates) depend on a single
comments GET so mock sequences and fail-open #485 non-list handling stay
stable. Structured PR-lookup failures belong only to
:func:`_fetch_pr_lease_comments_safe` used by conflict-fix push assessment
(#519).
"""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
@@ -4326,6 +4520,59 @@ def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> boo
raise
@mcp.tool()
def gitea_capture_branches_worktree_snapshot(
open_pr_branches: list[str] | None = None,
active_lock_branch: str | None = None,
leased_paths: list[str] | None = None,
worktree_path: str | None = None,
) -> dict:
"""Read-only: capture ``branches/`` and worktree audit snapshot (#404)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
root = PROJECT_ROOT
if worktree_path:
root = os.path.realpath(os.path.abspath(worktree_path))
git_root = _get_git_root(root)
if git_root:
root = git_root
snapshot = worktree_cleanup_audit.capture_branches_worktree_snapshot(
root,
open_pr_branches=open_pr_branches,
active_lock_branch=active_lock_branch,
leased_paths=leased_paths,
)
return {"success": True, "snapshot": snapshot}
@mcp.tool()
def gitea_assess_worktree_cleanup_integrity(
before_snapshot: dict,
after_snapshot: dict,
removals: list[dict] | None = None,
explained_missing: dict[str, str] | None = None,
) -> dict:
"""Read-only: reconcile cleanup before/after snapshots (#404)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"integrity_passed": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
return worktree_cleanup_audit.assess_worktree_cleanup_integrity(
before=before_snapshot,
after=after_snapshot,
removals=removals,
explained_missing=explained_missing,
)
@mcp.tool()
def gitea_reconcile_merged_cleanups(
dry_run: bool = True,
@@ -4677,6 +4924,90 @@ def gitea_scan_already_landed_open_prs(
}
@mcp.tool()
def gitea_audit_worktree_cleanup(
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS,
) -> dict:
"""Read-only: classify every session-owned worktree under ``branches/`` (#401).
Audits the local ``branches/`` directory, classifying each worktree as
active open PR, active issue work, dirty local, clean stale removable,
detached review leftover, or unsafe/unknown. Open PR branch heads are
fetched live so a worktree tied to an open PR is never marked removable;
the active issue-lock branch is read from the local lock file and treated
as active work. Deletes nothing and mutates no Gitea state.
Fails closed if the live open-PR list cannot be fetched: without it,
removability cannot be proven, so no candidates are returned.
Args:
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
ttl_hours: Age (hours) after which a clean issue/conflict-fix
worktree becomes stale-removable (default from
GITEA_WORKTREE_TTL_HOURS).
Returns:
dict with per-worktree classifications, counts, removable
candidates, and the ``git worktree list`` verification proof.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"performed": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
try:
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
open_prs = api_get_all(f"{repo_api_url(h, o, r)}/pulls?state=open", auth)
except Exception as exc:
return {
"success": False,
"performed": False,
"open_pr_state_verified": False,
"reasons": [
"could not fetch live open PRs; removability unverified "
f"(fail closed): {_redact(str(exc))}"
],
}
open_pr_branches = {
str((pr.get("head") or {}).get("ref"))
for pr in open_prs
if (pr.get("head") or {}).get("ref")
}
active_issue_branches: set[str] = set()
lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE)
if lock and lock.get("branch_name"):
active_issue_branches.add(str(lock["branch_name"]).strip())
report = worktree_cleanup_audit.audit_branches_directory(
PROJECT_ROOT,
open_pr_branches=open_pr_branches,
active_issue_branches=active_issue_branches,
now=datetime.now(timezone.utc),
ttl_hours=ttl_hours,
)
return {
"success": True,
"performed": False,
"open_pr_state_verified": True,
"task_mode": "work-issue",
**report,
}
@mcp.tool()
def gitea_reconcile_already_landed_pr(
pr_number: int,
@@ -5374,7 +5705,10 @@ def gitea_acquire_reviewer_pr_lease(
"target_branch_sha": target_branch_sha,
"repo": repo_label,
"comment_id": posted.get("id"),
})
}, lease_provenance=merger_lease_adoption.build_lease_provenance(
source=merger_lease_adoption.SOURCE_ACQUIRE,
comment_id=posted.get("id"),
))
return {
"success": True,
"acquired": True,
@@ -5386,6 +5720,175 @@ def gitea_acquire_reviewer_pr_lease(
}
@mcp.tool()
def gitea_adopt_merger_pr_lease(
pr_number: int,
worktree: str,
expected_head_sha: str,
issue_number: int | None = None,
session_id: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Adopt an active reviewer PR lease for a merger session (#536).
Merger sessions must not manually seed ``reviewer_pr_lease._SESSION_LEASE``.
This tool posts durable adoption proof on the PR thread and records
sanctioned in-session lease provenance so ``gitea_merge_pr`` can proceed
after a separate reviewer session held the lease.
Args:
pr_number: Open PR to adopt.
worktree: Merger workspace under ``branches/``.
expected_head_sha: Approved head the merger will pin during merge.
issue_number: Optional linked issue for the adoption record.
session_id: Optional merger session id (generated when omitted).
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict with adoption proof, session lease, and block reasons when the
guarded preconditions are not met.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"adopted": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
return {
"success": False,
"adopted": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
merge_block = _profile_operation_gate("gitea.pr.merge")
if merge_block:
return {
"success": False,
"adopted": False,
"reasons": merge_block,
"permission_report": _permission_block_report("gitea.pr.merge"),
}
_verify_role_mutation_workspace(
remote, worktree=worktree, task="adopt_merger_pr_lease"
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
profile_name = profile.get("profile_name") or "unknown"
identity = _authenticated_username(remote) or profile.get("username") or ""
sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
repo_label = f"{o}/{r}"
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
pr_state = (pr_live.get("state") or "").strip().lower()
pr_open = pr_state == "open" and not pr_live.get("merged")
live_head = (pr_live.get("head", {}) or {}).get("sha") or pr_live.get(
"head_sha"
)
feedback = gitea_get_pr_review_feedback(
pr_number=pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
)
approval_at_head = bool(feedback.get("approval_at_current_head"))
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
assessment = merger_lease_adoption.assess_adopt_merger_lease(
comments,
pr_number=pr_number,
adopter_identity=identity,
adopter_profile=profile_name,
adopter_session_id=sid,
repo=repo_label,
issue_number=issue_number,
worktree=worktree,
expected_head_sha=expected_head_sha,
live_head_sha=live_head,
approval_at_head=approval_at_head,
pr_open=pr_open,
)
if not assessment.get("adopt_allowed"):
return {
"success": False,
"adopted": False,
"pr_number": pr_number,
"reasons": assessment.get("reasons") or [],
"active_lease": assessment.get("active_lease"),
"expected_head_sha": expected_head_sha,
"live_head_sha": live_head,
}
active = assessment.get("active_lease") or {}
body = assessment["adoption_body"]
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "adopt_merger_pr_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
provenance = merger_lease_adoption.build_lease_provenance(
source=merger_lease_adoption.SOURCE_ADOPT,
comment_id=posted.get("id"),
adopted_from_session_id=active.get("session_id"),
adopted_from_profile=active.get("profile"),
adopted_from_reviewer_identity=active.get("reviewer_identity"),
adoption_reason=merger_lease_adoption.DEFAULT_ADOPTION_REASON,
)
session_lease = reviewer_pr_lease.record_session_lease({
"pr_number": pr_number,
"issue_number": issue_number or active.get("issue_number"),
"session_id": sid,
"reviewer_identity": identity,
"profile": profile_name,
"worktree": worktree,
"phase": "adopted",
"candidate_head": live_head,
"target_branch": active.get("target_branch") or "master",
"target_branch_sha": active.get("target_branch_sha"),
"repo": repo_label,
"comment_id": posted.get("id"),
}, lease_provenance=provenance)
return {
"success": True,
"adopted": True,
"pr_number": pr_number,
"session_id": sid,
"adoption_comment_id": posted.get("id"),
"adopted_from_session_id": active.get("session_id"),
"adopted_from_profile": active.get("profile"),
"adopted_from_reviewer_identity": active.get("reviewer_identity"),
"adoption_reason": merger_lease_adoption.DEFAULT_ADOPTION_REASON,
"expected_head_sha": expected_head_sha,
"live_head_sha": live_head,
"session_lease": session_lease,
"lease_provenance": provenance,
"reasons": [],
}
@mcp.tool()
def gitea_heartbeat_reviewer_pr_lease(
pr_number: int,
@@ -5446,6 +5949,17 @@ def gitea_heartbeat_reviewer_pr_lease(
):
posted = api_request("POST", comment_url, auth, {"body": body})
prior_provenance = (session.get("lease_provenance") or {}).copy()
heartbeat_provenance = merger_lease_adoption.build_lease_provenance(
source=merger_lease_adoption.SOURCE_HEARTBEAT,
comment_id=posted.get("id"),
adopted_from_session_id=prior_provenance.get("adopted_from_session_id"),
adopted_from_profile=prior_provenance.get("adopted_from_profile"),
adopted_from_reviewer_identity=prior_provenance.get(
"adopted_from_reviewer_identity"
),
adoption_reason=prior_provenance.get("adoption_reason"),
)
updated = reviewer_pr_lease.record_session_lease({
**session,
"phase": phase,
@@ -5453,7 +5967,8 @@ def gitea_heartbeat_reviewer_pr_lease(
"candidate_head": candidate_head or session.get("candidate_head"),
"target_branch_sha": target_branch_sha or session.get("target_branch_sha"),
"last_comment_id": posted.get("id"),
})
"comment_id": posted.get("id"),
}, lease_provenance=heartbeat_provenance)
return {
"success": True,
"posted": True,
@@ -5614,6 +6129,118 @@ def gitea_cleanup_post_merge_moot_lease(
return report
@mcp.tool()
def gitea_release_reviewer_pr_lease(
pr_number: int,
worktree: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Safely release an active reviewer lease owned by the current session on an open PR.
This provides a canonical way to clean up reviewer leases when a review fails
before submission.
Args:
pr_number: The PR number whose reviewer lease to release.
worktree: Optional worktree path (resolves automatically if not supplied).
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict reporting release status.
"""
_verify_role_mutation_workspace(
remote, worktree=worktree, task="review_pr"
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo
)
active = reviewer_pr_lease.find_active_reviewer_lease(
comments, pr_number=pr_number
)
if not active:
return {
"success": True,
"released": False,
"reasons": ["no active reviewer lease found on PR"],
}
identity = _authenticated_username(remote) or ""
session = reviewer_pr_lease.get_session_lease() or {}
session_id = session.get("session_id")
owner_identity = active.get("reviewer_identity")
owner_session_id = active.get("session_id")
# Authorize release: identity matches, session_id matches, or lease is expired/reclaimable
authorized = False
reasons = []
if owner_identity and identity and owner_identity == identity:
authorized = True
elif owner_session_id and session_id and owner_session_id == session_id:
authorized = True
else:
freshness = reviewer_pr_lease.classify_lease_freshness(active)
if freshness in {"expired", "reclaimable"}:
authorized = True
if not authorized:
return {
"success": False,
"released": False,
"reasons": [
f"unauthorized to release active lease: owned by {owner_identity} "
f"(session {owner_session_id})"
],
}
# Format and post release comment
body = reviewer_pr_lease.format_lease_body(
repo=active.get("repo") or f"{o}/{r}",
pr_number=pr_number,
issue_number=active.get("issue_number"),
reviewer_identity=owner_identity or identity,
profile=active.get("profile") or "reviewer",
session_id=owner_session_id or session_id or "unknown",
worktree=active.get("worktree") or "",
phase="released",
candidate_head=active.get("candidate_head"),
target_branch=active.get("target_branch") or "master",
target_branch_sha=active.get("target_branch_sha"),
blocker="manual-release",
)
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "release_reviewer_pr_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
reviewer_pr_lease.clear_session_lease()
return {
"success": True,
"released": True,
"comment_id": posted.get("id"),
"reasons": [],
}
@mcp.tool()
def gitea_list_issue_comments(
issue_number: int,
@@ -5685,6 +6312,7 @@ def gitea_create_issue_comment(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Post a markdown comment to a Gitea issue's discussion thread.
@@ -5706,6 +6334,7 @@ def gitea_create_issue_comment(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional path to verify branches-only guard.
Returns:
dict with 'success', 'comment_id', and 'issue_number' ('url' only
@@ -5714,7 +6343,7 @@ def gitea_create_issue_comment(
(permission blocks also carry a structured 'permission_report',
#142).
"""
verify_preflight_purity(remote, task="comment_issue")
verify_preflight_purity(remote, worktree_path=worktree_path, task="comment_issue")
gate_reasons = _profile_operation_gate("gitea.issue.comment")
reasons = list(gate_reasons)
if not (body or "").strip():
@@ -6958,14 +7587,42 @@ def gitea_get_runtime_context(
return result
@mcp.tool()
def gitea_record_pre_review_command(
command: str,
cwd: str | None = None,
classification: str | None = None,
) -> dict:
"""Classify and record a command executed before workflow load (#403).
Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands
may be recorded as allowed; boundary violations block reviewer mutations.
"""
recorded = review_workflow_boundary.record_pre_review_command(
command,
cwd=cwd,
project_root=PROJECT_ROOT,
classification=classification,
)
boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT)
return {
"success": True,
"recorded": recorded,
"boundary_status": boundary_state.get("boundary_status"),
"boundary_clean": boundary_state.get("boundary_clean"),
"reasons": list(boundary_state.get("reasons") or []),
}
@mcp.tool()
def gitea_load_review_workflow(
prompt_text: str | None = None,
) -> dict:
"""Load and record canonical review-merge workflow proof for this session (#389).
"""Load and record canonical review-merge workflow proof for this session (#389, #403).
Read-only with respect to Gitea API; records in-process workflow source/hash
proof required before reviewer review or merge mutations.
proof and session boundary state required before reviewer review or merge
mutations.
"""
try:
recorded = review_workflow_load.record_review_workflow_load(
@@ -6977,8 +7634,11 @@ def gitea_load_review_workflow(
"reasons": [str(exc)],
"recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
}
boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT)
helper = review_workflow_boundary.workflow_load_helper_result(
recorded, PROJECT_ROOT)
return {
"success": True,
"success": not boundary_reasons,
"loaded": True,
"workflow_source": recorded["workflow_source"],
"task_mode": recorded["task_mode"],
@@ -6990,7 +7650,14 @@ def gitea_load_review_workflow(
"prompt_conflicts_with_workflow"],
"prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
"workflow_load_proof_present": True,
"reasons": [],
"boundary_status": recorded.get("boundary_status"),
"boundary_clean": recorded.get("boundary_clean"),
"workflow_load_helper_result": helper,
"reasons": boundary_reasons,
"recovery_handoff": (
review_workflow_load.recovery_handoff_without_replay()
if boundary_reasons else []
),
}
@@ -7485,22 +8152,39 @@ def gitea_assess_conflict_fix_push(
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comments = _list_pr_lease_comments(
fetched = _fetch_pr_lease_comments_safe(
pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
require_open=True,
)
return pr_work_lease.assess_conflict_fix_push(
if not fetched["success"]:
return {
"push_allowed": False,
"block": True,
"reasons": fetched["reasons"],
"pr_lookup": fetched.get("pr_lookup"),
"resolved_repo": fetched.get("resolved_repo"),
"remote": fetched.get("remote"),
"pr_number": pr_number,
"assessment_failed": True,
}
assessment = pr_work_lease.assess_conflict_fix_push(
pr_number=pr_number,
comments=comments,
comments=fetched["comments"],
branch_head_before=branch_head_before,
branch_head_after=branch_head_after,
worktree_path=worktree_path,
push_cwd=push_cwd,
is_fast_forward=is_fast_forward,
)
assessment["pr_lookup"] = fetched.get("pr_lookup")
assessment["resolved_repo"] = fetched.get("resolved_repo")
assessment["remote"] = fetched.get("remote")
assessment["live_pr_head_sha"] = fetched.get("head_sha")
return assessment
@mcp.tool()
@@ -8138,6 +8822,7 @@ def gitea_resolve_task_capability(
init_review_decision_lock(
remote if remote in REMOTES else None,
task,
force=False,
)
record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)