feat: add prgs-reconciler profile model and role detection (Closes #304) #388

Merged
sysadmin merged 6 commits from feat/issue-304-reconciler-profile into master 2026-07-07 10:27:13 -05:00
5 changed files with 203 additions and 7 deletions
+1 -1
View File
@@ -23,7 +23,7 @@ launched with exactly one static execution profile:
|-----------------------------|----------------|-------------| |-----------------------------|----------------|-------------|
| `gitea-author` | an author profile | implement issues, push branches, open PRs, comment | | `gitea-author` | an author profile | implement issues, push branches, open PRs, comment |
| `gitea-reviewer` | a reviewer profile | review, approve/request changes, merge | | `gitea-reviewer` | a reviewer profile | review, approve/request changes, merge |
| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#310) | | `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#304 profile; #310 close tool) |
Properties: Properties:
+13 -6
View File
@@ -226,11 +226,12 @@ explicit operator-directed closure of a contaminated PR). If `close_pr` ever
resolves as unknown, agents must fail closed rather than fall back to the resolves as unknown, agents must fail closed rather than fall back to the
edit path. edit path.
## Reconciler profile for already-landed open PRs (#310) ## Reconciler profile for already-landed open PRs (#304 / #310)
Normal author and reviewer profiles must not gain broad PR-close authority. Normal author and reviewer profiles must not gain broad `gitea.pr.close`
Already-landed open PRs (head SHA is an ancestor of the target branch) need a authority. Already-landed open PRs (head SHA is an ancestor of the target
dedicated reconciler profile such as `prgs-reconciler` with: branch) need a dedicated reconciler profile such as `prgs-reconciler` with a
narrow operation set:
- `gitea.read` - `gitea.read`
- `gitea.pr.comment` - `gitea.pr.comment`
@@ -238,8 +239,14 @@ dedicated reconciler profile such as `prgs-reconciler` with:
- `gitea.issue.close` - `gitea.issue.close`
- `gitea.pr.close` - `gitea.pr.close`
Use the `gitea-reconciler` MCP namespace (static profile launch) and the Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
`gitea_reconcile_already_landed_pr` tool. The resolver task is `gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
`gitea.repo.commit`.
Launch a static `gitea-reconciler` MCP namespace with
`GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
`reconciler_profile.assess_reconciler_profile` (#304). Use the
`gitea_reconcile_already_landed_pr` tool (#310). The resolver task is
`reconcile_already_landed_pr`. PR close is allowed only after live PR fetch, `reconcile_already_landed_pr`. PR close is allowed only after live PR fetch,
fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
heads are not already landed cannot be closed through this path. heads are not already landed cannot be closed through this path.
+8
View File
@@ -504,6 +504,7 @@ import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402 import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402 import issue_claim_heartbeat # noqa: E402
import merged_cleanup_reconcile # noqa: E402 import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402 import reconciliation_workflow # noqa: E402
import review_merge_state_machine # noqa: E402 import review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402 import native_mcp_preference # noqa: E402
@@ -4472,12 +4473,15 @@ def gitea_create_issue_comment(
def _role_kind(allowed, forbidden) -> str: def _role_kind(allowed, forbidden) -> str:
"""Classify the active profile from its normalized operations. """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; 'author' can create PRs / push branches; 'reviewer' can approve/merge;
'reconciler' can close already-landed PRs via ``gitea.pr.close`` without 'reconciler' can close already-landed PRs via ``gitea.pr.close`` without
review/author powers; 'mixed' can do both (a config smell the review/author powers; 'mixed' can do both (a config smell the
reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do
neither. neither.
""" """
if reconciler_profile.is_reconciler_profile(allowed, forbidden):
return "reconciler"
def can(op): def can(op):
return gitea_config.check_operation(op, allowed, forbidden)[0] return gitea_config.check_operation(op, allowed, forbidden)[0]
review = can("gitea.pr.approve") or can("gitea.pr.merge") review = can("gitea.pr.approve") or can("gitea.pr.merge")
@@ -5642,6 +5646,10 @@ def gitea_get_runtime_context(
"preflight_block_reasons": preflight["preflight_block_reasons"], "preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"), "preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities, "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(), "shell_health": native_mcp_preference.shell_health_status(),
} }
+94
View File
@@ -0,0 +1,94 @@
"""Reconciler profile model for already-landed PR closure (#304).
Defines the narrowly scoped operation set for a dedicated reconciler profile
such as ``prgs-reconciler``. Close MCP tooling and ancestry gates ship in the
#310 stack; this module validates profile shape only.
"""
from __future__ import annotations
import gitea_config
RECONCILER_REQUIRED_OPERATIONS = (
"gitea.read",
"gitea.pr.close",
)
RECONCILER_RECOMMENDED_OPERATIONS = (
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
)
RECONCILER_FORBIDDEN_OPERATIONS = (
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.review",
"gitea.pr.create",
"gitea.branch.push",
"gitea.repo.commit",
)
def _normalized_allowed(allowed: list[str]) -> set[str]:
normalized: set[str] = set()
for entry in allowed or []:
try:
normalized.add(gitea_config.normalize_operation(entry))
except gitea_config.ConfigError:
continue
return normalized
def _forbidden_in_allowed(allowed: list[str], ops: tuple[str, ...]) -> list[str]:
"""Return forbidden ops that are explicitly listed in *allowed*."""
allowed_n = _normalized_allowed(allowed)
present: list[str] = []
for op in ops:
try:
if gitea_config.normalize_operation(op) in allowed_n:
present.append(op)
except gitea_config.ConfigError:
continue
return present
def is_reconciler_profile(allowed: list[str], forbidden: list[str]) -> bool:
"""Return True when *allowed*/*forbidden* describe a reconciler profile."""
def can(op: str) -> bool:
return gitea_config.check_operation(op, allowed, forbidden)[0]
if not can("gitea.pr.close"):
return False
if _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
return False
return True
def assess_reconciler_profile(allowed: list[str], forbidden: list[str]) -> dict:
"""Validate reconciler profile operations (read-only, fail closed)."""
allowed = list(allowed or [])
forbidden = list(forbidden or [])
reasons: list[str] = []
for op in RECONCILER_REQUIRED_OPERATIONS:
ok, _ = gitea_config.check_operation(op, allowed, forbidden)
if not ok:
reasons.append(f"missing required operation {op}")
for op in _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
reasons.append(f"forbidden operation must not be allowed: {op}")
missing_recommended = [
op for op in RECONCILER_RECOMMENDED_OPERATIONS
if not gitea_config.check_operation(op, allowed, forbidden)[0]
]
return {
"is_reconciler_profile": is_reconciler_profile(allowed, forbidden),
"valid": not reasons and is_reconciler_profile(allowed, forbidden),
"reasons": reasons,
"missing_recommended_operations": missing_recommended,
"required_operations": list(RECONCILER_REQUIRED_OPERATIONS),
"forbidden_operations": list(RECONCILER_FORBIDDEN_OPERATIONS),
}
+87
View File
@@ -0,0 +1,87 @@
"""Tests for reconciler profile model (#304)."""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import gitea_mcp_server as mcp_server
import migrate_profiles
import reconciler_profile
PRGS_RECONCILER_ALLOWED = [
"gitea.read",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
"gitea.pr.close",
]
PRGS_RECONCILER_FORBIDDEN = [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.create",
"gitea.branch.push",
]
class TestReconcilerProfileModel(unittest.TestCase):
def test_prgs_reconciler_shape_valid(self):
result = reconciler_profile.assess_reconciler_profile(
PRGS_RECONCILER_ALLOWED,
PRGS_RECONCILER_FORBIDDEN,
)
self.assertTrue(result["is_reconciler_profile"])
self.assertTrue(result["valid"])
self.assertEqual(result["reasons"], [])
def test_author_profile_not_reconciler(self):
allowed = [
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
"gitea.issue.comment",
]
forbidden = ["gitea.pr.approve", "gitea.pr.merge"]
self.assertFalse(
reconciler_profile.is_reconciler_profile(allowed, forbidden)
)
def test_reviewer_profile_not_reconciler(self):
allowed = [
"gitea.read",
"gitea.pr.review",
"gitea.pr.approve",
"gitea.pr.merge",
]
forbidden = ["gitea.pr.create", "gitea.branch.push"]
self.assertFalse(
reconciler_profile.is_reconciler_profile(allowed, forbidden)
)
def test_broad_close_on_author_invalid(self):
allowed = PRGS_RECONCILER_ALLOWED + ["gitea.pr.create"]
result = reconciler_profile.assess_reconciler_profile(
allowed,
PRGS_RECONCILER_FORBIDDEN,
)
self.assertFalse(result["valid"])
self.assertFalse(result["is_reconciler_profile"])
def test_role_kind_detects_reconciler(self):
role = mcp_server._role_kind(
PRGS_RECONCILER_ALLOWED,
PRGS_RECONCILER_FORBIDDEN,
)
self.assertEqual(role, "reconciler")
def test_migrate_infer_role_reconciler(self):
self.assertEqual(
migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
"reconciler",
)
if __name__ == "__main__":
unittest.main()