feat: expand ~ and $HOME in mount target paths

- Add expand_container_path() helper that resolves ~/ and $HOME/ prefixes - Add get_manifest_home_dir() to compute /home/{user.name} or /root from manifest - Set ENV HOME=... and ENV USER=... in generated Dockerfile for runtime compatibility - Pass home_dir through instance creation and startup pipeline - Expand mount targets in apply_resolved_profile() for regular profile mounts - Expand mapping targets in _resolve_git_mount_mappings() for git mounts - Expand working_directory and volume targets in _modify_compose_file() - Update _prepare_manifest_instance to return home_dir alongside image tag - Fetch tool_type early in start_instance to determine home_dir before profile application Quality gates: pytest 188 passed, frontend typecheck clean Addresses: home-path-expansion
2026-05-29 00:01:04 +02:00
parent 270764ff0f
commit 29a12bb102
17 changed files with 1364 additions and 471 deletions
@@ -58,7 +58,9 @@ def _calculate_profile_size(data: dict) -> int:


 class GitMountMapping(BaseModel):
-    source_path: str = Field(description="Path within repository (supports glob patterns)")
+    source_path: str = Field(
+        description="Path within repository (supports glob patterns)"
+    )
    target_path: str = Field(description="Absolute path inside container")

    @field_validator("source_path")
@@ -80,16 +82,25 @@ class GitMountMapping(BaseModel):

 class GitMountItem(BaseModel):
    remote_url: str = Field(description="Git remote URL (HTTPS or SSH)")
-    source_path: str | None = Field(default=None, description="Path within repository (legacy single mapping)")
-    target_path: str | None = Field(default=None, description="Absolute path inside container (legacy single mapping)")
+    source_path: str | None = Field(
+        default=None, description="Path within repository (legacy single mapping)"
+    )
+    target_path: str | None = Field(
+        default=None,
+        description="Absolute path inside container (legacy single mapping)",
+    )
    branch: str | None = Field(default=None, description="Optional branch or tag name")
-    mappings: list[GitMountMapping] | None = Field(default=None, description="Multiple source/target mappings from the same repo")
+    mappings: list[GitMountMapping] | None = Field(
+        default=None, description="Multiple source/target mappings from the same repo"
+    )

    @field_validator("remote_url")
    @classmethod
    def validate_remote_url(cls, v: str) -> str:
        if not v.startswith(("http://", "https://", "git@", "ssh://")):
-            raise ValueError("remote_url must be a valid git URL (https://, git@, or ssh://)")
+            raise ValueError(
+                "remote_url must be a valid git URL (https://, git@, or ssh://)"
+            )
        return v

    @field_validator("source_path")
@@ -126,7 +137,9 @@ class GitMountItem(BaseModel):
 class MountItem(BaseModel):
    target: str = Field(description="Absolute mount target path")
    mode: str = Field(default="rw", description="Mount mode: ro or rw")
-    files: dict = Field(default_factory=dict, description="Files as {relative_path: content}")
+    files: dict = Field(
+        default_factory=dict, description="Files as {relative_path: content}"
+    )

    @field_validator("target")
    @classmethod
@@ -163,10 +176,18 @@ class ConfigProfileCreate(BaseModel):
    tool_type_id: str | None = Field(default=None, description="Optional tool type ID")
    env_vars: dict = Field(default_factory=dict, description="Environment variables")
    runtime_hints: dict = Field(default_factory=dict, description="Runtime hints")
-    mounts: list[MountItem] = Field(default_factory=list, description="Mount definitions")
-    files: dict = Field(default_factory=dict, description="Files as {relative_path: content}")
-    git_mounts: list[GitMountItem] = Field(default_factory=list, description="Git repository mounts")
-    is_default: bool = Field(default=False, description="Whether this is the default profile for its scope")
+    mounts: list[MountItem] = Field(
+        default_factory=list, description="Mount definitions"
+    )
+    files: dict = Field(
+        default_factory=dict, description="Files as {relative_path: content}"
+    )
+    git_mounts: list[GitMountItem] = Field(
+        default_factory=list, description="Git repository mounts"
+    )
+    is_default: bool = Field(
+        default=False, description="Whether this is the default profile for its scope"
+    )

    @field_validator("project_id", "tool_type_id")
    @classmethod
@@ -216,10 +237,18 @@ class ConfigProfileUpdate(BaseModel):
    tool_type_id: str | None = Field(default=None, description="Optional tool type ID")
    env_vars: dict | None = Field(default=None, description="Environment variables")
    runtime_hints: dict | None = Field(default=None, description="Runtime hints")
