"""Manifest compiler: transforms ToolDefinitionManifest into Dockerfile + Compose.""" import hashlib import json import shlex from copy import deepcopy from typing import Any import yaml from src.services.docker import sort_volumes_by_specificity def resolve_base(manifest: dict) -> dict: """Merge a base definition into a tool manifest. If the manifest has base_definition_id, the base manifest is loaded and merged. Tool-specific values override base values. Args: manifest: The tool manifest JSON (may reference a base) Returns: A fully resolved manifest with base values merged in. """ result = deepcopy(manifest) base_definition_id = result.pop("base_definition_id", None) result.pop("base_version", None) if base_definition_id: # This will be provided by the caller (they have the DB session) # For now, we assume the manifest has been pre-resolved # or the caller provides the base manifest separately. pass return result def deep_merge(base: dict, override: dict) -> dict: """Deep merge two manifests. Arrays are concatenated; dicts are merged. Args: base: The base manifest. override: The tool-specific overrides. Returns: Merged manifest. """ merged = deepcopy(base) for key, value in override.items(): if key == "mounts" and isinstance(value, list): # Concatenate mount arrays existing = merged.get("mounts", []) merged["mounts"] = existing + deepcopy(value) elif key == "scripts" and isinstance(value, dict): # Merge script categories if "scripts" not in merged: merged["scripts"] = {} for script_key, script_value in value.items(): existing = merged["scripts"].get(script_key, []) merged["scripts"][script_key] = existing + deepcopy(script_value) elif key == "packages" and isinstance(value, dict): # Union package arrays if "packages" not in merged: merged["packages"] = {} for pkg_key, pkg_value in value.items(): if ( pkg_key in merged["packages"] and isinstance(merged["packages"][pkg_key], list) and isinstance(pkg_value, list) ): merged["packages"][pkg_key] = merged["packages"][ pkg_key ] + deepcopy(pkg_value) else: merged["packages"][pkg_key] = deepcopy(pkg_value) elif key == "env" and isinstance(value, dict): # Dict merge: override wins on key conflict if "env" not in merged: merged["env"] = {} merged["env"].update(deepcopy(value)) elif ( isinstance(value, dict) and key in merged and isinstance(merged[key], dict) ): # Generic dict merge merged[key] = {**merged[key], **deepcopy(value)} else: # Override entirely merged[key] = deepcopy(value) return merged def compile_dockerfile(manifest: dict) -> str: """Compile a resolved manifest into a Dockerfile string. Args: manifest: Fully resolved manifest JSON. Returns: Dockerfile content. """ lines: list[str] = [] # FROM base_image = manifest.get("base_image", "ubuntu:24.04") lines.append(f"FROM {base_image}") lines.append("") # Build-time environment env = manifest.get("env", {}) for key, value in env.items(): lines.append(f"ENV {key}={shlex.quote(value)}") if env: lines.append("") # System packages (apt) apt_packages = manifest.get("packages", {}).get("apt", []) if manifest.get("user"): # Ensure sudo is available for permission-fixing startup scripts apt_packages = list(apt_packages) if "sudo" not in apt_packages: apt_packages.append("sudo") if apt_packages: lines.append("RUN apt-get update && apt-get install -y \\") for pkg in apt_packages[:-1]: lines.append(f" {pkg} \\") lines.append(f" {apt_packages[-1]} \\") lines.append(" && rm -rf /var/lib/apt/lists/*") lines.append("") # Node.js node = manifest.get("packages", {}).get("node") if node: version = node.get("version", "20") lines.append( f"RUN curl -fsSL https://deb.nodesource.com/setup_{version}.x | bash - && \\" ) lines.append(" apt-get install -y nodejs && \\") lines.append(" rm -rf /var/lib/apt/lists/*") lines.append("") # NPM global packages npm_packages = manifest.get("packages", {}).get("npm_global", []) if npm_packages: pkg_list = " ".join(shlex.quote(p) for p in npm_packages) lines.append(f"RUN npm install -g {pkg_list}") lines.append("") # Pip packages pip_packages = manifest.get("packages", {}).get("pip", []) if pip_packages: pkg_list = " ".join(shlex.quote(p) for p in pip_packages) lines.append(f"RUN pip install {pkg_list}") lines.append("") # User creation user = manifest.get("user") if user: name = user["name"] uid = user["uid"] gid = user["gid"] create_home = "-m " if user.get("create_home", True) else "" shell = user.get("shell", "/bin/bash") 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("") # Ensure home directory exists and is writable by the user lines.append( f"RUN mkdir -p {home} && chown {name}:{name} {home} && chmod 755 {home}" ) lines.append("") # Configure passwordless sudo so startup scripts can fix permissions lines.append( f'RUN echo "{name} ALL=(ALL) NOPASSWD:ALL" > /etc/sudoers.d/{name} && chmod 0440 /etc/sudoers.d/{name}' ) lines.append("") # Build scripts build_scripts = manifest.get("scripts", {}).get("build", []) for script in build_scripts: # Normalize multi-line scripts into single RUN command stripped_lines = [ line.strip() for line in script.strip().split("\n") if line.strip() ] if stripped_lines: normalized = " && ".join(stripped_lines) lines.append(f"RUN {normalized}") if build_scripts: lines.append("") # After build scripts, ensure everything in home is owned by the user if user and build_scripts: lines.append(f"RUN chown -R {name}:{name} {home}") lines.append("") # Create mount target directories mounts = manifest.get("mounts", []) if mounts: dirs = [mount["target"] for mount in mounts] dir_str = " ".join(dirs) lines.append(f"RUN mkdir -p {dir_str}") if user: lines.append(f"RUN chown -R {user['name']}:{user['name']} {dir_str}") lines.append("") # Entrypoint for startup scripts startup_scripts = manifest.get("scripts", {}).get("startup", []) if startup_scripts: lines.append( "COPY .headquarter/entrypoint.sh /usr/local/bin/headquarter-entrypoint" ) lines.append("RUN chmod +x /usr/local/bin/headquarter-entrypoint") lines.append("") # Switch to runtime user if user: lines.append(f"USER {user['name']}") lines.append(f"WORKDIR /home/{user['name']}") lines.append("") # Entrypoint and CMD runtime = manifest.get("runtime", {}) if startup_scripts: lines.append('ENTRYPOINT ["/usr/local/bin/headquarter-entrypoint"]') command = runtime.get("command", ["/bin/bash"]) cmd_json = json.dumps(command) lines.append(f"CMD {cmd_json}") return "\n".join(lines) def compile_entrypoint(manifest: dict) -> str: """Generate the startup entrypoint script from startup scripts. Args: manifest: Fully resolved manifest JSON. Returns: Shell script content. """ lines = ["#!/bin/bash", "set -e", ""] startup_scripts = manifest.get("scripts", {}).get("startup", []) for script in startup_scripts: lines.append(script) lines.append("") lines.append('exec "$@"') return "\n".join(lines) def compile_compose(manifest: dict, variables: dict[str, Any]) -> str: """Compile a resolved manifest into a Docker Compose string. Args: manifest: Fully resolved manifest JSON. variables: Resolved values: IMAGE_TAG, INSTANCE_NAME, REPO_PATH, etc. Returns: Docker Compose YAML content. """ runtime = manifest.get("runtime", {}) user = manifest.get("user") interface_type = manifest["interface_type"] service: dict[str, Any] = { "image": variables["IMAGE_TAG"], "container_name": variables["INSTANCE_NAME"], "restart": "unless-stopped", } # Terminal-specific fields if runtime.get("stdin_open", False): service["stdin_open"] = True if runtime.get("tty", False): service["tty"] = True if runtime.get("working_dir"): service["working_dir"] = runtime["working_dir"] # User override if user: service["user"] = f"{user['uid']}:{user['gid']}" # Ports for web tools default_port = manifest.get("default_port") if interface_type == "web" and default_port: service["ports"] = [f"{variables['TOOL_PORT']}:{default_port}"] # Environment env = manifest.get("env", {}) if env: service["environment"] = dict(env) # Merge extra env from config extra_env = variables.get("EXTRA_ENV", {}) if extra_env: if "environment" not in service: service["environment"] = {} service["environment"].update(extra_env) # Volumes from mount schema volumes = [] for mount in manifest.get("mounts", []): source = resolve_mount_source(mount, variables) if not source: continue target = mount["target"] readonly = ":ro" if mount.get("readonly", False) else "" volumes.append(f"{source}:{target}{readonly}") # Append extra volumes from tool config / config profile for vol in variables.get("EXTRA_VOLUMES", []): vol_str = f"{vol['source']}:{vol['target']}" if vol.get("readonly"): vol_str += ":ro" volumes.append(vol_str) if volumes: service["volumes"] = sort_volumes_by_specificity(volumes) compose = {"services": {"app": service}} result = yaml.dump(compose, default_flow_style=False) # Debug: log mount resolution so we can diagnose missing mounts import logging logger = logging.getLogger(__name__) logger.debug( "compile_compose: REPO_PATH=%s SSH_PATH=%s EXTRA_VOLUMES=%s mounts=%s volumes=%s", variables.get("REPO_PATH", ""), variables.get("SSH_PATH", ""), variables.get("EXTRA_VOLUMES", []), manifest.get("mounts", []), volumes, ) return result def resolve_mount_source(mount: dict, variables: dict[str, Any]) -> str: """Resolve a mount's source_type to an actual host path. Args: mount: Mount definition from manifest. variables: Resolved variables dict. Returns: Host path string, or empty string if unresolved. """ source_type = mount.get("source_type", "host_path") if source_type == "repo": return variables.get("REPO_PATH", "") elif source_type == "ssh_key": return variables.get("SSH_PATH", "") elif source_type == "instance": instance_dir = variables.get("INSTANCE_DIR", "") mount_name = mount.get("name", "unknown") return f"{instance_dir}/mounts/{mount_name}" elif source_type == "git_mount": ref = mount.get("git_mount_ref", "default") return variables.get(f"GIT_MOUNT_{ref}", "") elif source_type == "host_path": return mount.get("source", "") 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. Args: tool_name: Human-readable tool name. manifest: Fully resolved manifest JSON. Returns: Docker image tag string. """ # Canonicalize: sort keys, stable JSON canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":")) hash_suffix = hashlib.sha256(canonical.encode()).hexdigest()[:8] safe_name = tool_name.lower().replace(" ", "-").replace("_", "-") return f"headquarter/{safe_name}-{hash_suffix}:latest" def merge_with_config(manifest: dict, profile: dict | None = None) -> dict: """Merge ConfigProfile overrides into a manifest. Args: manifest: Base manifest from tool definition. profile: Resolved ConfigProfile (optional). Returns: Manifest with overrides applied. """ result = deepcopy(manifest) extra_env: dict[str, str] = {} extra_volumes: list[dict] = [] # Apply ConfigProfile if profile: if profile.get("environment_variables"): extra_env.update(profile["environment_variables"]) if profile.get("mounts"): extra_volumes.extend(profile["mounts"]) # Profile hints override everything hints = profile.get("hints", {}) if hints.get("start_command"): result["runtime"] = result.get("runtime", {}) result["runtime"]["command"] = hints["start_command"].split() if hints.get("working_directory"): result["runtime"] = result.get("runtime", {}) result["runtime"]["working_dir"] = hints["working_directory"] if hints.get("port_override"): result["default_port"] = hints["port_override"] # Store merged extras for the compose compiler result["_extra_env"] = extra_env result["_extra_volumes"] = extra_volumes return result