Gitea-Tools/tests/test_author_issue_bootstrap.py

"""Regression test suite for native author issue worktree bootstrap (#850)."""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import tempfile
import unittest
from unittest import mock

import author_issue_bootstrap
import task_capability_map


def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict:
    repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args
    os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir
    return author_issue_bootstrap.bootstrap_author_issue_worktree(
        issue_number=issue_num,
        canonical_repo_root=repo_dir,
        expected_base_sha=master_sha,
        idempotency_key=key,
        lock_dir=lock_dir,
        owner_session="session-concurrent-test",
        active_identity="jcwalker3",
        active_profile="prgs-author",
    )


class TestAuthorIssueBootstrap(unittest.TestCase):
    """Test suite covering AC1-AC10 and comment #14959 specification."""

    def setUp(self):
        self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_")
        self.repo_dir = os.path.join(self.tmp_dir, "repo")
        os.makedirs(self.repo_dir)

        # Initialize synthetic git repo
        subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True)
        subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True)
        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo_dir, check=True)

        readme = os.path.join(self.repo_dir, "README.md")
        with open(readme, "w", encoding="utf-8") as f:
            f.write("# Test Repo\n")
        subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True)
        subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True)

        rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True)
        self.master_sha = rev_res.stdout.strip()

        self.branches_dir = os.path.join(self.repo_dir, "branches")
        os.makedirs(self.branches_dir, exist_ok=True)
        self.lock_dir = os.path.join(self.tmp_dir, "locks")
        os.makedirs(self.lock_dir, exist_ok=True)
        self.journal_dir = os.path.join(self.tmp_dir, "journals")
        os.makedirs(self.journal_dir, exist_ok=True)
        os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir

    def tearDown(self):
        os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None)
        shutil.rmtree(self.tmp_dir, ignore_errors=True)

    def test_bootstrap_success_path(self):
        """AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof."""
        key = "test_key_success_1"
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            # assignment_id/lease_id omitted: optional unless verified live.
            expected_base_sha=self.master_sha,
            idempotency_key=key,
            remote="prgs",
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res.get("success"), f"Bootstrap failed: {res}")
        self.assertFalse(res.get("replayed"))
        self.assertEqual(res.get("issue_number"), 850)
        self.assertEqual(res.get("base_sha"), self.master_sha)
        self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path"))

        # Verify worktree directory exists and is registered
        worktree_path = res["worktree_path"]
        self.assertTrue(os.path.isdir(worktree_path))

        wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True)
        self.assertIn(worktree_path, wt_list.stdout)

        # Verify phase journal written
        journal = author_issue_bootstrap.load_phase_journal(key, journal_dir=self.lock_dir)
        self.assertIsNotNone(journal)
        self.assertTrue(journal.get("completed"))
        self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED)

    def test_idempotent_replay(self):
        """Item 2: Replaying with identical key returns cached transition without duplicate creation."""
        key = "test_key_idempotent_1"
        res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res1["success"], f"res1 failed: {res1}")
        self.assertFalse(res1.get("replayed"))

        # Second call
        res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res2["success"], f"res2 failed: {res2}")
        self.assertTrue(res2.get("replayed"))
        self.assertEqual(res1["worktree_path"], res2["worktree_path"])

    def test_stale_concurrency_pin_refusal(self):
        """Item 3: Mismatched expected base SHA fails closed without silent rebasing."""
        stale_sha = "0000000000000000000000000000000000000000"
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            expected_base_sha=stale_sha,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
        self.assertIn("exact_next_action", res)

    def test_path_outside_branches_root_refusal(self):
        """Item 6: Worktree path outside branches/ root is refused."""
        outside_path = os.path.join(self.tmp_dir, "outside_worktree")
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            worktree_path=outside_path,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root")

    def test_preexisting_dirty_worktree_preservation(self):
        """Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned."""
        branch = "fix/issue-850-dirty-test"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)

        # Create dirty untracked file
        dirty_file = os.path.join(wt_path, "dirty.txt")
        with open(dirty_file, "w") as f:
            f.write("dirty edits\n")

        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name=branch,
            worktree_path=wt_path,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree")

        # Prove dirty file is preserved byte-for-byte
        self.assertTrue(os.path.exists(dirty_file))
        with open(dirty_file, "r") as f:
            self.assertEqual(f.read(), "dirty edits\n")

    def test_compensating_recovery_on_failed_phase(self):
        """AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts."""
        key = "test_key_recovery_1"
        # Simulate partial progress in journal
        journal = {
            "idempotency_key": key,
            "issue_number": 850,
            "branch_name": "fix/issue-850-recovery-test",
            "worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"),
            "artifacts_created": {
                "branch_created": True,
                "worktree_dir_created": True,
                "worktree_registered": True,
                "lock_created": False,
            },
            "failure_reason": "simulated lock failure",
            "current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED,
            "completed": False,
        }
        # Create the branch and worktree manually to simulate partial state
        subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True)

        # Run compensating recovery
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir)
        self.assertTrue(rec["executed"])
        self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"])
        self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"])

        # Prove worktree directory and branch were rolled back
        self.assertFalse(os.path.exists(journal["worktree_path"]))
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False)
        self.assertNotEqual(branch_check.returncode, 0)

    def test_cross_process_concurrency(self):
        """Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption."""
        import concurrent.futures

        key = "test_concurrent_key_850"
        args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha)

        with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
            fut1 = executor.submit(_concurrent_bootstrap_worker, args)
            fut2 = executor.submit(_concurrent_bootstrap_worker, args)
            res1 = fut1.result(timeout=10)
            res2 = fut2.result(timeout=10)

        self.assertTrue(res1["success"], f"res1 failed: {res1}")
        self.assertTrue(res2["success"], f"res2 failed: {res2}")
        # One process performs creation, the other process receives idempotent replay
        replayed_count = sum(1 for r in (res1, res2) if r.get("replayed"))
        created_count = sum(1 for r in (res1, res2) if not r.get("replayed"))
        self.assertEqual(replayed_count, 1)
        self.assertEqual(created_count, 1)
        self.assertEqual(res1["worktree_path"], res2["worktree_path"])

    def test_interrupted_replay_preserves_artifacts_created_provenance(self):
        """Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically."""
        key = "test_key_interrupted_replay_1"
        branch = "fix/issue-850-interrupted-replay"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay")

        # Simulate Phase 2/3 completion where branch and worktree directory were created by this transition
        journal = {
            "idempotency_key": key,
            "issue_number": 850,
            "branch_name": branch,
            "worktree_path": wt_path,
            "active_identity": "jcwalker3",
            "active_profile": "prgs-author",
            "remote": "prgs",
            "org": "Scaled-Tech-Consulting",
            "repo": "Gitea-Tools",
            "phases": {
                author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"},
                author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True},
            },
            "artifacts_created": {
                "branch_created": True,
                "worktree_dir_created": True,
                "worktree_registered": True,
                "lock_created": False,
            },
            "current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED,
            "completed": False,
        }
        # Pre-create the branch and worktree on disk to simulate partial state after crash
        subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
        author_issue_bootstrap.save_phase_journal(journal, journal_dir=self.lock_dir)

        # Now resume/replay the transition but simulate lock binding failure during Phase 6
        with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")):
            res = author_issue_bootstrap.bootstrap_author_issue_worktree(
                issue_number=850,
                canonical_repo_root=self.repo_dir,
                branch_name=branch,
                worktree_path=wt_path,
                idempotency_key=key,
                lock_dir=self.lock_dir,
                owner_session="session-test-1234",
            )

        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed")

        # Verify that compensating recovery correctly deleted transition-created branch & worktree
        # because creation provenance was preserved across replay (NOT downgraded to False!)
        self.assertFalse(os.path.exists(wt_path))
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
        self.assertNotEqual(branch_check.returncode, 0)

    def test_transition_created_only_compensation(self):
        """Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created."""
        key = "test_key_preexisting_branch_compensation"
        preexisting_branch = "fix/issue-850-preexisting"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting")

        # Create branch BEFORE bootstrap (preexisting branch)
        subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True)

        # Call bootstrap with simulated failure during Phase 6 (lock binding)
        with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")):
            res = author_issue_bootstrap.bootstrap_author_issue_worktree(
                issue_number=850,
                canonical_repo_root=self.repo_dir,
                branch_name=preexisting_branch,
                worktree_path=wt_path,
                idempotency_key=key,
                lock_dir=self.lock_dir,
                owner_session="session-test-1234",
            )

        self.assertFalse(res["success"])
        # Worktree dir was created by transition -> removed by compensation
        self.assertFalse(os.path.exists(wt_path))

        # Preexisting branch was NOT created by transition -> MUST BE PRESERVED!
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False)
        self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!")

    def test_incompatible_idempotency_replay_refusal(self):
        """Review #525 Finding 4: Replaying key with incompatible parameters returns refusal."""
        key = "test_key_incompatible_replay"
        res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name="fix/issue-850-param-a",
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res1["success"])

        # Second call with different branch_name
        res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name="fix/issue-850-param-b",
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res2["success"])
        self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay")

    def test_exact_next_action_satisfiable_via_mcp(self):
        """Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands."""
        key = "test_key_next_action_mcp"
        stale_sha = "0000000000000000000000000000000000000000"
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            expected_base_sha=stale_sha,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        next_action = res.get("exact_next_action", "")
        self.assertNotIn("scripts/worktree-start", next_action)
        self.assertNotIn("git worktree add", next_action)
        self.assertNotIn("bash", next_action.lower())

    def test_missing_owner_session_refusal(self):
        """Finding D: Missing owner_session context fails closed with typed refusal and zero mutation."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session=None,
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_owner_session")
        self.assertIn("exact_next_action", res)

    def test_symlink_lock_file_refusal(self):
        """Finding C: BootstrapTransitionLock refuses to follow symlinks."""
        key = "test_symlink_lock_key"
        safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key)
        lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock")
        target_file = os.path.join(self.tmp_dir, "fake_target")
        with open(target_file, "w") as f:
            f.write("target")
        os.symlink(target_file, lock_path)

        with self.assertRaises(RuntimeError) as ctx:
            with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir):
                pass
        self.assertIn("symlink", str(ctx.exception).lower())

    def test_lock_directory_escape_refusal(self):
        """Finding C: BootstrapTransitionLock refuses keys that escape lock directory."""
        with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"):
            with self.assertRaises(RuntimeError) as ctx:
                author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir)
        self.assertIn("escapes", str(ctx.exception).lower())

    def test_missing_active_identity_refusal(self):
        """F-5: Missing active_identity parameter fails closed."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session="session-test-1234",
            active_identity=None,
            active_profile="prgs-author",
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_active_identity")

    def test_missing_active_profile_refusal(self):
        """F-5: Missing active_profile parameter fails closed."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session="session-test-1234",
            active_identity="jcwalker3",
            active_profile=None,
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_active_profile")

    def test_dirty_worktree_preserved_during_recovery(self):
        """F-4: Compensating recovery does not delete dirty worktree."""
        branch = "fix/issue-850-rec-dirty"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
        dirty_file = os.path.join(wt_path, "dirty.txt")
        with open(dirty_file, "w") as f:
            f.write("uncommitted work")

        journal = {
            "idempotency_key": "test_dirty_rec",
            "issue_number": 850,
            "branch_name": branch,
            "worktree_path": wt_path,
            "artifacts_created": {
                "worktree_dir_created": True,
                "worktree_registered": True,
            },
            "failure_reason": "test dirty recovery",
        }
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
        self.assertTrue(os.path.exists(wt_path))
        self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"])

    def test_branch_with_commits_preserved_during_recovery(self):
        """F-4: Compensating recovery does not delete branch with author commits."""
        branch = "fix/issue-850-rec-commits"
        subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
        # Add a commit on the branch
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
        cfile = os.path.join(wt_path, "commit.txt")
        with open(cfile, "w") as f:
            f.write("author commit")
        subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True)
        subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True)

        journal = {
            "idempotency_key": "test_commits_rec",
            "issue_number": 850,
            "branch_name": branch,
            "resolved_base_sha": self.master_sha,
            "artifacts_created": {
                "branch_created": True,
            },
            "failure_reason": "test commit branch recovery",
        }
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
        self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!")
        self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"])

    def test_task_capability_map_integration(self):
        """Verify task_capability_map has bootstrap_author_issue_worktree configured correctly."""
        self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author")
        self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create")
        self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree"))
        self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue"))

    def test_unverified_assignment_lease_ids_fail_closed(self):
        """Review #531 Finding 4: fabricated assignment/lease IDs are refused."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            assignment_id="asn-fabricated",
            lease_id="lease-fabricated",
            expected_base_sha=self.master_sha,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertIn(
            res.get("reason_code"),
            {
                "unknown_lease_id",
                "assignment_lease_lookup_failed",
                "incomplete_assignment_lease_ids",
            },
        )

    def test_partial_assignment_lease_ids_fail_closed(self):
        """Review #531 Finding 4: one of assignment_id/lease_id alone is incomplete."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            assignment_id="asn-only",
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids")

    def test_stale_diverged_branch_is_not_accepted_via_merge_base(self):
        """Review #531 Finding 3: any common ancestor is not enough; require master ⊆ branch."""
        branch = "fix/issue-850-stale-divergent"
        # Create branch from current master, then advance master so branch lacks tip.
        subprocess.run(
            ["git", "-C", self.repo_dir, "branch", branch, self.master_sha],
            check=True,
            capture_output=True,
        )
        # Make a new commit on master (orphan path so branch does not contain it).
        marker = os.path.join(self.repo_dir, "master-advance.txt")
        with open(marker, "w") as f:
            f.write("advance master\n")
        subprocess.run(["git", "-C", self.repo_dir, "add", "master-advance.txt"], check=True, capture_output=True)
        subprocess.run(
            ["git", "-C", self.repo_dir, "commit", "-m", "advance master past branch"],
            check=True,
            capture_output=True,
        )
        new_master = subprocess.run(
            ["git", "-C", self.repo_dir, "rev-parse", "HEAD"],
            capture_output=True,
            text=True,
            check=True,
        ).stdout.strip()
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name=branch,
            expected_base_sha=new_master,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "incompatible_existing_branch")

    def test_compensating_recovery_attempts_lease_release(self):
        """Review #531 Finding 5: recovery invokes lease release when lease_id is present."""
        from unittest import mock

        journal = {
            "idempotency_key": "test_lease_rec",
            "issue_number": 850,
            "owner_session": "session-test-1234",
            "lease_id": "lease-abc",
            "branch_name": "fix/issue-850-lease-rec",
            "artifacts_created": {"lock_created": True},
            "failure_reason": "simulated",
            "completed": False,
        }
        with mock.patch.object(
            author_issue_bootstrap.lease_lifecycle,
            "release_lease",
            return_value={"success": True},
        ) as rel, mock.patch.object(
            author_issue_bootstrap.control_plane_db,
            "ControlPlaneDB",
            return_value=mock.Mock(),
        ):
            rec = author_issue_bootstrap.run_compensating_recovery(
                journal, self.repo_dir, journal_dir=self.lock_dir
            )
        self.assertTrue(rec["executed"])
        rel.assert_called_once()
        self.assertIn("lease:lease-abc", rec["rolled_back"])


class TestCanonicalRootNoStringSplit(unittest.TestCase):
    def test_fallback_uses_commonpath_not_substring_split(self):
        """Review #531 Finding 2: no norm.split('/branches/') fallback."""
        import inspect
        import author_mutation_worktree as amw

        src = inspect.getsource(amw.resolve_canonical_repo_root)
        self.assertNotIn('split("/branches/")', src)
        self.assertNotIn("split('/branches/')", src)

        # Fallback recovers repo root from a nested branches worktree path.
        with tempfile.TemporaryDirectory() as tmp:
            repo = os.path.join(tmp, "repo")
            wt = os.path.join(repo, "branches", "fix-issue-850-x")
            os.makedirs(wt)
            # git unavailable path: pass missing workspace so fallback is used.
            root = amw.resolve_canonical_repo_root("/missing/path", wt)
            self.assertEqual(root, os.path.realpath(repo))


if __name__ == "__main__":
    unittest.main()