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

Merge branch 'master' into feat/issue-389-review-workflow-load-proof

Browse Source
This commit is contained in:
sysadmin
2026-07-07 12:05:17 -05:00
parent 8a8cffbdf5 1a1e679246
commit 0d41760ea3
55 changed files with 7203 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+722 -29
View File
@@ -21,6 +21,7 @@ import json
import functools
import contextlib
import subprocess
from datetime import datetime, timedelta, timezone
# Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
@@ -397,6 +398,28 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
_preflight_resolved_role = resolved_role
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
"""#274: author mutations must run from a branches/ session worktree."""
if _preflight_resolved_role == "reviewer":
return
workspace = 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),
)
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
assessment = author_mutation_worktree.assess_author_mutation_worktree(
workspace_path=workspace,
project_root=PROJECT_ROOT,
current_branch=git_state.get("current_branch"),
)
if assessment["block"]:
raise RuntimeError(
author_mutation_worktree.format_author_mutation_worktree_error(assessment)
)
def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
"""Verify that identity and capability were verified prior to session edits."""
global _preflight_reviewer_violation_files
@@ -426,32 +449,33 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
"file edits before mutation (fail closed). "
f"{_format_preflight_workspace_details(details)}"
)
return
if _preflight_whoami_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
f"gitea_whoami verification (fail closed). Offending files: "
f"{_format_preflight_files(_preflight_whoami_violation_files)}"
)
if _preflight_capability_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
f"gitea_resolve_task_capability verification (fail closed). Offending files: "
f"{_format_preflight_files(_preflight_capability_violation_files)}"
)
if _preflight_resolved_role == "reviewer":
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:
else:
if _preflight_whoami_violation:
raise RuntimeError(
"Reviewer role violation: Reviewer profile is forbidden from modifying "
"tracked workspace files (fail closed). Offending files: "
f"{_format_preflight_files(reviewer_delta)}"
"Pre-flight order violation: Workspace file edits occurred before "
f"gitea_whoami verification (fail closed). Offending files: "
f"{_format_preflight_files(_preflight_whoami_violation_files)}"
)
if _preflight_capability_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
f"gitea_resolve_task_capability verification (fail closed). Offending files: "
f"{_format_preflight_files(_preflight_capability_violation_files)}"
)
if _preflight_resolved_role == "reviewer":
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 "
"tracked workspace files (fail closed). Offending files: "
f"{_format_preflight_files(reviewer_delta)}"
)
_enforce_branches_only_author_mutation(worktree_path)
from mcp.server.fastmcp import FastMCP # noqa: E402
@@ -477,14 +501,145 @@ import review_proofs # noqa: E402
import review_workflow_load # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
import review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
# consumed by gitea_create_pr and scripts/worktree-start.
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
VALID_WORK_LEASE_OPERATIONS = frozenset({
AUTHOR_ISSUE_WORK_LEASE,
"review_pr_work",
"fix_pr_changes",
"cleanup_branch_work",
"issue_filing_work",
"recovery_work",
})
def _work_lease_now() -> datetime:
return datetime.now(timezone.utc)
def _work_lease_timestamp(value: datetime) -> str:
return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
text = (value or "").strip()
if not text:
return None
try:
return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
except ValueError:
return None
def _load_existing_issue_lock() -> dict | None:
if not os.path.exists(ISSUE_LOCK_FILE):
return None
try:
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else None
except Exception:
return None
def _work_lease_claimant(host: str | None) -> dict:
profile = get_profile()
username = _IDENTITY_CACHE.get(host) if host else None
if username is None and host and not _preflight_in_test_mode():
username = _authenticated_username(host)
return {
"username": username,
"profile": profile.get("profile_name"),
}
def _build_author_issue_work_lease(
*,
issue_number: int,
branch_name: str,
worktree_path: str,
host: str | None,
) -> dict:
created = _work_lease_now()
expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
return {
"operation_type": AUTHOR_ISSUE_WORK_LEASE,
"issue_number": issue_number,
"pr_number": None,
"branch": branch_name,
"worktree_path": worktree_path,
"claimant": _work_lease_claimant(host),
"created_at": _work_lease_timestamp(created),
"expires_at": _work_lease_timestamp(expires),
"last_heartbeat_at": _work_lease_timestamp(created),
}
def _active_work_lease_block(
existing_lock: dict | None,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
operation_type: str,
) -> str | None:
if not existing_lock:
return None
existing_issue = existing_lock.get("issue_number")
existing_branch = existing_lock.get("branch_name")
existing_worktree = existing_lock.get("worktree_path")
lease = existing_lock.get("work_lease")
existing_operation = (
lease.get("operation_type")
if isinstance(lease, dict)
else AUTHOR_ISSUE_WORK_LEASE
)
if existing_issue != issue_number or existing_operation != operation_type:
return None
same_owner = (
existing_branch == branch_name
and os.path.realpath(str(existing_worktree or "")) == os.path.realpath(worktree_path)
)
expires_at = (
_parse_work_lease_timestamp(lease.get("expires_at"))
if isinstance(lease, dict)
else None
)
if expires_at and expires_at <= _work_lease_now():
return (
f"Issue #{issue_number} has an expired {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
"Recovery review is required before takeover (fail closed)"
)
if same_owner:
return None
return (
f"Issue #{issue_number} already has an active {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}' "
"(fail closed)"
)
def _branch_entry_name(branch: dict | str) -> str:
if isinstance(branch, str):
return branch
if not isinstance(branch, dict):
return ""
return str(branch.get("name") or branch.get("ref") or "")
def _reveal_endpoints() -> bool:
@@ -932,6 +1087,16 @@ def gitea_lock_issue(
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, PROJECT_ROOT
)
active_lease_block = _active_work_lease_block(
_load_existing_issue_lock(),
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
operation_type=AUTHOR_ISSUE_WORK_LEASE,
)
if active_lease_block:
raise RuntimeError(active_lease_block)
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
verify_preflight_purity(remote, worktree_path=resolved_worktree)
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
@@ -977,6 +1142,25 @@ def gitea_lock_issue(
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
)
branch_url = f"{repo_api_url(h, o, r)}/branches"
try:
branches = api_get_all(branch_url, auth)
except Exception as e:
raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
for branch in branches:
name = _branch_entry_name(branch)
if expected_pattern in name:
raise ValueError(
f"Issue #{issue_number} already has matching branch '{name}' "
"(fail closed)"
)
work_lease = _build_author_issue_work_lease(
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
host=h,
)
data = {
"issue_number": issue_number,
"branch_name": branch_name,
@@ -984,6 +1168,7 @@ def gitea_lock_issue(
"org": o,
"repo": r,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
}
try:
@@ -1004,6 +1189,7 @@ def gitea_lock_issue(
"issue_number": issue_number,
"branch_name": branch_name,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
}
if agent_artifacts:
result["warnings"] = [
@@ -1060,7 +1246,7 @@ def gitea_create_pr(
)
if blocked:
return blocked
verify_preflight_purity(remote)
verify_preflight_purity(remote, worktree_path=worktree_path)
h, o, r = _resolve(remote, host, org, repo)
# ── Issue Lock Validation (Issue #194 / #196) ──
@@ -3305,16 +3491,32 @@ def gitea_delete_branch(
repo: Override the repository name.
Returns:
dict with 'success' and 'message'.
dict with 'success' and 'message'; on a permission block,
'success'/'performed' False, 'reasons', and a structured
'permission_report' (#142) with no API call made.
"""
gate_reasons = _profile_operation_gate("gitea.branch.delete")
if gate_reasons:
return {
"success": False,
"performed": False,
"required_permission": "gitea.branch.delete",
"reasons": gate_reasons,
"permission_report": _permission_block_report("gitea.branch.delete"),
}
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
import urllib.parse
encoded_branch = urllib.parse.quote(branch, safe="")
url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}"
request_metadata = {
"branch": branch,
"required_permission": "gitea.branch.delete",
}
with _audited("delete_branch", host=h, remote=remote, org=o, repo=r,
target_branch=branch, request_metadata={"branch": branch}):
target_branch=branch, request_metadata=request_metadata):
api_request("DELETE", url, auth)
return {"success": True, "message": f"Remote branch '{branch}' deleted."}
@@ -3454,6 +3656,357 @@ def gitea_reconcile_merged_cleanups(
return {"success": True, "performed": True, **report}
@mcp.tool()
def gitea_assess_already_landed_reconciliation(
pr_number: int,
target_branch: str = "master",
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only: assess already-landed reconciliation for one open PR (#301).
Fetches the PR live, refreshes the target branch, checks ancestor proof,
and returns a capability plan for the active profile. Does not review,
approve, request changes, merge, or mutate Gitea state.
Args:
pr_number: Open PR to assess.
target_branch: Branch used for ancestry proof (default master).
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict with eligibility proof, capability plan, and safe next action.
"""
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"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr=pr,
project_root=PROJECT_ROOT,
remote=remote,
target_branch=target_branch,
)
profile = get_profile()
capabilities = reconciliation_workflow.profile_reconciliation_capabilities(
profile.get("allowed_operations"),
profile.get("forbidden_operations"),
)
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment=assessment,
capabilities=capabilities,
)
return {
"success": True,
"performed": False,
"pr_number": pr_number,
"assessment": assessment,
"capabilities": capabilities,
"plan": plan,
"workflow_source": "skills/llm-project-workflow/workflows/reconcile-landed-pr.md",
"task_mode": "reconcile-landed-pr",
"review_merge_allowed": False,
}
@mcp.tool()
def gitea_scan_already_landed_open_prs(
target_branch: str = "master",
limit: int = 50,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only: list open PRs already landed on the target branch (#301).
Traverses open PR inventory with pagination proof, assesses each candidate,
and returns PRs classified as ``ALREADY_LANDED_RECONCILE_REQUIRED``. Does
not review, approve, merge, or mutate Gitea state.
Args:
target_branch: Branch used for ancestry proof (default master).
limit: Max open PRs to inspect across pages.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict with candidates, pagination metadata, and target branch SHA 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"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
open_prs: list[dict] = []
current_page = 1
pages_fetched = 0
last_meta: dict = {}
per_page = min(50, max(1, int(limit)))
while pages_fetched < 100 and len(open_prs) < limit:
raw_page, last_meta = api_fetch_page(
url, auth, page=current_page, limit=per_page
)
pages_fetched += 1
remaining = limit - len(open_prs)
open_prs.extend(raw_page[:remaining])
if last_meta["is_final_page"] or len(open_prs) >= limit:
break
current_page += 1
pagination = {
"page": 1,
"per_page": per_page,
"returned_count": len(open_prs),
"has_more": False,
"next_page": None,
"is_final_page": bool(last_meta.get("is_final_page")),
"pages_fetched": pages_fetched,
"inventory_complete": bool(last_meta.get("is_final_page")),
"total_count": len(open_prs) if last_meta.get("is_final_page") else None,
}
target_fetch = reconciliation_workflow.fetch_target_branch(
PROJECT_ROOT, remote, target_branch
)
candidates: list[dict] = []
for pr in open_prs:
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr=pr,
project_root=PROJECT_ROOT,
remote=remote,
target_branch=target_branch,
target_fetch=target_fetch,
)
if assessment.get("eligibility_class") == (
reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED
):
candidates.append(assessment)
return {
"success": True,
"performed": False,
"target_branch": target_branch,
"target_branch_sha": target_fetch.get("target_branch_sha"),
"git_ref_mutations": (
[target_fetch["git_fetch_command"]]
if target_fetch.get("git_fetch_command")
else []
),
"candidates": candidates,
"candidate_count": len(candidates),
"pagination": pagination,
"task_mode": "reconcile-landed-pr",
"review_merge_allowed": False,
}
@mcp.tool()
def gitea_reconcile_already_landed_pr(
pr_number: int,
close_pr: bool = False,
close_linked_issue: bool = False,
post_comment: bool = False,
comment_body: str = "",
target_branch: str = "master",
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Reconcile an open PR whose head is already on the target branch (#310).
Read-only assessment always runs. PR close requires ``gitea.pr.close`` and
proof that the pinned PR head SHA is an ancestor of a freshly fetched
target branch. Arbitrary PR closure is denied.
Args:
pr_number: Open PR to reconcile.
close_pr: When True, close the PR after already-landed proof passes.
close_linked_issue: When True, close a linked issue after live fetch
when the issue is open and ``gitea.issue.close`` is allowed.
post_comment: When True, post ``comment_body`` on the PR thread.
comment_body: PR comment text when ``post_comment`` is True.
target_branch: Target branch used for ancestry proof (default master).
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict with eligibility proof, optional mutations, and recovery guidance.
"""
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"),
}
if close_pr:
close_block = _profile_operation_gate("gitea.pr.close")
if close_block:
return {
"success": False,
"performed": False,
"pr_number": pr_number,
"required_permission": "gitea.pr.close",
"reasons": close_block,
"permission_report": _permission_block_report("gitea.pr.close"),
"safe_next_action": (
"Launch the prgs-reconciler MCP namespace or configure a "
"profile with gitea.pr.close for already-landed cleanup."
),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
pr_url = f"{base}/pulls/{pr_number}"
pr = api_request("GET", pr_url, auth)
assessment = already_landed_reconcile.assess_already_landed_reconciliation(
pr=pr,
project_root=PROJECT_ROOT,
remote=remote,
target_branch=target_branch,
)
result: dict = {
"success": True,
"performed": False,
"pr_number": pr_number,
"pr_state": assessment.get("pr_state"),
"candidate_head_sha": assessment.get("candidate_head_sha"),
"target_branch": assessment.get("target_branch"),
"target_branch_sha": assessment.get("target_branch_sha"),
"ancestor_proof": assessment.get("ancestor_proof"),
"eligibility_class": assessment.get("eligibility_class"),
"linked_issue": assessment.get("linked_issue"),
"close_allowed": assessment.get("close_allowed"),
"git_ref_mutations": assessment.get("git_ref_mutations") or [],
"reasons": list(assessment.get("reasons") or []),
"pr_closed": False,
"issue_closed": False,
"pr_comment_posted": False,
}
if not assessment.get("close_allowed"):
result["success"] = False
result["safe_next_action"] = (
"Do not close this PR via the reconciler path until "
"already-landed proof passes."
)
return result
verify_preflight_purity(remote)
if post_comment and comment_body.strip():
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
result["success"] = False
result["reasons"].extend(comment_block)
result["permission_report"] = _permission_block_report(
"gitea.pr.comment"
)
return result
comment_url = f"{base}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "reconcile_already_landed_pr"},
):
api_request("POST", comment_url, auth, {"body": comment_body.strip()})
result["pr_comment_posted"] = True
result["performed"] = True
if close_pr:
patch_url = f"{base}/pulls/{pr_number}"
request_metadata = {
"source": "reconcile_already_landed_pr",
"required_permission": "gitea.pr.close",
"candidate_head_sha": assessment.get("candidate_head_sha"),
"target_branch_sha": assessment.get("target_branch_sha"),
"ancestor_proof": assessment.get("ancestor_proof"),
}
with _audited(
"close_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata=request_metadata,
):
closed = api_request("PATCH", patch_url, auth, {"state": "closed"})
result["pr_closed"] = True
result["performed"] = True
result["pr_state"] = closed.get("state", "closed")
linked_issue = assessment.get("linked_issue")
if close_linked_issue and linked_issue:
issue_block = _profile_operation_gate("gitea.issue.close")
if issue_block:
result["reasons"].extend(issue_block)
result["issue_close_blocked"] = True
return result
issue_url = f"{base}/issues/{linked_issue}"
issue = api_request("GET", issue_url, auth)
result["linked_issue_live_status"] = issue.get("state")
if (issue.get("state") or "").lower() == "open":
with _audited(
"close_issue",
host=h,
remote=remote,
org=o,
repo=r,
issue_number=linked_issue,
request_metadata={"source": "reconcile_already_landed_pr"},
):
api_request(
"PATCH",
issue_url,
auth,
{"state": "closed"},
)
result["issue_closed"] = True
result["performed"] = True
return result
@mcp.tool()
def gitea_close_issue(
issue_number: int,
@@ -3963,18 +4516,30 @@ def gitea_create_issue_comment(
def _role_kind(allowed, forbidden) -> str:
"""Classify the active profile from its normalized operations.
'reconciler' can close already-landed PRs without review/author powers;
'author' can create PRs / push branches; 'reviewer' can approve/merge;
'mixed' can do both (a config smell — the reviewer-deadlock invariant
forbids it in v2 configs); 'limited' can do neither.
'reconciler' can close already-landed PRs via ``gitea.pr.close`` without
review/author powers; 'mixed' can do both (a config smell — the
reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do
neither.
"""
if reconciler_profile.is_reconciler_profile(allowed, forbidden):
return "reconciler"
def can(op):
return gitea_config.check_operation(op, allowed, forbidden)[0]
review = can("gitea.pr.approve") or can("gitea.pr.merge")
author = can("gitea.pr.create") or can("gitea.branch.push")
reconciler = (
can("gitea.pr.close")
and not review
and not author
)
if review and author:
return "mixed"
if review:
return "reviewer"
if reconciler:
return "reconciler"
if author:
return "author"
return "limited"
@@ -4111,6 +4676,14 @@ _GUIDE_RULES = {
"inventory, summarize) needs no explicit authorization. Enforced "
"fail closed via subagent_gate.assess_subagent_delegation and "
"validate_subagent_report (#266)."),
"review_merge_state_machine": (
"PR review/merge must follow the enforced state machine: PRECHECK → "
"INVENTORY → SELECT_PR → PIN_HEAD_SHA → CREATE_BRANCHES_WORKTREE → "
"VALIDATE → REVIEW_DECISION → APPROVE_OR_REQUEST_CHANGES → "
"PRE_MERGE_RECHECK → MERGE → POST_MERGE_REPORT. Failed upstream "
"states forbid downstream approve/merge. infra_stop and capability "
"blocks stop immediately. Blocked recovery handoffs must restart the "
"full workflow — never replay approve/merge commands (#290)."),
"native_mcp_preference": (
"Gitea mutations must use native MCP tools first. When MCP is "
"available, block shell scripts, direct API calls, WebFetch, "
@@ -4221,6 +4794,28 @@ _PROJECT_SKILLS = {
"Do not merge unless the operator explicitly authorizes it.",
],
},
"gitea-reconcile-landed-pr": {
"description": "Reconcile open PRs whose heads are already on the "
"target branch without review or merge.",
"when_to_use": "An open PR blocks the review queue but its head SHA "
"is already an ancestor of master; use the dedicated "
"reconciliation path instead of review/merge.",
"required_operations": ["gitea.read"],
"status": "available",
"notes": "Load skills/llm-project-workflow/workflows/reconcile-landed-pr.md "
"first. PR close requires exact gitea.pr.close; review/merge "
"capabilities must not be used on this path.",
"steps": [
"Resolve task: gitea_resolve_task_capability(task='reconcile_landed_pr').",
"Load canonical workflow via mcp_get_skill_guide or "
"skills/llm-project-workflow/workflows/reconcile-landed-pr.md.",
"Scan queue: gitea_scan_already_landed_open_prs (pagination proof).",
"Assess one PR: gitea_assess_already_landed_reconciliation.",
"Follow the plan outcome: full close only with gitea.pr.close, "
"comment-then-stop when close is missing, recovery handoff otherwise.",
"Never approve, request changes, or merge on this path.",
],
},
"gitea-pr-merge": {
"description": "Gated merge of an approved pull request.",
"when_to_use": "Reviewer/merger profile with explicit operator "
@@ -4745,9 +5340,11 @@ _RUNTIME_CAPABILITY_TASKS = (
"create_issue",
"comment_issue",
"create_pr",
"work_issue",
"review_pr",
"merge_pr",
"close_issue",
"reconcile_already_landed_pr",
)
@@ -4796,9 +5393,11 @@ def _build_runtime_task_capabilities(
"create_issue": "can_create_issues",
"comment_issue": "can_comment_on_issues",
"create_pr": "can_author_prs",
"work_issue": "can_work_issues",
"review_pr": "can_review_prs",
"merge_pr": "can_merge_prs",
"close_issue": "can_close_issues",
"reconcile_already_landed_pr": "can_reconcile_already_landed_prs",
}
for task in _RUNTIME_CAPABILITY_TASKS:
permission = task_capability_map.required_permission(task)
@@ -4826,6 +5425,65 @@ def _build_runtime_task_capabilities(
@mcp.tool()
def gitea_assess_review_merge_state_machine(
target_state: str | None = None,
state_completion: dict[str, bool] | None = None,
pre_merge_gates: dict[str, bool] | None = None,
infra_stop: bool = False,
capability_blocked: bool = False,
recovery_handoff_text: str | None = None,
final_report_text: str | None = None,
) -> dict:
"""Read-only: assess enforced PR review/merge workflow state (#290)."""
completion = state_completion or {}
blockers = review_merge_state_machine.assess_workflow_blockers(
infra_stop=infra_stop,
capability_blocked=capability_blocked,
)
result = {
"workflow": review_merge_state_machine.workflow_status(
completion,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
),
"blockers": blockers,
"approve": review_merge_state_machine.can_approve(
completion,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
),
"merge": review_merge_state_machine.can_merge(
completion,
pre_merge_gates=pre_merge_gates,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
),
}
if target_state:
result["advancement"] = review_merge_state_machine.assess_state_advancement(
completion,
target_state=target_state,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
)
if recovery_handoff_text is not None:
result["recovery_handoff"] = (
review_merge_state_machine.assess_blocked_recovery_handoff(
recovery_handoff_text
)
)
if final_report_text is not None:
result["final_report_claims"] = (
review_merge_state_machine.assess_final_report_state_claims(
final_report_text,
state_completion=completion,
approve_completed=bool(completion.get("APPROVE_OR_REQUEST_CHANGES")),
merge_completed=bool(completion.get("MERGE")),
)
)
return result
@mcp.tool()
def gitea_assess_gitea_operation_path(
task: str,
@@ -4888,6 +5546,37 @@ def gitea_get_shell_health() -> dict:
return native_mcp_preference.shell_health_status()
@mcp.tool()
def gitea_validate_review_final_report(
report_text: str,
review_decision_lock: dict | None = None,
linked_issue_lock: dict | None = None,
validation_report: dict | None = None,
action_log: list[dict] | None = None,
mutations_observed: bool = False,
local_edits: bool = False,
validation_session: dict | None = None,
) -> dict:
"""Read-only: machine-validate reviewer final report schema before output (#391).
Composes the composable final-report validator with review-specific schema
gates (legacy fields, mutation categories, merge/issue claims, blocked
handoff replay, and narrative/handoff consistency).
"""
from review_final_report_schema import assess_review_final_report_schema
return assess_review_final_report_schema(
report_text,
review_decision_lock=review_decision_lock,
linked_issue_lock=linked_issue_lock,
validation_report=validation_report,
action_log=action_log,
mutations_observed=mutations_observed,
local_edits=local_edits,
validation_session=validation_session,
)
@mcp.tool()
def gitea_get_runtime_context(
remote: str = "dadeschools",
@@ -5006,6 +5695,10 @@ def gitea_get_runtime_context(
"preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities,
"reconciler_profile_assessment": reconciler_profile.assess_reconciler_profile(
allowed, forbidden
),
"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),