fix: robust worktree repo-path derivation and branch typing (review)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/hqt/sessions/service.py

@@ -86,19 +86,29 @@ class SessionService:
             return Path(sess.worktree_path)
         return self._project_path(db, sess.project_id)
 
-    def _repo_path_for_worktree(self, db: DBSession, sess: Session) -> Path:
+    def _repo_path_for_worktree(
+        self, db: DBSession, project_id: int, worktree_path: str, branch: str
+    ) -> Path:
         """Resolve the repo path for a worktree session.
 
         Prefer the project row; if it is gone, derive the repo from the
-        worktree layout (<repo>/.worktrees/<branch> → repo = parent.parent).
+        worktree layout. ``create_worktree`` builds the path as
+        ``<repo>/.worktrees/<branch>`` (see ``git.worktree``), so the repo is
+        recovered by stripping the ``.worktrees`` dir plus the branch's path
+        components. Branch names may contain slashes (e.g. ``feature/foo``),
+        which add extra path components, so we cannot assume a fixed depth.
 
-        Only meaningful for worktree sessions; ``sess.worktree_path`` must be set.
+        Callers must only invoke this for worktree sessions (``worktree_path``
+        and ``branch`` already established as set).
         """
-        project = db.get(Project, sess.project_id)
+        project = db.get(Project, project_id)
         if project is not None:
             return Path(project.path)
-        assert sess.worktree_path is not None
-        return Path(sess.worktree_path).parent.parent
+        repo = Path(worktree_path)
+        # One level for the ".worktrees" dir, plus one per branch path component.
+        for _ in range(len(Path(branch).parts) + 1):
+            repo = repo.parent
+        return repo
 
     async def _capture_session_id_with_retry(
         self,
@@ -269,11 +279,14 @@ class SessionService:
             # but whose branch still exists can be recreated before anything
             # else runs. If recreation fails, the branch is gone too.
             if sess.worktree_path and not Path(sess.worktree_path).exists():
+                branch = sess.worktree_branch
+                if branch is None:
+                    raise ServiceError(
+                        "Worktree session is missing its branch; delete the session"
+                    )
                 project_path = self._project_path(db, sess.project_id)
                 try:
-                    await worktree.add_worktree_for_branch(
-                        project_path, sess.worktree_branch
-                    )
+                    await worktree.add_worktree_for_branch(project_path, branch)
                 except ServiceError as exc:
                     raise ServiceError(
                         "Worktree and branch are gone; delete the session"
@@ -404,12 +417,18 @@ class SessionService:
                 await self.tmux.kill(sess.tmux_session_name)
 
             if remove_worktree and sess.worktree_path:
-                repo_path = self._repo_path_for_worktree(db, sess)
+                worktree_path = sess.worktree_path
+                branch = sess.worktree_branch
+                if branch is None:
+                    raise ServiceError("Worktree session is missing its branch")
+                repo_path = self._repo_path_for_worktree(
+                    db, sess.project_id, worktree_path, branch
+                )
                 # Let ServiceError propagate: do NOT delete the row on failure.
                 await worktree.remove_worktree(
                     repo_path,
-                    Path(sess.worktree_path),
-                    sess.worktree_branch,
+                    Path(worktree_path),
+                    branch,
                     force,
                 )
 
@@ -425,10 +444,14 @@ class SessionService:
             sess = db.get(Session, session_id)
             if sess is None or not sess.worktree_path:
                 return None
-            repo_path = self._repo_path_for_worktree(db, sess)
-            return await worktree.worktree_state(
-                repo_path, Path(sess.worktree_path), sess.worktree_branch
+            worktree_path = sess.worktree_path
+            branch = sess.worktree_branch
+            if branch is None:
+                raise ServiceError("Worktree session is missing its branch")
+            repo_path = self._repo_path_for_worktree(
+                db, sess.project_id, worktree_path, branch
             )
+            return await worktree.worktree_state(repo_path, Path(worktree_path), branch)
 
     async def _status_for(
         self, sess: Session, wi: WindowInfo | None, now: float

tests/test_sessions.py

@@ -1433,3 +1433,38 @@ async def test_worktree_state_for_worktree_session_returns_state(
 
     assert state is fake_worktree.state_result
     assert fake_worktree.state_calls == [(Path("/tmp/myproj"), WT_PATH, "feature-x")]
+
+
+@pytest.mark.asyncio
+async def test_repo_path_fallback_slash_branch_when_project_gone(
+    factory, db, tmux, harnesses, fake_worktree
+):
+    """When the project row is gone, the repo is derived from the worktree
+    layout even for slash-containing branch names (feature/foo).
+
+    create_worktree builds the path as <repo>/.worktrees/<branch>, so for a
+    branch like 'feature/foo' the path is <repo>/.worktrees/feature/foo. The
+    fallback must strip '.worktrees/feature/foo' and recover <repo>, not the
+    wrong parent.parent result (<repo>/.worktrees).
+    """
+    from hqt.db.models import Session
+
+    repo = Path("/tmp/myproj")
+    branch = "feature/foo"
+    # Mirror how worktree.create_worktree builds the path.
+    wt_path = repo / ".worktrees" / branch
+    sess = _seed_worktree_session(db, branch=branch, path=str(wt_path))
+    sess_id = sess.id
+
+    # Delete the project row so the fallback derivation is exercised.
+    db.query(Project).delete()
+    db.commit()
+    db.expire_all()
+    # Session row outlives the project (no FK cascade in this seed).
+    assert db.get(Session, sess_id) is not None
+
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
+    await service.worktree_state_for(sess_id)
+
+    # The fallback must recover the real repo, not <repo>/.worktrees.
+    assert fake_worktree.state_calls == [(repo, wt_path, branch)]