From 4ea618e8b442e6bbd7d5f93c4b790e4241bc5c0d Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:19:22 -0400 Subject: [PATCH 1/3] Add validation integrity verifier for PR-head vs diagnostic tests (#316) Separate official PR-head validation from post-edit diagnostic runs in reviewer reports. Block treating diagnostic passes as official validation and require contaminated sessions to report recovery-required status. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 + reviewer_validation_integrity.py | 148 ++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_validation_integrity.py | 130 +++++++++++++++++ 4 files changed, 292 insertions(+), 1 deletion(-) create mode 100644 reviewer_validation_integrity.py create mode 100644 tests/test_reviewer_validation_integrity.py diff --git a/review_proofs.py b/review_proofs.py index d6acb01..ef74ad4 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3413,3 +3413,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): "ref_mutating_commands": ref_commands, "fetch_targets": fetch_targets, } + + +def assess_validation_integrity_report(report_text, **kwargs): + """#316: separate official PR-head validation from diagnostic experiments.""" + from reviewer_validation_integrity import assess_validation_integrity_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_validation_integrity.py b/reviewer_validation_integrity.py new file mode 100644 index 0000000..038c457 --- /dev/null +++ b/reviewer_validation_integrity.py @@ -0,0 +1,148 @@ +"""PR-head vs diagnostic validation integrity verifier (#316).""" + +from __future__ import annotations + +import re +from typing import Any + +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only validation", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official validation|pr-head validation|validation integrity)", + re.IGNORECASE, +) +_OFFICIAL_COMMAND_RE = re.compile( + r"(?:official validation command|pr-head validation command|validation command)", + re.IGNORECASE, +) +_OFFICIAL_RESULT_RE = re.compile( + r"(?:official validation result|pr-head validation result|validation result|validation integrity status)", + re.IGNORECASE, +) +_DIRTY_AFTER_RE = re.compile( + r"(?:worktree dirty after (?:official )?validation|dirty (?:state )?after (?:official )?validation|" + r"validation worktree dirty after)", + re.IGNORECASE, +) +_DIAGNOSTIC_EDIT_RE = re.compile( + r"(?:diagnostic edit(?:ed)? path|file edits by reviewer|diagnostic local edit)", + re.IGNORECASE, +) +_DIAGNOSTIC_COMMAND_RE = re.compile( + r"(?:diagnostic (?:validation )?command|diagnostic test run|diagnostic result)", + re.IGNORECASE, +) +_SUGGESTED_FIX_RE = re.compile( + r"(?:diagnostic results? (?:were )?used only for suggested fix|suggested fix only|" + r"not used as official validation)", + re.IGNORECASE, +) +_INTEGRITY_STATUS_RE = re.compile( + r"validation integrity status\s*:\s*(passed|failed|not run|contaminated)", + re.IGNORECASE, +) +_POST_EDIT_AS_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit|" + r"full suite passed after diagnostic edit)", + re.IGNORECASE, +) + + +def _performed_edits(action_log: list[dict] | None) -> list[str]: + paths: list[str] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + path = entry.get("path") + if path and entry.get("kind", "file_edit") in {"file_edit", "edit", "write"}: + paths.append(str(path)) + return paths + + +def assess_validation_integrity_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Separate official PR-head validation from diagnostic edited-worktree tests (#316).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + diagnostic_edits = list(session.get("diagnostic_edits") or []) + if not diagnostic_edits: + diagnostic_edits = _performed_edits(action_log) + diagnostic_ran = bool( + session.get("diagnostic_validation_ran") + or session.get("diagnostic_runs") + ) + official_ran = bool(session.get("official_validation_ran", True)) + official_result = (session.get("official_result") or "").strip().lower() + diagnostic_result = (session.get("diagnostic_result") or "").strip().lower() + dirty_after = session.get("worktree_dirty_after_official") + contaminated = bool(session.get("contaminated")) + + if official_ran: + if not _OFFICIAL_VALIDATION_RE.search(text) and not _OFFICIAL_COMMAND_RE.search(text): + reasons.append("report missing official PR-head validation section") + if not _OFFICIAL_COMMAND_RE.search(text) and "validation:" not in text.lower(): + reasons.append("report missing official validation command") + if not _OFFICIAL_RESULT_RE.search(text): + reasons.append("report missing official validation result or integrity status") + if dirty_after is not None and not _DIRTY_AFTER_RE.search(text): + reasons.append( + "report missing worktree dirty state after official validation" + ) + elif dirty_after is None and official_ran and not _DIRTY_AFTER_RE.search(text): + reasons.append( + "report missing worktree dirty state after official validation" + ) + + if diagnostic_edits or diagnostic_ran: + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "post-edit diagnostic runs must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if diagnostic_edits and not _DIAGNOSTIC_EDIT_RE.search(text): + reasons.append("report missing diagnostic edit path") + if diagnostic_ran and not _DIAGNOSTIC_COMMAND_RE.search(text): + reasons.append("report missing diagnostic command/result") + if not _SUGGESTED_FIX_RE.search(text): + reasons.append( + "report must state diagnostic results were used only for suggested fix" + ) + if _POST_EDIT_AS_OFFICIAL_RE.search(text): + reasons.append( + "post-edit diagnostic results cannot be reported as official PR-head validation" + ) + if diagnostic_result == "pass" and official_result == "fail": + if "request_changes" not in text.lower() and "failed" not in text.lower(): + reasons.append( + "request-changes must cite unmodified PR-head failure, not diagnostic pass" + ) + + if contaminated: + status = _INTEGRITY_STATUS_RE.search(text) + if not status or "contaminated" not in status.group(1).lower(): + reasons.append( + "contaminated validation must report integrity status " + "'contaminated — recovery required'" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "separate official PR-head validation from diagnostic experiments; " + "report dirty-after-validation state" + if reasons + else "proceed" + ), + } diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index f8d3b04..6f13a84 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow(): def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report - assert callable(assess_create_issue_final_report) \ No newline at end of file + assert callable(assess_create_issue_final_report) + + +def test_validation_integrity_verifier_exported(): + from review_proofs import assess_validation_integrity_report + + assert callable(assess_validation_integrity_report) \ No newline at end of file diff --git a/tests/test_reviewer_validation_integrity.py b/tests/test_reviewer_validation_integrity.py new file mode 100644 index 0000000..5137057 --- /dev/null +++ b/tests/test_reviewer_validation_integrity.py @@ -0,0 +1,130 @@ +"""Tests for PR-head vs diagnostic validation integrity verifier (#316).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_integrity import assess_validation_integrity_report # noqa: E402 + + +def _official_only_report() -> str: + return "\n".join([ + "Official PR-head validation on unmodified worktree.", + "Official validation command: pytest tests/ -q", + "Official validation result: failed (3 failed)", + "Validation integrity status: failed", + "Worktree dirty after official validation: clean", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + _official_only_report(), + "Diagnostic local experiment — not PR-head validation", + "Diagnostic edit path: tests/test_example.py", + "File edits by reviewer: tests/test_example.py", + "Diagnostic command: pytest tests/test_example.py -q", + "Diagnostic result: passed after local fix", + "Diagnostic results used only for suggested fix; blocker remains PR-head failure.", + ]) + + +class TestValidationIntegrity(unittest.TestCase): + def test_official_only_passes(self): + result = assess_validation_integrity_report( + _official_only_report(), + validation_session={ + "official_validation_ran": True, + "official_result": "fail", + "worktree_dirty_after_official": False, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_failing_pr_head_with_diagnostic_pass_passes(self): + result = assess_validation_integrity_report( + _diagnostic_report(), + validation_session={ + "official_validation_ran": True, + "official_result": "fail", + "diagnostic_validation_ran": True, + "diagnostic_result": "pass", + "diagnostic_edits": ["tests/test_example.py"], + "worktree_dirty_after_official": False, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_post_edit_tests_as_official_blocks(self): + report = "\n".join([ + "Official validation result: passed after diagnostic edit", + "pytest passed after local fix in validation worktree", + ]) + result = assess_validation_integrity_report( + report, + validation_session={ + "official_validation_ran": True, + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_validation_ran": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_missing_dirty_after_validation_blocks(self): + report = _official_only_report().replace( + "Worktree dirty after official validation: clean", + "", + ) + result = assess_validation_integrity_report( + report, + validation_session={"official_validation_ran": True, "official_result": "fail"}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("dirty" in r.lower() for r in result["reasons"])) + + def test_diagnostic_without_label_blocks(self): + report = _diagnostic_report().replace( + "Diagnostic local experiment — not PR-head validation", + "", + ) + result = assess_validation_integrity_report( + report, + validation_session={ + "official_validation_ran": True, + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_validation_ran": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_contaminated_requires_integrity_status(self): + report = "Validation worktree was edited before official validation completed." + result = assess_validation_integrity_report( + report, + validation_session={"contaminated": True, "official_validation_ran": True}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"])) + + def test_action_log_edits_trigger_diagnostic_rules(self): + result = assess_validation_integrity_report( + _official_only_report(), + validation_session={"official_validation_ran": True, "official_result": "fail"}, + action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}], + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_integrity_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() From b71d3aeab38b1c05ff74f1c23e0d34f77e2a5ca4 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:31:24 -0400 Subject: [PATCH 2/3] Add precise mutation category verifier for controller handoffs (#319) Reject legacy Workspace mutations wording and require explicit file, worktree/index, and git ref mutation fields in reviewer controller handoffs. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 ++ reviewer_mutation_categories.py | 130 +++++++++++++++++++++ tests/test_llm_workflow_split.py | 6 + tests/test_reviewer_mutation_categories.py | 88 ++++++++++++++ 4 files changed, 231 insertions(+) create mode 100644 reviewer_mutation_categories.py create mode 100644 tests/test_reviewer_mutation_categories.py diff --git a/review_proofs.py b/review_proofs.py index e7c94e2..add5ddf 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3839,3 +3839,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_mutation_categories_report(report_text, **kwargs): + """#319: require precise mutation categories in reviewer controller handoffs.""" + from reviewer_mutation_categories import assess_mutation_categories_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_mutation_categories.py b/reviewer_mutation_categories.py new file mode 100644 index 0000000..a09e73b --- /dev/null +++ b/reviewer_mutation_categories.py @@ -0,0 +1,130 @@ +"""Precise mutation category verifier for reviewer controller handoffs (#319).""" + +from __future__ import annotations + +import re +from typing import Any + +_CONTROLLER_HANDOFF_RE = re.compile(r"controller handoff", re.IGNORECASE) +_WORKSPACE_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*workspace\s+mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_VAGUE_MUTATIONS_NONE_RE = re.compile( + r"^\s*[-*]?\s*mutations\s*:\s*none\s*$", + re.IGNORECASE | re.MULTILINE, +) + +_REQUIRED_CATEGORIES = ( + "file edits by reviewer", + "worktree/index mutations", + "git ref mutations", +) +_WORKTREE_FIELD_ALIASES = ( + "worktree/index mutations", + "worktree mutations", +) + + +def _has_category(text: str, category: str) -> bool: + pattern = re.compile( + rf"^\s*[-*]?\s*{re.escape(category)}\s*:", + re.IGNORECASE | re.MULTILINE, + ) + return bool(pattern.search(text)) + + +def _has_worktree_category(text: str) -> bool: + return any(_has_category(text, alias) for alias in _WORKTREE_FIELD_ALIASES) + + +def _normalize_for_consistency_check(text: str) -> str: + """Map #319 precise labels to #313 field names for consistency delegation.""" + normalized = text + if _has_category(text, "worktree/index mutations") and not _has_category( + text, "worktree mutations" + ): + normalized = re.sub( + r"(worktree/index mutations\s*:)", + "Worktree mutations:", + normalized, + flags=re.IGNORECASE, + ) + return normalized + + +def assess_mutation_categories_report( + report_text: str, + *, + handoff_session: dict | None = None, + observed_commands: list | None = None, +) -> dict[str, Any]: + """Require precise mutation categories in reviewer controller handoffs (#319).""" + text = report_text or "" + session = dict(handoff_session or {}) + reasons: list[str] = [] + lower = text.lower() + + is_controller = bool( + session.get("controller_handoff") + or _CONTROLLER_HANDOFF_RE.search(text) + ) + if not is_controller and not session.get("require_categories"): + return { + "proven": True, + "block": False, + "reasons": [], + "controller_handoff": False, + "safe_next_action": "proceed", + } + + if _WORKSPACE_MUTATIONS_RE.search(text): + reasons.append( + "controller handoffs must not use legacy 'Workspace mutations'; " + "use precise mutation category fields" + ) + + if _VAGUE_MUTATIONS_NONE_RE.search(text): + reasons.append( + "controller handoffs must not use vague 'Mutations: none'; " + "report each precise category separately" + ) + + missing = [ + category + for category in _REQUIRED_CATEGORIES + if category != "worktree/index mutations" and not _has_category(text, category) + ] + if not _has_worktree_category(text): + missing.append("worktree/index mutations") + + if missing: + reasons.append( + "controller handoff missing precise mutation categories: " + + ", ".join(missing) + ) + + commands = observed_commands or session.get("observed_commands") + if commands: + from review_proofs import assess_workspace_mutation_consistency + + consistency = assess_workspace_mutation_consistency( + _normalize_for_consistency_check(text), + commands, + ) + if not consistency.get("complete"): + reasons.extend(consistency.get("reasons") or []) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "controller_handoff": True, + "safe_next_action": ( + "replace Workspace mutations with file edits, worktree/index, git ref, " + "and other precise mutation category fields" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 9a96d17..ba4df08 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -134,3 +134,9 @@ def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof assert callable(assess_non_mergeable_skip_proof) + + +def test_mutation_categories_verifier_exported(): + from review_proofs import assess_mutation_categories_report + + assert callable(assess_mutation_categories_report) diff --git a/tests/test_reviewer_mutation_categories.py b/tests/test_reviewer_mutation_categories.py new file mode 100644 index 0000000..055b174 --- /dev/null +++ b/tests/test_reviewer_mutation_categories.py @@ -0,0 +1,88 @@ +"""Tests for precise mutation category verifier (#319).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_mutation_categories import assess_mutation_categories_report # noqa: E402 + + +def _precise_handoff() -> str: + return "\n".join([ + "Controller Handoff", + "File edits by reviewer: none", + "Worktree/index mutations: git worktree add branches/review-pr1", + "Git ref mutations: git fetch prgs master", + "MCP/Gitea mutations: gitea_view_pr", + "Review mutations: gitea_review_pr request_changes", + "Read-only diagnostics: git status, git diff", + ]) + + +class TestMutationCategories(unittest.TestCase): + def test_precise_categories_pass(self): + result = assess_mutation_categories_report(_precise_handoff()) + self.assertTrue(result["proven"], result["reasons"]) + + def test_legacy_workspace_mutations_blocks(self): + report = "\n".join([ + "Controller Handoff", + "Workspace mutations: none (no local file changes)", + "File edits by reviewer: none", + "Worktree/index mutations: none", + "Git ref mutations: none", + ]) + result = assess_mutation_categories_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_workspace_none_with_worktree_command_blocks(self): + report = "\n".join([ + "Controller Handoff", + "Workspace mutations: none (no local file changes)", + "File edits by reviewer: none", + "Worktree/index mutations: none", + "Git ref mutations: none", + ]) + result = assess_mutation_categories_report( + report, + observed_commands=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + + def test_file_edits_none_with_worktree_mutation_passes(self): + report = "\n".join([ + "Controller Handoff", + "File edits by reviewer: none", + "Worktree/index mutations: git reset --hard FETCH_HEAD", + "Git ref mutations: git fetch prgs", + ]) + result = assess_mutation_categories_report( + report, + observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_missing_required_categories_blocks(self): + report = "Controller Handoff\nFile edits by reviewer: none\n" + result = assess_mutation_categories_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("missing" in r.lower() for r in result["reasons"])) + + def test_non_controller_handoff_skips(self): + result = assess_mutation_categories_report( + "Workspace mutations: none\n", + ) + self.assertTrue(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_mutation_categories_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 275a362a4650e8c77186eaae70e6eb0cf28519d5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:41:33 -0400 Subject: [PATCH 3/3] feat: add issue claim heartbeat leases and stale-claim reconciliation (Closes #268) Structured claim/progress heartbeats post on issue claim and via gitea_post_heartbeat. Read-only gitea_reconcile_issue_claims inventories active, stale, phantom, and PR-backed claims; gitea_cleanup_stale_claims supports dry-run cleanup of reclaimable labels. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 275 ++++++++++++++++++++++++- issue_claim_heartbeat.py | 295 +++++++++++++++++++++++++++ task_capability_map.py | 12 ++ tests/test_issue_claim_heartbeat.py | 116 +++++++++++ tests/test_issue_write_tool_gates.py | 2 + tests/test_mcp_server.py | 4 +- 6 files changed, 702 insertions(+), 2 deletions(-) create mode 100644 issue_claim_heartbeat.py create mode 100644 tests/test_issue_claim_heartbeat.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b5415eb..dbe1c2e 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -476,6 +476,7 @@ import task_capability_map # noqa: E402 import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 +import issue_claim_heartbeat # noqa: E402 import merged_cleanup_reconcile # noqa: E402 @@ -5146,6 +5147,41 @@ def gitea_audit_config() -> dict: return report +def _post_structured_issue_comment( + *, + issue_number: int, + body: str, + remote: str, + host: str | None, + org: str | None, + repo: str | None, + audit_op: str = "create_issue_comment", +) -> dict: + """Post an issue-thread comment after permission gates already passed.""" + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments" + with _audited( + audit_op, + host=h, + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + request_metadata={"body_chars": len(body)}, + ): + data = api_request("POST", api, auth, {"body": body}) + result = { + "success": True, + "performed": True, + "comment_id": data["id"], + "issue_number": issue_number, + } + if _reveal_endpoints(): + result["url"] = data.get("html_url") + return result + + @mcp.tool() def gitea_mark_issue( issue_number: int, @@ -5155,6 +5191,8 @@ def gitea_mark_issue( org: str | None = None, repo: str | None = None, worktree_path: str | None = None, + branch_name: str | None = None, + profile: str | None = None, ) -> dict: """Claim or release an issue via the status:in-progress label. @@ -5204,7 +5242,31 @@ def gitea_mark_issue( request_metadata={"op": "add", "label": "status:in-progress"}): api_request("POST", f"{base}/issues/{issue_number}/labels", auth, {"labels": [label_id]}) - return {"success": True, "message": f"Issue #{issue_number} claimed."} + active_profile = (profile or get_profile().get("profile_name") or "unknown") + branch = (branch_name or "pending").strip() or "pending" + heartbeat_body = issue_claim_heartbeat.format_heartbeat_body( + kind="claim", + issue_number=issue_number, + branch=branch, + phase="claimed", + profile=active_profile, + next_action="create worktree and begin implementation", + ) + heartbeat = _post_structured_issue_comment( + issue_number=issue_number, + body=heartbeat_body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="claim_heartbeat", + ) + return { + "success": True, + "message": f"Issue #{issue_number} claimed.", + "heartbeat_posted": heartbeat.get("success", False), + "heartbeat_comment_id": heartbeat.get("comment_id"), + } else: with _audited("unlabel_issue", host=h, remote=remote, org=o, repo=r, issue_number=issue_number, @@ -5214,6 +5276,217 @@ def gitea_mark_issue( return {"success": True, "message": f"Issue #{issue_number} released."} +@mcp.tool() +def gitea_post_heartbeat( + issue_number: int, + branch: str, + phase: str, + pr: str = "none", + next_action: str = "none", + blocker: str = "none", + profile: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Post a structured progress heartbeat on a claimed issue (#268).""" + blocked = _profile_permission_block( + task_capability_map.required_permission("post_heartbeat")) + if blocked: + return blocked + verify_preflight_purity(remote) + active_profile = profile or get_profile().get("profile_name") + body = issue_claim_heartbeat.format_heartbeat_body( + kind="progress", + issue_number=issue_number, + branch=branch, + phase=phase, + profile=active_profile, + pr=pr, + next_action=next_action, + blocker=blocker, + ) + return _post_structured_issue_comment( + issue_number=issue_number, + body=body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="progress_heartbeat", + ) + + +@mcp.tool() +def gitea_reconcile_issue_claims( + state: str = "open", + stale_after_hours: int = 24, + heartbeat_lease_minutes: int = 30, + limit: int = 100, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Read-only inventory of issue claims and heartbeat lease status (#268).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + issues = api_get_all(f"{base}/issues?state={state}&type=issues", auth, limit=limit) + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + branches = api_get_all(f"{base}/branches", auth, limit=limit) + branch_names = [b.get("name") for b in branches if b.get("name")] + + comments_by_issue: dict[int, list[dict]] = {} + for issue in issues: + if not issue_claim_heartbeat.issue_has_in_progress_label(issue): + continue + number = int(issue["number"]) + api = f"{base}/issues/{number}/comments" + comments_by_issue[number] = api_request("GET", api, auth) or [] + + reclaim_after_minutes = max(heartbeat_lease_minutes * 2, 60) + inventory = issue_claim_heartbeat.build_claim_inventory( + issues=issues, + comments_by_issue=comments_by_issue, + open_prs=open_prs, + branch_names=branch_names, + heartbeat_lease_minutes=heartbeat_lease_minutes, + reclaim_after_minutes=reclaim_after_minutes, + ) + inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory) + inventory["success"] = True + inventory["performed"] = False + return inventory + + +@mcp.tool() +def gitea_cleanup_stale_claims( + dry_run: bool = True, + execute_confirmed: bool = False, + heartbeat_lease_minutes: int = 30, + limit: int = 100, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Propose or execute stale-claim cleanup for phantom/reclaimable issues (#268).""" + 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"), + } + + inventory = gitea_reconcile_issue_claims( + heartbeat_lease_minutes=heartbeat_lease_minutes, + limit=limit, + remote=remote, + host=host, + org=org, + repo=repo, + ) + if not inventory.get("success"): + return inventory + + plan = [ + entry + for entry in inventory.get("cleanup_plan") or [] + if entry.get("action") == "remove_status_in_progress_and_comment" + ] + report = { + "success": True, + "dry_run": dry_run, + "performed": False, + "planned_actions": plan, + "inventory_counts": inventory.get("counts"), + } + if dry_run: + return report + + if not execute_confirmed: + raise ValueError( + "execute_confirmed must be True when dry_run=False (fail closed)" + ) + + blocked = _profile_permission_block( + task_capability_map.required_permission("cleanup_stale_claims")) + if blocked: + return blocked + verify_preflight_purity(remote) + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + labels = api_request("GET", f"{base}/labels?limit=100", auth) + label_id = next( + (lb["id"] for lb in labels if lb.get("name") == "status:in-progress"), + None, + ) + if label_id is None: + raise RuntimeError("Label 'status:in-progress' not found") + + actions: list[dict] = [] + active_profile = get_profile().get("profile_name") + for entry in plan: + issue_number = int(entry["issue_number"]) + with _audited( + "unlabel_issue", + host=h, + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + request_metadata={"op": "cleanup_stale_claim", "label": "status:in-progress"}, + ): + api_request( + "DELETE", + f"{base}/issues/{issue_number}/labels/{label_id}", + auth, + ) + cleanup_body = issue_claim_heartbeat.format_heartbeat_body( + kind="cleanup", + issue_number=issue_number, + branch="none", + phase="stale-claim-cleanup", + profile=active_profile, + next_action="issue reclaimable by queue", + blocker=entry.get("status") or "stale", + ) + comment = _post_structured_issue_comment( + issue_number=issue_number, + body=cleanup_body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="cleanup_heartbeat", + ) + actions.append( + { + "issue_number": issue_number, + "label_removed": True, + "cleanup_comment_id": comment.get("comment_id"), + } + ) + + report["performed"] = True + report["actions"] = actions + return report + + @mcp.tool() def gitea_list_labels( remote: str = "dadeschools", diff --git a/issue_claim_heartbeat.py b/issue_claim_heartbeat.py new file mode 100644 index 0000000..3d16e1d --- /dev/null +++ b/issue_claim_heartbeat.py @@ -0,0 +1,295 @@ +"""Issue claim heartbeat leases and stale-claim reconciliation (#268). + +Structured issue-thread comments prove live ownership beyond the +``status:in-progress`` label alone. Queue inventory can classify claims as +active, stale, reclaimable, PR-backed, or phantom. +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Any + +MARKER = "" +IN_PROGRESS_LABEL = "status:in-progress" + +_KIND_CLAIM = "claim" +_KIND_PROGRESS = "progress" +_KIND_CLEANUP = "cleanup" + +_FIELD_RE = re.compile( + r"^\s*-\s*([a-z_]+)\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE) + + +def _parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +def format_heartbeat_body( + *, + kind: str, + issue_number: int, + branch: str, + phase: str, + profile: str | None = None, + pr: str = "none", + next_action: str = "none", + blocker: str = "none", +) -> str: + """Return a structured, machine-parseable issue comment body.""" + profile_value = (profile or "unknown").strip() or "unknown" + lines = [ + MARKER, + "**Issue claim heartbeat**", + f"- kind: {kind}", + f"- issue: #{issue_number}", + f"- branch: {branch}", + f"- phase: {phase}", + f"- profile: {profile_value}", + f"- pr: {pr}", + f"- blocker: {blocker}", + f"- next_action: {next_action}", + ] + return "\n".join(lines) + + +def parse_heartbeat_comment(body: str) -> dict[str, Any] | None: + """Parse one structured heartbeat comment, or None when not a heartbeat.""" + text = body or "" + if MARKER not in text: + return None + fields: dict[str, str] = {} + for match in _FIELD_RE.finditer(text): + fields[match.group(1).strip().lower()] = match.group(2).strip() + if not fields: + return None + issue_raw = fields.get("issue", "") + issue_digits = re.sub(r"[^\d]", "", issue_raw) + issue_number = int(issue_digits) if issue_digits.isdigit() else None + return { + "kind": fields.get("kind"), + "issue_number": issue_number, + "branch": fields.get("branch"), + "phase": fields.get("phase"), + "profile": fields.get("profile"), + "pr": fields.get("pr"), + "blocker": fields.get("blocker"), + "next_action": fields.get("next_action"), + "raw_fields": fields, + } + + +def extract_issue_heartbeats( + comments: list[dict], + *, + issue_number: int | None = None, +) -> list[dict]: + """Return parsed heartbeats newest-last, optionally filtered to *issue_number*.""" + heartbeats: list[dict] = [] + for comment in comments or []: + parsed = parse_heartbeat_comment(comment.get("body") or "") + if not parsed: + continue + if issue_number is not None and parsed.get("issue_number") != issue_number: + continue + heartbeats.append( + { + **parsed, + "comment_id": comment.get("id"), + "author": (comment.get("user") or {}).get("login") + or comment.get("author"), + "created_at": comment.get("created_at"), + "updated_at": comment.get("updated_at"), + } + ) + return heartbeats + + +def issue_has_in_progress_label(issue: dict) -> bool: + labels = issue.get("labels") or [] + for label in labels: + name = label if isinstance(label, str) else label.get("name") + if name == IN_PROGRESS_LABEL: + return True + return False + + +def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None: + pattern = f"issue-{issue_number}" + closes = f"closes #{issue_number}" + fixes = f"fixes #{issue_number}" + for pr in open_prs or []: + head = (pr.get("head") or {}).get("ref") or "" + text = f"{pr.get('title', '')} {pr.get('body', '')}".lower() + if pattern in head.lower(): + return pr + if closes in text or fixes in text: + return pr + return None + + +def _matching_branch_names(issue_number: int, branch_names: list[str]) -> list[str]: + pattern = f"issue-{issue_number}" + return [name for name in branch_names if pattern in (name or "").lower()] + + +def classify_issue_claim( + *, + issue: dict, + comments: list[dict], + open_prs: list[dict] | None = None, + branch_names: list[str] | None = None, + now: datetime | None = None, + heartbeat_lease_minutes: int = 30, + reclaim_after_minutes: int = 60, +) -> dict[str, Any]: + """Classify one issue's claim state from labels, heartbeats, PRs, and branches.""" + issue_number = int(issue.get("number") or 0) + open_prs = open_prs or [] + branch_names = branch_names or [] + current = now or datetime.now(timezone.utc) + + has_label = issue_has_in_progress_label(issue) + heartbeats = extract_issue_heartbeats(comments, issue_number=issue_number) + latest = heartbeats[-1] if heartbeats else None + latest_at = _parse_timestamp( + (latest or {}).get("updated_at") or (latest or {}).get("created_at") + ) + age_minutes = None + if latest_at: + age_minutes = int((current - latest_at).total_seconds() // 60) + + linked_pr = _linked_open_pr(issue_number, open_prs) + matching_branches = _matching_branch_names(issue_number, branch_names) + + if not has_label: + status = "not_claimed" + reasons = ["issue lacks status:in-progress label"] + elif linked_pr: + status = "awaiting_review" + reasons = [f"open PR #{linked_pr.get('number')} covers this issue"] + elif not heartbeats: + status = "phantom" + reasons = [ + "status:in-progress label present without structured claim heartbeat" + ] + elif latest_at is None: + status = "phantom" + reasons = ["heartbeat comments present but timestamps could not be parsed"] + elif age_minutes is not None and age_minutes <= heartbeat_lease_minutes: + status = "active" + reasons = [f"heartbeat age {age_minutes}m within {heartbeat_lease_minutes}m lease"] + elif matching_branches and age_minutes is not None and age_minutes <= reclaim_after_minutes: + status = "active" + reasons = [ + f"matching branch(es) {', '.join(matching_branches)} with heartbeat " + f"age {age_minutes}m" + ] + elif age_minutes is not None and age_minutes > reclaim_after_minutes and not matching_branches: + status = "reclaimable" + reasons = [ + f"no heartbeat within {reclaim_after_minutes}m and no matching branch" + ] + elif age_minutes is not None and age_minutes > heartbeat_lease_minutes: + status = "stale" + reasons = [ + f"heartbeat age {age_minutes}m exceeds {heartbeat_lease_minutes}m lease" + ] + else: + status = "active" + reasons = ["claim has structured heartbeat proof"] + + return { + "issue_number": issue_number, + "title": issue.get("title"), + "status": status, + "has_in_progress_label": has_label, + "heartbeat_count": len(heartbeats), + "latest_heartbeat": latest, + "heartbeat_age_minutes": age_minutes, + "linked_open_pr": linked_pr.get("number") if linked_pr else None, + "matching_branches": matching_branches, + "reasons": reasons, + "reclaimable": status == "reclaimable", + "stale": status in {"stale", "phantom", "reclaimable"}, + } + + +def build_claim_inventory( + *, + issues: list[dict], + comments_by_issue: dict[int, list[dict]], + open_prs: list[dict], + branch_names: list[str], + now: datetime | None = None, + heartbeat_lease_minutes: int = 30, + reclaim_after_minutes: int = 60, +) -> dict[str, Any]: + """Build a queue inventory report for in-progress issue claims.""" + entries: list[dict] = [] + for issue in issues or []: + if not issue_has_in_progress_label(issue): + continue + number = int(issue["number"]) + entry = classify_issue_claim( + issue=issue, + comments=comments_by_issue.get(number, []), + open_prs=open_prs, + branch_names=branch_names, + now=now, + heartbeat_lease_minutes=heartbeat_lease_minutes, + reclaim_after_minutes=reclaim_after_minutes, + ) + entries.append(entry) + + counts: dict[str, int] = {} + for entry in entries: + counts[entry["status"]] = counts.get(entry["status"], 0) + 1 + + return { + "entries": entries, + "counts": counts, + "heartbeat_lease_minutes": heartbeat_lease_minutes, + "reclaim_after_minutes": reclaim_after_minutes, + "in_progress_total": len(entries), + } + + +def build_cleanup_plan(inventory: dict[str, Any]) -> list[dict]: + """Return stale/reclaimable/phantom claims that may be cleaned up.""" + plan: list[dict] = [] + for entry in inventory.get("entries") or []: + if entry.get("status") in {"reclaimable", "phantom"}: + plan.append( + { + "issue_number": entry["issue_number"], + "status": entry["status"], + "action": "remove_status_in_progress_and_comment", + "reasons": list(entry.get("reasons") or []), + } + ) + elif entry.get("status") == "stale" and not entry.get("linked_open_pr"): + plan.append( + { + "issue_number": entry["issue_number"], + "status": entry["status"], + "action": "report_only", + "reasons": list(entry.get("reasons") or []), + } + ) + return plan \ No newline at end of file diff --git a/task_capability_map.py b/task_capability_map.py index 9d162c1..2b794e7 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -96,6 +96,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + "post_heartbeat": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "reconcile_issue_claims": { + "permission": "gitea.read", + "role": "author", + }, + "cleanup_stale_claims": { + "permission": "gitea.issue.comment", + "role": "author", + }, } # Issue-mutating MCP tools and their resolver task keys. diff --git a/tests/test_issue_claim_heartbeat.py b/tests/test_issue_claim_heartbeat.py new file mode 100644 index 0000000..c8a154e --- /dev/null +++ b/tests/test_issue_claim_heartbeat.py @@ -0,0 +1,116 @@ +"""Tests for issue claim heartbeat leases (#268).""" + +from __future__ import annotations + +import sys +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_claim_heartbeat as ich # noqa: E402 + + +class TestHeartbeatFormatting(unittest.TestCase): + def test_format_and_parse_roundtrip(self): + body = ich.format_heartbeat_body( + kind="claim", + issue_number=268, + branch="feat/issue-268-heartbeat-leases", + phase="claimed", + profile="prgs-author", + next_action="implement module", + ) + parsed = ich.parse_heartbeat_comment(body) + self.assertIsNotNone(parsed) + assert parsed is not None + self.assertEqual(parsed["issue_number"], 268) + self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases") + self.assertEqual(parsed["phase"], "claimed") + + +class TestClaimClassification(unittest.TestCase): + NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc) + + def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict: + ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat() + body = ich.format_heartbeat_body( + kind=kind, + issue_number=268, + branch="feat/issue-268-heartbeat-leases", + phase="implementation", + profile="prgs-author", + ) + return {"id": 1, "body": body, "created_at": ts, "updated_at": ts} + + def test_active_claim_within_lease(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=5)], + now=self.NOW, + ) + self.assertEqual(result["status"], "active") + + def test_stale_claim_without_branch(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=45)], + now=self.NOW, + ) + self.assertEqual(result["status"], "stale") + + def test_reclaimable_after_sixty_minutes_without_branch(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=90)], + now=self.NOW, + ) + self.assertEqual(result["status"], "reclaimable") + + def test_awaiting_review_when_open_pr_exists(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + open_prs = [ + { + "number": 400, + "title": "feat", + "body": "Closes #268", + "head": {"ref": "feat/issue-268-heartbeat-leases"}, + } + ] + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=120)], + open_prs=open_prs, + now=self.NOW, + ) + self.assertEqual(result["status"], "awaiting_review") + + def test_phantom_claim_without_heartbeat(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[{"id": 1, "body": "working on it"}], + now=self.NOW, + ) + self.assertEqual(result["status"], "phantom") + + +class TestInventory(unittest.TestCase): + def test_build_cleanup_plan_flags_phantom(self): + inventory = { + "entries": [ + {"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]}, + {"issue_number": 2, "status": "active", "reasons": []}, + ] + } + plan = ich.build_cleanup_plan(inventory) + self.assertEqual(len(plan), 1) + self.assertEqual(plan[0]["issue_number"], 1) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_issue_write_tool_gates.py b/tests/test_issue_write_tool_gates.py index afe865d..736b295 100644 --- a/tests/test_issue_write_tool_gates.py +++ b/tests/test_issue_write_tool_gates.py @@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase): return [{"id": 10, "name": "status:in-progress"}] if method == "POST" and "/issues/9/labels" in url: return [{"name": "status:in-progress"}] + if method == "POST" and "/issues/9/comments" in url: + return {"id": 501} if method == "GET" and url.endswith("/user"): return {"login": "author-user"} return {} diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 394ab03..99d7d73 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_start_adds_label(self, _auth, mock_api): - # First call: get labels; second call: add label + # labels, add label, post claim heartbeat comment mock_api.side_effect = [ [{"id": 10, "name": "status:in-progress"}], [{"name": "status:in-progress"}], + {"id": 99}, ] with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): result = gitea_mark_issue(issue_number=5, action="start") self.assertTrue(result["success"]) self.assertIn("claimed", result["message"]) + self.assertTrue(result.get("heartbeat_posted")) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)