Adjusted local project setup directory tree structure to map branches on project repos

roboco/api/routes/git.py

@@ -3,10 +3,27 @@ Git API Routes
 
 Git operations for agents working on code tasks.
 These endpoints are called by the Git MCP Server.
+
+Workspace Structure:
+    Each agent gets their own workspace (git clone) for a project:
+
+    {workspaces_root}/
+    └── {project-slug}/
+        └── {team}/
+            └── {agent-slug}/
+                └── [git repo files]
+
+    Example:
+        /data/workspaces/roboco/backend/be-dev-1/
+        /data/workspaces/roboco/backend/be-dev-2/
+
+    This allows multiple agents to work on the same project in parallel,
+    each on their own branch, without file conflicts.
 """
 
 import subprocess
 from pathlib import Path
+from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, Query, status
 
@@ -31,7 +48,9 @@ from roboco.api.schemas.git import (
     GitPushResponse,
     GitStatusResponse,
 )
+from roboco.config import settings
 from roboco.services.project import get_project_service
+from roboco.services.workspace import WorkspaceError, get_workspace_service
 
 router = APIRouter()
 
@@ -80,26 +99,82 @@ async def _run_git(
 async def _get_workspace(
     db: DbSession,
     project_slug: str,
+    agent_id: UUID | None = None,
 ) -> Path:
-    """Get the workspace path for a project."""
+    """
-    service = get_project_service(db)
+    Get the workspace path for an agent on a project.
-    project = await service.get_by_slug(project_slug)
+
+    Uses multi-agent workspace structure:
+        {workspaces_root}/{project_slug}/{team}/{agent_slug}/
+
+    If workspace doesn't exist and auto_clone is enabled, clones the repo.
+
+    Args:
+        db: Database session
+        project_slug: Project identifier
+        agent_id: Agent UUID (uses workspace service to resolve path)
+
+    Returns:
+        Path to the workspace directory
+
+    Raises:
+        HTTPException: If project not found or workspace setup fails
+    """
+    # Get project info
+    project_service = get_project_service(db)
+    project = await project_service.get_by_slug(project_slug)
     if not project:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Project '{project_slug}' not found",
         )
-    if not project.workspace_path:
+
+    # If no agent_id, fall back to legacy workspace_path (for backwards compat)
+    if agent_id is None:
+        if not project.workspace_path:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Project '{project_slug}' has no workspace configured "
+                "and no agent_id provided for dynamic workspace resolution",
+            )
+        workspace = Path(project.workspace_path)
+        if not workspace.exists():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Workspace path does not exist: {workspace}",
+            )
+        return workspace
+
+    # Use workspace service for multi-agent workspace resolution
+    workspace_service = get_workspace_service(db)
+
+    try:
+        if settings.workspace_auto_clone:
+            # Ensure workspace exists (clone if needed)
+            workspace = await workspace_service.ensure_workspace(
+                project_slug=project_slug,
+                agent_id=agent_id,
+                git_url=project.git_url,
+                default_branch=project.default_branch or "main",
+            )
+        else:
+            # Just resolve path, don't auto-clone
+            workspace = await workspace_service.resolve_workspace(
+                project_slug=project_slug,
+                agent_id=agent_id,
+            )
+            if not workspace.exists():
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=f"Workspace does not exist: {workspace}. "
+                    "Clone the repository first or enable auto_clone.",
+                )
+    except WorkspaceError as e:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Project '{project_slug}' has no workspace configured",
+            detail=str(e),
-        )
+        ) from e
-    workspace = Path(project.workspace_path)
+
-    if not workspace.exists():
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Workspace path does not exist: {workspace}",
-        )
     return workspace
 
 
@@ -111,12 +186,12 @@ async def _get_workspace(
 @router.get("/status", response_model=GitStatusResponse)
 async def get_git_status(
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
     project_slug: str = Query(...),
     _task_id: str | None = Query(default=None),
 ) -> GitStatusResponse:
     """Get git status for a project."""
-    workspace = await _get_workspace(db, project_slug)
+    workspace = await _get_workspace(db, project_slug, agent.agent_id)
 
     # Get current branch
     branch_result = await _run_git(workspace, ["branch", "--show-current"])
@@ -174,13 +249,13 @@ async def get_git_status(
 @router.get("/log", response_model=GitLogResponse)
 async def get_git_log(
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
     project_slug: str = Query(...),
     limit: int = Query(default=10, le=50),
     branch: str | None = Query(default=None),
 ) -> GitLogResponse:
     """Get git log for a project."""
-    workspace = await _get_workspace(db, project_slug)
+    workspace = await _get_workspace(db, project_slug, agent.agent_id)
 
     # Get current branch if not specified
     if not branch:
@@ -222,12 +297,12 @@ async def get_git_log(
 @router.get("/branches", response_model=GitBranchListResponse)
 async def list_branches(
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
     project_slug: str = Query(...),
     include_remote: bool = Query(default=False),
 ) -> GitBranchListResponse:
     """List git branches for a project."""
-    workspace = await _get_workspace(db, project_slug)
+    workspace = await _get_workspace(db, project_slug, agent.agent_id)
 
     # Get current branch
     current_result = await _run_git(workspace, ["branch", "--show-current"])
@@ -271,13 +346,13 @@ async def list_branches(
 @router.get("/diff", response_model=GitDiffResponse)
 async def get_git_diff(
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
     project_slug: str = Query(...),
     staged: bool = Query(default=False),
     file_path: str | None = Query(default=None),
 ) -> GitDiffResponse:
     """Get git diff for a project."""
-    workspace = await _get_workspace(db, project_slug)
+    workspace = await _get_workspace(db, project_slug, agent.agent_id)
 
     args = ["diff"]
     if staged:
@@ -312,10 +387,10 @@ async def get_git_diff(
 async def create_commit(
     data: GitCommitRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> GitCommitResponse:
-    """Create a git commit."""
+    """Create a git commit and link it to the task."""
-    workspace = await _get_workspace(db, data.project_slug)
+    workspace = await _get_workspace(db, data.project_slug, agent.agent_id)
 
     # Stage files
     if data.files:
@@ -347,6 +422,34 @@ async def create_commit(
                 if "file" in part:
                     files_changed = int(part.strip().split()[0])
 
+    # Link commit to task (ensures traceability)
+    from roboco.services.task import get_task_service
+    from roboco.services.work_session import get_work_session_service
+
+    try:
+        task_uuid = UUID(data.task_id)
+
+        # Add commit to task record
+        task_service = get_task_service(db)
+        await task_service.add_commit(
+            task_id=task_uuid,
+            commit_hash=commit_hash,
+            message=data.message,  # Store original message without prefix
+            author_id=agent.agent_id,
+        )
+
+        # If task has a work session, add commit there too
+        task = await task_service.get(task_uuid)
+        if task and task.work_session_id:
+            work_session_service = get_work_session_service(db)
+            await work_session_service.add_commit(task.work_session_id, commit_hash)
+
+        await db.commit()
+    except Exception:
+        # Don't fail the commit response if linking fails
+        # The commit was still made successfully
+        pass
+
     return GitCommitResponse(
         commit_hash=commit_hash,
         message=message,
@@ -360,10 +463,10 @@ async def create_commit(
 async def push_commits(
     data: GitPushRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> GitPushResponse:
     """Push commits to remote."""
-    workspace = await _get_workspace(db, data.project_slug)
+    workspace = await _get_workspace(db, data.project_slug, agent.agent_id)
 
     # Get current branch
     branch_result = await _run_git(workspace, ["branch", "--show-current"])
@@ -401,7 +504,7 @@ async def create_branch(
     agent: CurrentAgentContext,
 ) -> GitCreateBranchResponse:
     """Create a task branch (PM only)."""
-    workspace = await _get_workspace(db, data.project_slug)
+    workspace = await _get_workspace(db, data.project_slug, agent.agent_id)
 
     # Get team from agent context
     team = agent.team or "unknown"
@@ -433,10 +536,10 @@ async def create_branch(
 async def checkout_branch(
     data: GitCheckoutRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> GitCheckoutResponse:
     """Checkout a branch."""
-    workspace = await _get_workspace(db, data.project_slug)
+    workspace = await _get_workspace(db, data.project_slug, agent.agent_id)
 
     await _run_git(workspace, ["checkout", data.branch])
 
@@ -450,7 +553,7 @@ async def checkout_branch(
 async def create_pull_request(
     data: GitCreatePRRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> GitCreatePRResponse:
     """Create a pull request using GitHub CLI.
 
