alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: store workspaces as {workspace_id}/{repo_name} for natural git clone layout

Browse Source 
Working copies were stored as /data/working-copies/{repo_id}/{workspace_name}/,
so git clone was forced into a user-named directory. That meant the container
mount basename was the workspace name (e.g. main) instead of the repo name.

- Generate the workspace UUID before cloning and clone into
  /data/working-copies/{workspace_id}/ so git creates {repo_name}/ naturally
- Set workspace.path to /data/working-copies/{workspace_id}/{repo_name}/
- Update _migrate_clone_into_workspace() to use the same layout
- _get_repository_mount_name() now prefers workspace.path basename and only
  falls back to remote URL / repo.name for legacy repo-only instances
- Update unit tests to assert workspace path basename is used for mounts

Quality gates:
- pytest tests/unit: 219 passed
- ruff: clean on changed files
- mypy: clean on changed files
This commit is contained in:
Developer
2026-06-15 09:40:59 +00:00
parent 6e33e8e4e9
commit b26ed7c3e4
27 changed files with 176 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/services/shared/workspace_manager.py
+86 -30
View File
@@ -17,6 +17,7 @@ from sqlalchemy import select
from src.models import Workspace
from src.services.git.git_service import GitService
from src.services.shared.ssh_keys import _get_fernet
from src.utils.git_url_parser import extract_base_repo_url
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
@@ -47,9 +48,32 @@ class WorkspaceManager:
BASE_PATH = "/data/working-copies"
def _workspace_path(self, repo_id: uuid.UUID, name: str) -> str:
"""Return the filesystem path for a workspace."""
return os.path.join(self.BASE_PATH, str(repo_id), name)
@staticmethod
def _repo_directory_name(repo: "GitRepository") -> str:
"""Return the directory name git would create for a standard clone.
Prefers the name parsed from the remote URL and falls back to the
user-provided repository name when no remote URL is available.
"""
if repo.remote_url:
base_url = extract_base_repo_url(repo.remote_url) or repo.remote_url
name = base_url.rstrip("/").split("/")[-1]
if name.endswith(".git"):
name = name[:-4]
if name:
return name
return repo.name
def _workspace_path(self, workspace_id: uuid.UUID, repo: "GitRepository") -> str:
"""Return the filesystem path for a workspace.
Layout: /data/working-copies/{workspace_id}/{repo_name}/
The workspace_id prevents collisions between workspaces, and the
repo_name matches the directory git clone naturally creates.
"""
return os.path.join(
self.BASE_PATH, str(workspace_id), self._repo_directory_name(repo)
)
async def create(
self,
@@ -74,24 +98,31 @@ class WorkspaceManager:
Raises:
RuntimeError: If git clone fails.
"""
path = self._workspace_path(repo.id, name)
parent = os.path.dirname(path)
os.makedirs(parent, exist_ok=True)
# Ensure container users (various UIDs) can write to workspace dirs
with contextlib.suppress(OSError):
os.chmod(parent, 0o777)
logger.info(
"Creating workspace: name=%s, repo=%s, branch=%s", name, repo.id, branch
)
if not repo.remote_url:
raise ValueError("Repository has no remote URL")
repo_dir_name = self._repo_directory_name(repo)
workspace_id = uuid.uuid4()
parent_path = os.path.join(self.BASE_PATH, str(workspace_id))
clone_target = parent_path # git clone creates {repo_dir_name}/ inside this
logger.info(
"Creating workspace: id=%s name=%s repo=%s branch=%s",
workspace_id,
name,
repo.id,
branch,
)
# Remove stale directory from previous failed/aborted clone
if os.path.exists(path):
logger.warning("Removing stale workspace directory: %s", path)
shutil.rmtree(path, ignore_errors=True)
if os.path.exists(parent_path):
logger.warning("Removing stale workspace directory: %s", parent_path)
shutil.rmtree(parent_path, ignore_errors=True)
os.makedirs(parent_path, exist_ok=True)
# Ensure container users (various UIDs) can write to workspace dirs
with contextlib.suppress(OSError):
os.chmod(parent_path, 0o777)
# Load SSH key if repo has one
ssh_key = None
@@ -108,19 +139,38 @@ class WorkspaceManager:
ssh_key_obj.private_key_encrypted.encode()
).decode()
await GitService.clone(repo.remote_url, branch, path, ssh_key=ssh_key)
self._make_world_writable(path)
await GitService.clone(repo.remote_url, branch, clone_target, ssh_key=ssh_key)
expected_path = os.path.join(parent_path, repo_dir_name)
if not os.path.isdir(expected_path):
# git clone can create a different directory name than expected for
# some URL shapes; fall back to the single directory git created.
entries = [
entry
for entry in os.listdir(parent_path)
if os.path.isdir(os.path.join(parent_path, entry))
]
if len(entries) == 1:
expected_path = os.path.join(parent_path, entries[0])
else:
raise RuntimeError(
f"Expected git clone to create {repo_dir_name}/ under "
f"{parent_path}, but found: {entries}"
)
self._make_world_writable(expected_path)
workspace = Workspace(
id=workspace_id,
name=name,
repo_id=repo.id,
user_id=user_id,
branch=branch,
path=path,
path=expected_path,
status="ready",
last_sync_at=datetime.now(),
)
logger.info("Workspace created: %s", workspace.id)
logger.info("Workspace created: %s at %s", workspace.id, workspace.path)
return workspace
async def delete(
@@ -375,24 +425,30 @@ class WorkspaceManager:
if not os.path.exists(clone_path):
raise RuntimeError(f"Clone path not found: {clone_path}")
path = self._workspace_path(repo.id, name)
parent = os.path.dirname(path)
os.makedirs(parent, exist_ok=True)
workspace_id = uuid.uuid4()
parent_path = os.path.join(self.BASE_PATH, str(workspace_id))
repo_dir_name = self._repo_directory_name(repo)
target_path = os.path.join(parent_path, repo_dir_name)
os.makedirs(parent_path, exist_ok=True)
with contextlib.suppress(OSError):
os.chmod(parent, 0o777)
os.chmod(parent_path, 0o777)
if os.path.exists(path):
shutil.rmtree(path, ignore_errors=True)
if os.path.exists(target_path):
shutil.rmtree(target_path, ignore_errors=True)
shutil.move(clone_path, path)
self._make_world_writable(path)
# Move the existing clone into the repo-named subdirectory so the
# workspace path matches the natural git clone layout.
shutil.move(clone_path, target_path)
self._make_world_writable(target_path)
workspace = Workspace(
id=workspace_id,
name=name,
repo_id=repo.id,
user_id=instance.owner_id,
branch=instance.branch or "main",
path=path,
path=target_path,
status="ready",
last_sync_at=datetime.now(),
)