"""Tests for self-propagating canonical handoffs (#626).""" from __future__ import annotations import os import sys import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from final_report_validator import assess_final_report_validator # noqa: E402 from self_propagating_handoff import ( # noqa: E402 HANDOFF_FIELDS, NEXT_ACTOR_BY_STATE, WORKFLOW_STATES, assess_controller_decision, assess_durable_state_update, assess_final_report_self_propagating_handoff, assess_handoff_live_state, assess_merge_completion_transition, assess_role_continuation, assess_self_propagating_handoff, assess_thread_recoverability, assess_workflow_failure_escalation, parse_self_propagating_handoff, render_self_propagating_handoff, ) REPO = "Scaled-Tech-Consulting/Gitea-Tools" AUTHOR_PROMPT = ( "Review PR #900 on Scaled-Tech-Consulting/Gitea-Tools for issue 626 at head " "aaaa111. Validate the branch, then submit an independent review verdict." ) REVIEWER_PROMPT = ( "Merge PR #900 on Scaled-Tech-Consulting/Gitea-Tools for issue 626 once the " "approval at head aaaa111 still applies to the live head." ) MERGER_PROMPT = ( "Accept or reject the merged work for issue 626 on " "Scaled-Tech-Consulting/Gitea-Tools; verify acceptance criteria then close." ) CONTROLLER_PROMPT = ( "Address the controller's requested changes for issue 626 on " "Scaled-Tech-Consulting/Gitea-Tools, then hand back to an independent reviewer." ) def build_handoff(**overrides): """Render a valid author -> reviewer handoff, with overrides applied.""" values = { "REPOSITORY": REPO, "ISSUE": "626", "PR": "900", "WORKFLOW_STATE": "needs-review", "HEAD_SHA": "aaaa111", "BASE_BRANCH": "master", "BASE_OR_MERGE_SHA": "bbbb222", "ACTING_ROLE": "author", "ACTING_IDENTITY": "jcwalker3 (prgs-author)", "COMPLETED_ACTIONS": "implemented AC1-AC9; opened PR #900", "VALIDATION_EVIDENCE": "pytest tests/test_self_propagating_handoff.py: 20 passed", "MUTATION_LEDGER": "branch pushed; PR #900 opened; comment 13547 posted", "BLOCKERS": "none", "NEXT_ACTOR": "reviewer", "NEXT_ACTION": "independently review PR #900 at head aaaa111", "PROHIBITED_ACTIONS": "merge, self-approve, force-push", "NEXT_PROMPT": AUTHOR_PROMPT, "WORKFLOW_FAILURE_ISSUES": "none", "LAST_UPDATED": "2026-07-21T03:55:00Z", } values.update(overrides) return render_self_propagating_handoff(**values) class RenderAndParseTests(unittest.TestCase): def test_render_emits_every_canonical_field(self): body = build_handoff() parsed = parse_self_propagating_handoff(body) self.assertIsNotNone(parsed) for name in HANDOFF_FIELDS: self.assertIn(name, parsed) def test_render_rejects_unknown_workflow_state(self): with self.assertRaises(ValueError): build_handoff(WORKFLOW_STATE="almost-done") def test_every_state_maps_to_exactly_one_actor(self): self.assertEqual(set(WORKFLOW_STATES), set(NEXT_ACTOR_BY_STATE)) def test_absent_block_parses_as_none(self): self.assertIsNone(parse_self_propagating_handoff("no handoff here")) class AuthorToReviewerTests(unittest.TestCase): """Scenario 1: author -> reviewer.""" def test_valid_author_handoff_passes(self): result = assess_self_propagating_handoff(build_handoff()) self.assertTrue(result["valid"], result["reasons"]) self.assertEqual(result["next_actor"], "reviewer") self.assertFalse(result["terminal"]) def test_reviewer_may_continue_author_handoff(self): result = assess_role_continuation( handoff=build_handoff(), actor_role="reviewer" ) self.assertTrue(result["allowed"], result["reasons"]) self.assertIn("review", result["allowed_actions"]) def test_author_may_not_continue_its_own_handoff(self): result = assess_role_continuation( handoff=build_handoff(), actor_role="author" ) self.assertTrue(result["block"]) self.assertEqual(result["expected_actor"], "reviewer") def test_next_actor_must_match_declared_state(self): result = assess_self_propagating_handoff( build_handoff(NEXT_ACTOR="merger") ) self.assertTrue(result["block"]) self.assertTrue( any("does not match state" in reason for reason in result["reasons"]) ) class ReviewerToMergerTests(unittest.TestCase): """Scenario 2: reviewer -> merger.""" def build(self, **overrides): values = { "WORKFLOW_STATE": "approved-awaiting-merge", "ACTING_ROLE": "reviewer", "ACTING_IDENTITY": "reviewer-bot (prgs-reviewer)", "COMPLETED_ACTIONS": "review 500 APPROVED at aaaa111", "NEXT_ACTOR": "merger", "NEXT_ACTION": "merge PR #900 at approved head aaaa111", "PROHIBITED_ACTIONS": "re-review, commit, push", "NEXT_PROMPT": REVIEWER_PROMPT, } values.update(overrides) return build_handoff(**values) def test_reviewer_handoff_is_valid(self): result = assess_self_propagating_handoff(self.build()) self.assertTrue(result["valid"], result["reasons"]) self.assertEqual(result["next_actor"], "merger") def test_merger_may_continue(self): result = assess_role_continuation(handoff=self.build(), actor_role="merger") self.assertTrue(result["allowed"], result["reasons"]) self.assertIn("merge", result["allowed_actions"]) class MergerToControllerTests(unittest.TestCase): """Scenario 3: merger -> controller.""" def test_merge_success_stops_at_controller_boundary(self): result = assess_merge_completion_transition(merge_succeeded=True) self.assertEqual(result["next_state"], "merged-awaiting-controller") self.assertEqual(result["next_actor"], "controller") self.assertTrue(result["next_prompt_required"]) def test_configured_auto_accept_may_complete(self): result = assess_merge_completion_transition( merge_succeeded=True, controller_auto_accept=True ) self.assertEqual(result["next_state"], "complete") self.assertFalse(result["next_prompt_required"]) def test_failed_merge_keeps_the_work_item_with_the_merger(self): result = assess_merge_completion_transition(merge_succeeded=False) self.assertEqual(result["next_state"], "approved-awaiting-merge") def test_merger_handoff_names_the_controller(self): body = build_handoff( WORKFLOW_STATE="merged-awaiting-controller", ACTING_ROLE="merger", ACTING_IDENTITY="merger-bot (prgs-merger)", COMPLETED_ACTIONS="merged PR #900 as cccc333", BASE_OR_MERGE_SHA="cccc333", NEXT_ACTOR="controller", NEXT_ACTION="verify acceptance criteria and close issue 626", PROHIBITED_ACTIONS="reopen the PR, re-merge", NEXT_PROMPT=MERGER_PROMPT, ) result = assess_self_propagating_handoff(body) self.assertTrue(result["valid"], result["reasons"]) self.assertEqual(result["next_actor"], "controller") class ControllerBackToAuthorTests(unittest.TestCase): """Scenario 4: controller -> author.""" def test_request_corrections_returns_to_author(self): result = assess_controller_decision(decision="request_corrections") self.assertFalse(result["block"]) self.assertEqual(result["next_state"], "needs-author") self.assertTrue(result["next_prompt_required"]) def test_return_to_actor_requires_a_named_target(self): result = assess_controller_decision(decision="return_to_actor") self.assertTrue(result["block"]) def test_return_to_reviewer_is_supported(self): result = assess_controller_decision( decision="return_to_actor", return_to="reviewer" ) self.assertEqual(result["next_state"], "needs-review") def test_unknown_decision_fails_closed(self): result = assess_controller_decision(decision="looks-fine") self.assertTrue(result["block"]) def test_controller_handoff_back_to_author_validates(self): body = build_handoff( WORKFLOW_STATE="needs-author", ACTING_ROLE="controller", ACTING_IDENTITY="controller (operator)", COMPLETED_ACTIONS="reviewed merged work; requested corrections", NEXT_ACTOR="author", NEXT_ACTION="address controller corrections on issue 626", PROHIBITED_ACTIONS="close the issue, merge", NEXT_PROMPT=CONTROLLER_PROMPT, ) result = assess_self_propagating_handoff(body) self.assertTrue(result["valid"], result["reasons"]) class StaleHeadRejectionTests(unittest.TestCase): """Scenario 5: stale-head rejection.""" def test_changed_head_invalidates_a_merge_handoff(self): body = build_handoff( WORKFLOW_STATE="approved-awaiting-merge", ACTING_ROLE="reviewer", NEXT_ACTOR="merger", NEXT_ACTION="merge PR #900 at approved head aaaa111", NEXT_PROMPT=REVIEWER_PROMPT, ) result = assess_handoff_live_state( handoff=body, live={"pr_head_sha": "dddd444", "pr_state": "open"}, ) self.assertTrue(result["block"]) self.assertIn("changed_pr_head", result["kinds"]) self.assertEqual(result["recovered_state"], "needs-review") def test_stale_approval_blocks_the_merger(self): body = build_handoff( WORKFLOW_STATE="approved-awaiting-merge", NEXT_ACTOR="merger", NEXT_ACTION="merge PR #900", HEAD_SHA="dddd444", NEXT_PROMPT=REVIEWER_PROMPT, ) result = assess_handoff_live_state( handoff=body, live={"pr_head_sha": "dddd444", "approved_head_sha": "aaaa111"}, ) self.assertTrue(result["block"]) self.assertIn("stale_approval", result["kinds"]) def test_unchanged_head_is_not_blocked(self): result = assess_handoff_live_state( handoff=build_handoff(), live={ "pr_head_sha": "aaaa111", "pr_state": "open", "issue_state": "open", "base_branch": "master", "namespace_role": "reviewer", }, ) self.assertFalse(result["block"], result["reasons"]) def test_merged_pr_recovers_to_the_controller_boundary(self): result = assess_handoff_live_state( handoff=build_handoff(), live={"pr_head_sha": "aaaa111", "pr_state": "merged"}, ) self.assertIn("pr_merged", result["kinds"]) self.assertEqual(result["recovered_state"], "merged-awaiting-controller") def test_reopened_issue_invalidates_a_complete_handoff(self): body = build_handoff( WORKFLOW_STATE="complete", ACTING_ROLE="controller", NEXT_ACTOR="none", NEXT_ACTION="none", NEXT_PROMPT="none", ) result = assess_handoff_live_state( handoff=body, live={"issue_state": "open"} ) self.assertIn("issue_reopened", result["kinds"]) self.assertEqual(result["recovered_state"], "needs-author") def test_foreign_lease_and_worktree_faults_are_detected(self): result = assess_handoff_live_state( handoff=build_handoff(), live={ "pr_head_sha": "aaaa111", "lease": {"status": "expired", "session_id": "other-session"}, "actor_session_id": "my-session", "worktree": {"present": False, "dirty": True}, "namespace_role": "author", "runtime_stale": True, "base_branch": "dev", "conflicting_canonical_comments": True, }, ) for kind in ( "stale_lease", "foreign_lease", "missing_worktree", "dirty_worktree", "namespace_mismatch", "stale_runtime", "changed_base", "conflicting_canonical_comments", ): self.assertIn(kind, result["kinds"]) class BlockedInfrastructurePathTests(unittest.TestCase): """Scenario 6: blocked infrastructure path.""" def build(self, **overrides): values = { "WORKFLOW_STATE": "blocked", "PR": "none", "HEAD_SHA": "none", "ACTING_ROLE": "author", "COMPLETED_ACTIONS": "attempted native publish; MCP mutation rejected", "BLOCKERS": "gitea_create_pr rejected: namespace unreachable", "NEXT_ACTOR": "operator", "NEXT_ACTION": "restore the author MCP namespace", "PROHIBITED_ACTIONS": "raw git push, curl, force-push", "NEXT_PROMPT": ( "Repair the author MCP namespace for " "Scaled-Tech-Consulting/Gitea-Tools so issue 626 can publish " "natively, then hand back to the author." ), "WORKFLOW_FAILURE_ISSUES": "#640", } values.update(overrides) return build_handoff(**values) def test_blocked_handoff_without_pr_is_valid(self): result = assess_self_propagating_handoff(self.build()) self.assertTrue(result["valid"], result["reasons"]) self.assertEqual(result["next_actor"], "operator") def test_blocked_requires_a_concrete_blocker(self): result = assess_self_propagating_handoff(self.build(BLOCKERS="none")) self.assertTrue(result["block"]) self.assertTrue( any("BLOCKERS" in reason for reason in result["reasons"]) ) def test_operator_is_the_only_authorized_continuation(self): self.assertTrue( assess_role_continuation(handoff=self.build(), actor_role="operator")[ "allowed" ] ) self.assertTrue( assess_role_continuation(handoff=self.build(), actor_role="merger")["block"] ) class FinalClosureTests(unittest.TestCase): """Scenario 7: final successful closure.""" def build(self, **overrides): values = { "WORKFLOW_STATE": "complete", "ACTING_ROLE": "controller", "ACTING_IDENTITY": "controller (operator)", "COMPLETED_ACTIONS": "verified acceptance criteria; closed issue 626", "BASE_OR_MERGE_SHA": "cccc333", "NEXT_ACTOR": "none", "NEXT_ACTION": "none", "PROHIBITED_ACTIONS": "reopen without new evidence", "NEXT_PROMPT": "none", } values.update(overrides) return build_handoff(**values) def test_terminal_handoff_is_valid_without_a_next_prompt(self): result = assess_self_propagating_handoff(self.build()) self.assertTrue(result["valid"], result["reasons"]) self.assertTrue(result["terminal"]) def test_terminal_handoff_must_not_manufacture_more_work(self): result = assess_self_propagating_handoff( self.build(NEXT_PROMPT=CONTROLLER_PROMPT) ) self.assertTrue(result["block"]) self.assertTrue( any("must not carry a NEXT_PROMPT" in r for r in result["reasons"]) ) def test_no_role_may_continue_a_complete_workflow(self): result = assess_role_continuation(handoff=self.build(), actor_role="author") self.assertTrue(result["block"]) def test_controller_acceptance_requires_full_closure_proof(self): partial = assess_controller_decision( decision="accept", closure_proof={"acceptance_criteria_satisfied": True}, ) self.assertTrue(partial["block"]) self.assertEqual(partial["next_state"], "merged-awaiting-controller") full = assess_controller_decision( decision="accept", closure_proof={ "acceptance_criteria_satisfied": True, "cleanup_complete": True, "canonical_final_state_posted": True, "issue_closed_through_workflow": True, }, ) self.assertFalse(full["block"]) self.assertEqual(full["next_state"], "complete") self.assertFalse(full["next_prompt_required"]) class IncompleteHandoffRejectionTests(unittest.TestCase): """Scenario 8: incomplete handoff rejection.""" def test_missing_block_is_rejected(self): result = assess_self_propagating_handoff("Work is done, ping the reviewer.") self.assertTrue(result["block"]) self.assertFalse(result["present"]) def test_missing_field_is_rejected(self): body = build_handoff() body = "\n".join( line for line in body.splitlines() if not line.startswith("MUTATION_LEDGER:") ) result = assess_self_propagating_handoff(body) self.assertTrue(result["block"]) self.assertIn("MUTATION_LEDGER", result["missing_fields"]) def test_placeholder_field_is_rejected(self): result = assess_self_propagating_handoff( build_handoff(VALIDATION_EVIDENCE="TBD") ) self.assertTrue(result["block"]) def test_stub_next_prompt_is_rejected(self): result = assess_self_propagating_handoff(build_handoff(NEXT_PROMPT="review it")) self.assertTrue(result["block"]) self.assertTrue( any("ready-to-run" in reason for reason in result["reasons"]) ) def test_prompt_depending_on_outside_chat_is_rejected(self): prompt = ( "Continue issue 626 on Scaled-Tech-Consulting/Gitea-Tools using the " "previous chat for the missing details." ) result = assess_thread_recoverability(build_handoff(NEXT_PROMPT=prompt)) self.assertTrue(result["block"]) def test_prompt_must_name_repository_and_issue(self): prompt = ( "Please review the pull request at the current head and submit an " "independent verdict when validation passes." ) result = assess_thread_recoverability(build_handoff(NEXT_PROMPT=prompt)) self.assertTrue(result["block"]) def test_self_contained_prompt_is_recoverable(self): self.assertFalse(assess_thread_recoverability(build_handoff())["block"]) def test_chat_only_report_is_not_durable(self): result = assess_durable_state_update( handoff_text=build_handoff(), posted_comment_id=None, canonical_state_posted=False, ) self.assertTrue(result["block"]) self.assertEqual(len(result["reasons"]), 2) def test_posted_handoff_is_durable(self): result = assess_durable_state_update( handoff_text=build_handoff(), posted_comment_id=13550, canonical_state_posted=True, ) self.assertTrue(result["durable"], result["reasons"]) class WorkflowFailureEscalationTests(unittest.TestCase): """Scenario 9: duplicate workflow-failure issue handling.""" def failure(self, **overrides): values = { "signature": "lease-cleanup-internal-error", "classification": "mcp-tool-defect", "linked_issue": "718", "temporary_impact": "lease cleanup unavailable this session", "next_valid_actor": "operator", "recovery_prompt": "restart the namespace and re-run lease cleanup", } values.update(overrides) return values def test_complete_failure_record_passes(self): result = assess_workflow_failure_escalation( failures=[self.failure()], active_issue_number=626 ) self.assertTrue(result["escalated"], result["reasons"]) def test_incomplete_failure_record_fails_closed(self): result = assess_workflow_failure_escalation( failures=[self.failure(recovery_prompt="")], active_issue_number=626 ) self.assertTrue(result["block"]) def test_folding_into_the_active_issue_is_rejected(self): result = assess_workflow_failure_escalation( failures=[self.failure(linked_issue="626")], active_issue_number=626 ) self.assertTrue(result["block"]) self.assertTrue( any("folded into the active work item" in r for r in result["reasons"]) ) def test_known_signature_reuses_the_existing_issue(self): result = assess_workflow_failure_escalation( failures=[self.failure()], active_issue_number=626, existing_failure_issues=[ {"signature": "lease-cleanup-internal-error", "number": 718} ], ) self.assertTrue(result["escalated"], result["reasons"]) self.assertEqual(result["reused_issues"], [ {"signature": "lease-cleanup-internal-error", "issue": "718"} ]) def test_duplicate_issue_for_known_signature_is_rejected(self): result = assess_workflow_failure_escalation( failures=[self.failure(linked_issue="799")], active_issue_number=626, existing_failure_issues=[ {"signature": "lease-cleanup-internal-error", "number": 718} ], ) self.assertTrue(result["block"]) self.assertTrue( any("reuse the existing issue #718" in r for r in result["reasons"]) ) def test_same_signature_twice_in_one_session_is_rejected(self): result = assess_workflow_failure_escalation( failures=[self.failure(), self.failure()], active_issue_number=626 ) self.assertTrue(result["block"]) def test_no_failures_is_not_an_error(self): result = assess_workflow_failure_escalation( failures=[], active_issue_number=626 ) self.assertTrue(result["escalated"]) class FinalReportIntegrationTests(unittest.TestCase): def test_report_without_the_protocol_is_not_applicable(self): result = assess_final_report_self_propagating_handoff("## Controller Handoff\n") self.assertFalse(result["applicable"]) self.assertFalse(result["block"]) def test_report_with_a_complete_handoff_passes(self): result = assess_final_report_self_propagating_handoff(build_handoff()) self.assertTrue(result["applicable"]) self.assertFalse(result["block"], result["reasons"]) def test_report_with_an_incomplete_handoff_blocks(self): result = assess_final_report_self_propagating_handoff( build_handoff(NEXT_ACTION="") ) self.assertTrue(result["block"]) def test_validator_blocks_an_incomplete_handoff_in_a_work_issue_report(self): report = build_handoff(MUTATION_LEDGER="TBD") result = assess_final_report_validator(report, "work_issue") self.assertTrue(result["blocked"]) self.assertTrue( any( finding["rule_id"] == "shared.self_propagating_handoff" for finding in result["findings"] ) ) def test_validator_ignores_reports_that_predate_the_protocol(self): result = assess_final_report_validator("plain legacy report", "work_issue") self.assertFalse( any( finding["rule_id"] == "shared.self_propagating_handoff" for finding in result["findings"] ) ) if __name__ == "__main__": unittest.main()