@@ -460,7 +563,7 @@ async def create_pull_request(
     - Developer sets pr_created=True (this endpoint)
     - When BOTH are true, task transitions to awaiting_pm_review
     """
-    workspace = await _get_workspace(db, data.project_slug)
+    workspace = await _get_workspace(db, data.project_slug, agent.agent_id)
 
     # Get current branch
     branch_result = await _run_git(workspace, ["branch", "--show-current"])
@@ -538,10 +641,10 @@ async def create_pull_request(
 async def merge_pull_request(
     data: GitMergePRRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> GitMergePRResponse:
     """Merge a pull request using GitHub CLI (PM only)."""
-    workspace = await _get_workspace(db, data.project_slug)
+    workspace = await _get_workspace(db, data.project_slug, agent.agent_id)
 
     # Merge PR using gh CLI (wrapped in thread to avoid blocking)
     import asyncio

roboco/api/routes/tasks.py

@@ -41,6 +41,7 @@ from roboco.api.schemas.tasks import (
     TaskSessionLinkResponse,
     TaskUpdate,
     TeamTasksQuery,
+    enrich_task_with_context,
     task_list_to_response,
     task_to_response,
     transform_update_data,
@@ -305,7 +306,7 @@ async def get_task(
     task_id: UUID,
     db: DbSession,
 ) -> TaskResponse:
-    """Get a specific task with linked sessions."""
+    """Get a specific task with full context (sessions, work session, project)."""
     service = get_task_service(db)
     task = await service.get(task_id)
     if not task:
@@ -331,6 +332,9 @@ async def get_task(
         if link.session and link.session.group and link.session.group.channel
     ]
 
+    # Enrich with work session and project context
+    response = await enrich_task_with_context(response, db)
+
     return response
 
 

roboco/api/routes/test.py

@@ -3,10 +3,14 @@ Test API Routes
 
 Test and CI/CD operations for agents working on code tasks.
 These endpoints are called by the Test MCP Server.
+
+Uses multi-agent workspace structure - each agent gets their own
+workspace at: {workspaces_root}/{project_slug}/{team}/{agent_slug}/
 """
 
 import subprocess
 from pathlib import Path
+from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, Query, status
 
@@ -26,7 +30,9 @@ from roboco.api.schemas.test import (
     TypecheckRequest,
     TypecheckResponse,
 )
+from roboco.config import settings
 from roboco.services.project import get_project_service
+from roboco.services.workspace import WorkspaceError, get_workspace_service
 
 router = APIRouter()
 
@@ -43,8 +49,13 @@ _TYPE_ERROR_PARTS_MIN = 3
 async def _get_project_and_workspace(
     db: DbSession,
     project_slug: str,
+    agent_id: UUID | None = None,
 ) -> tuple[object, Path]:
-    """Get the project and workspace path."""
+    """
+    Get the project and workspace path for an agent.
+
+    Uses multi-agent workspace structure if agent_id is provided.
+    """
     service = get_project_service(db)
     project = await service.get_by_slug(project_slug)
     if not project:
@@ -52,17 +63,51 @@ async def _get_project_and_workspace(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Project '{project_slug}' not found",
         )
-    if not project.workspace_path:
+
+    # If no agent_id, fall back to legacy workspace_path
+    if agent_id is None:
+        if not project.workspace_path:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Project '{project_slug}' has no workspace configured "
+                "and no agent_id provided for dynamic workspace resolution",
+            )
+        workspace = Path(project.workspace_path)
+        if not workspace.exists():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Workspace path does not exist: {workspace}",
+            )
+        return project, workspace
+
+    # Use workspace service for multi-agent workspace resolution
+    workspace_service = get_workspace_service(db)
+
+    try:
+        if settings.workspace_auto_clone:
+            workspace = await workspace_service.ensure_workspace(
+                project_slug=project_slug,
+                agent_id=agent_id,
+                git_url=project.git_url,
+                default_branch=project.default_branch or "main",
+            )
+        else:
+            workspace = await workspace_service.resolve_workspace(
+                project_slug=project_slug,
+                agent_id=agent_id,
+            )
+            if not workspace.exists():
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=f"Workspace does not exist: {workspace}. "
+                    "Clone the repository first or enable auto_clone.",
+                )
+    except WorkspaceError as e:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Project '{project_slug}' has no workspace configured",
+            detail=str(e),
-        )
+        ) from e
-    workspace = Path(project.workspace_path)
+
-    if not workspace.exists():
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Workspace path does not exist: {workspace}",
-        )
     return project, workspace
 
 
