opencode-container/opencode-container.py

#!/usr/bin/env python3

import hashlib
import logging
import os
import subprocess
import sys
import time
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(message)s',
    stream=sys.stderr
)
logger = logging.getLogger(__name__)


class OpenCodeContainer:
    IMAGE = "opencode-container:latest"
    PROJECT_ID_LEN = 12
    START_TIMEOUT = 3.0

    def __init__(self):
        self.project_path = Path.cwd().resolve()
        self.project_id = self._project_id(self.project_path)

        # Store host user info once
        self.host_username = os.environ.get('USER', 'dev')
        self.host_uid = os.getuid()
        self.host_gid = os.getgid()

        self.container_name = f"oc-{self.project_path.name}-{self.project_id}"
        self.docker_context_dir = Path(__file__).resolve().parent

    def _get_xdg_data_home(self) -> Path:
        """Get XDG_DATA_HOME with fallback to ~/.local/share"""
        xdg_data = os.environ.get('XDG_DATA_HOME')
        if xdg_data:
            return Path(xdg_data)
        return Path.home() / '.local' / 'share'

    @property
    def container_home_path(self) -> Path:
        """Container home directory in XDG_DATA_HOME"""
        return self._get_xdg_data_home() / 'opencode-container' / 'container-home'

    # =========================
    # Public entrypoint
    # =========================

    def run(self, args: list[str]) -> None:
        # Ensure container home directory exists
        self.container_home_path.mkdir(parents=True, exist_ok=True)

        # Check if this project already has a running container
        if self.container_exists() and self.container_running():
            logger.error(f"Project '{self.project_path.name}' already has a running OpenCode container.")
            logger.error(f"Container name: {self.container_name}")
            logger.error("Wait for the current instance to finish or manually stop it with:")
            logger.error(f"  docker stop {self.container_name}")
            sys.exit(1)

        # Pre-create project directory structure to prevent root-owned directories
        try:
            relative_path = self.project_path.relative_to(Path.home())
            (self.container_home_path / relative_path).mkdir(parents=True, exist_ok=True)
        except ValueError:
            # Project is outside home directory - no action needed
            pass

        if not self.image_exists():
            self.build_image()

        if not self.container_exists():
            self.create_container()

        if not self.start_container():
            logger.warning("Recreating container due to failed start")
            self.remove_container()
            self.create_container()
            if not self.start_container():
                raise RuntimeError("Container failed to start after recreation")

        try:
            self.exec_opencode(args)
        finally:
            self.stop_container()

    # =========================
    # Image handling
    # =========================

    def image_exists(self) -> bool:
        return subprocess.run(
            ["docker", "image", "inspect", self.IMAGE],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        ).returncode == 0

    def build_image(self) -> None:
        logger.info(f"Building image '{self.IMAGE}' with user {self.host_username} ({self.host_uid}:{self.host_gid})")
        self._run([
            "docker", "build",
            "--build-arg", f"USERNAME={self.host_username}",
            "--build-arg", f"UID={self.host_uid}",
            "--build-arg", f"GID={self.host_gid}",
            "-t", self.IMAGE,
            str(self.docker_context_dir),
        ])

    # =========================
    # Container lifecycle
    # =========================

    def create_container(self) -> None:
        logger.info(f"Creating container '{self.container_name}'")

        self._run([
            "docker", "create",
            "--name", self.container_name,
            "--user", f"{self.host_uid}:{self.host_gid}",
            "--volume", f"{self.project_path}:{self.project_path}",
            "--volume", f"{self.container_home_path}:/home/{self.host_username}",
            self.IMAGE,
            "sh", "-c", "sleep infinity",
        ])

    def start_container(self) -> bool:
        subprocess.run(
            ["docker", "start", self.container_name],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        return self.wait_until_running()

    def stop_container(self) -> None:
        if self.container_running():
            self._run(["docker", "stop", "-t", "0", self.container_name])

    def remove_container(self) -> None:
        subprocess.run(
            ["docker", "rm", "-f", self.container_name],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )

    # =========================
    # Execution
    # =========================

    def exec_opencode(self, args: list[str]) -> None:
        subprocess.run([
            "docker", "exec",
            "-it",
            "-w", str(self.project_path),
            self.container_name,
            "opencode",
            *args,
        ], check=True)

    # =========================
    # State checks
    # =========================

    def container_exists(self) -> bool:
        return subprocess.run(
            ["docker", "inspect", self.container_name],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        ).returncode == 0

    def container_running(self) -> bool:
        result = subprocess.run(
            ["docker", "inspect", "-f", "{{.State.Running}}", self.container_name],
            capture_output=True,
            text=True,
        )
        return result.returncode == 0 and result.stdout.strip() == "true"

    def wait_until_running(self) -> bool:
        deadline = time.time() + self.START_TIMEOUT
        while time.time() < deadline:
            if self.container_running():
                return True
            time.sleep(0.1)
        return False

    # =========================
    # Helpers
    # =========================

    @staticmethod
    def _project_id(path: Path) -> str:
        return hashlib.sha256(str(path).encode()).hexdigest()[:12]

    @staticmethod
    def _run(cmd: list[str]) -> None:
        subprocess.run(cmd, check=True)


def main() -> None:
    OpenCodeContainer().run(sys.argv[1:])


if __name__ == "__main__":
    main()