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 #385

Browse Source 
Merge prgs/master; keep work-issue author routing (#139) alongside
reconciler task entries from master (#309/#310).
This commit is contained in:
sysadmin
2026-07-07 10:34:23 -04:00
parent d66d465e31 273eba7fb3
commit 10e64f6683
35 changed files with 3674 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+310 -26
View File
@@ -397,6 +397,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 +448,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
@@ -476,8 +499,11 @@ import task_capability_map # noqa: E402
import review_proofs # 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 review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402
@@ -3427,6 +3453,186 @@ def gitea_reconcile_merged_cleanups(
return {"success": True, "performed": True, **report}
@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,
@@ -3937,17 +4143,26 @@ def _role_kind(allowed, forbidden) -> str:
"""Classify the active profile from its normalized operations.
'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.
"""
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"
@@ -4084,6 +4299,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, "
@@ -4720,6 +4943,7 @@ _RUNTIME_CAPABILITY_TASKS = (
"review_pr",
"merge_pr",
"close_issue",
"reconcile_already_landed_pr",
)
@@ -4772,6 +4996,7 @@ def _build_runtime_task_capabilities(
"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)
@@ -4799,6 +5024,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,