@@ -102,12 +147,14 @@ async def _run_command(
 @router.get("/status", response_model=TestStatusResponse)
 async def get_test_status(
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
     project_slug: str = Query(...),
     _task_id: str | None = Query(default=None),
 ) -> TestStatusResponse:
     """Get test status for a project."""
-    _project, _workspace = await _get_project_and_workspace(db, project_slug)
+    _project, _workspace = await _get_project_and_workspace(
+        db, project_slug, agent.agent_id
+    )
 
     # Return status - in a full implementation this would query stored results
     return TestStatusResponse(
@@ -127,10 +174,12 @@ async def get_test_status(
 async def run_tests(
     data: TestRunRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> TestRunResponse:
     """Run tests for a project."""
-    project, workspace = await _get_project_and_workspace(db, data.project_slug)
+    project, workspace = await _get_project_and_workspace(
+        db, data.project_slug, agent.agent_id
+    )
 
     test_cmd = getattr(project, "test_command", None)
     if not test_cmd:
@@ -190,10 +239,12 @@ async def run_tests(
 async def run_lint(
     data: LintRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> LintResponse:
     """Run linter for a project."""
-    project, workspace = await _get_project_and_workspace(db, data.project_slug)
+    project, workspace = await _get_project_and_workspace(
+        db, data.project_slug, agent.agent_id
+    )
 
     lint_cmd = getattr(project, "lint_command", None)
     if not lint_cmd:
@@ -253,10 +304,12 @@ async def run_lint(
 async def run_format(
     data: FormatRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> FormatResponse:
     """Run formatter for a project."""
-    project, workspace = await _get_project_and_workspace(db, data.project_slug)
+    project, workspace = await _get_project_and_workspace(
+        db, data.project_slug, agent.agent_id
+    )
 
     format_cmd = getattr(project, "format_command", None)
     if not format_cmd:
@@ -295,10 +348,12 @@ async def run_format(
 async def run_typecheck(
     data: TypecheckRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> TypecheckResponse:
     """Run type checker for a project."""
-    project, workspace = await _get_project_and_workspace(db, data.project_slug)
+    project, workspace = await _get_project_and_workspace(
+        db, data.project_slug, agent.agent_id
+    )
 
     typecheck_cmd = getattr(project, "typecheck_command", None)
     if not typecheck_cmd:
@@ -350,10 +405,12 @@ async def run_typecheck(
 async def run_build(
     data: BuildRequest,
     db: DbSession,
-    _agent: CurrentAgentContext,
+    agent: CurrentAgentContext,
 ) -> BuildResponse:
     """Run build command for a project."""
-    project, workspace = await _get_project_and_workspace(db, data.project_slug)
+    project, workspace = await _get_project_and_workspace(
+        db, data.project_slug, agent.agent_id
+    )
 
     build_cmd = getattr(project, "build_command", None)
     if not build_cmd:

roboco/api/schemas/tasks.py

@@ -61,6 +61,29 @@ class TaskSessionLinkResponse(BaseModel):
     relationship_type: str
 
 
+class WorkSessionSummaryInTask(BaseModel):
+    """Work session info embedded in task response."""
+
+    id: UUID
+    branch_name: str
+    status: str
+    commits: list[str] = []
+    files_modified: list[str] = []
+    pr_number: int | None = None
+    pr_url: str | None = None
+    pr_status: str | None = None
+
+
+class ProjectSummaryInTask(BaseModel):
+    """Project info embedded in task response."""
+
+    id: UUID
+    name: str
+    slug: str
+    git_url: str
+    default_branch: str
+
+
 class SubTaskResponse(BaseModel):
     """A sub-task within a task plan."""
 
@@ -244,6 +267,13 @@ class TaskResponse(BaseModel):
     # Linked Sessions (for agent context)
     sessions: list[TaskSessionLinkResponse] = []
 
+    # Git/Development Context (for full traceability)
+    project: ProjectSummaryInTask | None = None
+    work_session: WorkSessionSummaryInTask | None = None
+    branch_name: str | None = None
+    pr_number: int | None = None
+    pr_url: str | None = None
+
     class Config:
         from_attributes = True
 
@@ -526,9 +556,76 @@ def task_to_response(task: "TaskTable") -> TaskResponse:
         # Review Status
         self_verified=task.self_verified,
         qa_verified=task.qa_verified,
+        # Git context from task record
+        branch_name=getattr(task, "branch_name", None),
+        pr_number=getattr(task, "pr_number", None),
+        pr_url=getattr(task, "pr_url", None),
     )
 
 
+async def enrich_task_with_context(
+    task_response: TaskResponse,
+    db: Any,
+    include_project: bool = True,
+    include_work_session: bool = True,
+) -> TaskResponse:
+    """
+    Enrich a TaskResponse with related context (project, work session).
+
+    Call this when full traceability context is needed.
+    """
+    from sqlalchemy import select  # noqa: PLC0415
+
+    from roboco.db.tables import ProjectTable, WorkSessionTable  # noqa: PLC0415
+
+    task_dict = task_response.model_dump()
+
+    # Get project info if task has project_id
+    if include_project:
+        # Task doesn't have project_id in response yet, need to fetch from related data
+        # This requires knowing the project_id from the task record
+        pass  # TODO: Add project_id to TaskResponse or fetch via work session
+
+    # Get work session info
+    if include_work_session and hasattr(task_response, "id"):
+        query = select(WorkSessionTable).where(
+            WorkSessionTable.task_id == task_response.id
+        )
+        result = await db.execute(query)
+        work_session = result.scalar_one_or_none()
+
+        if work_session:
+            task_dict["work_session"] = WorkSessionSummaryInTask(
+                id=work_session.id,
+                branch_name=work_session.branch_name,
+                status=work_session.status.value if work_session.status else "unknown",
+                commits=list(work_session.commits or []),
+                files_modified=list(work_session.files_modified or []),
+                pr_number=work_session.pr_number,
+                pr_url=work_session.pr_url,
+                pr_status=work_session.pr_status,
+            )
+
+            # Also get project info from work session
+            if include_project and work_session.project_id:
+                proj_query = select(ProjectTable).where(
+                    ProjectTable.id == work_session.project_id
+                )
+                proj_result = await db.execute(proj_query)
+                project = proj_result.scalar_one_or_none()
+
+                if project:
+                    task_dict["project"] = ProjectSummaryInTask(
+                        id=project.id,
+                        name=project.name,
+                        slug=project.slug,
+                        git_url=project.git_url,
+                        default_branch=project.default_branch,
+                    )
+
+    return TaskResponse(**task_dict)
+
+
 def task_list_to_response(tasks: list["TaskTable"]) -> list[TaskResponse]:
     """Convert list of TaskTable to list of TaskResponse."""
     return [task_to_response(t) for t in tasks]

roboco/config.py

@@ -213,6 +213,23 @@ class Settings(BaseSettings):
     session_max_content_length: int = Field(default=50000, ge=1)
     message_max_length: int = Field(default=10000, ge=1)
 
+    # ==========================================================================
+    # Workspaces (Multi-Agent Git)
+    # ==========================================================================
+    workspaces_root: str = Field(
+        default="/data/workspaces",
+        description="Root directory for all agent workspaces",
+    )
+    workspace_auto_clone: bool = Field(
+        default=True,
+        description="Automatically clone repos when workspace is first accessed",
+    )
+    workspace_clone_timeout: int = Field(
+        default=300,
+        ge=30,
+        description="Timeout in seconds for git clone operations",
+    )
+
 
 @lru_cache
 def get_settings() -> Settings:

roboco/models/base.py

@@ -33,6 +33,7 @@ class TaskStatus(str, Enum):
     AWAITING_CEO_APPROVAL = "awaiting_ceo_approval"  # PMs approved, CEO decides
     COMPLETED = "completed"
     CANCELLED = "cancelled"
+    QUARANTINED = "quarantined"  # Special state for problematic tasks
 
 
 class TaskType(str, Enum):

roboco/services/workspace.py

@@ -0,0 +1,311 @@
+"""
+Workspace Service
+
+Manages multi-agent workspaces for git operations.
+
+Each agent gets their own workspace (git clone) for a project, allowing
+parallel development without conflicts:
+
+    /data/workspaces/
+    └── {project-slug}/
+        └── {team}/
+            └── {agent-slug}/
+                └── [git repo files]
+
+Example:
+    /data/workspaces/roboco/backend/be-dev-1/
+    /data/workspaces/roboco/backend/be-dev-2/
+    /data/workspaces/roboco/frontend/fe-dev-1/
+"""
+
+import asyncio
+import subprocess
+from pathlib import Path
+from uuid import UUID
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from roboco.config import settings
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.logging import get_logger
+from roboco.models.base import Team
+
+logger = get_logger(__name__)
+
+
+class WorkspaceError(Exception):
+    """Raised when workspace operations fail."""
+
+    pass
+
+
+class WorkspaceService:
+    """
+    Service for managing agent workspaces.
+
+    Workspaces follow the structure:
+        {workspaces_root}/{project_slug}/{team}/{agent_slug}/
+
+    This allows:
+    - Multiple agents to work on the same project in parallel
+    - Each agent has their own git working tree
+    - Agents can be on different branches simultaneously
+    - No file locking conflicts between agents
+    """
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+        self.root = Path(settings.workspaces_root)
+
+    def get_workspace_path(
+        self,
+        project_slug: str,
+        team: Team | str,
+        agent_slug: str,
+    ) -> Path:
+        """
+        Compute the workspace path for an agent on a project.
+
+        Args:
+            project_slug: Project identifier (e.g., 'roboco')
+            team: Agent's team (e.g., Team.BACKEND or 'backend')
+            agent_slug: Agent identifier (e.g., 'be-dev-1')
+
+        Returns:
+            Path to the workspace directory
+
+        Example:
+            >>> get_workspace_path('roboco', Team.BACKEND, 'be-dev-1')
+            Path('/data/workspaces/roboco/backend/be-dev-1')
+        """
+        team_str = team.value if isinstance(team, Team) else str(team)
+        return self.root / project_slug / team_str / agent_slug
+
+    async def resolve_workspace(
+        self,
+        project_slug: str,
+        agent_id: UUID | str,
+    ) -> Path:
+        """
+        Resolve workspace path from project slug and agent ID.
+
+        Looks up the agent to get team and slug, then computes path.
+
+        Args:
+            project_slug: Project identifier
+            agent_id: Agent UUID or slug
+
+        Returns:
+            Path to the workspace directory
+
+        Raises:
+            WorkspaceError: If agent not found
+        """
+        from sqlalchemy import select
+
+        # Look up agent
+        agent_id_str = str(agent_id)
+
+        # Try by UUID first, then by slug
+        query = select(AgentTable)
+        try:
+            agent_uuid = UUID(agent_id_str)
+            query = query.where(AgentTable.id == agent_uuid)
+        except ValueError:
+            query = query.where(AgentTable.slug == agent_id_str)
+
+        result = await self.session.execute(query)
+        agent = result.scalar_one_or_none()
+
+        if not agent:
+            raise WorkspaceError(f"Agent not found: {agent_id}")
+
+        team = agent.team if agent.team else Team.BACKEND
+        return self.get_workspace_path(project_slug, team, agent.slug)
+
+    async def ensure_workspace(
+        self,
+        project_slug: str,
+        agent_id: UUID | str,
+        git_url: str | None = None,
+        default_branch: str = "main",
+    ) -> Path:
+        """
+        Ensure workspace exists, cloning if necessary.
+
+        Args:
+            project_slug: Project identifier
+            agent_id: Agent UUID or slug
+            git_url: Git URL to clone (fetched from project if not provided)
+            default_branch: Default branch to checkout
+
+        Returns:
+            Path to the workspace directory
+
+        Raises:
+            WorkspaceError: If workspace creation fails
+        """
+        workspace = await self.resolve_workspace(project_slug, agent_id)
+
+        # Check if already exists
+        if (workspace / ".git").exists():
+            logger.debug(
+                "Workspace already exists",
+                workspace=str(workspace),
+                project=project_slug,
+            )
+            return workspace
+
+        # Get git URL if not provided
+        if not git_url:
+            from sqlalchemy import select
+
+            result = await self.session.execute(
+                select(ProjectTable).where(ProjectTable.slug == project_slug)
+            )
+            project = result.scalar_one_or_none()
+            if not project:
+                raise WorkspaceError(f"Project not found: {project_slug}")
+            git_url = project.git_url
+            default_branch = project.default_branch or default_branch
+
+        # Clone the repository
+        await self._clone_repo(workspace, git_url, default_branch)
+        return workspace
+
+    async def _clone_repo(
+        self,
+        workspace: Path,
+        git_url: str,
+        default_branch: str,
+    ) -> None:
+        """
+        Clone a git repository to the workspace.
+
+        Args:
+            workspace: Target directory
+            git_url: Git URL to clone
+            default_branch: Branch to checkout
+
+        Raises:
+            WorkspaceError: If clone fails
+        """
+        # Create parent directories
+        workspace.parent.mkdir(parents=True, exist_ok=True)
+
+        logger.info(
+            "Cloning repository",
+            workspace=str(workspace),
+            git_url=git_url,
+            branch=default_branch,
+        )
+
+        def _do_clone() -> subprocess.CompletedProcess[str]:
+            return subprocess.run(
+                [
+                    "git",
+                    "clone",
+                    "--branch",
+                    default_branch,
+                    "--single-branch",
+                    git_url,
+                    str(workspace),
+                ],
+                capture_output=True,
+                text=True,
+                timeout=settings.workspace_clone_timeout,
+                check=True,
+            )
+
+        try:
+            await asyncio.to_thread(_do_clone)
+            logger.info(
+                "Repository cloned successfully",
+                workspace=str(workspace),
+            )
+        except subprocess.CalledProcessError as e:
+            raise WorkspaceError(
+                f"Failed to clone repository: {e.stderr or e.stdout}"
+            ) from e
+        except subprocess.TimeoutExpired as e:
+            raise WorkspaceError(
+                f"Clone timed out after {settings.workspace_clone_timeout}s"
+            ) from e
+
+    async def workspace_exists(
+        self,
+        project_slug: str,
+        agent_id: UUID | str,
+    ) -> bool:
+        """Check if a workspace exists for the given project and agent."""
+        try:
+            workspace = await self.resolve_workspace(project_slug, agent_id)
+            return (workspace / ".git").exists()
+        except WorkspaceError:
+            return False
+
+    async def list_workspaces(self, project_slug: str) -> list[dict]:
+        """
+        List all workspaces for a project.
+
+        Returns:
+            List of workspace info dicts with team, agent, and path
+        """
+        project_dir = self.root / project_slug
+        if not project_dir.exists():
+            return []
+
+        workspaces = []
+        for team_dir in project_dir.iterdir():
+            if not team_dir.is_dir():
+                continue
+            for agent_dir in team_dir.iterdir():
+                if not agent_dir.is_dir():
+                    continue
+                if (agent_dir / ".git").exists():
+                    workspaces.append(
+                        {
+                            "team": team_dir.name,
+                            "agent": agent_dir.name,
+                            "path": str(agent_dir),
+                            "exists": True,
+                        }
+                    )
+        return workspaces
+
+    async def delete_workspace(
+        self,
+        project_slug: str,
+        agent_id: UUID | str,
+    ) -> bool:
+        """
+        Delete a workspace (use with caution).
+
+        Args:
+            project_slug: Project identifier
+            agent_id: Agent UUID or slug
+
+        Returns:
+            True if deleted, False if didn't exist
+        """
+        import shutil
+
+        workspace = await self.resolve_workspace(project_slug, agent_id)
+        if not workspace.exists():
+            return False
+
+        logger.warning(
+            "Deleting workspace",
+            workspace=str(workspace),
+        )
+
+        def _do_delete() -> None:
+            shutil.rmtree(workspace)
+
+        await asyncio.to_thread(_do_delete)
+        return True
+
+
+def get_workspace_service(session: AsyncSession) -> WorkspaceService:
+    """Factory function to get workspace service."""
+    return WorkspaceService(session)

uv.lock

@@ -847,14 +847,14 @@ wheels = [
 
 [[package]]
 name = "faker"
-version = "39.0.0"
+version = "40.1.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "tzdata" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/30/b9/0897fb5888ddda099dc0f314a8a9afb5faa7e52eaf6865c00686dfb394db/faker-39.0.0.tar.gz", hash = "sha256:ddae46d3b27e01cea7894651d687b33bcbe19a45ef044042c721ceac6d3da0ff", size = 1941757, upload-time = "2025-12-17T19:19:04.762Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/d7/1d/aa43ef59589ddf3647df918143f1bac9eb004cce1c43124ee3347061797d/faker-40.1.0.tar.gz", hash = "sha256:c402212a981a8a28615fea9120d789e3f6062c0c259a82bfb8dff5d273e539d2", size = 1948784, upload-time = "2025-12-29T18:06:00.659Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/eb/5a/26cdb1b10a55ac6eb11a738cea14865fa753606c4897d7be0f5dc230df00/faker-39.0.0-py3-none-any.whl", hash = "sha256:c72f1fca8f1a24b8da10fcaa45739135a19772218ddd61b86b7ea1b8c790dce7", size = 1980775, upload-time = "2025-12-17T19:19:02.926Z" },
+    { url = "https://files.pythonhosted.org/packages/fc/23/e22da510e1ec1488966330bf76d8ff4bd535cbfc93660eeb7657761a1bb2/faker-40.1.0-py3-none-any.whl", hash = "sha256:a616d35818e2a2387c297de80e2288083bc915e24b7e39d2fb5bc66cce3a929f", size = 1985317, upload-time = "2025-12-29T18:05:58.831Z" },
 ]
 
 [[package]]
@@ -2456,15 +2456,15 @@ wheels = [
 
 [[package]]
 name = "pdfminer-six"
-version = "20251228"
+version = "20251230"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "charset-normalizer" },
     { name = "cryptography" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/16/65/1ea9a0a4b0bf0e711b5ec40ec4478a3dc597955a81bafdb46ed657f88bc5/pdfminer_six-20251228.tar.gz", hash = "sha256:5972b2babc5dd576a58634023b47e41ee827b505e36793cbfe37ac899000a1fb", size = 7391349, upload-time = "2025-12-28T14:32:26.76Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/46/9a/d79d8fa6d47a0338846bb558b39b9963b8eb2dfedec61867c138c1b17eeb/pdfminer_six-20251230.tar.gz", hash = "sha256:e8f68a14c57e00c2d7276d26519ea64be1b48f91db1cdc776faa80528ca06c1e", size = 8511285, upload-time = "2025-12-30T15:49:13.104Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c3/aa/4ec00440997c382093b492e50eb117e134dd739eb6b9a1d752e68bbd3072/pdfminer_six-20251228-py3-none-any.whl", hash = "sha256:d365fb6dc41c5b8d04bd63622a3f468be24f3fecf54107e36e2451cf1bde870a", size = 5622067, upload-time = "2025-12-28T14:32:24.762Z" },
+    { url = "https://files.pythonhosted.org/packages/65/d7/b288ea32deb752a09aab73c75e1e7572ab2a2b56c3124a5d1eb24c62ceb3/pdfminer_six-20251230-py3-none-any.whl", hash = "sha256:9ff2e3466a7dfc6de6fd779478850b6b7c2d9e9405aa2a5869376a822771f485", size = 6591909, upload-time = "2025-12-30T15:49:10.76Z" },
 ]
 
 [[package]]