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 into feat/issue-139-role-aware-work-issue-routing and
keep both work-issue task routing entries and reconcile-landed-pr entries
in task_capability_map.py.
This commit is contained in:
sysadmin
2026-07-07 11:13:45 -04:00
parent 10e64f6683 9f4dd4c39b
commit f75f2327f5
15 changed files with 1496 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_final_report_validator.py
+194 -7
View File
@@ -57,26 +57,47 @@ def _review_handoff(**overrides):
return text
def _reconcile_handoff(**overrides):
def _reconcile_handoff(drop=(), **overrides):
lines = [
"## Controller Handoff",
"",
"- Task: reconcile already-landed PR #99",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: reconciler",
"- Identity: sysadmin / prgs-author",
"- Role/profile: reconciler / prgs-reconciler",
"- Identity: sysadmin",
"- Selected PR: #99",
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
"- PR live state: open",
"- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Target branch: master",
"- Target branch SHA: 5e023dc71b0e2b813a0b1eee6b0f42630c47a95c",
"- Ancestor proof: candidate head is ancestor of target branch SHA",
"- Linked issue: #98",
"- Linked issue live status: not verified in this session",
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Missing capabilities: gitea.pr.close",
"- PR comments posted: 1 reconciliation comment on PR #99",
"- Issue comments posted: none",
"- PRs closed: none",
"- Issues closed: none",
"- File edits by reconciler: none",
"- Git ref mutations: git fetch prgs master",
"- Worktree mutations: none",
"- MCP/Gitea mutations: PR comment posted",
"- External-state mutations: none",
"- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
"- Reconciliation mutations: PR comment posted",
"- Current status: reconciliation complete",
"- Safe next action: none",
"- Blocker: missing gitea.pr.close capability",
"- Safe next action: hand off to a profile with gitea.pr.close",
"- No review/merge confirmation: confirmed",
]
text = "\n".join(lines)
kept = [
line
for line in lines
if not any(line.startswith(f"- {name}:") for name in drop)
]
text = "\n".join(kept)
for key, value in overrides.items():
if f"- {key}:" in text:
text = text.replace(f"- {key}:", f"- {key}: {value}")
@@ -194,6 +215,20 @@ class TestReviewPrRules(unittest.TestCase):
)
)
def test_already_landed_oldest_eligible_pr_blocks(self):
report = (
_review_handoff()
+ "\nAlready landed.\nOldest eligible PR: PR #278."
)
result = assess_final_report_validator(report, "review_pr")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reviewer.already_landed_eligible_wording"
for f in result["findings"]
)
)
def test_already_landed_gate_blocks_review_terminal_states(self):
cases = {
"APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
@@ -242,7 +277,6 @@ class TestReviewPrRules(unittest.TestCase):
)
result = assess_final_report_validator(report, "review_pr")
self.assertEqual(result["grade"], "A")
def test_git_fetch_must_not_be_readonly_only(self):
report = (
_review_handoff()
@@ -307,6 +341,159 @@ class TestReconciliationRules(unittest.TestCase):
)
class TestCanonicalReconcileSchema(unittest.TestCase):
"""Issue #307: reconciliation workflows emit only the canonical schema."""
def test_comment_only_reconciliation_passes(self):
result = assess_final_report_validator(
_reconcile_handoff(),
"reconcile_already_landed",
)
self.assertEqual(result["grade"], "A")
self.assertEqual(result["findings"], [])
def test_successful_close_reconciliation_passes(self):
report = _reconcile_handoff().replace(
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
).replace(
"- Missing capabilities: gitea.pr.close",
"- Missing capabilities: none",
).replace(
"- PRs closed: none",
"- PRs closed: #99",
).replace(
"- Blocker: missing gitea.pr.close capability",
"- Blocker: none",
)
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "A")
def test_blocked_reconciliation_passes(self):
report = _reconcile_handoff().replace(
"- Capabilities proven: gitea.read, gitea.pr.comment",
"- Capabilities proven: gitea.read",
).replace(
"- Missing capabilities: gitea.pr.close",
"- Missing capabilities: gitea.pr.close, gitea.pr.comment",
).replace(
"- PR comments posted: 1 reconciliation comment on PR #99",
"- PR comments posted: none",
).replace(
"- MCP/Gitea mutations: PR comment posted",
"- MCP/Gitea mutations: none",
).replace(
"- Reconciliation mutations: PR comment posted",
"- Reconciliation mutations: none",
)
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "A")
def test_missing_capabilities_field_required(self):
report = _reconcile_handoff(drop=("Missing capabilities",))
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "downgraded")
self.assertTrue(
any(
f["rule_id"] == "reconcile.controller_handoff"
and "Missing capabilities" in f["reason"]
for f in result["findings"]
)
)
def test_missing_ancestor_proof_and_candidate_sha_downgrade(self):
report = _reconcile_handoff(drop=("Ancestor proof", "Candidate head SHA"))
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertEqual(result["grade"], "downgraded")
reasons = " ".join(f["reason"] for f in result["findings"])
self.assertIn("Ancestor proof", reasons)
self.assertIn("Candidate head SHA", reasons)
def test_stale_issue_lock_proof_blocks(self):
report = _reconcile_handoff() + "\n- Issue lock proof: locked before diff"
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "issue lock proof"
for f in result["findings"]
)
)
def test_stale_claim_comment_status_blocks(self):
report = _reconcile_handoff() + "\n- Claim/comment status: claimed"
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "claim/comment status"
for f in result["findings"]
)
)
def test_stale_workspace_mutations_blocks_without_observed_mutations(self):
report = _reconcile_handoff() + "\n- Workspace mutations: None"
result = assess_final_report_validator(report, "reconcile_already_landed")
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "workspace mutations"
for f in result["findings"]
)
)
def test_pr_number_opened_allowed_when_session_opened_pr(self):
report = _reconcile_handoff() + "\n- PR number opened: #12"
result = assess_final_report_validator(
report,
"reconcile_already_landed",
session_pr_opened=True,
)
self.assertFalse(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "pr number opened"
for f in result["findings"]
)
)
def test_pr_number_opened_still_blocked_without_session_proof(self):
report = _reconcile_handoff() + "\n- PR number opened: #12"
result = assess_final_report_validator(
report,
"reconcile_already_landed",
session_pr_opened=False,
)
self.assertTrue(result["blocked"])
self.assertTrue(
any(
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
and f["field"] == "pr number opened"
for f in result["findings"]
)
)
def test_action_log_fetch_requires_git_ref_mutation_entry(self):
report = _reconcile_handoff().replace(
"- Git ref mutations: git fetch prgs master",
"- Git ref mutations: none",
)
result = assess_final_report_validator(
report,
"reconcile_already_landed",
action_log=[{"command": "git fetch prgs master", "performed": True}],
)
self.assertTrue(
any(
f["rule_id"] == "reviewer.git_fetch_readonly"
for f in result["findings"]
)
)
class TestEntryPoint(unittest.TestCase):
def test_unknown_task_kind_blocks(self):
result = assess_final_report_validator("report", "unknown_mode")
tests/test_llm_workflow_split.py
+16
View File
@@ -141,6 +141,16 @@ def test_start_issue_template_references_work_issue_workflow():
assert "workflows/work-issue.md" in text
def test_reconcile_skill_registered_in_mcp_server():
from mcp_server import _PROJECT_SKILLS
assert "gitea-reconcile-landed-pr" in _PROJECT_SKILLS
steps = _PROJECT_SKILLS["gitea-reconcile-landed-pr"]["steps"]
joined = " ".join(steps).lower()
assert "gitea_scan_already_landed_open_prs" in joined
assert "gitea_assess_already_landed_reconciliation" in joined
def test_pr_queue_cleanup_template_references_workflow():
text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text(
encoding="utf-8"
@@ -251,6 +261,12 @@ def test_worktree_ownership_verifier_exported():
assert callable(assess_worktree_ownership_report)
def test_already_landed_classification_verifier_exported():
from review_proofs import assess_already_landed_classification_report
assert callable(assess_already_landed_classification_report)
def test_pr_queue_cleanup_verifier_exported():
from review_proofs import assess_pr_queue_cleanup_report
tests/test_mcp_server.py
+53
View File
@@ -3050,10 +3050,18 @@ class TestIssueLocking(unittest.TestCase):
mock_api.return_value = [] # no open PRs
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
self.assertEqual(res["work_lease"]["issue_number"], 196)
self.assertEqual(res["work_lease"]["branch"], "feat/issue-196-mutations")
self.assertIn("created_at", res["work_lease"])
self.assertIn("expires_at", res["work_lease"])
self.assertIn("last_heartbeat_at", res["work_lease"])
self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
lock = json.load(f)
self.assertIn("worktree_path", lock)
self.assertIn("work_lease", lock)
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
@@ -3094,6 +3102,51 @@ class TestIssueLocking(unittest.TestCase):
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
mock_api.side_effect = [
[],
[{"name": "feat/issue-196-existing-work"}],
]
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has matching branch", str(ctx.exception))
def test_lock_issue_blocks_active_same_operation_lease(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}, f)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2000-01-01T00:00:00Z",
},
}, f)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("Recovery review is required before takeover", str(ctx.exception))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_from_clean_scratch_worktree(self, _auth, _api):
tests/test_reconciliation_mcp_tools.py
+106
View File
@@ -0,0 +1,106 @@
"""Tests for reconciliation MCP assessment tools (#301)."""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
from mcp_server import (
gitea_assess_already_landed_reconciliation,
gitea_scan_already_landed_open_prs,
)
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
],
"forbidden_operations": ["gitea.pr.close"],
"audit_label": "prgs-author",
}
OPEN_PR = {
"number": 278,
"title": "Landed (Closes #263)",
"body": "",
"state": "open",
"head": {"ref": "feat/x", "sha": "a" * 40},
"base": {"ref": "master"},
}
class TestReconciliationMcpTools(unittest.TestCase):
def setUp(self):
self.mock_api = patch("mcp_server.api_request").start()
self.mock_auth = patch(
"mcp_server.get_auth_header", return_value="token test"
).start()
patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE).start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
mcp_server._IDENTITY_CACHE.clear()
def test_assess_returns_plan_without_mutations(self):
self.mock_api.return_value = OPEN_PR
with patch(
"mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
return_value={
"reconciliation_allowed": True,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"candidate_head_sha": OPEN_PR["head"]["sha"],
"target_branch_sha": "deadbeef",
"linked_issue": 263,
"review_merge_allowed": False,
},
):
res = gitea_assess_already_landed_reconciliation(
pr_number=278, remote="prgs"
)
self.assertTrue(res["success"])
self.assertFalse(res["performed"])
self.assertEqual(res["plan"]["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
self.assertFalse(res["review_merge_allowed"])
patch_calls = [
c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
]
self.assertEqual(patch_calls, [])
def test_scan_returns_candidates_with_pagination(self):
with patch(
"mcp_server.api_fetch_page",
return_value=([OPEN_PR], {
"page": 1,
"per_page": 50,
"is_final_page": True,
"has_more": False,
"next_page": None,
}),
), patch(
"mcp_server.reconciliation_workflow.fetch_target_branch",
return_value={
"success": True,
"target_branch_sha": "deadbeef",
"git_fetch_command": "git fetch prgs master",
},
), patch(
"mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
return_value={
"pr_number": 278,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"reconciliation_allowed": True,
},
):
res = gitea_scan_already_landed_open_prs(remote="prgs", limit=50)
self.assertTrue(res["success"])
self.assertEqual(res["candidate_count"], 1)
self.assertTrue(res["pagination"]["inventory_complete"])
if __name__ == "__main__":
unittest.main()
tests/test_reconciliation_workflow.py
+149
View File
@@ -0,0 +1,149 @@
"""Tests for already-landed reconciliation workflow helpers (#301)."""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import reconciliation_workflow
from review_proofs import assess_reconcile_workflow_source
class TestReconciliationAssessment(unittest.TestCase):
def test_already_landed_open_pr(self):
with patch(
"reconciliation_workflow.is_head_ancestor_of_ref",
return_value=True,
):
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr={
"number": 278,
"state": "open",
"title": "Fix (Closes #263)",
"body": "",
"head": {"ref": "feat/x", "sha": "abc" * 13 + "a"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
"git_fetch_command": "git fetch prgs master",
},
)
self.assertTrue(assessment["reconciliation_allowed"])
self.assertEqual(
assessment["eligibility_class"],
reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
)
def test_not_landed_pr(self):
with patch(
"reconciliation_workflow.is_head_ancestor_of_ref",
return_value=False,
):
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr={
"number": 1,
"state": "open",
"title": "WIP",
"body": "",
"head": {"ref": "feat/y", "sha": "fff" * 13 + "f"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
},
)
self.assertFalse(assessment["reconciliation_allowed"])
class TestReconciliationPlan(unittest.TestCase):
def test_full_close_when_close_pr_allowed(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": True,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
},
capabilities={
"close_pr": True,
"comment_pr": True,
"close_issue": True,
"comment_issue": True,
},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_FULL_RECONCILE,
)
self.assertTrue(plan["close_pr_allowed"])
self.assertFalse(plan["review_merge_allowed"])
def test_comment_then_stop_without_close(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": True,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
},
capabilities={
"close_pr": False,
"comment_pr": True,
},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_PARTIAL_COMMENT,
)
def test_recovery_handoff_without_comment_or_close(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": True,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
},
capabilities={"close_pr": False, "comment_pr": False},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF,
)
def test_not_landed_no_action(self):
plan = reconciliation_workflow.resolve_reconciliation_plan(
assessment={
"reconciliation_allowed": False,
"eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED,
"reasons": ["not ancestor"],
},
capabilities={"close_pr": True},
)
self.assertEqual(
plan["outcome"],
reconciliation_workflow.OUTCOME_NOT_LANDED,
)
class TestReconcileWorkflowSource(unittest.TestCase):
def test_valid_report_passes(self):
report = (
"Task mode: reconcile-landed-pr\n"
"Workflow source: workflows/reconcile-landed-pr.md\n"
)
result = assess_reconcile_workflow_source(report)
self.assertTrue(result["complete"])
def test_missing_workflow_fails(self):
result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr")
self.assertFalse(result["complete"])
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_already_landed_classification.py
+81
View File
@@ -0,0 +1,81 @@
import unittest
from reviewer_already_landed_classification import ( # noqa: E402
ALREADY_LANDED_ELIGIBILITY_CLASS,
assess_already_landed_classification_report,
)
class TestAlreadyLandedClassification(unittest.TestCase):
def test_non_landed_report_passes(self):
result = assess_already_landed_classification_report(
"Selected the next eligible PR: PR #286."
)
self.assertTrue(result["proven"])
self.assertFalse(result["landed_context"])
def test_oldest_eligible_pr_on_landed_blocks(self):
result = assess_already_landed_classification_report(
"PR is already landed.\n"
"Oldest eligible PR: PR #278."
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_next_eligible_pr_on_landed_blocks(self):
result = assess_already_landed_classification_report(
"Ancestor proof: passed\n"
"Selected the next eligible PR: PR #278."
)
self.assertFalse(result["proven"])
def test_eligible_for_review_on_landed_blocks(self):
result = assess_already_landed_classification_report(
"Already landed on master and eligible for review next."
)
self.assertFalse(result["proven"])
def test_proper_landed_wording_passes(self):
result = assess_already_landed_classification_report(
"Oldest open PR requiring action: PR #278\n"
f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
"Candidate head SHA: abc123\n"
"Reviewed head SHA: none\n"
"Queue blocked by already-landed PR requiring reconciliation",
selected_pr_already_landed=True,
)
self.assertTrue(result["proven"])
def test_pinned_reviewed_head_without_validation_blocks(self):
result = assess_already_landed_classification_report(
"Already landed.\n"
"Pinned reviewed head: abc123def456",
selected_pr_already_landed=True,
)
self.assertFalse(result["proven"])
def test_wrong_eligibility_class_blocks(self):
result = assess_already_landed_classification_report(
"Already landed.\n"
"Eligibility class: REVIEW_ELIGIBLE",
selected_pr_already_landed=True,
)
self.assertFalse(result["proven"])
def test_missing_queue_blocker_blocks(self):
result = assess_already_landed_classification_report(
f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
"Candidate head SHA: abc123\n"
"Reviewed head SHA: none",
selected_pr_already_landed=True,
)
self.assertFalse(result["proven"])
def test_exported_from_review_proofs(self):
from review_proofs import assess_already_landed_classification_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()