Files
headquarter/apps/api/src/api/tool_instances.py
T
alex 0c839e8c6f fix: terminal 4004 infinite reconnect loop for pi-agent tool type
- Add stdin_open: true and tty: true to dockerfile-based compose generation.
  Without these, bash (PID 1) exits immediately, causing a container restart
  loop that makes the instance invisible to docker ps and triggers 4004.
- Treat WebSocket close codes 4001/4003/4004 as permanent errors in the
  frontend. Stop retrying and show the server reason to the user.
- Prevent visibilitychange handler from resetting retry attempts after a
  permanent error has occurred.
- Use docker ps -a in get_container_id/get_container_name to find
  stopped/exited containers for diagnostics.

Quality gates: tsc --noEmit (pass), pytest (98 passed, 4 pre-existing failures)
2026-05-28 09:33:35 +02:00

1895 lines
70 KiB
Python

"""Tool instance API endpoints."""
import logging
import os
import subprocess
import uuid
from datetime import datetime
import httpx
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
logger = logging.getLogger(__name__)
from src.auth.dependencies import _get_owned_project, _get_user, get_current_user_id, get_db_session
from src.models.git_repository import GitRepository
from src.models.project import Project
from src.models.ssh_key import SSHKey
from src.models.config_profile import ConfigProfile
from src.models.tool_config import ToolConfig
from src.models.tool_instance import ToolInstance
from src.models.tool_type import ToolType
from src.models.user import User
from src.services.docker import (
check_tunnel_health,
connect_container_to_network,
ensure_instance_directory,
execute_compose_command,
find_free_port,
get_container_id,
get_container_logs,
get_container_name,
get_container_status,
recreate_tunnel,
render_compose_template,
start_cloudflared_tunnel,
stop_cloudflared_tunnel,
wait_for_container_running,
write_compose_file,
write_config_files,
write_env_file,
)
from src.services.clone import check_dirty_state, clone_repository, remove_clone_directory
from src.services.docker_build import build_image
from src.services.config_profile_resolver import (
apply_resolved_profile,
resolve_profile,
ConfigProfileCycleError,
ResolvedProfile,
)
from src.services.readiness_probe import execute_probe
from src.services.ssh_keys import prepare_ssh_key_files, cleanup_ssh_key_files
import asyncio
import glob as glob_module
async def _resolve_git_mounts(
session: AsyncSession,
resolved: ResolvedProfile,
instance_dir: str | None = None,
working_directory: str | None = None,
) -> 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))
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
async def _resolve_single_git_mount(
session: AsyncSession,
git_mount: dict,
instance_dir: str | None = None,
working_directory: str | None = None,
) -> 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).
"""
remote_url = git_mount.get("remote_url")
source_path = git_mount.get("source_path", ".")
target_path = git_mount.get("target_path")
branch = git_mount.get("branch")
if not remote_url or not target_path:
logger.warning("Invalid git mount skipped: missing remote_url or target_path")
return []
# Resolve relative target paths against working directory
if target_path and 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
)
return []
target_path = os.path.join(working_directory, target_path)
logger.debug("Resolved relative target path to %s", target_path)
if not instance_dir:
logger.warning("Git mount skipped: no instance_dir provided for cloning")
return []
# Generate a unique directory name from the URL
import hashlib
url_hash = hashlib.md5(remote_url.encode()).hexdigest()[:12]
repo_name = remote_url.split("/")[-1].replace(".git", "") or "repo"
clone_parent = os.path.join(instance_dir, "git-mounts", f"{repo_name}-{url_hash}")
# clone_repository always creates 'repo-clone' inside the given directory
repo_path = os.path.join(clone_parent, "repo-clone")
# Clone or pull the repository
if not os.path.exists(repo_path):
try:
os.makedirs(clone_parent, exist_ok=True)
repo_path = await asyncio.to_thread(
clone_repository,
remote_url,
None, # No SSH key for now - can be added later
clone_parent,
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)
return []
else:
# Repo exists - pull latest updates
try:
await asyncio.to_thread(_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 = await asyncio.to_thread(_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
)
# 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 %s", source_path, remote_url)
return []
volume_mounts = []
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
final_target = target_path
else:
# Multiple matches: append relative path to target
rel_path = os.path.relpath(matched_path, repo_path)
final_target = os.path.join(target_path, rel_path)
volume_mounts.append({
"source": matched_path,
"target": final_target,
"type": "bind",
})
logger.debug("Added git mount: %s -> %s (url: %s)", matched_path, final_target, remote_url)
return volume_mounts
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"])
class CreateInstanceRequest(BaseModel):
"""Request body for creating a tool instance."""
model_config = {"extra": "ignore"}
tool_type_id: str = Field(description="UUID of the tool type to instantiate")
display_name: str | None = Field(default=None, description="Optional display name for the instance")
clone_mode: str = Field(default="mount", description="Repository access mode: 'mount' or 'clone'")
branch: str | None = Field(default="main", description="Branch to clone (when clone_mode='clone')")
new_branch: str | None = Field(default=None, description="Create a new local branch after cloning")
config_profile_id: str | None = Field(default=None, description="Optional config profile ID for launch")
class StartInstanceRequest(BaseModel):
"""Request body for starting a tool instance."""
model_config = {"extra": "ignore"}
config_profile_id: str | None = Field(default=None, description="Config profile ID to apply, or null for none")
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 profile_id is None:
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,
) -> 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"] = working_directory
if extra_volumes:
if "volumes" not in service_config:
service_config["volumes"] = []
for vol in extra_volumes:
source = vol.get("source", "")
target = vol.get("target", "")
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}")
break # Only modify the first service
# Write back
compose_file.write_text(yaml.dump(compose_data, default_flow_style=False))
@router.post(
"/{project_id}/repositories/{repo_id}/instances",
summary="Create tool instance",
description="Create a new tool instance for a repository.",
)
async def create_instance(
project_id: uuid.UUID,
repo_id: uuid.UUID,
data: CreateInstanceRequest,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Create a new tool instance for a repository.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
tool_type_id: UUID of the tool type to instantiate.
display_name: Optional display name for the instance.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with instance details.
"""
logger.debug(
"Creating instance: project_id=%s, repo_id=%s, tool_type_id=%s, display_name=%s",
project_id,
repo_id,
data.tool_type_id,
data.display_name,
)
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
repo = await session.get(GitRepository, repo_id)
if repo is None or repo.project_id != project_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="repository not found"
)
tool_type_id = uuid.UUID(data.tool_type_id)
tool_type = await session.get(ToolType, tool_type_id)
if tool_type is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="tool type not found"
)
# Validate config profile if provided
selected_profile_id = await _validate_config_profile(
session, data.config_profile_id, user_id, project_id, tool_type_id
)
try:
# Validate clone mode requirements
if data.clone_mode == "clone":
if not repo.remote_url:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="repository does not have a remote URL for cloning"
)
if not repo.ssh_key_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="repository must have an SSH key assigned for clone mode"
)
# Generate unique name
instance_name = f"{tool_type.name}-{repo.name}-{uuid.uuid4().hex[:8]}"
instance_display = data.display_name or f"{tool_type.display_name} - {repo.name}"
# Create instance directory
instance_dir = ensure_instance_directory(instance_name)
compose_path = os.path.join(instance_dir, "docker-compose.yml")
# Find free port
tool_port = find_free_port()
# Determine repo path based on clone mode
if data.clone_mode == "clone":
# Get SSH key for cloning
ssh_key = await session.get(SSHKey, repo.ssh_key_id)
if ssh_key is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="repository SSH key not found"
)
# Prepare SSH key for clone operation
ssh_key_path = None
try:
ssh_dir = prepare_ssh_key_files(instance_dir, ssh_key)
ssh_key_path = os.path.join(ssh_dir, "id_ed25519")
# Clone repository
clone_path = clone_repository(
remote_url=repo.remote_url,
ssh_key_path=ssh_key_path,
instance_dir=instance_dir,
branch=data.branch or "main",
)
repo_path = clone_path
except Exception as exc:
logger.exception("Failed to clone repository: %s", exc)
cleanup_ssh_key_files(instance_dir)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to clone repository: {exc}"
)
else:
repo_path = repo.path
# Verify cloned repo has files
if data.clone_mode == "clone" and repo_path:
try:
repo_contents = os.listdir(repo_path)
if not repo_contents or (len(repo_contents) == 1 and repo_contents[0] == ".git"):
logger.error("Cloned repository at %s appears empty", repo_path)
raise RuntimeError("Cloned repository is empty")
logger.debug("Verified cloned repo at %s has %d items", repo_path, len(repo_contents))
except Exception as exc:
logger.exception("Failed to verify cloned repository: %s", exc)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Cloned repository verification failed: {exc}"
)
# Create new local branch if requested
if data.clone_mode == "clone" and data.new_branch:
try:
result = subprocess.run(
["git", "-C", repo_path, "checkout", "-b", data.new_branch],
capture_output=True,
text=True,
)
if result.returncode != 0:
logger.error("Failed to create branch %s: %s", data.new_branch, result.stderr)
raise RuntimeError(f"Failed to create branch: {result.stderr}")
logger.debug("Created local branch %s in cloned repository", data.new_branch)
except Exception as exc:
logger.exception("Failed to create local branch: %s", exc)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to create local branch: {exc}"
)
# Handle based on definition type
if tool_type.definition_type == "dockerfile":
# Build image from Dockerfile
image_tag = f"headquarter/{instance_name}:latest".lower()
if tool_type.dockerfile_template:
returncode, stdout, stderr = await asyncio.to_thread(
build_image,
instance_dir=instance_dir,
dockerfile=tool_type.dockerfile_template,
tag=image_tag,
build_context=tool_type.build_context,
)
if returncode != 0:
logger.error("Failed to build image for instance %s: %s", instance_name, stderr)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to build Docker image: {stderr[:500]}",
)
logger.info("Successfully built image %s for instance %s", image_tag, instance_name)
# Generate compose for dockerfile-built image
# Only include ports if tool requires one (skip for terminal-only tools)
ports_section = f""" ports:
- "{tool_port}:{tool_type.default_port}"
""" if tool_type.default_port and tool_type.default_port > 0 else ""
compose_content = f"""version: "3.8"
services:
app:
image: {image_tag}
container_name: {instance_name.lower()}
stdin_open: true
tty: true
{ports_section} volumes:
- {repo_path}:/workspace
restart: unless-stopped
"""
write_compose_file(instance_dir, compose_content)
else:
# Render compose template
variables = {
"REPO_PATH": repo_path,
"INSTANCE_NAME": instance_name,
"INSTANCE_ID": instance_name,
"TOOL_NAME": instance_name,
"TOOL_PORT": tool_port,
"USER_ID": str(user_id),
"PROJECT_ID": str(project_id),
}
compose_content = render_compose_template(tool_type.compose_template, variables)
# Safety check: for clone mode, ensure repo is mounted in compose file
if data.clone_mode == "clone" and repo_path:
import yaml
compose_data = yaml.safe_load(compose_content)
repo_mounted = False
if compose_data and "services" in compose_data:
for svc in compose_data["services"].values():
volumes = svc.get("volumes", [])
for vol in volumes:
vol_str = str(vol)
if repo_path in vol_str:
repo_mounted = True
break
if repo_mounted:
break
if not repo_mounted:
logger.warning(
"Compose template for tool type %s does not mount repo path; adding default mount",
tool_type.name,
)
# Add default mount to first service
if compose_data and "services" in compose_data:
for svc in compose_data["services"].values():
if "volumes" not in svc:
svc["volumes"] = []
svc["volumes"].append(f"{repo_path}:/workspace")
break
compose_content = yaml.dump(compose_data, default_flow_style=False)
write_compose_file(instance_dir, compose_content)
# Create database record
instance = ToolInstance(
name=instance_name,
display_name=instance_display,
tool_type_id=tool_type_id,
repository_id=repo_id,
project_id=project_id,
owner_id=user_id,
status="pending",
compose_path=compose_path,
port=tool_port,
clone_mode=data.clone_mode,
branch=data.new_branch if data.new_branch else (data.branch if data.clone_mode == "clone" else None),
selected_config_profile_id=selected_profile_id,
)
session.add(instance)
await session.commit()
await session.refresh(instance)
return {
"id": str(instance.id),
"name": instance.name,
"display_name": instance.display_name,
"tool_type_id": str(instance.tool_type_id),
"status": instance.status,
"clone_mode": instance.clone_mode,
"branch": instance.branch,
"selected_config_profile_id": str(instance.selected_config_profile_id) if instance.selected_config_profile_id else None,
"created_at": instance.created_at.isoformat(),
}
except Exception as exc:
logger.exception("Failed to create instance: %s", exc)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to create instance: {exc}",
)
@router.get(
"/{project_id}/repositories/{repo_id}/instances",
summary="List instances",
description="List all tool instances for a repository.",
)
async def list_instances(
project_id: uuid.UUID,
repo_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""List all instances for a repository.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary containing list of instances.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
repo = await session.get(GitRepository, repo_id)
if repo is None or repo.project_id != project_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="repository not found"
)
result = await session.execute(
select(ToolInstance)
.where(ToolInstance.repository_id == repo_id)
.where(ToolInstance.owner_id == user_id)
.order_by(ToolInstance.created_at.desc())
)
instances = result.scalars().all()
instances_data = []
for i in instances:
tool_type = await session.get(ToolType, i.tool_type_id)
instances_data.append({
"id": str(i.id),
"name": i.name,
"display_name": i.display_name,
"tool_type_id": str(i.tool_type_id),
"tool_type_name": tool_type.name if tool_type else "unknown",
"tool_type_interfaces": [tool_type.interface_type] if tool_type else [],
"status": i.status,
"url": i.url,
"port": i.port,
"clone_mode": i.clone_mode,
"branch": i.branch,
"created_at": i.created_at.isoformat(),
})
return {"instances": instances_data}
@router.get(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}",
summary="Get instance",
description="Get a specific instance with real-time status from Docker.",
)
async def get_instance(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Get a specific instance with real-time status.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with instance details and current status.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Get real-time status from Docker
if instance.container_id:
docker_status = get_container_status(instance.container_id)
if docker_status == "running" and instance.status != "running":
instance.status = "running"
await session.commit()
elif docker_status == "exited" and instance.status == "running":
instance.status = "stopped"
instance.last_stopped_at = datetime.now()
await session.commit()
return {
"id": str(instance.id),
"name": instance.name,
"display_name": instance.display_name,
"tool_type_id": str(instance.tool_type_id),
"status": instance.status,
"container_id": instance.container_id,
"compose_path": instance.compose_path,
"url": instance.url,
"port": instance.port,
"clone_mode": instance.clone_mode,
"branch": instance.branch,
"selected_config_profile_id": str(instance.selected_config_profile_id) if instance.selected_config_profile_id else None,
"last_started_at": instance.last_started_at.isoformat() if instance.last_started_at else None,
"last_stopped_at": instance.last_stopped_at.isoformat() if instance.last_stopped_at else None,
"created_at": instance.created_at.isoformat(),
}
@router.post(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/start",
summary="Start instance",
description="Start a tool instance using Docker Compose.",
)
async def start_instance(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
data: StartInstanceRequest | None = None,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Start a tool instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance to start.
data: Optional start configuration including config profile selection.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with status and URL of the running instance.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Validate and store config profile selection
if data and data.config_profile_id is not None:
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
await session.commit()
if not instance.compose_path or not os.path.exists(instance.compose_path):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="compose file not found"
)
instance.status = "building"
await session.commit()
logger.info("Starting instance %s (name=%s)", instance.id, instance.name)
# Fetch tool configs for this tool type
env_vars = {}
config_files = {}
port_override = None
start_command = None
working_directory = None
extra_env_vars = {}
extra_volumes = []
config_query = select(ToolConfig).where(
ToolConfig.user_id == user_id,
ToolConfig.tool_type_id == instance.tool_type_id,
).where(
(ToolConfig.project_id == project_id) | (ToolConfig.project_id.is_(None))
)
config_result = await session.execute(config_query)
configs = config_result.scalars().all()
logger.debug("Found %d tool configs for instance %s", len(configs), instance.id)
for config in configs:
if config.config_type == "env":
env_vars[config.key] = config.value
elif config.config_type == "file" and config.file_path:
config_files[config.file_path] = config.value
# Handle new config fields
if config.port_override:
port_override = config.port_override
if config.start_command:
start_command = config.start_command
if config.working_directory:
working_directory = config.working_directory
if config.environment_variables:
extra_env_vars.update(config.environment_variables)
if config.volumes:
extra_volumes.extend(config.volumes)
# Merge extra env vars
env_vars.update(extra_env_vars)
# Apply selected config profile if any
instance_dir = os.path.dirname(instance.compose_path)
if instance.selected_config_profile_id is not None:
try:
resolved = await resolve_profile(session, instance.selected_config_profile_id)
profile_env, profile_files, profile_mounts, profile_hints = apply_resolved_profile(
instance_dir, resolved
)
# Profile env vars override tool config env vars
env_vars.update(profile_env)
# Profile files are written by apply_resolved_profile
config_files.update(profile_files)
# 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(session, resolved, instance_dir, working_directory)
extra_volumes.extend(git_mount_volumes)
# Profile runtime hints override tool config values
if profile_hints.get("start_command"):
start_command = profile_hints["start_command"]
if profile_hints.get("working_directory"):
working_directory = profile_hints["working_directory"]
if profile_hints.get("port_override"):
port_override = profile_hints["port_override"]
logger.debug(
"Applied config profile %s to instance %s (env=%d, files=%d, mounts=%d, git_mounts=%d)",
resolved.profile_name,
instance.id,
len(profile_env),
len(profile_files),
len(profile_mounts),
len(git_mount_volumes),
)
except ConfigProfileCycleError as exc:
logger.error("Cycle detected in config profile for instance %s: %s", instance.id, exc)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Config profile cycle detected: {exc}",
)
else:
logger.debug("No config profile selected for instance %s", instance.id)
# Write env file and config files
env_file_path = None
if env_vars:
env_file_path = write_env_file(instance_dir, env_vars)
logger.debug("Wrote env file for instance %s: %s", instance.id, env_file_path)
if config_files:
write_config_files(instance_dir, config_files)
logger.debug("Wrote %d config files for instance %s", len(config_files), instance.id)
# Mount SSH key for clone-mode instances
if instance.clone_mode == "clone":
repo = await session.get(GitRepository, instance.repository_id)
if repo and repo.ssh_key_id:
ssh_key = await session.get(SSHKey, repo.ssh_key_id)
if ssh_key:
try:
ssh_dir = prepare_ssh_key_files(instance_dir, ssh_key)
extra_volumes.append({
"source": ssh_dir,
"target": "/root/.ssh",
"type": "ro",
})
logger.debug("Mounted SSH key for clone-mode instance %s", instance.id)
except Exception as exc:
logger.error("Failed to prepare SSH key for instance %s: %s", instance.id, exc)
# 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(instance.compose_path, port_override, start_command, working_directory, extra_volumes)
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)
# Execute docker compose up with env file
logger.debug("Running docker compose up for instance %s (compose_path=%s)", instance.id, instance.compose_path)
returncode, stdout, stderr = execute_compose_command(
instance.compose_path, "up", env_file=env_file_path
)
logger.debug("Docker compose up completed for instance %s: returncode=%d, stdout=%s, stderr=%s",
instance.id, returncode, stdout[:200] if stdout else "", stderr[:500] if stderr else "")
if returncode != 0:
instance.status = "error"
await session.commit()
logger.error("Failed to start instance %s: %s", instance.id, stderr)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"failed to start instance: {stderr}",
)
# Get container ID and name
container_id = get_container_id(instance.name)
if container_id:
instance.container_id = container_id
logger.debug("Container ID for instance %s: %s", instance.id, container_id)
container_name = get_container_name(instance.name)
if container_name:
instance.container_name = container_name
logger.debug("Container name for instance %s: %s", instance.id, container_name)
# Connect container to backend network so API can reach it
logger.debug("Connecting container %s to backend network...", container_name)
connected = connect_container_to_network(container_name, "backend")
if connected:
logger.debug("Successfully connected %s to backend network", container_name)
else:
logger.warning("Failed to connect %s to backend network", container_name)
# Verify container reached running state
if instance.container_id:
instance.status = "starting"
instance.last_started_at = datetime.now()
await session.commit()
logger.debug("Instance %s: verifying container startup...", instance.id)
startup_result = wait_for_container_running(instance.container_id, timeout=30, interval=2.0)
if not startup_result["success"]:
# Container failed to start
error_msg = f"Container failed to start: status={startup_result['status']}"
if startup_result["exit_code"] is not None:
error_msg += f", exit_code={startup_result['exit_code']}"
# Get logs for debugging
logs = get_container_logs(instance.container_id, tail=50)
instance.status = "error"
await session.commit()
logger.error(
"Instance %s container startup failed after %.1fs: %s\nLogs:\n%s",
instance.id,
startup_result["waited_seconds"],
error_msg,
logs,
)
return {
"status": "error",
"error": error_msg,
"logs": logs,
}
logger.debug(
"Instance %s container started successfully after %.1fs",
instance.id,
startup_result["waited_seconds"],
)
# Execute readiness probe if configured
tool_type = await session.get(ToolType, instance.tool_type_id)
if tool_type and instance.container_id:
# Determine probe command
probe_command = None
probe_timeout = 30
probe_interval = 2
if tool_type.readiness_probe:
probe_config = tool_type.readiness_probe
probe_command = probe_config.get("command", "")
probe_timeout = probe_config.get("timeout", 30)
probe_interval = probe_config.get("interval", 2)
elif tool_type.interface_type == "web":
# Default probe for web tools
probe_command = f"curl -f http://localhost:{tool_type.default_port or 8080}"
probe_timeout = 30
probe_interval = 2
if probe_command:
instance.status = "probing"
await session.commit()
logger.debug(
"Executing readiness probe for instance %s: command='%s', timeout=%d, interval=%d",
instance.id, probe_command, probe_timeout, probe_interval
)
success, probe_logs = await execute_probe(
container_id=instance.container_id,
command=probe_command,
timeout=probe_timeout,
interval=probe_interval,
)
# Store probe result
instance.probe_result = {
"success": success,
"command": probe_command,
"logs": probe_logs,
"timestamp": datetime.now().isoformat(),
}
if not success:
instance.status = "unhealthy"
await session.commit()
logger.error(
"Readiness probe failed for instance %s after %ds: %s",
instance.id,
probe_timeout,
"\n".join(probe_logs),
)
return {
"status": "unhealthy",
"error": f"Readiness probe failed after {probe_timeout}s",
"probe_logs": probe_logs,
}
logger.info("Readiness probe succeeded for instance %s", instance.id)
instance.status = "running"
await session.commit()
logger.info("Instance %s is now running", instance.id)
# Get tool type for default port
tool_type = await session.get(ToolType, instance.tool_type_id)
if not tool_type:
logger.error("Tool type %s not found", instance.tool_type_id)
instance.status = "error"
await session.commit()
return {
"status": "error",
"error": f"Tool type '{instance.tool_type_id}' not found",
}
instance_port = tool_type.default_port or 0
logger.debug("Tool type for instance %s: name=%s, default_port=%s, interface_type=%s",
instance.id, tool_type.name, instance_port, tool_type.interface_type)
# Only create Cloudflare tunnel for web-enabled tools
if tool_type.interface_type == "web":
# Create temporary Cloudflare tunnel for public access
try:
logger.debug("Creating temporary tunnel for instance %s (container=%s, port=%d)",
instance.id, instance.container_name, instance_port)
tunnel_info = start_cloudflared_tunnel(
container_name=instance.container_name or instance.name,
port=instance_port,
)
instance.tunnel_id = tunnel_info["pid"]
instance.public_url = tunnel_info["url"]
instance.url = tunnel_info["url"]
await session.commit()
logger.debug(
"Created temporary tunnel for instance %s: pid=%s, url=%s",
instance.id,
tunnel_info["pid"],
tunnel_info["url"],
)
except Exception as exc:
import traceback
error_msg = str(exc)
error_trace = traceback.format_exc()
logger.error(
"Failed to create tunnel for instance %s: %s\nTraceback:\n%s",
instance.id,
error_msg,
error_trace,
)
instance.status = "error"
instance.url = None
await session.commit()
return {
"status": "error",
"error": f"Failed to create tunnel: {error_msg}",
}
else:
# Terminal-only tool - no tunnel needed
logger.info("Instance %s is terminal-only (no web interface), skipping tunnel creation", instance.id)
instance.url = None
instance.public_url = None
await session.commit()
return {"status": instance.status, "url": instance.url}
@router.post(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/stop",
summary="Stop instance",
description="Stop a running tool instance.",
)
async def stop_instance(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Stop a tool instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance to stop.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with the stopped status.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Stop Cloudflare tunnel if exists
if instance.tunnel_id:
try:
stop_cloudflared_tunnel(instance.tunnel_id)
logger.debug("Stopped tunnel for instance %s (pid=%s)", instance.id, instance.tunnel_id)
except Exception as exc:
logger.warning("Failed to stop tunnel for instance %s: %s", instance.id, exc)
if instance.compose_path and os.path.exists(instance.compose_path):
execute_compose_command(instance.compose_path, "stop")
instance.status = "stopped"
instance.last_stopped_at = datetime.now()
instance.url = None
instance.public_url = None
instance.tunnel_id = None
await session.commit()
return {"status": instance.status}
@router.post(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/restart",
summary="Restart instance",
description="Restart a tool instance.",
)
async def restart_instance(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Restart a tool instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance to restart.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with status and URL of the restarted instance.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Stop old tunnel if exists
if instance.tunnel_id:
try:
stop_cloudflared_tunnel(instance.tunnel_id)
logger.debug("Stopped old tunnel for instance %s (pid=%s)", instance.id, instance.tunnel_id)
except Exception as exc:
logger.warning("Failed to stop old tunnel for instance %s: %s", instance.id, exc)
# Re-apply stored config profile on restart
if instance.compose_path and os.path.exists(instance.compose_path):
instance_dir = os.path.dirname(instance.compose_path)
if instance.selected_config_profile_id is not None:
try:
resolved = await resolve_profile(session, instance.selected_config_profile_id)
profile_env, profile_files, profile_mounts, profile_hints = apply_resolved_profile(
instance_dir, resolved
)
# Write env file with resolved profile env vars
if profile_env:
write_env_file(instance_dir, profile_env)
logger.debug(
"Re-applied config profile %s on restart for instance %s",
resolved.profile_name,
instance.id,
)
except ConfigProfileCycleError as exc:
logger.error(
"Cycle detected in stored config profile for instance %s: %s",
instance.id,
exc,
)
returncode, stdout, stderr = execute_compose_command(
instance.compose_path, "restart"
)
if returncode == 0:
instance.status = "running"
instance.last_started_at = datetime.now()
# Get tool type for default port
tool_type = await session.get(ToolType, instance.tool_type_id)
if not tool_type or not tool_type.default_port:
logger.error("Tool type %s has no default_port configured. Cannot create tunnel.",
instance.tool_type_id)
instance.status = "error"
await session.commit()
return {
"status": "error",
"error": f"Tool type '{tool_type.name if tool_type else 'unknown'}' has no port configured",
}
instance_port = tool_type.default_port
# Only create tunnel for web-enabled tools
if tool_type.interface_type == "web":
# Create new temporary tunnel
try:
tunnel_info = start_cloudflared_tunnel(
container_name=instance.container_name or instance.name,
port=instance_port,
)
instance.tunnel_id = tunnel_info["pid"]
instance.public_url = tunnel_info["url"]
instance.url = tunnel_info["url"]
logger.debug(
"Created new tunnel for instance %s: %s",
instance.id,
tunnel_info["url"],
)
except Exception as exc:
logger.warning(
"Failed to create tunnel for instance %s: %s",
instance.id,
exc,
)
instance.status = "error"
instance.url = None
await session.commit()
return {
"status": "error",
"error": f"Failed to create tunnel: {exc}",
}
else:
# Terminal-only tool
instance.url = None
instance.public_url = None
await session.commit()
return {"status": instance.status, "url": instance.url}
instance.status = "error"
await session.commit()
return {"status": instance.status}
@router.delete(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}",
summary="Delete instance",
description="Delete a tool instance and remove its Docker containers and files.",
)
async def delete_instance(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
force: bool = False,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> None:
"""Delete a tool instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance to delete.
user_id: ID of the authenticated user.
session: Database session.
Returns:
None with 204 status code.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Check dirty state for clone-mode instances
if instance.clone_mode == "clone" and not force:
instance_dir = os.path.dirname(instance.compose_path) if instance.compose_path else None
if instance_dir:
clone_path = os.path.join(instance_dir, "repo-clone")
if os.path.exists(clone_path):
is_dirty, changed_files = check_dirty_state(clone_path)
if is_dirty:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={
"message": "Repository has uncommitted changes",
"changed_files": changed_files,
"force_required": True,
},
)
# Stop Cloudflare tunnel if exists
if instance.tunnel_id:
try:
stop_cloudflared_tunnel(instance.tunnel_id)
logger.debug("Stopped tunnel for instance %s (pid=%s)", instance.id, instance.tunnel_id)
except Exception as exc:
logger.warning("Failed to stop tunnel for instance %s: %s", instance.id, exc)
# Stop and remove container
if instance.compose_path and os.path.exists(instance.compose_path):
execute_compose_command(instance.compose_path, "down")
# Remove instance directory (includes clone and SSH keys)
if instance.compose_path:
instance_dir = os.path.dirname(instance.compose_path)
if os.path.exists(instance_dir):
import shutil
shutil.rmtree(instance_dir)
await session.delete(instance)
await session.commit()
@router.get(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/logs",
summary="Get instance logs",
description="Get container logs for a tool instance.",
)
async def get_instance_logs(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
tail: int = 100,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Get container logs for an instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance.
tail: Number of log lines to return (default: 100).
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary containing the container logs.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
if not instance.container_id:
return {"logs": "No container running"}
logs = get_container_logs(instance.container_id, tail)
return {"logs": logs}
@router.post(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/recreate-tunnel",
summary="Recreate tunnel",
description="Recreate the temporary Cloudflare tunnel for a running instance.",
)
async def recreate_tunnel_endpoint(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Recreate the temporary tunnel for an instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with new URL and status.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
if instance.status != "running":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="instance must be running to recreate tunnel",
)
# Validate tunnel is actually broken before recreating
if instance.url:
tunnel_health = check_tunnel_health(instance.url)
if tunnel_health["tunnel_status"] == "error_response":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Tunnel is working but application returned HTTP {tunnel_health.get('status_code')}. Recreating the tunnel will not fix this issue.",
)
elif tunnel_health["tunnel_status"] == "healthy":
return {"status": "healthy", "url": instance.url, "message": "Tunnel is already healthy"}
# Get tool type for default port
tool_type = await session.get(ToolType, instance.tool_type_id)
instance_port = tool_type.default_port if tool_type and tool_type.default_port else 8080
try:
tunnel_info = recreate_tunnel(
container_name=instance.container_name or instance.name,
port=instance_port,
old_pid=instance.tunnel_id,
)
instance.tunnel_id = tunnel_info["pid"]
instance.public_url = tunnel_info["url"]
instance.url = tunnel_info["url"]
await session.commit()
logger.debug(
"Recreated tunnel for instance %s: pid=%s, url=%s",
instance.id,
tunnel_info["pid"],
tunnel_info["url"],
)
return {"status": "healthy", "url": instance.url}
except Exception as exc:
logger.exception("Failed to recreate tunnel for instance %s", instance.id)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to recreate tunnel: {str(exc)}",
)
@router.get(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/health",
summary="Check instance health",
description="Check container and tunnel health for an instance.",
)
async def check_instance_tunnel_health(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Check health for an instance (container + tunnel).
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary with container_status, tunnel_status, probe_status, and overall healthy flag.
"""
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Check container status
container_info = {"status": "not_found", "exit_code": None, "health": None}
if instance.container_id:
container_info = get_container_status(instance.container_id)
# Build response
response = {
"healthy": False,
"container_status": container_info["status"],
"container_health": container_info["health"],
"tunnel_status": "not_applicable",
"tunnel_status_code": None,
"probe_status": "not_applicable",
"last_probe_output": None,
"error": None,
}
# Determine probe status
if instance.status == "probing":
response["probe_status"] = "pending"
elif instance.probe_result:
response["probe_status"] = "success" if instance.probe_result.get("success") else "failed"
response["last_probe_output"] = "\n".join(instance.probe_result.get("logs", []))
# Check tunnel health if instance has a URL and is web-enabled
if instance.url and instance.status in ("running", "unhealthy"):
tunnel_health = check_tunnel_health(instance.url)
response["tunnel_status"] = tunnel_health["tunnel_status"]
response["tunnel_status_code"] = tunnel_health.get("status_code")
if tunnel_health.get("error"):
response["error"] = tunnel_health["error"]
# Overall healthy: web tools need running container + healthy tunnel;
# terminal tools only need running container
container_healthy = container_info["status"] == "running"
if instance.url:
tunnel_healthy = response["tunnel_status"] == "healthy"
response["healthy"] = container_healthy and tunnel_healthy
else:
response["healthy"] = container_healthy
# If container is not running, override error message
if not container_healthy:
response["error"] = f"Container is {container_info['status']}"
if container_info["exit_code"] is not None:
response["error"] += f" (exit code: {container_info['exit_code']})"
return response
@router.get(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
)
@router.post(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
include_in_schema=False,
)
@router.put(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
include_in_schema=False,
)
@router.delete(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
include_in_schema=False,
)
@router.patch(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
include_in_schema=False,
)
@router.head(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
include_in_schema=False,
)
@router.options(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
description="Proxy HTTP requests to a running tool instance.",
include_in_schema=False,
)
async def proxy_to_instance(
request: Request,
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
path: str = "",
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> Response:
"""Proxy requests to a running tool instance.
Args:
request: The incoming HTTP request.
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance.
path: The path to proxy to the instance.
user_id: ID of the authenticated user.
session: Database session.
Returns:
Response from the proxied instance.
"""
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
# Verify ownership
if instance.owner_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="not authorized to access this instance",
)
if instance.status != "running" or not instance.container_name:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="instance is not running",
)
# Build target URL
target_url = f"http://{instance.container_name}:{instance.port}"
if path:
target_url += f"/{path}"
# Get query string
query_string = str(request.query_params)
if query_string:
target_url += f"?{query_string}"
# Forward headers (excluding host)
headers = dict(request.headers)
headers.pop("host", None)
headers.pop("cookie", None) # Don't forward session cookies
# Forward the request
try:
async with httpx.AsyncClient() as client:
body = await request.body()
response = await client.request(
method=request.method,
url=target_url,
headers=headers,
content=body,
follow_redirects=False,
timeout=30.0,
)
except Exception as exc:
logger.error("Proxy error: %s", exc)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"failed to reach instance: {exc}",
)
# Build response
response_headers = dict(response.headers)
# Remove hop-by-hop headers
for header in ["content-encoding", "transfer-encoding", "connection"]:
response_headers.pop(header, None)
return Response(
content=response.content,
status_code=response.status_code,
headers=response_headers,
)
from fastapi import APIRouter as FastAPIRouter
sessions_router = FastAPIRouter(prefix="/users", tags=["sessions"])
@sessions_router.get(
"/me/sessions",
summary="Get user sessions",
description="Get all active sessions (running instances) for the current user.",
)
async def get_user_sessions(
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Get all active sessions for the current user.
Args:
user_id: ID of the authenticated user.
session: Database session.
Returns:
Dictionary containing list of active sessions with instance details.
"""
_user = await _get_user(session, user_id)
result = await session.execute(
select(ToolInstance)
.where(ToolInstance.owner_id == user_id)
.where(ToolInstance.status.in_(["running", "building", "pending", "stopped", "error"]))
.order_by(ToolInstance.created_at.desc())
)
instances = result.scalars().all()
sessions = []
for instance in instances:
tool_type = await session.get(ToolType, instance.tool_type_id)
repo = await session.get(GitRepository, instance.repository_id)
project = await session.get(Project, instance.project_id)
sessions.append({
"id": str(instance.id),
"display_name": instance.display_name,
"tool_type_name": tool_type.name if tool_type else "unknown",
"tool_icon": tool_type.name if tool_type else "code",
"tool_type_interfaces": [tool_type.interface_type] if tool_type else [],
"repository_name": repo.name if repo else "unknown",
"repository_id": str(instance.repository_id),
"project_name": project.name if project else "unknown",
"project_id": str(instance.project_id),
"status": instance.status,
"url": instance.url,
"clone_mode": instance.clone_mode,
"branch": instance.branch,
"selected_config_profile_id": str(instance.selected_config_profile_id) if instance.selected_config_profile_id else None,
"created_at": instance.created_at.isoformat() if instance.created_at else None,
})
return {"sessions": sessions}