fix(bootstrap): address PR #853 REQUEST_CHANGES findings (#850)

- Remove /branches/ string-split fallback in resolve_canonical_repo_root;
  recover roots via commonpath ancestry only (review #531 F2).
- Refuse existing branches that do not contain live master; no weak
  merge-base acceptance (F3).
- Verify caller-supplied assignment_id/lease_id against the control plane
  or fail closed (F4).
- Compensating recovery releases bound workflow leases via lease_lifecycle (F5).
- Regression tests for each finding.

tests/test_author_issue_bootstrap.py

@@ -69,8 +69,7 @@ class TestAuthorIssueBootstrap(unittest.TestCase):
         res = author_issue_bootstrap.bootstrap_author_issue_worktree(
             issue_number=850,
             canonical_repo_root=self.repo_dir,
-            assignment_id="asn-12345",
-            lease_id="lease-67890",
+            # assignment_id/lease_id omitted: optional unless verified live.
             expected_base_sha=self.master_sha,
             idempotency_key=key,
             remote="prgs",
@@ -477,6 +476,125 @@ class TestAuthorIssueBootstrap(unittest.TestCase):
         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()

tests/test_workspace_guard_alignment.py

@@ -35,8 +35,10 @@ class TestCanonicalRepoRoot(unittest.TestCase):
         self.assertEqual(root, CONTROL_ROOT)
 
     def test_falls_back_when_git_unavailable(self):
+        # When fallback is a path under <repo>/branches/<worktree>, recover
+        # <repo> via commonpath ancestry (never string-split on "/branches/").
         root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT)
-        self.assertEqual(root, os.path.realpath(MCP_PROCESS_ROOT))
+        self.assertEqual(root, os.path.realpath(CONTROL_ROOT))
 
 
 class TestWorkspaceRepoMembership(unittest.TestCase):