diff --git a/apps/api/src/api/tool/tool_instances.py b/apps/api/src/api/tool/tool_instances.py index 5206d08..255e914 100644 --- a/apps/api/src/api/tool/tool_instances.py +++ b/apps/api/src/api/tool/tool_instances.py @@ -83,681 +83,28 @@ from src.services.shared.permission_fixer import ( from src.services.shared.readiness_probe import execute_probe from src.services.shared.ssh_keys import cleanup_ssh_key_files, prepare_ssh_key_files from src.schemas.tool import CreateInstanceRequest, StartInstanceRequest +from src.services.tool.instance_service import ( + checkout_branch, + clone_git_repo, + ensure_backend_network_in_compose, + ensure_container_name_in_compose, + ensure_web_bind_address, + expand_glob_source, + modify_compose_file, + normalize_git_mount, + pull_repository_updates, + resolve_git_mount_mappings, + resolve_git_mounts, + resolve_single_git_mount, + sanitize_compose_file, + validate_config_profile, +) + logger = logging.getLogger(__name__) _event_bus = InstanceEventBus() -async def _resolve_git_mounts( - session: AsyncSession, - 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. - - Looks up repository paths, auto-clones if needed, handles branch checkout, - expands glob patterns, and prepares bind mount entries. - Logs warnings for missing repos or invalid paths (non-blocking). - """ - if not resolved.git_mounts: - return [] - - # Process all git mounts concurrently - tasks = [] - for git_mount in resolved.git_mounts: - tasks.append( - _resolve_single_git_mount( - session, git_mount, instance_dir, working_directory, home_dir - ) - ) - - results = await asyncio.gather(*tasks, return_exceptions=True) - - volume_mounts = [] - for result in results: - if isinstance(result, Exception): - logger.warning("Git mount failed: %s", result) - continue - if result: - volume_mounts.extend(result) - - return volume_mounts - - -def _normalize_git_mount(entry: dict) -> dict: - """Normalize a git mount entry to the unified mappings format. - - Converts legacy source_path + target_path into a single-entry mappings array. - """ - entry = dict(entry) - if "mappings" not in entry or not entry.get("mappings"): - source = entry.get("source_path", ".") - target = entry.get("target_path") - if target is not None: - entry["mappings"] = [{"source_path": source, "target_path": target}] - entry.pop("source_path", None) - entry.pop("target_path", None) - return entry - - -def _clone_git_repo( - remote_url: str, - branch: str | None, - clone_parent: str, -) -> str: - """Clone or pull a git repository. - - Returns the path to the cloned repo (repo-clone directory). - """ - import hashlib - - url_hash = hashlib.md5(remote_url.encode()).hexdigest()[:12] - repo_name = remote_url.split("/")[-1].replace(".git", "") or "repo" - clone_dir = os.path.join(clone_parent, "git-mounts", f"{repo_name}-{url_hash}") - repo_path = os.path.join(clone_dir, "repo-clone") - - if not os.path.exists(repo_path): - try: - os.makedirs(clone_dir, exist_ok=True) - repo_path = clone_repository( - remote_url, - None, # No SSH key for now - can be added later - clone_dir, - branch or "main", - ) - logger.debug("Cloned git mount repository %s to %s", remote_url, repo_path) - except Exception as exc: - logger.warning("Clone failed for git mount %s: %s", remote_url, exc) - raise - else: - # Repo exists - pull latest updates - try: - _pull_repository_updates(repo_path, remote_url) - logger.debug("Pulled updates for git mount %s", remote_url) - except Exception as exc: - logger.warning("Failed to pull updates for %s: %s", remote_url, exc) - - # Handle branch checkout if specified - if branch and repo_path: - success = _checkout_branch(repo_path, branch) - if success: - logger.debug("Checked out branch %s for %s", branch, remote_url) - else: - logger.warning( - "Branch %s not found in %s, using current branch", branch, remote_url - ) - - return repo_path - - -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. - - Returns a flat list of volume mount dicts. - """ - volume_mounts = [] - - for mapping in mappings: - source_path = mapping.get("source_path", ".") - target_path = mapping.get("target_path") - - if not target_path: - 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("/"): - if not working_directory: - logger.warning( - "Git mount skipped: target_path '%s' is relative but no working_directory is configured. " - "Set working_directory in the tool config or use an absolute path.", - target_path, - ) - continue - final_target = os.path.join(working_directory, target_path) - - # Build source path and expand globs - if source_path and source_path != ".": - source_full = os.path.join(repo_path, source_path) - else: - source_full = repo_path - - # Expand glob patterns - matched_paths = _expand_glob_source(source_full, repo_path) - - if not matched_paths: - logger.warning( - "Git mount skipped: no files matched source path %s in repo", - source_path, - ) - continue - - for matched_path in matched_paths: - if not os.path.exists(matched_path): - continue - - # Determine target path for this match - if len(matched_paths) == 1: - # Single match: mount directly to target_path - mount_target = final_target - else: - # Multiple matches: append relative path to target - rel_path = os.path.relpath(matched_path, repo_path) - mount_target = os.path.join(final_target, rel_path) - - volume_mounts.append( - { - "source": matched_path, - "target": mount_target, - "type": "bind", - } - ) - logger.debug( - "Added git mount: %s -> %s", - matched_path, - mount_target, - ) - - return volume_mounts - - -async def _resolve_single_git_mount( - session: AsyncSession, - 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. - - Clones directly from remote_url, no database lookup needed. - Returns a list of volume mounts (one for each matched file/directory). - """ - git_mount = _normalize_git_mount(git_mount) - remote_url = git_mount.get("remote_url") - branch = git_mount.get("branch") - mappings = git_mount.get("mappings", []) - - if not remote_url: - logger.warning("Invalid git mount skipped: missing remote_url") - return [] - - if not mappings: - logger.warning("Invalid git mount skipped: no mappings") - return [] - - if not instance_dir: - logger.warning("Git mount skipped: no instance_dir provided for cloning") - return [] - - # Clone or pull the repository - try: - repo_path = await asyncio.to_thread( - _clone_git_repo, remote_url, branch, instance_dir - ) - except Exception: - return [] - - # Resolve all mappings from the cloned repo - return _resolve_git_mount_mappings(repo_path, mappings, working_directory, home_dir) - - -def _checkout_branch(repo_path: str, branch: str) -> bool: - """Checkout a specific branch in a git repository. - - Returns True if checkout succeeded, False if it failed. - On failure, the repository remains on its current branch. - """ - import subprocess - - # First try to checkout existing branch - result = subprocess.run( - ["git", "-C", repo_path, "checkout", branch], - capture_output=True, - text=True, - ) - - if result.returncode != 0: - # Try fetching and checking out - subprocess.run( - ["git", "-C", repo_path, "fetch", "origin", branch], - capture_output=True, - text=True, - ) - result = subprocess.run( - ["git", "-C", repo_path, "checkout", "-b", branch, f"origin/{branch}"], - capture_output=True, - text=True, - ) - - if result.returncode != 0: - logger.warning( - "Failed to checkout branch %s in %s: %s", - branch, - repo_path, - result.stderr.strip(), - ) - return False - - return True - - -def _pull_repository_updates(repo_path: str, remote_url: str) -> None: - """Pull latest updates from remote repository. - - Used when starting a new container with an existing cloned repository - to ensure the latest code is mounted. - """ - import subprocess - - # Fetch latest changes - result = subprocess.run( - ["git", "-C", repo_path, "fetch", "origin"], - capture_output=True, - text=True, - ) - - if result.returncode != 0: - raise RuntimeError(f"Failed to fetch updates: {result.stderr}") - - # Pull changes for current branch - result = subprocess.run( - ["git", "-C", repo_path, "pull", "origin"], - capture_output=True, - text=True, - ) - - if result.returncode != 0: - raise RuntimeError(f"Failed to pull updates: {result.stderr}") - - -def _expand_glob_source(source_path: str, repo_path: str) -> list[str]: - """Expand glob patterns in source path. - - Returns a list of matched absolute paths. - Limits results to prevent abuse. - """ - MAX_GLOB_MATCHES = 100 - - # Check if path contains glob characters - if not any(c in source_path for c in "*?["): - # No glob pattern: return single path if it exists - return [source_path] if os.path.exists(source_path) else [] - - # Expand glob pattern - matched = glob_module.glob(source_path, recursive=True) - total_matched = len(matched) - - # Filter to only paths within the repo and limit count - results = [] - for path in matched: - abs_path = os.path.abspath(path) - if abs_path.startswith(os.path.abspath(repo_path)): - results.append(abs_path) - if len(results) >= MAX_GLOB_MATCHES: - logger.warning( - "Glob pattern matched %d files, limited to %d", - total_matched, - MAX_GLOB_MATCHES, - ) - break - - return results - - -router = APIRouter(prefix="/projects", tags=["tool-instances"]) - - -async def _validate_config_profile( - session: AsyncSession, - profile_id: str | None, - user_id: uuid.UUID, - project_id: uuid.UUID, - tool_type_id: uuid.UUID, -) -> uuid.UUID | None: - """Validate a config profile selection. - - Args: - session: Database session. - profile_id: Profile ID string or None. - user_id: Authenticated user ID. - project_id: Project ID for compatibility check. - tool_type_id: Tool type ID for compatibility check. - - Returns: - Validated UUID or None. - - Raises: - HTTPException: If profile is not found, not owned, or incompatible. - """ - if not profile_id: - return None - - try: - profile_uuid = uuid.UUID(profile_id) - except ValueError: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Invalid config profile ID: {profile_id}", - ) - - profile = await session.get(ConfigProfile, profile_uuid) - if profile is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"Config profile not found: {profile_id}", - ) - - if profile.user_id != user_id: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Not authorized to use this config profile", - ) - - # Check compatibility: profile must be portable or match project/tool - is_compatible = ( - (profile.project_id is None and profile.tool_type_id is None) - or (profile.project_id == project_id) - or (profile.tool_type_id == tool_type_id) - or (profile.project_id == project_id and profile.tool_type_id == tool_type_id) - ) - - if not is_compatible: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Selected config profile is not compatible with this project and tool type", - ) - - return profile_uuid - - -def _sanitize_compose_file(compose_path: str) -> None: - """Remove invalid port mappings (target port 0) from compose file.""" - import yaml - from pathlib import Path - - compose_file = Path(compose_path) - if not compose_file.exists(): - return - - content = compose_file.read_text() - compose_data = yaml.safe_load(content) - - if not compose_data or "services" not in compose_data: - return - - modified = False - for service_name, service_config in compose_data["services"].items(): - if "ports" in service_config: - valid_ports = [] - for port_mapping in service_config["ports"]: - if isinstance(port_mapping, str) and ":" in port_mapping: - parts = port_mapping.split(":") - if len(parts) == 2: - host_port, container_port = parts - # Skip invalid mappings (target port 0 or empty) - if container_port == "0" or not container_port: - modified = True - continue - valid_ports.append(port_mapping) - - if valid_ports: - service_config["ports"] = valid_ports - else: - del service_config["ports"] - modified = True - break # Only check first service - - if modified: - compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) - - -def _modify_compose_file( - compose_path: str, - port_override: int | None = None, - 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 - from pathlib import Path - - compose_file = Path(compose_path) - content = compose_file.read_text() - compose_data = yaml.safe_load(content) - - if not compose_data or "services" not in compose_data: - return - - # Apply modifications to the first service - for service_name, service_config in compose_data["services"].items(): - if port_override and "ports" in service_config: - # Update port mapping - for i, port_mapping in enumerate(service_config["ports"]): - if isinstance(port_mapping, str) and ":" in port_mapping: - host_port, container_port = port_mapping.split(":", 1) - service_config["ports"][i] = f"{port_override}:{container_port}" - break - - if start_command: - service_config["command"] = start_command - - if 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 = 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}") - else: - service_config["volumes"].append(f"{source}:{target}:{vol_type}") - - # Sort volumes so parent paths come before child paths - if service_config.get("volumes"): - service_config["volumes"] = sort_volumes_by_specificity( - service_config["volumes"] - ) - - break # Only modify the first service - - # Write back - compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) - - -def _ensure_container_name_in_compose(compose_path: str, container_name: str) -> None: - """Ensure compose file has explicit container_name for predictable naming. - - Docker Compose auto-generates container names from the project directory - when container_name is absent. This breaks tunnel connectivity because - get_container_name(instance.name) cannot find the container. We inject - container_name into every service so the container has a predictable name. - """ - import yaml - from pathlib import Path - - compose_file = Path(compose_path) - if not compose_file.exists(): - return - - content = compose_file.read_text() - compose_data = yaml.safe_load(content) - - if not compose_data or "services" not in compose_data: - return - - modified = False - for svc_name, svc_config in compose_data["services"].items(): - if "container_name" not in svc_config: - svc_config["container_name"] = container_name.lower() - modified = True - - if modified: - compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) - logger.info( - "Injected container_name '%s' into compose file", - container_name.lower(), - ) - - -def _ensure_web_bind_address( - compose_path: str, tool_type_name: str, default_port: int -) -> None: - """Auto-inject bind address for known web tools that default to 127.0.0.1. - - Many web tools (code-server, jupyter) bind to localhost by default, - making them inaccessible from the Docker network. This function detects - known tool images and injects the correct --bind-addr or --ip flag. - """ - import yaml - from pathlib import Path - - if default_port <= 0: - return - - KNOWN_BIND_FIXES: dict[str, str] = { - "code-server": f"--bind-addr 0.0.0.0:{default_port}", - "jupyter-notebook": f"start-notebook.sh --ip=0.0.0.0 --port={default_port} --no-browser", - } - - bind_command = KNOWN_BIND_FIXES.get(tool_type_name) - if not bind_command: - return - - compose_file = Path(compose_path) - if not compose_file.exists(): - return - - content = compose_file.read_text() - compose_data = yaml.safe_load(content) - - if not compose_data or "services" not in compose_data: - return - - for service_config in compose_data["services"].values(): - image = service_config.get("image", "") - if not image: - continue - - # LSIO images already bind to 0.0.0.0 — command override breaks s6 init - if "linuxserver" in image: - existing_command = service_config.get("command", "") - if "--bind-addr" in existing_command or "--host" in existing_command: - del service_config["command"] - compose_file.write_text( - yaml.dump(compose_data, default_flow_style=False) - ) - logger.warning( - "Removed broken command override from LSIO image: %s", - existing_command, - ) - return - return - - # Check if the image matches a known tool - is_code_server = tool_type_name == "code-server" and ( - "code-server" in image or "coder" in image - ) - is_jupyter = tool_type_name == "jupyter-notebook" and ( - "jupyter" in image or "notebook" in image - ) - if not is_code_server and not is_jupyter: - continue - - existing_command = service_config.get("command", "") - if existing_command: - # Already correct — nothing to do - if bind_command in existing_command: - return - # Fix broken or outdated bind flags - if ( - "--bind-addr" in existing_command - or "--host" in existing_command - or "--ip=" in existing_command - ): - service_config["command"] = bind_command - compose_file.write_text( - yaml.dump(compose_data, default_flow_style=False) - ) - logger.warning( - "Replaced broken bind address for %s: %s → %s", - tool_type_name, - existing_command, - bind_command, - ) - return - # Some other command override exists — don't touch it - return - - # No command yet — inject the correct bind address - service_config["command"] = bind_command - compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) - logger.info("Injected bind address for %s: %s", tool_type_name, bind_command) - return - - -def _ensure_backend_network_in_compose(compose_path: str) -> None: - """Inject the backend network into the compose file so compose up attaches it. - - Instead of running 'docker network connect' after container creation (which - is prone to race conditions and silent failures), we declare the network in - the compose file itself. Docker Compose then connects the container to the - network atomically during 'docker compose up'. - """ - import yaml - from pathlib import Path - - compose_file = Path(compose_path) - if not compose_file.exists(): - return - - content = compose_file.read_text() - compose_data = yaml.safe_load(content) - - if not compose_data or "services" not in compose_data: - return - - network_name = get_backend_network_name() - modified = False - - for svc_config in compose_data["services"].values(): - existing = svc_config.get("networks", []) - if network_name not in existing: - svc_config["networks"] = existing + [network_name] - modified = True - break # Only modify first service - - # Declare the network as external at the top level - if "networks" not in compose_data: - compose_data["networks"] = {} - if network_name not in compose_data["networks"]: - compose_data["networks"][network_name] = {"external": True} - modified = True - - if modified: - compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) - logger.info("Injected backend network '%s' into compose file", network_name) - - @router.post( "/{project_id}/repositories/{repo_id}/instances", summary="Create tool instance", @@ -807,7 +154,7 @@ async def create_instance( ) # Validate config profile if provided - selected_profile_id = await _validate_config_profile( + selected_profile_id = await validate_config_profile( session, data.config_profile_id, user_id, project_id, tool_type_id ) @@ -1298,145 +645,6 @@ async def get_instance( } -async def _prepare_manifest_instance( - session: AsyncSession, - instance: ToolInstance, - instance_dir: str, - repo_path: str, - env_vars: dict, - extra_volumes: list, - working_directory: str | None, -) -> 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, home_dir) - """ - from src.models import ToolDefinitionManifest - - tool_type = await session.get(ToolType, instance.tool_type_id) - manifest_def = await session.get(ToolDefinitionManifest, tool_type.manifest_id) - - if not manifest_def: - raise RuntimeError(f"Manifest not found for tool type {tool_type.id}") - - manifest = dict(manifest_def.manifest) - - # Resolve base if referenced - if manifest_def.base_definition_id: - base_def = await session.get( - ToolDefinitionManifest, manifest_def.base_definition_id - ) - if base_def: - base_manifest = dict(base_def.manifest) - manifest = resolve_base(deep_merge(base_manifest, manifest)) - else: - logger.warning( - "Base definition %s not found for manifest %s", - manifest_def.base_definition_id, - manifest_def.id, - ) - - manifest = merge_with_config(manifest) - - # Resolve extra env and volumes from merge_with_config - extra_env = manifest.pop("_extra_env", {}) - extra_cfg_volumes = manifest.pop("_extra_volumes", []) - env_vars.update(extra_env) - extra_volumes.extend(extra_cfg_volumes) - - # Compute image tag - image_tag = compute_image_tag(tool_type.name, manifest) - - # Check if image already exists - check = subprocess.run( - ["docker", "images", "-q", image_tag], - capture_output=True, - text=True, - ) - image_exists = check.returncode == 0 and check.stdout.strip() - - if not image_exists: - # Compile and build - dockerfile = compile_dockerfile(manifest) - entrypoint = compile_entrypoint(manifest) - - logger.debug( - "Compiled Dockerfile for instance %s (%d chars)", - instance.id, - len(dockerfile), - ) - - build_ctx = { - "Dockerfile": dockerfile, - ".headquarter/entrypoint.sh": entrypoint, - } - - returncode, stdout, stderr = await asyncio.to_thread( - build_image, - instance_dir=instance_dir, - dockerfile=dockerfile, - tag=image_tag, - build_context=build_ctx, - ) - - if returncode != 0: - raise RuntimeError(f"Docker build failed: {stderr}") - - logger.info("Built image %s for instance %s", image_tag, instance.id) - else: - logger.info("Reusing existing image %s for instance %s", image_tag, instance.id) - - # Prepare SSH path for mount resolution - ssh_path = "" - if instance.clone_mode == "clone": - ssh_path = os.path.join(instance_dir, ".ssh") - - # Resolve git mount variables from config profile - git_mount_vars = {} - if instance.selected_config_profile_id: - resolved_profile = await resolve_profile( - session, instance.selected_config_profile_id - ) - for gm in resolved_profile.git_mounts or []: - ref = gm.get("git_mount_ref", "default") - # The actual resolution happens in _resolve_git_mounts; we store placeholder - git_mount_vars[f"GIT_MOUNT_{ref}"] = "" - - variables = { - "IMAGE_TAG": image_tag, - "INSTANCE_NAME": instance.name.lower(), - "INSTANCE_DIR": instance_dir, - "REPO_PATH": repo_path, - "SSH_PATH": ssh_path, - "TOOL_PORT": instance.port or 0, - "EXTRA_ENV": env_vars, - "EXTRA_VOLUMES": extra_volumes, - **git_mount_vars, - } - - compose_content = compile_compose(manifest, variables) - - logger.debug( - "_prepare_manifest_instance for %s: repo_path=%s compose_volumes=%s", - instance.id, - repo_path or "", - manifest.get("mounts", []), - ) - logger.debug( - "Generated compose for %s:\n%s", - instance.id, - compose_content, - ) - - # Cache - instance.image_tag = image_tag - instance.manifest_compiled_at = datetime.now() - - home_dir = get_manifest_home_dir(manifest) - return image_tag, compose_content, manifest, home_dir - - @router.post( "/{project_id}/repositories/{repo_id}/instances/{instance_id}/start", summary="Start instance", @@ -1474,7 +682,7 @@ async def start_instance( # Validate and store config profile selection if data and data.config_profile_id is not None: - selected_profile_id = await _validate_config_profile( + selected_profile_id = await validate_config_profile( session, data.config_profile_id, user_id, project_id, instance.tool_type_id ) instance.selected_config_profile_id = selected_profile_id @@ -1552,7 +760,7 @@ async def start_instance( # Profile mounts are added to extra volumes extra_volumes.extend(profile_mounts) # Git repository mounts are resolved and added - git_mount_volumes = await _resolve_git_mounts( + git_mount_volumes = await resolve_git_mounts( session, resolved, instance_dir, working_directory, home_dir ) extra_volumes.extend(git_mount_volumes) @@ -1712,7 +920,7 @@ async def start_instance( compose_content, resolved_manifest, _home_dir, - ) = await _prepare_manifest_instance( + ) = await prepare_manifest_instance( session=session, instance=instance, instance_dir=instance_dir, @@ -1766,7 +974,7 @@ async def start_instance( # Modify compose file if needed (port override, start command, working dir, volumes) if port_override or start_command or working_directory or extra_volumes: - _modify_compose_file( + modify_compose_file( instance.compose_path, port_override, start_command, @@ -1777,17 +985,17 @@ async def start_instance( logger.debug("Modified compose file for instance %s", instance.id) # Sanitize compose file to remove invalid port mappings from old instances - _sanitize_compose_file(instance.compose_path) + sanitize_compose_file(instance.compose_path) # Auto-fix bind address for known web tools that default to localhost if tool_type and tool_type.interface_type == "web": - _ensure_web_bind_address( + ensure_web_bind_address( instance.compose_path, tool_type.name, tool_type.default_port ) # Ensure predictable container name for tunnel connectivity - _ensure_container_name_in_compose(instance.compose_path, instance.name) - _ensure_backend_network_in_compose(instance.compose_path) + ensure_container_name_in_compose(instance.compose_path, instance.name) + ensure_backend_network_in_compose(instance.compose_path) # Execute docker compose up with env file logger.debug( @@ -2251,14 +1459,14 @@ async def restart_instance( ) # Re-apply compose fixes in case they were updated since last start - _sanitize_compose_file(instance.compose_path) + sanitize_compose_file(instance.compose_path) tool_type = await session.get(ToolType, instance.tool_type_id) if tool_type and tool_type.interface_type == "web": - _ensure_web_bind_address( + ensure_web_bind_address( instance.compose_path, tool_type.name, tool_type.default_port ) - _ensure_container_name_in_compose(instance.compose_path, instance.name) - _ensure_backend_network_in_compose(instance.compose_path) + ensure_container_name_in_compose(instance.compose_path, instance.name) + ensure_backend_network_in_compose(instance.compose_path) returncode, stdout, stderr = execute_compose_command( instance.compose_path, "restart" diff --git a/apps/api/src/services/tool/instance_service.py b/apps/api/src/services/tool/instance_service.py new file mode 100644 index 0000000..c5f6f20 --- /dev/null +++ b/apps/api/src/services/tool/instance_service.py @@ -0,0 +1,882 @@ +"""Tool instance service functions.""" + +import asyncio +import glob as glob_module +import logging +import os +import subprocess +import uuid +from datetime import datetime + +import httpx +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.models import ConfigProfile, GitRepository, SSHKey, ToolInstance, ToolType +from src.services.git.clone import check_dirty_state, clone_repository +from src.services.config.config_profile_resolver import ( + ConfigProfileCycleError, + ResolvedProfile, + apply_resolved_profile, + expand_container_path, + resolve_profile, +) +from src.services.docker import ( + connect_container_to_network, + ensure_instance_directory, + execute_compose_command, + find_free_port, + get_backend_network_name, + get_container_id, + get_container_ip_on_network, + get_container_logs, + get_container_status, + is_container_on_network, + render_compose_template, + sort_volumes_by_specificity, + wait_for_container_running, + write_compose_file, + write_config_files, + write_env_file, +) +from src.services.shared.tunnel import ( + check_tunnel_health, + recreate_tunnel, + start_tunnel, + stop_tunnel, +) +from src.services.build.docker_build import build_image +from src.services.build.manifest_compiler import ( + compile_compose, + compile_dockerfile, + compile_entrypoint, + compute_image_tag, + deep_merge, + get_manifest_home_dir, + merge_with_config, + resolve_base, +) +from src.services.shared.permission_fixer import ( + apply_mount_permissions, + apply_ssh_permissions, +) +from src.services.shared.readiness_probe import execute_probe +from src.services.shared.ssh_keys import cleanup_ssh_key_files, prepare_ssh_key_files +from src.services.instance.event_bus import InstanceEventBus +from src.services.instance.lifecycle_hooks import publish_lifecycle_event +from src.auth.dependencies import _get_owned_project, _get_user + +logger = logging.getLogger(__name__) +_event_bus = InstanceEventBus() + +async def resolve_git_mounts( + session: AsyncSession, + 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. + + Looks up repository paths, auto-clones if needed, handles branch checkout, + expands glob patterns, and prepares bind mount entries. + Logs warnings for missing repos or invalid paths (non-blocking). + """ + if not resolved.git_mounts: + return [] + + # Process all git mounts concurrently + tasks = [] + for git_mount in resolved.git_mounts: + tasks.append( + resolve_single_git_mount( + session, git_mount, instance_dir, working_directory, home_dir + ) + ) + + results = await asyncio.gather(*tasks, return_exceptions=True) + + volume_mounts = [] + for result in results: + if isinstance(result, Exception): + logger.warning("Git mount failed: %s", result) + continue + if result: + volume_mounts.extend(result) + + return volume_mounts + + +def normalize_git_mount(entry: dict) -> dict: + """Normalize a git mount entry to the unified mappings format. + + Converts legacy source_path + target_path into a single-entry mappings array. + """ + entry = dict(entry) + if "mappings" not in entry or not entry.get("mappings"): + source = entry.get("source_path", ".") + target = entry.get("target_path") + if target is not None: + entry["mappings"] = [{"source_path": source, "target_path": target}] + entry.pop("source_path", None) + entry.pop("target_path", None) + return entry + + +def clone_git_repo( + remote_url: str, + branch: str | None, + clone_parent: str, +) -> str: + """Clone or pull a git repository. + + Returns the path to the cloned repo (repo-clone directory). + """ + import hashlib + + url_hash = hashlib.md5(remote_url.encode()).hexdigest()[:12] + repo_name = remote_url.split("/")[-1].replace(".git", "") or "repo" + clone_dir = os.path.join(clone_parent, "git-mounts", f"{repo_name}-{url_hash}") + repo_path = os.path.join(clone_dir, "repo-clone") + + if not os.path.exists(repo_path): + try: + os.makedirs(clone_dir, exist_ok=True) + repo_path = clone_repository( + remote_url, + None, # No SSH key for now - can be added later + clone_dir, + branch or "main", + ) + logger.debug("Cloned git mount repository %s to %s", remote_url, repo_path) + except Exception as exc: + logger.warning("Clone failed for git mount %s: %s", remote_url, exc) + raise + else: + # Repo exists - pull latest updates + try: + pull_repository_updates(repo_path, remote_url) + logger.debug("Pulled updates for git mount %s", remote_url) + except Exception as exc: + logger.warning("Failed to pull updates for %s: %s", remote_url, exc) + + # Handle branch checkout if specified + if branch and repo_path: + success = checkout_branch(repo_path, branch) + if success: + logger.debug("Checked out branch %s for %s", branch, remote_url) + else: + logger.warning( + "Branch %s not found in %s, using current branch", branch, remote_url + ) + + return repo_path + + +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. + + Returns a flat list of volume mount dicts. + """ + volume_mounts = [] + + for mapping in mappings: + source_path = mapping.get("source_path", ".") + target_path = mapping.get("target_path") + + if not target_path: + 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("/"): + if not working_directory: + logger.warning( + "Git mount skipped: target_path '%s' is relative but no working_directory is configured. " + "Set working_directory in the tool config or use an absolute path.", + target_path, + ) + continue + final_target = os.path.join(working_directory, target_path) + + # Build source path and expand globs + if source_path and source_path != ".": + source_full = os.path.join(repo_path, source_path) + else: + source_full = repo_path + + # Expand glob patterns + matched_paths = expand_glob_source(source_full, repo_path) + + if not matched_paths: + logger.warning( + "Git mount skipped: no files matched source path %s in repo", + source_path, + ) + continue + + for matched_path in matched_paths: + if not os.path.exists(matched_path): + continue + + # Determine target path for this match + if len(matched_paths) == 1: + # Single match: mount directly to target_path + mount_target = final_target + else: + # Multiple matches: append relative path to target + rel_path = os.path.relpath(matched_path, repo_path) + mount_target = os.path.join(final_target, rel_path) + + volume_mounts.append( + { + "source": matched_path, + "target": mount_target, + "type": "bind", + } + ) + logger.debug( + "Added git mount: %s -> %s", + matched_path, + mount_target, + ) + + return volume_mounts + + +async def resolve_single_git_mount( + session: AsyncSession, + 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. + + Clones directly from remote_url, no database lookup needed. + Returns a list of volume mounts (one for each matched file/directory). + """ + git_mount = normalize_git_mount(git_mount) + remote_url = git_mount.get("remote_url") + branch = git_mount.get("branch") + mappings = git_mount.get("mappings", []) + + if not remote_url: + logger.warning("Invalid git mount skipped: missing remote_url") + return [] + + if not mappings: + logger.warning("Invalid git mount skipped: no mappings") + return [] + + if not instance_dir: + logger.warning("Git mount skipped: no instance_dir provided for cloning") + return [] + + # Clone or pull the repository + try: + repo_path = await asyncio.to_thread( + clone_git_repo, remote_url, branch, instance_dir + ) + except Exception: + return [] + + # Resolve all mappings from the cloned repo + return resolve_git_mount_mappings(repo_path, mappings, working_directory, home_dir) + + +def checkout_branch(repo_path: str, branch: str) -> bool: + """Checkout a specific branch in a git repository. + + Returns True if checkout succeeded, False if it failed. + On failure, the repository remains on its current branch. + """ + import subprocess + + # First try to checkout existing branch + result = subprocess.run( + ["git", "-C", repo_path, "checkout", branch], + capture_output=True, + text=True, + ) + + if result.returncode != 0: + # Try fetching and checking out + subprocess.run( + ["git", "-C", repo_path, "fetch", "origin", branch], + capture_output=True, + text=True, + ) + result = subprocess.run( + ["git", "-C", repo_path, "checkout", "-b", branch, f"origin/{branch}"], + capture_output=True, + text=True, + ) + + if result.returncode != 0: + logger.warning( + "Failed to checkout branch %s in %s: %s", + branch, + repo_path, + result.stderr.strip(), + ) + return False + + return True + + +def pull_repository_updates(repo_path: str, remote_url: str) -> None: + """Pull latest updates from remote repository. + + Used when starting a new container with an existing cloned repository + to ensure the latest code is mounted. + """ + import subprocess + + # Fetch latest changes + result = subprocess.run( + ["git", "-C", repo_path, "fetch", "origin"], + capture_output=True, + text=True, + ) + + if result.returncode != 0: + raise RuntimeError(f"Failed to fetch updates: {result.stderr}") + + # Pull changes for current branch + result = subprocess.run( + ["git", "-C", repo_path, "pull", "origin"], + capture_output=True, + text=True, + ) + + if result.returncode != 0: + raise RuntimeError(f"Failed to pull updates: {result.stderr}") + + +def expand_glob_source(source_path: str, repo_path: str) -> list[str]: + """Expand glob patterns in source path. + + Returns a list of matched absolute paths. + Limits results to prevent abuse. + """ + MAX_GLOB_MATCHES = 100 + + # Check if path contains glob characters + if not any(c in source_path for c in "*?["): + # No glob pattern: return single path if it exists + return [source_path] if os.path.exists(source_path) else [] + + # Expand glob pattern + matched = glob_module.glob(source_path, recursive=True) + total_matched = len(matched) + + # Filter to only paths within the repo and limit count + results = [] + for path in matched: + abs_path = os.path.abspath(path) + if abs_path.startswith(os.path.abspath(repo_path)): + results.append(abs_path) + if len(results) >= MAX_GLOB_MATCHES: + logger.warning( + "Glob pattern matched %d files, limited to %d", + total_matched, + MAX_GLOB_MATCHES, + ) + break + + return results + + +router = APIRouter(prefix="/projects", tags=["tool-instances"]) + + +async def validate_config_profile( + session: AsyncSession, + profile_id: str | None, + user_id: uuid.UUID, + project_id: uuid.UUID, + tool_type_id: uuid.UUID, +) -> uuid.UUID | None: + """Validate a config profile selection. + + Args: + session: Database session. + profile_id: Profile ID string or None. + user_id: Authenticated user ID. + project_id: Project ID for compatibility check. + tool_type_id: Tool type ID for compatibility check. + + Returns: + Validated UUID or None. + + Raises: + HTTPException: If profile is not found, not owned, or incompatible. + """ + if not profile_id: + return None + + try: + profile_uuid = uuid.UUID(profile_id) + except ValueError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid config profile ID: {profile_id}", + ) + + profile = await session.get(ConfigProfile, profile_uuid) + if profile is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Config profile not found: {profile_id}", + ) + + if profile.user_id != user_id: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Not authorized to use this config profile", + ) + + # Check compatibility: profile must be portable or match project/tool + is_compatible = ( + (profile.project_id is None and profile.tool_type_id is None) + or (profile.project_id == project_id) + or (profile.tool_type_id == tool_type_id) + or (profile.project_id == project_id and profile.tool_type_id == tool_type_id) + ) + + if not is_compatible: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Selected config profile is not compatible with this project and tool type", + ) + + return profile_uuid + + +def sanitize_compose_file(compose_path: str) -> None: + """Remove invalid port mappings (target port 0) from compose file.""" + import yaml + from pathlib import Path + + compose_file = Path(compose_path) + if not compose_file.exists(): + return + + content = compose_file.read_text() + compose_data = yaml.safe_load(content) + + if not compose_data or "services" not in compose_data: + return + + modified = False + for service_name, service_config in compose_data["services"].items(): + if "ports" in service_config: + valid_ports = [] + for port_mapping in service_config["ports"]: + if isinstance(port_mapping, str) and ":" in port_mapping: + parts = port_mapping.split(":") + if len(parts) == 2: + host_port, container_port = parts + # Skip invalid mappings (target port 0 or empty) + if container_port == "0" or not container_port: + modified = True + continue + valid_ports.append(port_mapping) + + if valid_ports: + service_config["ports"] = valid_ports + else: + del service_config["ports"] + modified = True + break # Only check first service + + if modified: + compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) + + +def modify_compose_file( + compose_path: str, + port_override: int | None = None, + 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 + from pathlib import Path + + compose_file = Path(compose_path) + content = compose_file.read_text() + compose_data = yaml.safe_load(content) + + if not compose_data or "services" not in compose_data: + return + + # Apply modifications to the first service + for service_name, service_config in compose_data["services"].items(): + if port_override and "ports" in service_config: + # Update port mapping + for i, port_mapping in enumerate(service_config["ports"]): + if isinstance(port_mapping, str) and ":" in port_mapping: + host_port, container_port = port_mapping.split(":", 1) + service_config["ports"][i] = f"{port_override}:{container_port}" + break + + if start_command: + service_config["command"] = start_command + + if 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 = 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}") + else: + service_config["volumes"].append(f"{source}:{target}:{vol_type}") + + # Sort volumes so parent paths come before child paths + if service_config.get("volumes"): + service_config["volumes"] = sort_volumes_by_specificity( + service_config["volumes"] + ) + + break # Only modify the first service + + # Write back + compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) + + +def ensure_container_name_in_compose(compose_path: str, container_name: str) -> None: + """Ensure compose file has explicit container_name for predictable naming. + + Docker Compose auto-generates container names from the project directory + when container_name is absent. This breaks tunnel connectivity because + get_container_name(instance.name) cannot find the container. We inject + container_name into every service so the container has a predictable name. + """ + import yaml + from pathlib import Path + + compose_file = Path(compose_path) + if not compose_file.exists(): + return + + content = compose_file.read_text() + compose_data = yaml.safe_load(content) + + if not compose_data or "services" not in compose_data: + return + + modified = False + for svc_name, svc_config in compose_data["services"].items(): + if "container_name" not in svc_config: + svc_config["container_name"] = container_name.lower() + modified = True + + if modified: + compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) + logger.info( + "Injected container_name '%s' into compose file", + container_name.lower(), + ) + + +def ensure_web_bind_address( + compose_path: str, tool_type_name: str, default_port: int +) -> None: + """Auto-inject bind address for known web tools that default to 127.0.0.1. + + Many web tools (code-server, jupyter) bind to localhost by default, + making them inaccessible from the Docker network. This function detects + known tool images and injects the correct --bind-addr or --ip flag. + """ + import yaml + from pathlib import Path + + if default_port <= 0: + return + + KNOWN_BIND_FIXES: dict[str, str] = { + "code-server": f"--bind-addr 0.0.0.0:{default_port}", + "jupyter-notebook": f"start-notebook.sh --ip=0.0.0.0 --port={default_port} --no-browser", + } + + bind_command = KNOWN_BIND_FIXES.get(tool_type_name) + if not bind_command: + return + + compose_file = Path(compose_path) + if not compose_file.exists(): + return + + content = compose_file.read_text() + compose_data = yaml.safe_load(content) + + if not compose_data or "services" not in compose_data: + return + + for service_config in compose_data["services"].values(): + image = service_config.get("image", "") + if not image: + continue + + # LSIO images already bind to 0.0.0.0 — command override breaks s6 init + if "linuxserver" in image: + existing_command = service_config.get("command", "") + if "--bind-addr" in existing_command or "--host" in existing_command: + del service_config["command"] + compose_file.write_text( + yaml.dump(compose_data, default_flow_style=False) + ) + logger.warning( + "Removed broken command override from LSIO image: %s", + existing_command, + ) + return + return + + # Check if the image matches a known tool + is_code_server = tool_type_name == "code-server" and ( + "code-server" in image or "coder" in image + ) + is_jupyter = tool_type_name == "jupyter-notebook" and ( + "jupyter" in image or "notebook" in image + ) + if not is_code_server and not is_jupyter: + continue + + existing_command = service_config.get("command", "") + if existing_command: + # Already correct — nothing to do + if bind_command in existing_command: + return + # Fix broken or outdated bind flags + if ( + "--bind-addr" in existing_command + or "--host" in existing_command + or "--ip=" in existing_command + ): + service_config["command"] = bind_command + compose_file.write_text( + yaml.dump(compose_data, default_flow_style=False) + ) + logger.warning( + "Replaced broken bind address for %s: %s → %s", + tool_type_name, + existing_command, + bind_command, + ) + return + # Some other command override exists — don't touch it + return + + # No command yet — inject the correct bind address + service_config["command"] = bind_command + compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) + logger.info("Injected bind address for %s: %s", tool_type_name, bind_command) + return + + +def ensure_backend_network_in_compose(compose_path: str) -> None: + """Inject the backend network into the compose file so compose up attaches it. + + Instead of running 'docker network connect' after container creation (which + is prone to race conditions and silent failures), we declare the network in + the compose file itself. Docker Compose then connects the container to the + network atomically during 'docker compose up'. + """ + import yaml + from pathlib import Path + + compose_file = Path(compose_path) + if not compose_file.exists(): + return + + content = compose_file.read_text() + compose_data = yaml.safe_load(content) + + if not compose_data or "services" not in compose_data: + return + + network_name = get_backend_network_name() + modified = False + + for svc_config in compose_data["services"].values(): + existing = svc_config.get("networks", []) + if network_name not in existing: + svc_config["networks"] = existing + [network_name] + modified = True + break # Only modify first service + + # Declare the network as external at the top level + if "networks" not in compose_data: + compose_data["networks"] = {} + if network_name not in compose_data["networks"]: + compose_data["networks"][network_name] = {"external": True} + modified = True + + if modified: + compose_file.write_text(yaml.dump(compose_data, default_flow_style=False)) + logger.info("Injected backend network '%s' into compose file", network_name) + + + +async def prepare_manifest_instance( + session: AsyncSession, + instance: ToolInstance, + instance_dir: str, + repo_path: str, + env_vars: dict, + extra_volumes: list, + working_directory: str | None, +) -> 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, home_dir) + """ + from src.models import ToolDefinitionManifest + + tool_type = await session.get(ToolType, instance.tool_type_id) + manifest_def = await session.get(ToolDefinitionManifest, tool_type.manifest_id) + + if not manifest_def: + raise RuntimeError(f"Manifest not found for tool type {tool_type.id}") + + manifest = dict(manifest_def.manifest) + + # Resolve base if referenced + if manifest_def.base_definition_id: + base_def = await session.get( + ToolDefinitionManifest, manifest_def.base_definition_id + ) + if base_def: + base_manifest = dict(base_def.manifest) + manifest = resolve_base(deep_merge(base_manifest, manifest)) + else: + logger.warning( + "Base definition %s not found for manifest %s", + manifest_def.base_definition_id, + manifest_def.id, + ) + + manifest = merge_with_config(manifest) + + # Resolve extra env and volumes from merge_with_config + extra_env = manifest.pop("_extra_env", {}) + extra_cfg_volumes = manifest.pop("_extra_volumes", []) + env_vars.update(extra_env) + extra_volumes.extend(extra_cfg_volumes) + + # Compute image tag + image_tag = compute_image_tag(tool_type.name, manifest) + + # Check if image already exists + check = subprocess.run( + ["docker", "images", "-q", image_tag], + capture_output=True, + text=True, + ) + image_exists = check.returncode == 0 and check.stdout.strip() + + if not image_exists: + # Compile and build + dockerfile = compile_dockerfile(manifest) + entrypoint = compile_entrypoint(manifest) + + logger.debug( + "Compiled Dockerfile for instance %s (%d chars)", + instance.id, + len(dockerfile), + ) + + build_ctx = { + "Dockerfile": dockerfile, + ".headquarter/entrypoint.sh": entrypoint, + } + + returncode, stdout, stderr = await asyncio.to_thread( + build_image, + instance_dir=instance_dir, + dockerfile=dockerfile, + tag=image_tag, + build_context=build_ctx, + ) + + if returncode != 0: + raise RuntimeError(f"Docker build failed: {stderr}") + + logger.info("Built image %s for instance %s", image_tag, instance.id) + else: + logger.info("Reusing existing image %s for instance %s", image_tag, instance.id) + + # Prepare SSH path for mount resolution + ssh_path = "" + if instance.clone_mode == "clone": + ssh_path = os.path.join(instance_dir, ".ssh") + + # Resolve git mount variables from config profile + git_mount_vars = {} + if instance.selected_config_profile_id: + resolved_profile = await resolve_profile( + session, instance.selected_config_profile_id + ) + for gm in resolved_profile.git_mounts or []: + ref = gm.get("git_mount_ref", "default") + # The actual resolution happens in resolve_git_mounts; we store placeholder + git_mount_vars[f"GIT_MOUNT_{ref}"] = "" + + variables = { + "IMAGE_TAG": image_tag, + "INSTANCE_NAME": instance.name.lower(), + "INSTANCE_DIR": instance_dir, + "REPO_PATH": repo_path, + "SSH_PATH": ssh_path, + "TOOL_PORT": instance.port or 0, + "EXTRA_ENV": env_vars, + "EXTRA_VOLUMES": extra_volumes, + **git_mount_vars, + } + + compose_content = compile_compose(manifest, variables) + + logger.debug( + "_prepare_manifest_instance for %s: repo_path=%s compose_volumes=%s", + instance.id, + repo_path or "", + manifest.get("mounts", []), + ) + logger.debug( + "Generated compose for %s:\n%s", + instance.id, + compose_content, + ) + + # Cache + instance.image_tag = image_tag + instance.manifest_compiled_at = datetime.now() + + home_dir = get_manifest_home_dir(manifest) + return image_tag, compose_content, manifest, home_dir + +