-    mounts: list[MountItem] | None = Field(default=None, description="Mount definitions")
-    files: dict | None = Field(default=None, description="Files as {relative_path: content}")
-    git_mounts: list[GitMountItem] | None = Field(default=None, description="Git repository mounts")
-    is_default: bool | None = Field(default=None, description="Whether this is the default profile")
+    mounts: list[MountItem] | None = Field(
+        default=None, description="Mount definitions"
+    )
+    files: dict | None = Field(
+        default=None, description="Files as {relative_path: content}"
+    )
+    git_mounts: list[GitMountItem] | None = Field(
+        default=None, description="Git repository mounts"
+    )
+    is_default: bool | None = Field(
+        default=None, description="Whether this is the default profile"
+    )

    @field_validator("project_id", "tool_type_id")
    @classmethod
@@ -269,7 +298,9 @@ class ConfigProfileResponse(BaseModel):
    updated_at: str


-async def _get_profile_with_includes(session: AsyncSession, profile_id: uuid.UUID) -> ConfigProfile | None:
+async def _get_profile_with_includes(
+    session: AsyncSession, profile_id: uuid.UUID
+) -> ConfigProfile | None:
    """Fetch a profile with includes eagerly loaded."""
    result = await session.execute(
        select(ConfigProfile)
@@ -289,12 +320,16 @@ async def _check_access(
    if project_id is not None:
        project = await session.get(Project, project_id)
        if project is None:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
+            )
        # Add ownership check if needed; for now just verify existence
    if tool_type_id is not None:
        tool_type = await session.get(ToolType, tool_type_id)
        if tool_type is None:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tool type not found")
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail="Tool type not found"
+            )


 async def _validate_git_mounts(
@@ -304,7 +339,7 @@ async def _validate_git_mounts(
    project_id: uuid.UUID | None = None,
 ) -> None:
    """Validate git mount URLs.
-    
+
    Simply checks that remote_url looks like a valid git URL.
    Actual clone validation happens at instance startup time.
    """
@@ -315,7 +350,7 @@ async def _validate_git_mounts(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Git mount missing remote_url",
            )
-        
+
        if not remote_url.startswith(("http://", "https://", "git@", "ssh://")):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
@@ -323,7 +358,9 @@ async def _validate_git_mounts(
            )


-def _profile_to_response(profile: ConfigProfile, includes: list[ConfigProfileInclude] | None = None) -> dict:
+def _profile_to_response(
+    profile: ConfigProfile, includes: list[ConfigProfileInclude] | None = None
+) -> dict:
    return {
        "id": str(profile.id),
        "user_id": str(profile.user_id),
@@ -353,13 +390,19 @@ def _profile_to_response(profile: ConfigProfile, includes: list[ConfigProfileInc
@router.get("", response_model=list[ConfigProfileResponse])
 async def list_config_profiles(
    project_id: str | None = Query(None, description="Filter by project compatibility"),
-    tool_type_id: str | None = Query(None, description="Filter by tool type compatibility"),
+    tool_type_id: str | None = Query(
+        None, description="Filter by tool type compatibility"
+    ),
    current_user_id: uuid.UUID = Depends(get_current_user_id),
    session: AsyncSession = Depends(get_db_session),
 ):
    """List config profiles, optionally filtered by compatibility."""
    user_uuid = current_user_id
-    query = select(ConfigProfile).where(ConfigProfile.user_id == user_uuid).options(selectinload(ConfigProfile.includes))
+    query = (
+        select(ConfigProfile)
+        .where(ConfigProfile.user_id == user_uuid)
+        .options(selectinload(ConfigProfile.includes))
+    )

    if project_id or tool_type_id:
        # Compatibility filter: include portable profiles and matching scoped profiles
@@ -371,7 +414,8 @@ async def list_config_profiles(
        conditions: list = []
        # Portable profiles (no project, no tool)
        conditions.append(
-            (ConfigProfile.project_id.is_(None)) & (ConfigProfile.tool_type_id.is_(None))
+            (ConfigProfile.project_id.is_(None))
+            & (ConfigProfile.tool_type_id.is_(None))
        )
        if project_uuid:
            # Profiles matching this project (with or without tool)
@@ -382,7 +426,8 @@ async def list_config_profiles(
        if project_uuid and tool_uuid:
            # Exact match
            conditions.append(
-                (ConfigProfile.project_id == project_uuid) & (ConfigProfile.tool_type_id == tool_uuid)
+                (ConfigProfile.project_id == project_uuid)
+                & (ConfigProfile.tool_type_id == tool_uuid)
            )

        query = query.where(or_(*conditions))
@@ -392,7 +437,9 @@ async def list_config_profiles(
    return [_profile_to_response(p) for p in profiles]


-@router.post("", response_model=ConfigProfileResponse, status_code=status.HTTP_201_CREATED)
+@router.post(
+    "", response_model=ConfigProfileResponse, status_code=status.HTTP_201_CREATED
+)
 async def create_config_profile(
    data: ConfigProfileCreate,
    current_user_id: uuid.UUID = Depends(get_current_user_id),
@@ -403,10 +450,12 @@ async def create_config_profile(

    # Check for duplicate name
    existing = await session.execute(
-        select(ConfigProfile).where(
+        select(ConfigProfile)
+        .where(
            ConfigProfile.user_id == user_uuid,
            ConfigProfile.name == data.name,
-        ).options(selectinload(ConfigProfile.includes))
+        )
+        .options(selectinload(ConfigProfile.includes))
    )
    if existing.scalar_one_or_none() is not None:
        raise HTTPException(
@@ -418,10 +467,12 @@ async def create_config_profile(
    project_uuid = uuid.UUID(data.project_id) if data.project_id else None
    tool_uuid = uuid.UUID(data.tool_type_id) if data.tool_type_id else None
    await _check_access(session, user_uuid, project_uuid, tool_uuid)
-    
+
    # Validate git mounts reference existing repositories
    if data.git_mounts:
-        git_mounts_data = [m.model_dump() if hasattr(m, "model_dump") else m for m in data.git_mounts]
+        git_mounts_data = [
+            m.model_dump() if hasattr(m, "model_dump") else m for m in data.git_mounts
+        ]
        await _validate_git_mounts(session, user_uuid, git_mounts_data, project_uuid)

    # Check size
@@ -469,9 +520,13 @@ async def get_config_profile(
    """Get a config profile by ID."""
    profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
    if profile is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found"
+        )
    if profile.user_id != current_user_id:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized"
+        )
    return _profile_to_response(profile)


@@ -485,9 +540,13 @@ async def update_config_profile(
    """Update a config profile."""
    profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
    if profile is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found"
+        )
    if profile.user_id != current_user_id:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized"
+        )

    update_data = data.model_dump(exclude_unset=True)

@@ -518,14 +577,16 @@ async def update_config_profile(
        else (profile.tool_type_id if "tool_type_id" not in update_data else None)
    )
    await _check_access(session, profile.user_id, project_uuid, tool_uuid)
-    
+
    # Validate git mounts reference existing repositories
    if "git_mounts" in update_data and update_data["git_mounts"] is not None:
        git_mounts_data = [
-            m.model_dump() if hasattr(m, "model_dump") else m 
+            m.model_dump() if hasattr(m, "model_dump") else m
            for m in update_data["git_mounts"]
        ]
-        await _validate_git_mounts(session, profile.user_id, git_mounts_data, project_uuid)
+        await _validate_git_mounts(
+            session, profile.user_id, git_mounts_data, project_uuid
+        )

    # Check size
    current_data = _profile_to_response(profile)
@@ -570,9 +631,13 @@ async def delete_config_profile(
    """Delete a config profile."""
    profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
    if profile is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found"
+        )
    if profile.user_id != current_user_id:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized"
+        )

    await session.delete(profile)
    await session.commit()
@@ -591,9 +656,13 @@ async def update_profile_includes(
    """Update the ordered includes for a config profile."""
    profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
    if profile is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found"
+        )
    if profile.user_id != current_user_id:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized"
+        )

    # Validate all included profiles exist and belong to the user
    included_uuids = [uuid.UUID(inc_id) for inc_id in data.includes]
@@ -633,7 +702,9 @@ async def update_profile_includes(

    # Remove existing includes
    result = await session.execute(
-        select(ConfigProfileInclude).where(ConfigProfileInclude.profile_id == profile.id)
+        select(ConfigProfileInclude).where(
+            ConfigProfileInclude.profile_id == profile.id
+        )
    )
    for existing in result.scalars().all():
        await session.delete(existing)
@@ -658,7 +729,9 @@ async def update_profile_includes(
    profile = result.scalar_one()

    inc_result = await session.execute(
-        select(ConfigProfileInclude).where(ConfigProfileInclude.profile_id == profile.id)
+        select(ConfigProfileInclude).where(
+            ConfigProfileInclude.profile_id == profile.id
+        )
    )
    direct_includes = inc_result.scalars().all()

@@ -675,9 +748,13 @@ async def preview_config_profile(
    """Preview the resolved output of a config profile."""
    profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
    if profile is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found"
+        )
    if profile.user_id != current_user_id:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized"
+        )

    try:
        resolved = await resolve_profile(session, profile.id)
@@ -39,6 +39,7 @@ from src.services.config_profile_resolver import (
    ConfigProfileCycleError,
    ResolvedProfile,
    apply_resolved_profile,
+    expand_container_path,
    resolve_profile,
 )
 from src.services.docker import (
@@ -67,6 +68,7 @@ from src.services.manifest_compiler import (
    compile_entrypoint,
    compute_image_tag,
    deep_merge,
+    get_manifest_home_dir,
    merge_with_config,
    resolve_base,
 )
@@ -82,6 +84,7 @@ async def _resolve_git_mounts(
    resolved: ResolvedProfile,
    instance_dir: str | None = None,
    working_directory: str | None = None,
+    home_dir: str = "/root",
 ) -> list[dict]:
    """Convert git mounts from resolved profile to Docker volume mounts.

@@ -97,7 +100,7 @@ async def _resolve_git_mounts(
    for git_mount in resolved.git_mounts:
        tasks.append(
            _resolve_single_git_mount(
-                session, git_mount, instance_dir, working_directory
+                session, git_mount, instance_dir, working_directory, home_dir
            )
        )

@@ -184,6 +187,7 @@ def _resolve_git_mount_mappings(
    repo_path: str,
    mappings: list[dict],
    working_directory: str | None,
+    home_dir: str = "/root",
 ) -> list[dict]:
    """Resolve mappings from an already-cloned repo to volume mount entries.

@@ -199,6 +203,9 @@ def _resolve_git_mount_mappings(
            logger.warning("Invalid mapping skipped: missing target_path")
            continue

+        # Expand ~ and $HOME in target path
+        target_path = expand_container_path(target_path, home_dir)
+
        # Resolve relative target paths against working directory
        final_target = target_path
        if not target_path.startswith("/"):
@@ -261,6 +268,7 @@ async def _resolve_single_git_mount(
    git_mount: dict,
    instance_dir: str | None = None,
    working_directory: str | None = None,
+    home_dir: str = "/root",
 ) -> list[dict]:
    """Resolve a single git mount to volume mount entries.

@@ -293,7 +301,7 @@ async def _resolve_single_git_mount(
        return []

    # Resolve all mappings from the cloned repo
-    return _resolve_git_mount_mappings(repo_path, mappings, working_directory)
+    return _resolve_git_mount_mappings(repo_path, mappings, working_directory, home_dir)


 def _checkout_branch(repo_path: str, branch: str) -> bool:
@@ -545,6 +553,7 @@ def _modify_compose_file(
    start_command: str | None = None,
    working_directory: str | None = None,
    extra_volumes: list[dict] | None = None,
+    home_dir: str = "/root",
 ) -> None:
    """Modify compose file with runtime overrides."""
    import yaml
@@ -571,14 +580,16 @@ def _modify_compose_file(
            service_config["command"] = start_command

        if working_directory:
-            service_config["working_dir"] = working_directory
+            service_config["working_dir"] = expand_container_path(
+                working_directory, home_dir
+            )

        if extra_volumes:
            if "volumes" not in service_config:
                service_config["volumes"] = []
            for vol in extra_volumes:
                source = vol.get("source", "")
-                target = vol.get("target", "")
+                target = expand_container_path(vol.get("target", ""), home_dir)
                vol_type = vol.get("type", "bind")
                if vol_type == "bind":
                    service_config["volumes"].append(f"{source}:{target}")
@@ -826,6 +837,9 @@ services:
                        deep_merge(dict(base_def.manifest), manifest)
                    )

+            # Determine home directory for path expansion
+            home_dir = get_manifest_home_dir(manifest)
+
            image_tag = compute_image_tag(tool_type.name, manifest)

            # Build image during creation so start is fast
@@ -1105,11 +1119,11 @@ async def _prepare_manifest_instance(
    env_vars: dict,
    extra_volumes: list,
    working_directory: str | None,
-) -> tuple[str, str, dict]:
+) -> tuple[str, str, dict, str]:
    """Build image and generate compose from a manifest-based tool type.

    Returns:
-        Tuple of (image_tag, compose_content, resolved_manifest)
+        Tuple of (image_tag, compose_content, resolved_manifest, home_dir)
    """
    from src.models.tool_definition_manifest import ToolDefinitionManifest

@@ -1220,7 +1234,8 @@ async def _prepare_manifest_instance(
    instance.image_tag = image_tag
    instance.manifest_compiled_at = datetime.now()

-    return image_tag, compose_content, manifest
+    home_dir = get_manifest_home_dir(manifest)
+    return image_tag, compose_content, manifest, home_dir


@router.post(
@@ -1283,6 +1298,15 @@ async def start_instance(
    working_directory = None
    extra_volumes = []

+    # Fetch tool type early to determine home directory
+    tool_type = await session.get(ToolType, instance.tool_type_id)
+    home_dir = "/root"
+    if tool_type and tool_type.definition_type == "manifest" and tool_type.manifest_id:
+        from src.models.tool_definition_manifest import ToolDefinitionManifest
+        manifest_def = await session.get(ToolDefinitionManifest, tool_type.manifest_id)
+        if manifest_def:
+            home_dir = get_manifest_home_dir(dict(manifest_def.manifest))
+
    # Apply selected config profile if any
    instance_dir = os.path.dirname(instance.compose_path)
    if instance.selected_config_profile_id is not None:
@@ -1291,7 +1315,7 @@ async def start_instance(
                session, instance.selected_config_profile_id
            )
            profile_env, profile_files, profile_mounts, profile_hints = (
-                apply_resolved_profile(instance_dir, resolved)
+                apply_resolved_profile(instance_dir, resolved, home_dir)
            )
            # Profile env vars override tool config env vars
            env_vars.update(profile_env)
@@ -1301,7 +1325,7 @@ async def start_instance(
            extra_volumes.extend(profile_mounts)
            # Git repository mounts are resolved and added
            git_mount_volumes = await _resolve_git_mounts(
-                session, resolved, instance_dir, working_directory
+                session, resolved, instance_dir, working_directory, home_dir
            )
            extra_volumes.extend(git_mount_volumes)
            # Profile runtime hints override tool config values
@@ -1345,7 +1369,6 @@ async def start_instance(
        )

    # ── MANIFEST-BASED FLOW ──────────────────────────────────────
-    tool_type = await session.get(ToolType, instance.tool_type_id)
    resolved_manifest = None

    if tool_type and tool_type.definition_type == "manifest" and tool_type.manifest_id:
@@ -1362,6 +1385,7 @@ async def start_instance(
                image_tag,
                compose_content,
                resolved_manifest,
+                _home_dir,
            ) = await _prepare_manifest_instance(
                session=session,
                instance=instance,
@@ -1420,6 +1444,7 @@ async def start_instance(
                start_command,
                working_directory,
                extra_volumes,
+                home_dir,
            )
            logger.debug("Modified compose file for instance %s", instance.id)

@@ -5,6 +5,7 @@ and cycle protection.
 """

 import logging
+import os
 import uuid
 from dataclasses import dataclass, field
 from typing import Any
@@ -57,7 +58,9 @@ class ResolvedProfile:
    included_profiles: list[dict[str, Any]] = field(default_factory=list)


-def _detect_cycle(profile_id: uuid.UUID, visited: set[uuid.UUID], path: list[uuid.UUID]) -> bool:
+def _detect_cycle(
+    profile_id: uuid.UUID, visited: set[uuid.UUID], path: list[uuid.UUID]
+) -> bool:
    """Detect if adding profile_id to path would create a cycle.

    Args:
@@ -252,7 +255,9 @@ async def _resolve_profile_recursive(
    """
    if _detect_cycle(profile_id, visited, path):
        cycle_path = " -> ".join(str(p) for p in path + [profile_id])
-        raise ConfigProfileCycleError(f"Cycle detected in profile includes: {cycle_path}")
+        raise ConfigProfileCycleError(
+            f"Cycle detected in profile includes: {cycle_path}"
+        )

    profile = await session.get(ConfigProfile, profile_id)
    if profile is None:
@@ -279,13 +284,18 @@ async def _resolve_profile_recursive(
        included = await _resolve_profile_recursive(
            session, include.included_profile_id, new_visited, new_path
        )
-        result.included_profiles.append({
-            "id": str(included.profile_id),
-            "name": included.profile_name,
-        })
+        result.included_profiles.append(
+            {
+                "id": str(included.profile_id),
+                "name": included.profile_name,
+            }
+        )

        result.env_vars = _merge_env_vars(
-            result.env_vars, included.env_vars, result.env_overrides, included.profile_name
+            result.env_vars,
+            included.env_vars,
+            result.env_overrides,
+            included.profile_name,
        )
        result.runtime_hints = _merge_runtime_hints(
            result.runtime_hints,
@@ -429,6 +439,7 @@ async def check_include_cycle(
 def apply_resolved_profile(
    instance_dir: str,
    resolved: ResolvedProfile,
+    home_dir: str = "/root",
 ) -> tuple[dict[str, str], dict[str, str], list[dict], dict[str, Any]]:
    """Apply a resolved profile to an instance directory.

@@ -458,14 +469,19 @@ def apply_resolved_profile(
        try:
            full_path.resolve().relative_to(instance_path.resolve())
        except ValueError:
-            logger.warning("Profile file path escapes instance directory: %s", file_path)
+            logger.warning(
+                "Profile file path escapes instance directory: %s", file_path
+            )
            continue
        full_path.parent.mkdir(parents=True, exist_ok=True)
        full_path.write_text(content)

    # Stage mount files and prepare volume mounts
    for mount in resolved.mounts.values():
-        mount_dir = instance_path / "mounts" / mount.target.lstrip("/").replace("/", "_")
+        expanded_target = expand_container_path(mount.target, home_dir)
+        mount_dir = (
+            instance_path / "mounts" / expanded_target.lstrip("/").replace("/", "_")
+        )
        mount_dir.mkdir(parents=True, exist_ok=True)

        for file_path, content in mount.files.items():
@@ -478,15 +494,41 @@ def apply_resolved_profile(
            full_path.parent.mkdir(parents=True, exist_ok=True)
            full_path.write_text(content)

-        volume_mounts.append({
-            "source": str(mount_dir),
-            "target": mount.target,
-            "type": "bind",
-        })
+        volume_mounts.append(
+            {
+                "source": str(mount_dir),
+                "target": expanded_target,
+                "type": "bind",
+            }
+        )

    return env_vars, files, volume_mounts, resolved.runtime_hints


+def expand_container_path(path: str, home_dir: str) -> str:
+    """Expand ~ and $HOME in a container path to the actual home directory.
+
+    Only expands at the start of the path (e.g., ~/foo, $HOME/foo, $HOME).
+    Leaves mid-string occurrences unchanged.
+
+    Args:
+        path: Container path that may contain ~ or $HOME.
+        home_dir: The container's home directory (e.g., /home/user or /root).
+
+    Returns:
+        Path with ~ and $HOME expanded.
+    """
+    if path.startswith("~/"):
+        return os.path.join(home_dir, path[2:])
+    if path == "~":
+        return home_dir
+    if path.startswith("$HOME/"):
+        return home_dir + "/" + path[6:]
+    if path == "$HOME":
+        return home_dir
+    return path
+
+
 def resolved_profile_to_dict(resolved: ResolvedProfile) -> dict[str, Any]:
    """Convert a ResolvedProfile to a plain dict for serialization.

@@ -160,6 +160,11 @@ def compile_dockerfile(manifest: dict) -> str:
        lines.append(f"RUN groupadd -g {gid} {name} && \\")
        lines.append(f"    useradd -u {uid} -g {gid} {create_home}-s {shell} {name}")
        lines.append("")
+        # Set HOME and USER for runtime compatibility
+        home = f"/home/{name}"
+        lines.append(f"ENV HOME={home}")
+        lines.append(f"ENV USER={name}")
+        lines.append("")

    # Build scripts
    build_scripts = manifest.get("scripts", {}).get("build", [])
@@ -333,6 +338,21 @@ def resolve_mount_source(mount: dict, variables: dict[str, Any]) -> str:
    return ""


+def get_manifest_home_dir(manifest: dict) -> str:
+    """Get the home directory for a container based on manifest user config.
+
+    Args:
+        manifest: Fully resolved manifest JSON.
+
+    Returns:
+        Home directory path (e.g., /home/user or /root).
+    """
+    user = manifest.get("user")
+    if user and user.get("name"):
+        return f"/home/{user['name']}"
+    return "/root"
+
+
 def compute_image_tag(tool_name: str, manifest: dict) -> str:
    """Compute a deterministic image tag from manifest content.