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

refactor: extract tool_instances helpers to service

Browse Source 
- Extract 700+ lines of helper functions to services/tool/instance_service.py
- Slim tool_instances.py from 2900 to 2108 lines
- Extracted functions: resolve_git_mounts, normalize_git_mount, clone_git_repo,
  resolve_git_mount_mappings, resolve_single_git_mount, checkout_branch,
  pull_repository_updates, expand_glob_source, validate_config_profile,
  sanitize_compose_file, modify_compose_file, ensure_container_name_in_compose,
  ensure_web_bind_address, ensure_backend_network_in_compose,
  prepare_manifest_instance

Quality gates: py_compile passes
This commit is contained in:
Developer
2026-06-05 20:23:39 +00:00
parent 183e910afd
commit b4aa4c5fcb
2 changed files with 912 additions and 822 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/api/tool/tool_instances.py
+30 -822
View File
@@ -83,681 +83,28 @@ from src.services.shared.permission_fixer import (
from src.services.shared.readiness_probe import execute_probe 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.shared.ssh_keys import cleanup_ssh_key_files, prepare_ssh_key_files
from src.schemas.tool import CreateInstanceRequest, StartInstanceRequest 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__) logger = logging.getLogger(__name__)
_event_bus = InstanceEventBus() _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( @router.post(
"/{project_id}/repositories/{repo_id}/instances", "/{project_id}/repositories/{repo_id}/instances",
summary="Create tool instance", summary="Create tool instance",
@@ -807,7 +154,7 @@ async def create_instance(
) )
# Validate config profile if provided # 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 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 "<empty>",
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( @router.post(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/start", "/{project_id}/repositories/{repo_id}/instances/{instance_id}/start",
summary="Start instance", summary="Start instance",
@@ -1474,7 +682,7 @@ async def start_instance(
# Validate and store config profile selection # Validate and store config profile selection
if data and data.config_profile_id is not None: 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 session, data.config_profile_id, user_id, project_id, instance.tool_type_id
) )
instance.selected_config_profile_id = selected_profile_id instance.selected_config_profile_id = selected_profile_id
@@ -1552,7 +760,7 @@ async def start_instance(
# Profile mounts are added to extra volumes # Profile mounts are added to extra volumes
extra_volumes.extend(profile_mounts) extra_volumes.extend(profile_mounts)
# Git repository mounts are resolved and added # 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 session, resolved, instance_dir, working_directory, home_dir
) )
extra_volumes.extend(git_mount_volumes) extra_volumes.extend(git_mount_volumes)
@@ -1712,7 +920,7 @@ async def start_instance(
compose_content, compose_content,
resolved_manifest, resolved_manifest,
_home_dir, _home_dir,
) = await _prepare_manifest_instance( ) = await prepare_manifest_instance(
session=session, session=session,
instance=instance, instance=instance,
instance_dir=instance_dir, instance_dir=instance_dir,
@@ -1766,7 +974,7 @@ async def start_instance(
# Modify compose file if needed (port override, start command, working dir, volumes) # Modify compose file if needed (port override, start command, working dir, volumes)
if port_override or start_command or working_directory or extra_volumes: if port_override or start_command or working_directory or extra_volumes:
_modify_compose_file( modify_compose_file(
instance.compose_path, instance.compose_path,
port_override, port_override,
start_command, start_command,
@@ -1777,17 +985,17 @@ async def start_instance(
logger.debug("Modified compose file for instance %s", instance.id) logger.debug("Modified compose file for instance %s", instance.id)
# Sanitize compose file to remove invalid port mappings from old instances # 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 # Auto-fix bind address for known web tools that default to localhost
if tool_type and tool_type.interface_type == "web": 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 instance.compose_path, tool_type.name, tool_type.default_port
) )
# Ensure predictable container name for tunnel connectivity # Ensure predictable container name for tunnel connectivity
_ensure_container_name_in_compose(instance.compose_path, instance.name) ensure_container_name_in_compose(instance.compose_path, instance.name)
_ensure_backend_network_in_compose(instance.compose_path) ensure_backend_network_in_compose(instance.compose_path)
# Execute docker compose up with env file # Execute docker compose up with env file
logger.debug( logger.debug(
@@ -2251,14 +1459,14 @@ async def restart_instance(
) )
# Re-apply compose fixes in case they were updated since last start # 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) tool_type = await session.get(ToolType, instance.tool_type_id)
if tool_type and tool_type.interface_type == "web": 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 instance.compose_path, tool_type.name, tool_type.default_port
) )
_ensure_container_name_in_compose(instance.compose_path, instance.name) ensure_container_name_in_compose(instance.compose_path, instance.name)
_ensure_backend_network_in_compose(instance.compose_path) ensure_backend_network_in_compose(instance.compose_path)
returncode, stdout, stderr = execute_compose_command( returncode, stdout, stderr = execute_compose_command(
instance.compose_path, "restart" instance.compose_path, "restart"
apps/api/src/services/tool/instance_service.py
+882
View File
@@ -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 "<empty>",
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