Make Workspace(path) work from inside a branch

Signed-off-by: Cong Wang <cwang@multikernel.io>

Author: congwang-mk

src/branching/core/registry.py

@@ -38,15 +38,15 @@ def get_fs(fstype: str) -> Optional[Type[FSBackend]]:
     return _registry.get(fstype)
 
 
-def detect_fs_for_mount(path: Path) -> Type[FSBackend]:
+def detect_fs_for_mount(path: Path) -> tuple[Type[FSBackend], Path]:
     """
     Auto-detect filesystem type for a mountpoint and return its implementation.
 
     Args:
-        path: Path to mounted filesystem
+        path: Path to mounted filesystem (may be inside a mount)
 
     Returns:
-        FSBackend subclass for the detected filesystem
+        Tuple of (FSBackend subclass, resolved mount root path)
 
     Raises:
         BranchingError: If no mount found or filesystem type not supported
@@ -77,7 +77,7 @@ def detect_fs_for_mount(path: Path) -> Type[FSBackend]:
             f"Supported types: {', '.join(_registry.keys()) or 'none'}"
         )
 
-    return fs_class
+    return fs_class, mount_info.mountpoint.resolve()
 
 
 def list_supported() -> list[str]:

src/branching/core/workspace.py

@@ -42,7 +42,7 @@ def __init__(self, path: str | Path):
             BranchingError: If no mount found or filesystem not supported
         """
         self._path = Path(path).resolve()
-        self._fs = detect_fs_for_mount(self._path)
+        self._fs, self._mount_root = detect_fs_for_mount(self._path)
 
     @property
     def path(self) -> Path:
@@ -77,7 +77,7 @@ def _generate_mountpoint(self, name: str) -> Path:
         """
         if self._fs.single_mount():
             # BranchFS: view switches in-place, same mount root
-            return self._path
+            return self._mount_root
         else:
             # DaxFS: each branch gets its own mount
             return self._path.parent / f"{self._path.name}_{name}"
@@ -108,7 +108,7 @@ def branch(
             parent_branch="/main",
             on_success=on_success,
             on_error=on_error,
-            mount_root=self._path,
+            mount_root=self._mount_root,
         )
 
     def __repr__(self) -> str:

src/branching/fs/_mount.py

@@ -75,20 +75,37 @@ def parse_mounts(path: str = PROC_MOUNTS) -> List[MountInfo]:
 
 def find_mount(mountpoint: Path, fstype: Optional[str] = None) -> Optional[MountInfo]:
     """
-    Find mount info for a specific mountpoint.
+    Find mount info for the nearest enclosing mountpoint.
+
+    Walks up from *mountpoint* toward the root, returning the first
+    (nearest) mount that matches.  This lets callers pass a path
+    *inside* a mount (e.g. a branch virtual path) and still find the
+    underlying filesystem.
 
     Args:
-        mountpoint: Path to the mountpoint
+        mountpoint: Path to look up (exact mountpoint or path inside one)
         fstype: Optional filesystem type filter
 
     Returns:
         MountInfo if found, None otherwise
     """
     mountpoint = mountpoint.resolve()
-    for mount in parse_mounts():
-        if mount.mountpoint.resolve() == mountpoint:
-            if fstype is None or mount.fstype == fstype:
-                return mount
+    mounts = parse_mounts()
+    # Build map: resolved mountpoint → MountInfo (last wins for overlays)
+    mount_map: Dict[Path, MountInfo] = {}
+    for m in mounts:
+        mp = m.mountpoint.resolve()
+        if fstype is None or m.fstype == fstype:
+            mount_map[mp] = m
+    # Walk up from path to root, return nearest enclosing mount
+    path = mountpoint
+    while True:
+        if path in mount_map:
+            return mount_map[path]
+        parent = path.parent
+        if parent == path:
+            break
+        path = parent
     return None
 
 

tests/test_resource_limits.py

@@ -407,7 +407,7 @@ def abort(cls, mountpoint):
                 pass
 
         with patch("branching.core.workspace.detect_fs_for_mount") as mock:
-            mock.return_value = MockFSBackend
+            mock.return_value = (MockFSBackend, Path("/tmp/test_ws"))
             return Workspace("/tmp/test_ws")
 
     def test_speculate_passes_resource_limits(self):
@@ -520,7 +520,7 @@ def abort(cls, mountpoint):
                 pass
 
         with patch("branching.core.workspace.detect_fs_for_mount") as mock:
-            mock.return_value = MockFSBackend
+            mock.return_value = (MockFSBackend, Path("/tmp/test_ws"))
             return Workspace("/tmp/test_ws")
 
     def test_speculate_passes_group_limits(self):

tests/test_speculate.py

@@ -57,10 +57,10 @@ def reset_mock():
     MockFSBackend.reset()
 
 
-def _make_workspace():
+def _make_workspace(path="/tmp/test_ws"):
     with patch("branching.core.workspace.detect_fs_for_mount") as mock_detect:
-        mock_detect.return_value = MockFSBackend
-        return Workspace("/tmp/test_ws")
+        mock_detect.return_value = (MockFSBackend, Path(path))
+        return Workspace(path)
 
 
 class TestSpeculate:
@@ -1066,3 +1066,63 @@ def test_outcome_defaults(self):
         assert o.winner is None
         assert o.all_results == []
         assert not o.committed
+
+
+class SingleMountMockFSBackend(MockFSBackend):
+    """Mock FS backend that uses single-mount semantics (like BranchFS)."""
+
+    @classmethod
+    def single_mount(cls) -> bool:
+        return True
+
+
+class TestSubBranching:
+    """Workspace(path) works when path is inside an existing mount."""
+
+    def test_workspace_from_branch_path(self):
+        """Candidate constructs Workspace(branch_path) and sub-branches."""
+        mount_root = "/mnt/ws"
+        branch_path = "/mnt/ws/@uuid0"
+
+        with patch("branching.core.workspace.detect_fs_for_mount") as mock_detect:
+            mock_detect.return_value = (SingleMountMockFSBackend, Path(mount_root))
+            ws = Workspace(branch_path)
+
+        # Public API: path reflects where the candidate works
+        assert ws.path == Path(branch_path)
+
+        # Sub-branching works and creates branches on the real mount
+        b = ws.branch("sub")
+        assert b.name == "sub"
+        # create_branch receives the mount root, not the branch virtual path
+        with b:
+            assert SingleMountMockFSBackend._branches_created[-1] == "sub"
+
+    def test_workspace_at_mount_root_unchanged(self):
+        """Normal usage (path == mount root) behaves the same as before."""
+        with patch("branching.core.workspace.detect_fs_for_mount") as mock_detect:
+            mock_detect.return_value = (SingleMountMockFSBackend, Path("/mnt/ws"))
+            ws = Workspace("/mnt/ws")
+        assert ws.path == Path("/mnt/ws")
+
+        b = ws.branch("feat")
+        assert b.name == "feat"
+        with b:
+            assert SingleMountMockFSBackend._branches_created[-1] == "feat"
+
+    def test_sub_branch_speculate(self):
+        """End-to-end: candidate sub-branches via Workspace and Speculate."""
+        mount_root = "/mnt/ws"
+        branch_path = "/mnt/ws/@uuid0"
+
+        with patch("branching.core.workspace.detect_fs_for_mount") as mock_detect:
+            mock_detect.return_value = (SingleMountMockFSBackend, Path(mount_root))
+            ws = Workspace(branch_path)
+
+        def candidate(path: Path) -> bool:
+            return True
+
+        spec = Speculate([candidate], first_wins=True)
+        outcome = spec(ws)
+        assert outcome.committed
+        assert outcome.winner.success

tests/test_workspace.py

@@ -39,15 +39,15 @@ def single_mount(cls) -> bool:
 
 @patch("branching.core.workspace.detect_fs_for_mount")
 def test_workspace_creation(mock_detect):
-    mock_detect.return_value = MockFSBackend
+    mock_detect.return_value = (MockFSBackend, Path("/tmp/test_ws"))
     ws = Workspace("/tmp/test_ws")
     assert ws.path == Path("/tmp/test_ws").resolve()
     assert ws.fstype == "mockfs"
 
 
 @patch("branching.core.workspace.detect_fs_for_mount")
 def test_workspace_branch_mountpoint_generation(mock_detect):
-    mock_detect.return_value = MockFSBackend
+    mock_detect.return_value = (MockFSBackend, Path("/tmp/test_ws"))
     ws = Workspace("/tmp/test_ws")
     b = ws.branch("feat")
     # Mount-per-branch: sibling directory
@@ -56,7 +56,7 @@ def test_workspace_branch_mountpoint_generation(mock_detect):
 
 @patch("branching.core.workspace.detect_fs_for_mount")
 def test_workspace_single_mount_branch(mock_detect):
-    mock_detect.return_value = SingleMountMockFSBackend
+    mock_detect.return_value = (SingleMountMockFSBackend, Path("/tmp/test_ws"))
     ws = Workspace("/tmp/test_ws")
     b = ws.branch("feat")
     # Single mount: same path
@@ -65,7 +65,7 @@ def test_workspace_single_mount_branch(mock_detect):
 
 @patch("branching.core.workspace.detect_fs_for_mount")
 def test_workspace_repr(mock_detect):
-    mock_detect.return_value = MockFSBackend
+    mock_detect.return_value = (MockFSBackend, Path("/tmp/test_ws"))
     ws = Workspace("/tmp/test_ws")
     assert "Workspace" in repr(ws)
     assert "mockfs" in repr(ws)