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 feat/issue-483-role-boundaries and preserve PR intent
alongside compatible master guardrails.
This commit is contained in:
sysadmin
2026-07-08 11:10:15 -04:00
parent 9e29d00292 6b45212659
commit 3b4f222a7d
27 changed files with 3099 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+460 -45
View File
@@ -175,12 +175,76 @@ _preflight_reviewer_violation_files: list[str] = []
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
import namespace_workspace_binding as nwb # noqa: E402
def _preflight_in_test_mode() -> bool:
return "pytest" in sys.modules or "unittest" in sys.modules
def _reviewer_session_worktree() -> str | None:
import reviewer_pr_lease as _reviewer_pr_lease
session = _reviewer_pr_lease.get_session_lease()
if not session:
return None
worktree = (session.get("worktree") or "").strip()
return worktree or None
def _effective_workspace_role() -> str:
"""Resolve the namespace key used for workspace binding (#510)."""
profile = get_profile()
role = _preflight_resolved_role
if not role:
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()
workspace, _source = nwb.resolve_namespace_workspace(
role_kind=role,
worktree_path=worktree_path,
process_project_root=PROJECT_ROOT,
session_lease_worktree=(
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
),
profile_name=get_profile().get("profile_name"),
)
return workspace
def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
"""Canonical namespace workspace + repository root for guards (#460/#510)."""
role = _effective_workspace_role()
return nwb.resolve_namespace_mutation_context(
role_kind=role,
worktree_path=worktree_path,
process_project_root=PROJECT_ROOT,
session_lease_worktree=(
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
),
profile_name=get_profile().get("profile_name"),
)
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
"""Backward-compatible alias for namespace workspace context."""
return _resolve_namespace_mutation_context(worktree_path)
def _ensure_process_start_porcelain() -> str:
"""Capture the shared-worktree baseline once per MCP process (#252)."""
global _process_start_porcelain
@@ -189,26 +253,6 @@ def _ensure_process_start_porcelain() -> str:
return _process_start_porcelain
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the workspace root inspected by pre-flight guards."""
return author_mutation_worktree.resolve_mutation_workspace(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
"""Canonical workspace + repository root for runtime_context and guards (#460)."""
return author_mutation_worktree.resolve_author_mutation_context(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
def _get_git_root(path: str) -> str | None:
try:
res = subprocess.run(
@@ -276,7 +320,7 @@ def _format_preflight_files(files: list[str]) -> str:
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
ctx = _resolve_author_mutation_context(worktree_path)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
inspected_root = _get_git_root(workspace)
process_root = ctx["process_project_root"]
@@ -294,6 +338,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
"dirty_files": list(dirty_files),
"dirty_scope": dirty_scope,
"workspace_roots_aligned": ctx["roots_aligned"],
"workspace_role_kind": ctx.get("workspace_role_kind"),
"workspace_binding_source": ctx.get("workspace_binding_source"),
"ignored_bindings": list(ctx.get("ignored_bindings") or []),
}
if not ctx["roots_aligned"]:
details["workspace_root_mismatch"] = (
@@ -304,13 +351,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
def _format_preflight_workspace_details(details: dict) -> str:
return (
f"MCP server process root: {details.get('mcp_server_process_root')}; "
f"active task workspace root: {details.get('active_task_workspace_root')}; "
f"inspected git root: {details.get('inspected_git_root')}; "
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
f"dirty scope: {details.get('dirty_scope')}"
)
parts = [
f"MCP server process root: {details.get('mcp_server_process_root')}",
f"active task workspace root: {details.get('active_task_workspace_root')}",
f"workspace role: {details.get('workspace_role_kind')}",
f"binding source: {details.get('workspace_binding_source')}",
f"inspected git root: {details.get('inspected_git_root')}",
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}",
f"dirty scope: {details.get('dirty_scope')}",
]
ignored = details.get("ignored_bindings") or []
if ignored:
parts.append(f"ignored foreign bindings: {'; '.join(ignored)}")
return "; ".join(parts)
def assess_preflight_status(worktree_path: str | None = None) -> dict:
@@ -333,6 +386,25 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
"Active task workspace has tracked file edits before mutation "
f"({_format_preflight_workspace_details(workspace_details)})"
)
role = _effective_workspace_role()
if role in nwb.NON_AUTHOR_ROLES:
binding = nwb.assess_metadata_only_worktree_binding(
role_kind=role,
declared_worktree_path=worktree_path,
mutation_workspace=_resolve_preflight_workspace_path(worktree_path),
process_project_root=PROJECT_ROOT,
profile_name=get_profile().get("profile_name"),
)
if binding.get("block"):
reasons.append(binding["reasons"][0])
reasons.append(
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=binding["mutation_workspace"],
binding_source="MCP server process root (default)",
reasons=binding.get("reasons"),
)
)
return {
"preflight_ready": not reasons,
"preflight_block_reasons": reasons,
@@ -466,13 +538,14 @@ def record_preflight_check(
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
"""#274: author file/branch mutations must run from a branches/ worktree.
Reviewer, reconciler, and merger roles are exempt: reconciler ``close_pr``
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
and merger metadata mutations must not require ``GITEA_AUTHOR_WORKTREE``
(#468, #483).
(#468, #483). Non-author namespaces use dedicated workspace env vars (#510).
"""
if _preflight_resolved_role in ("reviewer", "reconciler", "merger"):
role = _effective_workspace_role()
if role in nwb.NON_AUTHOR_ROLES:
return
ctx = _resolve_author_mutation_context(worktree_path)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
assessment = author_mutation_worktree.assess_author_mutation_worktree(
@@ -520,11 +593,12 @@ def verify_preflight_purity(
f"'{task}' (fail closed)"
)
ctx = _resolve_author_mutation_context(worktree_path)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
canonical_root = ctx["canonical_repo_root"]
process_root = ctx["process_project_root"]
real_workspace = os.path.realpath(workspace)
role = ctx.get("workspace_role_kind") or _effective_workspace_role()
if real_workspace != process_root:
if not _preflight_in_test_mode():
@@ -541,11 +615,15 @@ def verify_preflight_purity(
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
if dirty_files:
details = _preflight_workspace_details(workspace, dirty_files)
raise RuntimeError(
"Pre-flight order violation: Active task workspace has tracked "
"file edits before mutation (fail closed). "
f"{_format_preflight_workspace_details(details)}"
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=workspace,
binding_source=ctx.get("workspace_binding_source")
or "unknown binding source",
dirty_files=dirty_files,
ignored_bindings=ctx.get("ignored_bindings"),
)
)
else:
if _preflight_whoami_violation:
@@ -561,14 +639,14 @@ def verify_preflight_purity(
f"{_format_preflight_files(_preflight_capability_violation_files)}"
)
if _preflight_resolved_role == "reviewer":
if role in {"reviewer", "merger"}:
current = _get_workspace_porcelain()
baseline = _preflight_capability_baseline_porcelain or ""
reviewer_delta = _new_tracked_changes_since(baseline, current)
_preflight_reviewer_violation_files = reviewer_delta
if reviewer_delta:
raise RuntimeError(
"Reviewer role violation: Reviewer profile is forbidden from modifying "
f"{role.title()} role violation: profile is forbidden from modifying "
"tracked workspace files (fail closed). Offending files: "
f"{_format_preflight_files(reviewer_delta)}"
)
@@ -576,6 +654,45 @@ def verify_preflight_purity(
_enforce_branches_only_author_mutation(worktree_path)
_clear_preflight_capability_state()
def _verify_role_mutation_workspace(
remote: str | None = None,
*,
worktree_path: str | None = None,
worktree: str | None = None,
task: str | None = None,
) -> str:
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
role = _effective_workspace_role()
git_state = issue_lock_worktree.read_worktree_git_state(
_resolve_preflight_workspace_path(worktree_path)
)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=role,
worktree_path=worktree_path,
worktree=worktree,
process_project_root=PROJECT_ROOT,
session_lease_worktree=(
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
),
profile_name=get_profile().get("profile_name"),
current_branch=git_state.get("current_branch"),
)
if assessment["block"]:
raise RuntimeError(
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=assessment["mutation_workspace"],
binding_source=assessment.get("workspace_binding_source")
or "unknown binding source",
reasons=assessment.get("reasons"),
ignored_bindings=assessment.get("ignored_bindings"),
)
)
resolved = assessment["mutation_workspace"]
verify_preflight_purity(remote, worktree_path=resolved, task=task)
return resolved
from mcp.server.fastmcp import FastMCP # noqa: E402
from gitea_auth import ( # noqa: E402
@@ -597,6 +714,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_load # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import issue_lock_provenance # noqa: E402
@@ -612,6 +730,7 @@ import reviewer_pr_lease # noqa: E402
import merged_cleanup_reconcile # 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
@@ -1329,7 +1448,6 @@ def gitea_create_issue(
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
@mcp.tool()
def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
"""Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484)."""
try:
@@ -1340,6 +1458,7 @@ def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
)
@mcp.tool()
def gitea_lock_issue(
issue_number: int,
branch_name: str,
@@ -2279,6 +2398,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
"""Seed read-only-until-ready state for reviewer PR review tasks."""
if task != "review_pr":
return
review_workflow_load.clear_review_workflow_load()
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip()
session_lock = (
@@ -2305,6 +2425,11 @@ def init_review_decision_lock(remote: str | None, task: str | None):
})
def _review_workflow_load_gate_reasons() -> list[str]:
"""Fail closed when canonical review workflow was not loaded (#389)."""
return review_workflow_load.review_workflow_load_blockers(PROJECT_ROOT)
def check_review_decision_gate(
pr_number: int,
action: str,
@@ -2315,7 +2440,10 @@ def check_review_decision_gate(
repo: str | None = None,
) -> list[str]:
"""Fail closed unless validation completed and the final decision is ready."""
reasons = []
reasons = list(_review_workflow_load_gate_reasons())
if reasons:
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
return reasons
lock = _load_review_decision_lock()
if lock is None:
reasons.append(
@@ -2723,10 +2851,14 @@ def _evaluate_pr_review_submission(
*,
live: bool,
final_review_decision_ready: bool = False,
worktree_path: str | None = None,
) -> dict:
"""Shared gate chain for live submit and dry-run review tools."""
verify_preflight_purity(remote, task="review_pr")
_verify_role_mutation_workspace(
remote, worktree_path=worktree_path, task="review_pr"
)
action = (action or "").strip().lower()
workflow_blockers = _review_workflow_load_gate_reasons() if live else []
result = {
"requested_action": action,
"performed": False,
@@ -2742,6 +2874,10 @@ def _evaluate_pr_review_submission(
"reasons": [],
}
reasons = result["reasons"]
if workflow_blockers:
reasons.extend(workflow_blockers)
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
return result
if action not in _REVIEW_ACTIONS:
reasons.append(
@@ -2960,6 +3096,13 @@ def gitea_mark_final_review_decision(
}
org = resolved_org
repo = resolved_repo
workflow_blockers = _review_workflow_load_gate_reasons()
if workflow_blockers:
return {
"marked_ready": False,
"reasons": workflow_blockers + (
review_workflow_load.recovery_handoff_without_replay()),
}
hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
if hard_stop:
return {"marked_ready": False, "reasons": hard_stop}
@@ -3125,6 +3268,7 @@ def gitea_dry_run_pr_review(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Validate review submission mechanics without a live PR mutation."""
return _evaluate_pr_review_submission(
@@ -3137,6 +3281,7 @@ def gitea_dry_run_pr_review(
org=org,
repo=repo,
live=False,
worktree_path=worktree_path,
)
@@ -3152,6 +3297,7 @@ def gitea_submit_pr_review(
org: str | None = None,
repo: str | None = None,
final_review_decision_ready: bool = False,
worktree_path: str | None = None,
) -> dict:
"""Gated PR review mutation: comment findings, request changes, or approve.
@@ -3170,6 +3316,7 @@ def gitea_submit_pr_review(
repo=repo,
live=True,
final_review_decision_ready=final_review_decision_ready,
worktree_path=worktree_path,
)
@@ -3530,6 +3677,7 @@ def gitea_merge_pr(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Gated merge of a Gitea pull request (#16).
@@ -3576,6 +3724,10 @@ def gitea_merge_pr(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Merger workspace path under ``branches/`` or clean
control checkout; defaults to ``GITEA_MERGER_WORKTREE``,
``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores
foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510).
Returns:
dict describing the attempt: performed, authenticated user, profile
@@ -3583,12 +3735,18 @@ def gitea_merge_pr(
reasons/gates passed or blocked, and merge result / merge commit if
available. Never secrets.
"""
verify_preflight_purity(remote, task="merge_pr")
_verify_role_mutation_workspace(
remote, worktree_path=worktree_path, task="merge_pr"
)
allowed = get_profile()["allowed_operations"]
forbidden = get_profile()["forbidden_operations"]
active_role = _role_kind(allowed, forbidden)
if active_role == "reviewer":
raise RuntimeError("Reviewer profile cannot merge. Use a merger profile/session after formal approval at current head.")
raise RuntimeError(
"Reviewer profile cannot merge. Use a merger profile/session "
"after formal approval at current head."
)
workflow_blockers = _review_workflow_load_gate_reasons()
do = (do or "").strip().lower()
result = {
"performed": False,
@@ -3610,6 +3768,10 @@ def gitea_merge_pr(
"explicit_operator_auth": confirmation,
}
reasons = result["reasons"]
if workflow_blockers:
reasons.extend(workflow_blockers)
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
return result
# Gate 1 — valid merge method (no API call on a bad method).
if do not in _MERGE_METHODS:
@@ -4121,6 +4283,18 @@ def gitea_delete_branch(
"permission_report": _permission_block_report("gitea.branch.delete"),
}
audit_allowed, audit_reasons = (
audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
)
if not audit_allowed:
return {
"success": False,
"performed": False,
"required_permission": "gitea.branch.delete",
"reasons": audit_reasons,
"audit_phase": audit_reconciliation_mode.current_phase(),
}
verify_preflight_purity(remote, task="delete_branch")
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
@@ -4190,6 +4364,18 @@ def gitea_reconcile_merged_cleanups(
"execute_confirmed must be True when dry_run=False (fail closed)"
)
if not dry_run:
exec_allowed, exec_reasons = (
audit_reconciliation_mode.check_cleanup_execution_allowed()
)
if not exec_allowed:
return {
"success": False,
"performed": False,
"reasons": exec_reasons,
"audit_phase": audit_reconciliation_mode.current_phase(),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
@@ -4272,6 +4458,53 @@ def gitea_reconcile_merged_cleanups(
return {"success": True, "performed": True, **report}
@mcp.tool()
def gitea_authorize_reconciliation_cleanup_phase(
operator_approved: bool = False,
workflow_authorized: bool = False,
delete_capability_proven: bool = False,
safe_to_delete_remote: bool = False,
safe_to_remove_worktree: bool = False,
before_state: str = "",
after_state: str = "",
) -> dict:
"""Authorize cleanup phase after audit-only reconciliation (#419).
Requires operator or workflow approval, exact delete_branch capability proof,
branch/worktree safety proof, and before/after state snapshots. Audit phase
forbids branch deletion, worktree removal, pushes, and issue/PR mutations.
"""
delete_gate = _profile_operation_gate("gitea.branch.delete")
capability_ok = not bool(delete_gate)
if delete_capability_proven and delete_gate:
return {
"authorized": False,
"performed": False,
"delete_capability_verified": False,
"reasons": [
"delete_capability_proven=true but active profile lacks "
"gitea.branch.delete",
] + delete_gate,
"audit_phase": audit_reconciliation_mode.current_phase(),
}
result = audit_reconciliation_mode.authorize_cleanup_phase(
operator_approved=operator_approved,
workflow_authorized=workflow_authorized,
delete_capability_proven=delete_capability_proven and capability_ok,
safety_proof={
"safe_to_delete_remote": safe_to_delete_remote,
"safe_to_remove_worktree": safe_to_remove_worktree,
},
before_after_snapshot={
"before": (before_state or "").strip(),
"after": (after_state or "").strip(),
},
)
result["performed"] = bool(result.get("authorized"))
result["delete_capability_verified"] = capability_ok
return result
@mcp.tool()
def gitea_assess_already_landed_reconciliation(
pr_number: int,
@@ -5075,7 +5308,7 @@ def gitea_acquire_reviewer_pr_lease(
"permission_report": _permission_block_report("gitea.pr.comment"),
}
verify_preflight_purity(remote, task="review_pr")
_verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
@@ -5085,6 +5318,13 @@ def gitea_acquire_reviewer_pr_lease(
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
# Refuse merge-oriented lease acquisition/adoption on an already-merged or
# closed PR: the lease is moot and adopting it for merge work is unsafe (#515).
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
pr_merged_or_closed = bool(
pr_live.get("merged") or pr_live.get("merged_at")
) or (str(pr_live.get("state") or "").strip().lower() == "closed")
assessment = reviewer_pr_lease.assess_acquire_lease(
comments,
pr_number=pr_number,
@@ -5097,6 +5337,7 @@ def gitea_acquire_reviewer_pr_lease(
candidate_head=candidate_head,
target_branch=target_branch,
target_branch_sha=target_branch_sha,
pr_merged_or_closed=pr_merged_or_closed,
)
if not assessment.get("acquire_allowed"):
return {
@@ -5104,6 +5345,7 @@ def gitea_acquire_reviewer_pr_lease(
"acquired": False,
"reasons": assessment.get("reasons") or [],
"existing_lease": assessment.get("existing_lease"),
"post_merge_moot": assessment.get("post_merge_moot", False),
}
body = assessment["lease_body"]
@@ -5252,6 +5494,126 @@ def gitea_assess_reviewer_pr_lease(
}
@mcp.tool()
def gitea_cleanup_post_merge_moot_lease(
pr_number: int,
apply: bool = False,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Safely resolve a moot reviewer lease left on an ALREADY-MERGED/closed PR (#515).
Read-first and fail-safe. This tool never merges and never adopts a lease.
It only acts when the live PR state is merged/closed, and it never steals or
force-cleans an active *foreign* lease while the PR is still open. When
``apply`` is true and a lease is still active on a merged/closed PR, it posts
a terminal ``phase: released`` lease marker (``blocker: post-merge-moot``) —
an append-only comment that neutralises the moot lease without deleting any
other session's comment.
Args:
pr_number: The PR whose lingering lease to assess/clean.
apply: When false (default) report only (read-only). When true, post the
terminal released marker if — and only if — cleanup is allowed.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict reporting PR merged/closed state, merge_commit_sha, linked-issue
closure state, whether the lease is moot, whether cleanup was performed
or skipped (and why), and ``no_merge_or_adoption`` True — this path never
merges or adopts.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"cleanup_performed": False,
"no_merge_or_adoption": True,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
assessment = reviewer_pr_lease.assess_post_merge_moot_lease(
comments,
pr_number=pr_number,
pr_merged=bool(pr_live.get("merged") or pr_live.get("merged_at")),
pr_state=pr_live.get("state"),
merge_commit_sha=pr_live.get("merge_commit_sha"),
)
# Best-effort linked-issue closure state for the report.
active = assessment.get("active_lease") or {}
issue_no = active.get("issue_number")
linked_issue_state = None
if issue_no:
try:
issue = api_request(
"GET", f"{repo_api_url(h, o, r)}/issues/{issue_no}", auth) or {}
linked_issue_state = issue.get("state")
except Exception:
linked_issue_state = None
report = {
"success": True,
"pr_number": pr_number,
"pr_state": assessment.get("pr_state"),
"pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
"merge_commit_sha": assessment.get("merge_commit_sha"),
"linked_issue_number": issue_no,
"linked_issue_state": linked_issue_state,
"lease_moot": assessment.get("is_moot"),
"active_lease": assessment.get("active_lease"),
"cleanup_allowed": assessment.get("cleanup_allowed"),
"cleanup_performed": False,
"no_merge_or_adoption": True,
"mode": "apply" if apply else "read_only",
"reasons": assessment.get("reasons") or [],
}
if not apply:
return report
if not assessment.get("cleanup_allowed"):
report["cleanup_skipped_reason"] = (
assessment.get("reasons") or ["cleanup not allowed"]
)
return report
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
report["success"] = False
report["reasons"] = comment_block
report["permission_report"] = _permission_block_report("gitea.pr.comment")
return report
verify_preflight_purity(remote)
body = assessment["release_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": "cleanup_post_merge_moot_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
report["cleanup_performed"] = True
report["released_comment_id"] = posted.get("id")
return report
@mcp.tool()
def gitea_list_issue_comments(
issue_number: int,
@@ -5653,6 +6015,8 @@ _PROJECT_SKILLS = {
"steps": [
"Resolve task first: gitea_resolve_task_capability(task='review_pr') "
"to confirm reviewer namespace and avoid author-profile blocks.",
"Load canonical workflow proof with gitea_load_review_workflow "
"before any review/merge mutation (#389).",
"Verify reviewer identity with gitea_whoami; the PR author "
"must be a different user.",
"Reconcile live queue state FIRST (do not trust prior handoffs): "
@@ -6584,6 +6948,8 @@ def gitea_get_runtime_context(
),
"role_kind": _role_kind(allowed, forbidden),
"shell_health": native_mcp_preference.shell_health_status(),
"workflow_load_proof": review_workflow_load.workflow_load_status(
PROJECT_ROOT),
}
if reveal and h:
@@ -6592,6 +6958,42 @@ def gitea_get_runtime_context(
return result
@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).
Read-only with respect to Gitea API; records in-process workflow source/hash
proof required before reviewer review or merge mutations.
"""
try:
recorded = review_workflow_load.record_review_workflow_load(
PROJECT_ROOT, prompt_text=prompt_text)
except OSError as exc:
return {
"success": False,
"loaded": False,
"reasons": [str(exc)],
"recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
}
return {
"success": True,
"loaded": True,
"workflow_source": recorded["workflow_source"],
"task_mode": recorded["task_mode"],
"workflow_hash": recorded["workflow_hash"],
"workflow_version": recorded["workflow_version"],
"final_report_schema_path": recorded["final_report_schema_path"],
"final_report_schema_hash": recorded["final_report_schema_hash"],
"prompt_conflicts_with_workflow": recorded[
"prompt_conflicts_with_workflow"],
"prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
"workflow_load_proof_present": True,
"reasons": [],
}
@mcp.tool()
def gitea_list_profiles() -> dict:
"""Read-only: list all Gitea MCP profiles with redacted metadata.
@@ -7760,7 +8162,20 @@ def gitea_resolve_task_capability(
}
if reason_msg:
result["reason"] = reason_msg
if task in ("review_pr", "merge_pr"):
result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
PROJECT_ROOT)
if not result["workflow_load_proof"].get("workflow_load_valid"):
guidance = (
"Call gitea_load_review_workflow before any reviewer review "
"or merge mutation."
)
if guidance not in task_role_guidance:
task_role_guidance.append(guidance)
role_session_router.sync_route_from_capability(result)
if audit_reconciliation_mode.check_audit_task_enters_phase(task):
phase_record = audit_reconciliation_mode.enter_audit_phase(task)
result["reconciliation_phase"] = phase_record.get("phase")
was_terminal = capability_stop_terminal.is_active()
terminal = capability_stop_terminal.sync_from_capability_result(result)
if terminal: