launch

tit.launch

Launch the TI-Toolbox container and open its browser UI.

Uses the same compose specification and container labels as Electron. Installed wheels use BUILTIN_SPEC; tests verify it matches docker-compose.yml.

LaunchError

Bases: RuntimeError

A failure the user can act on; the message is the remedy, not a stack trace.

StackSpec dataclass

StackSpec(image: str, working_dir: str, init: bool, volumes: tuple[str, ...], environment: tuple[tuple[str, str], ...], ports: tuple[str, ...], healthcheck: tuple[str, ...] = ())

The parts of the compose service this launcher realises.

Values still carry ${VAR} / ${VAR:-default} references; they are interpolated against :func:build_env when the argv is built, exactly as shared/composeFile.ts does for the app.

LaunchOptions dataclass

LaunchOptions(project: str, port: int = DEFAULT_PORT, image: str | None = None, timeout: float = 180.0, echo: object = print, repo_dir: str = '', server_reload: bool = False, static_dir: str = '', existing: str | None = None, container: str | None = None, session_project: str = '')

One launch invocation.

The last three are the dev overrides — the three variables the compose file exposes as ${TIT_REPO_DIR:-}, ${TIT_SERVER_RELOAD:-} and ${TIT_STATIC_DIR:-}, and the only difference between a user run and a developer run (see --dev). They default to off, which is the only correct answer for a user: repo_dir bind-mounts a host directory over /ti-toolbox, which is where the image installed its own tit from, so a non-empty value in a packaged or pip-installed run replaces the toolbox with whatever is at that path.

compose_path

compose_path() -> Path | None

Resolve an explicit compose file, the checkout copy, or the built-in fallback.

Standalone loaders pass their adjacent YAML through TIT_COMPOSE_FILE. An invalid explicit path fails instead of silently starting a different spec.

Source code in tit/launch.py

def compose_path() -> Path | None:
    """Resolve an explicit compose file, the checkout copy, or the built-in fallback.

    Standalone loaders pass their adjacent YAML through ``TIT_COMPOSE_FILE``.
    An invalid explicit path fails instead of silently starting a different spec.
    """
    override = os.environ.get("TIT_COMPOSE_FILE")
    if override:
        candidate = Path(override).expanduser().resolve()
        if not candidate.is_file():
            raise LaunchError(
                f"Compose file does not exist or is not a file: {candidate}"
            )
        return candidate
    candidate = Path(__file__).resolve().parent.parent / "docker-compose.yml"
    return candidate if candidate.is_file() else None

load_spec

load_spec(path: Path | None = None) -> StackSpec

Parse the compose file into a :class:StackSpec, or return :data:BUILTIN_SPEC.

path defaults to :func:compose_path. Passing a path that does not exist is an error (the caller asked for that file); omitting it and having no checkout is not.

Source code in tit/launch.py

def load_spec(path: Path | None = None) -> StackSpec:
    """Parse the compose file into a :class:`StackSpec`, or return :data:`BUILTIN_SPEC`.

    ``path`` defaults to :func:`compose_path`.  Passing a path that does not
    exist is an error (the caller asked for that file); omitting it and having
    no checkout is not.
    """
    if path is None:
        path = compose_path()
        if path is None:
            return BUILTIN_SPEC
    return parse_compose(path.read_text(encoding="utf-8"))

parse_compose

parse_compose(text: str) -> StackSpec

The services.tit block of a v3 compose file, as a :class:StackSpec.

Source code in tit/launch.py

def parse_compose(text: str) -> StackSpec:
    """The ``services.tit`` block of a v3 compose file, as a :class:`StackSpec`."""
    lines = [line.rstrip() for line in text.splitlines()]
    service = _block(lines, ["services", SERVICE_NAME])
    if service is None:
        raise LaunchError(f"the compose file has no services.{SERVICE_NAME} block")

    scalars: dict[str, str] = {}
    lists: dict[str, list[str]] = {}
    maps: dict[str, list[tuple[str, str]]] = {}
    for key, body, inline in _entries(service):
        if inline:
            scalars[key] = _unquote(_strip_comment(inline))
        elif body and body[0].lstrip().startswith("- "):
            lists[key] = [
                _unquote(_strip_comment(m.group("value")))
                for m in (_ITEM.match(line) for line in body)
                if m
            ]
        else:
            maps[key] = [
                (k, _unquote(_strip_comment(v)))
                for k, _sub, v in _entries(body)
                if v is not None
            ]

    health = maps.get("healthcheck", [])
    test = next((v for k, v in health if k == "test"), "")

    return StackSpec(
        image=scalars.get("image", ""),
        working_dir=scalars.get("working_dir", "/ti-toolbox"),
        init=scalars.get("init", "false").lower() == "true",
        volumes=tuple(lists.get("volumes", ())),
        environment=tuple(
            (k, v) for k, v in maps.get("environment", []) if v is not None
        ),
        ports=tuple(lists.get("ports", ())),
        healthcheck=_parse_flow_list(test),
    )

interpolate

interpolate(template: str, env: dict[str, str]) -> str

${VAR} / ${VAR:-default}, from env only (never os.environ).

Reading the ambient environment here is exactly the hazard shared/compose.ts documents for TIT_REPO_DIR: a stray variable in the user's shell must not bind-mount a host directory over the image's own /ti-toolbox.

Source code in tit/launch.py

def interpolate(template: str, env: dict[str, str]) -> str:
    """``${VAR}`` / ``${VAR:-default}``, from ``env`` only (never ``os.environ``).

    Reading the ambient environment here is exactly the hazard
    ``shared/compose.ts`` documents for ``TIT_REPO_DIR``: a stray variable in
    the user's shell must not bind-mount a host directory over the image's own
    ``/ti-toolbox``.
    """

    def replace(match: re.Match[str]) -> str:
        value = env.get(match.group("name"))
        if value is None or value == "":
            return match.group("default") or ""
        return value

    return _VAR.sub(replace, template)

hash8

hash8(text: str) -> str

The Electron app's 32-bit project hash (shared/compose.ts#hash8), reproduced.

Not a security boundary — it exists so both launchers derive the same container name and tit.project label for the same directory, which is what lets tit launch --status see a container the app started and the app attach to one this launcher started.

It runs over UTF-16 code units, as JavaScript's charCodeAt does: an emoji is two units, not one code point (tests/fixtures/project_hash.json).

Source code in tit/launch.py

def hash8(text: str) -> str:
    """The Electron app's 32-bit project hash (``shared/compose.ts#hash8``), reproduced.

    Not a security boundary — it exists so both launchers derive the *same*
    container name and ``tit.project`` label for the same directory, which is
    what lets ``tit launch --status`` see a container the app started and the
    app attach to one this launcher started.

    It runs over UTF-16 code units, as JavaScript's ``charCodeAt`` does: an emoji
    is two units, not one code point (``tests/fixtures/project_hash.json``).
    """
    mask = 0xFFFFFFFF

    def imul(a: int, b: int) -> int:
        product = ((a & mask) * (b & mask)) & mask
        return product - 0x100000000 if product & 0x80000000 else product

    data = text.encode("utf-16-be", "surrogatepass")
    units = [int.from_bytes(data[i : i + 2], "big") for i in range(0, len(data), 2)]
    h1 = 0xDEADBEEF ^ len(units)
    h2 = 0x41C6CE57 ^ len(units)
    for code in units:
        h1 = imul(h1 ^ code, 2654435761)
        h2 = imul(h2 ^ code, 1597334677)
    h1 = imul(h1 ^ ((h1 & mask) >> 16), 2246822507) ^ imul(
        h2 ^ ((h2 & mask) >> 13), 3266489909
    )
    return format(h1 & mask, "08x")

project_name

project_name(host_project_dir: str) -> str

ti-toolbox-<hash8> — the compose project name and tit.project label.

Source code in tit/launch.py

def project_name(host_project_dir: str) -> str:
    """``ti-toolbox-<hash8>`` — the compose project name and ``tit.project`` label."""
    return f"ti-toolbox-{hash8(host_project_dir)}"

container_name

container_name(host_project_dir: str) -> str

What docker compose itself would have named the service's container.

Source code in tit/launch.py

def container_name(host_project_dir: str) -> str:
    """What ``docker compose`` itself would have named the service's container."""
    return f"{project_name(host_project_dir)}-{SERVICE_NAME}-1"

default_image

default_image() -> str

Use the compose image default, or an explicit TIT_IMAGE_TAG override.

Runtime package versions can be prereleases with no matching image. The compose spec (and its wheel fallback) owns the image independently of that version.

Source code in tit/launch.py

def default_image() -> str:
    """Use the compose image default, or an explicit ``TIT_IMAGE_TAG`` override.

    Runtime package versions can be prereleases with no matching image. The compose
    spec (and its wheel fallback) owns the image independently of that version.
    """
    tag = os.environ.get("TIT_IMAGE_TAG", "").strip()
    if tag:
        return normalize_image(tag)
    return interpolate(load_spec().image, {})

normalize_image

normalize_image(reference: str) -> str

A bare tag (v3.0.0: no / and no :) names a release of the toolbox image; anything else is kept, a local tit:dev included. loader.sh's resolved_image applies the same rule, so every launcher hands Docker and the desktop app the same reference, never library/v3.0.0.

Source code in tit/launch.py

def normalize_image(reference: str) -> str:
    """A bare tag (``v3.0.0``: no ``/`` and no ``:``) names a release of the toolbox image;
    anything else is kept, a local ``tit:dev`` included. ``loader.sh``'s ``resolved_image``
    applies the same rule, so every launcher hands Docker and the desktop app the same
    reference, never ``library/v3.0.0``.
    """
    if "/" in reference or ":" in reference:
        return reference
    return f"{IMAGE_REPO}:{reference}"

parse_port

parse_port(value: object) -> int

A TCP port written in decimal digits, 1-65535; 08765 is 8765 (loader.sh's rule).

Source code in tit/launch.py

def parse_port(value: object) -> int:
    """A TCP port written in decimal digits, 1-65535; ``08765`` is 8765 (``loader.sh``'s rule)."""
    text = str(value)
    if not _PORT.fullmatch(text) or not 1 <= int(text) <= 65535:
        raise LaunchError("port must be 1–65535")
    return int(text)

parse_timeout

parse_timeout(value: object) -> float

Positive seconds: text written as N or N.N (loader.sh's rule), or a number.

A number is one already read (argparse, :class:LaunchOptions); it is not turned back into text, where str(1e-05) would fail the N.N rule loader.sh's text passed.

Source code in tit/launch.py

def parse_timeout(value: object) -> float:
    """Positive seconds: text written as ``N`` or ``N.N`` (``loader.sh``'s rule), or a number.

    A number is one already read (argparse, :class:`LaunchOptions`); it is not turned back
    into text, where ``str(1e-05)`` would fail the ``N.N`` rule loader.sh's text passed.
    """
    if isinstance(value, (int, float)):
        seconds = float(value)
        if not math.isfinite(seconds):
            raise LaunchError("timeout must be positive seconds")
    elif _SECONDS.fullmatch(str(value)):
        seconds = float(str(value))
    else:
        raise LaunchError("timeout must be positive seconds")
    if seconds <= 0:
        raise LaunchError("timeout must be positive")
    return seconds

user_config_path

user_config_path(*, host_platform: str | None = None, env: Mapping[str, str] | None = None, home: str | None = None) -> str

The host directory mounted at /root/.config/ti-toolbox (telemetry consent, prefs).

The documented table (docs/wiki/telemetry.md, PathManager.user_config_dir): macOS <home>/.config/ti-toolbox always, since a Dock launch never sees a shell's XDG_CONFIG_HOME; Windows %APPDATA%\ti-toolbox (else <home>\AppData\Roaming); Linux and the rest $XDG_CONFIG_HOME/ti-toolbox when it is set and non-empty, else <home>/.config/ti-toolbox. The desktop's userConfigDir and loader.sh's user_config_dir apply the same rule (tests/fixtures/user_config_dirs.json), so a telemetry opt-out made through one launcher is the one every launcher mounts.

Source code in tit/launch.py

def user_config_path(
    *,
    host_platform: str | None = None,
    env: Mapping[str, str] | None = None,
    home: str | None = None,
) -> str:
    """The host directory mounted at ``/root/.config/ti-toolbox`` (telemetry consent, prefs).

    The documented table (``docs/wiki/telemetry.md``, ``PathManager.user_config_dir``):
    macOS ``<home>/.config/ti-toolbox`` always, since a Dock launch never sees a shell's
    ``XDG_CONFIG_HOME``; Windows ``%APPDATA%\\ti-toolbox`` (else ``<home>\\AppData\\Roaming``);
    Linux and the rest ``$XDG_CONFIG_HOME/ti-toolbox`` when it is set and non-empty, else
    ``<home>/.config/ti-toolbox``. The desktop's ``userConfigDir`` and ``loader.sh``'s
    ``user_config_dir`` apply the same rule (``tests/fixtures/user_config_dirs.json``), so
    a telemetry opt-out made through one launcher is the one every launcher mounts.
    """
    host_platform = sys.platform if host_platform is None else host_platform
    env = os.environ if env is None else env
    home = str(Path.home()) if home is None else home
    if host_platform == "win32":
        base = env.get("APPDATA") or str(PureWindowsPath(home, "AppData", "Roaming"))
        return str(PureWindowsPath(base, "ti-toolbox"))
    xdg = "" if host_platform == "darwin" else env.get("XDG_CONFIG_HOME")
    base = xdg or str(PurePosixPath(home, ".config"))
    return str(PurePosixPath(base, "ti-toolbox"))

user_config_dir

user_config_dir() -> Path

:func:user_config_path for this host, created if absent.

Source code in tit/launch.py

def user_config_dir() -> Path:
    """:func:`user_config_path` for this host, created if absent."""
    path = Path(user_config_path())
    path.mkdir(parents=True, exist_ok=True)
    return path

build_env

build_env(*, host_project_dir: str, port: int, token: str, user_config: str, image_tag: str, repo_dir: str = '', static_dir: str = '', server_reload: bool = False) -> dict[str, str]

The map the compose file's ${VAR} references are interpolated from.

Mirrors shared/compose.ts#buildStackEnv. TIT_REPO_DIR and TIT_SERVER_RELOAD are always present and empty by default for the reason that file gives: an omitted key would let the user's own shell supply one. TIT_NO_TELEMETRY is the opposite case: the user's own opt-out, read from the host environment so it reaches the container.

Source code in tit/launch.py

def build_env(
    *,
    host_project_dir: str,
    port: int,
    token: str,
    user_config: str,
    image_tag: str,
    repo_dir: str = "",
    static_dir: str = "",
    server_reload: bool = False,
) -> dict[str, str]:
    """The map the compose file's ``${VAR}`` references are interpolated from.

    Mirrors ``shared/compose.ts#buildStackEnv``.  ``TIT_REPO_DIR`` and
    ``TIT_SERVER_RELOAD`` are always present and empty by default for the reason
    that file gives: an omitted key would let the user's own shell supply one.
    ``TIT_NO_TELEMETRY`` is the opposite case: the user's own opt-out, read from
    the host environment so it reaches the container.
    """
    return {
        "LOCAL_PROJECT_DIR": host_project_dir,
        "PROJECT_DIR_NAME": project_dir_name(host_project_dir),
        "TIT_USER_CONFIG": user_config,
        "TIT_HOST_OS": platform.system().lower(),
        "TIT_HOST_OS_VERSION": platform.release(),
        "TIT_HOST_ARCH": platform.machine(),
        "TIT_NO_TELEMETRY": os.environ.get("TIT_NO_TELEMETRY", ""),
        "TZ": os.environ.get("TZ", "UTC"),
        "TIT_SERVER_PORT": str(port),
        "TIT_SERVER_TOKEN": token,
        "TIT_IMAGE_TAG": image_tag,
        "TIT_REPO_DIR": repo_dir,
        "TIT_STATIC_DIR": static_dir,
        "TIT_SERVER_RELOAD": "1" if server_reload else "",
    }

build_run_argv

build_run_argv(spec: StackSpec, env: dict[str, str], *, host_project_dir: str, image: str) -> list[str]

The full docker run argv for one stack — pure, so it can be asserted in a test.

Every volume whose source interpolates to empty is dropped rather than handed to Docker as an empty bind source (composeFile.ts does the same); that is how the optional dev-repo mount switches itself off.

Source code in tit/launch.py

def build_run_argv(
    spec: StackSpec, env: dict[str, str], *, host_project_dir: str, image: str
) -> list[str]:
    """The full ``docker run`` argv for one stack — pure, so it can be asserted in a test.

    Every volume whose source interpolates to empty is dropped rather than
    handed to Docker as an empty bind source (``composeFile.ts`` does the same);
    that is how the optional dev-repo mount switches itself off.
    """
    name = container_name(host_project_dir)
    argv = [
        "docker",
        "run",
        "--detach",
        "--name",
        name,
        "--platform",
        PLATFORM,
        "--workdir",
        spec.working_dir,
    ]
    if spec.init:
        argv.append("--init")
    for key, value in (
        (LABEL_PROJECT, project_name(host_project_dir)),
        (LABEL_STACK, STACK_ID),
        (LABEL_SERVICE, SERVICE_NAME),
        (LABEL_HOST_DIR, host_project_dir),
    ):
        argv += ["--label", f"{key}={value}"]
    for entry in spec.volumes:
        resolved = interpolate(entry, env)
        source = resolved.split(":", 1)[0]
        if not source:
            continue
        argv += ["--volume", resolved]
    for key, template in spec.environment:
        argv += ["--env", f"{key}={interpolate(template, env)}"]
    for entry in spec.ports:
        argv += ["--publish", interpolate(entry, env)]
    if spec.healthcheck and spec.healthcheck[0] == "CMD":
        argv += [
            "--health-cmd",
            " ".join(interpolate(part, env) for part in spec.healthcheck[1:]),
            "--health-interval",
            "10s",
            "--health-timeout",
            "3s",
            "--health-start-period",
            "20s",
            "--health-retries",
            "6",
        ]
    argv.append(image)
    return argv

session_url

session_url(origin: str, token: str) -> str

The one URL a browser needs: it trades the token for the session cookie.

tit/server/auth.py mints a session id, sets it as an HttpOnly; SameSite=Strict cookie and redirects to /, so the token never has to be typed again and never appears in the address bar afterwards.

Source code in tit/launch.py

def session_url(origin: str, token: str) -> str:
    """The one URL a browser needs: it trades the token for the session cookie.

    ``tit/server/auth.py`` mints a session id, sets it as an
    ``HttpOnly; SameSite=Strict`` cookie and redirects to ``/``, so the token
    never has to be typed again and never appears in the address bar afterwards.
    """
    from urllib.parse import quote

    return f"{origin}/auth/session?token={quote(token, safe='')}"

find_free_port

find_free_port(preferred: int = DEFAULT_PORT, attempts: int = 64) -> int

First free TCP port at or above preferred on the loopback interface.

Source code in tit/launch.py

def find_free_port(preferred: int = DEFAULT_PORT, attempts: int = 64) -> int:
    """First free TCP port at or above ``preferred`` on the loopback interface."""
    parse_port(preferred)  # 0 would bind "any port"; a negative one overflows bind()
    for offset in range(attempts):
        candidate = preferred + offset
        if candidate > 65535:
            break
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
                sock.bind(("127.0.0.1", candidate))
            except OSError:
                continue
            return candidate
    raise LaunchError(f"no free port in {preferred}-{preferred + attempts - 1}")

probe_container_gpu

probe_container_gpu(image: str, echo: Callable[[str], None] = print) -> bool

Exercise CUDA in the selected image without any project mounts.

Source code in tit/launch.py

def probe_container_gpu(image: str, echo: Callable[[str], None] = print) -> bool:
    """Exercise CUDA in the selected image without any project mounts."""
    name = "ti-gpu-probe-" + secrets.token_hex(8)
    try:
        result = subprocess.run(
            [
                "docker",
                "run",
                "--name",
                name,
                "--rm",
                "--network",
                "none",
                "--platform",
                PLATFORM,
                "--gpus",
                "all",
                "--env",
                "NVIDIA_DRIVER_CAPABILITIES=compute,utility",
                "--entrypoint",
                "simnibs_python",
                image,
                "-c",
                CUDA_PROBE,
            ],
            capture_output=True,
            text=True,
            timeout=60,
            check=False,
        )
        if result.returncode == 0:
            echo("CUDA GPU verified; GPU access enabled for FastSurfer.")
            return True
        reason = (result.stderr or result.stdout).strip().splitlines()
        detail = reason[0] if reason else "CUDA computation failed"
    except (OSError, subprocess.TimeoutExpired) as exc:
        detail = str(exc)
    finally:
        # The GPU is optional: a failed cleanup must not replace the verdict with an exception.
        try:
            subprocess.run(
                ["docker", "rm", "--force", name],
                capture_output=True,
                timeout=15,
                check=False,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            echo(f"Warning: could not remove the GPU probe container {name}: {exc}")
    echo(
        f"Container GPU unavailable ({detail}). CPU remains available; "
        "Apple Silicon users can enable native Apple GPU in Pre-processing."
    )
    return False

require_docker

require_docker() -> None

Fail with the remedy, not a traceback, when the daemon is unreachable.

Source code in tit/launch.py

def require_docker() -> None:
    """Fail with the remedy, not a traceback, when the daemon is unreachable."""
    result = _docker("version", "--format", "{{.Server.APIVersion}}", check=False)
    if result.returncode != 0:
        detail = (result.stderr or "").strip()
        if "permission denied" in detail.lower():
            raise LaunchError(
                "Permission denied talking to the Docker socket. On Linux, add your user to the "
                "docker group (sudo usermod -aG docker $USER), then log out and back in. "
                "On macOS, restart Docker Desktop."
            )
        raise LaunchError(
            "Docker is installed but not running. Start Docker Desktop (or your Docker daemon) "
            f"and try again.{(' (' + detail + ')') if detail else ''}"
        )

find_container

find_container(host_project_dir: str) -> dict | None

The container this project owns, by label — the app's own attach rule.

Both labels are checked: tit.project is a 32-bit hash, so matching it alone could attach one project's UI to another project's container.

Source code in tit/launch.py

def find_container(host_project_dir: str) -> dict | None:
    """The container this project owns, by label — the app's own attach rule.

    Both labels are checked: ``tit.project`` is a 32-bit hash, so matching it
    alone could attach one project's UI to another project's container.
    """
    result = _docker(
        "ps",
        "--all",
        "--filter",
        f"label={LABEL_PROJECT}={project_name(host_project_dir)}",
        "--format",
        "{{.ID}}",
    )
    for cid in result.stdout.split():
        info = json.loads(_docker("inspect", cid).stdout)[0]
        labels = info["Config"].get("Labels") or {}
        recorded = labels.get(LABEL_HOST_DIR)
        if recorded and os.path.realpath(recorded) == os.path.realpath(
            host_project_dir
        ):
            return info
    return None

is_session

is_session(labels: dict[str, str]) -> bool

True for a container a TI-Toolbox launcher started, judged by its labels alone.

Service tit (tit.service, else the compose service; empty counts as absent) and one of tit.stack=ti-toolbox-v3, tit.project or tit.host_project_dir. The desktop's isToolboxContainer and loader.sh's session_ids apply the same rule (tests/fixtures/session_labels.json).

Source code in tit/launch.py

def is_session(labels: dict[str, str]) -> bool:
    """True for a container a TI-Toolbox launcher started, judged by its labels alone.

    Service ``tit`` (``tit.service``, else the compose service; empty counts as absent) and
    one of ``tit.stack=ti-toolbox-v3``, ``tit.project`` or ``tit.host_project_dir``. The
    desktop's ``isToolboxContainer`` and ``loader.sh``'s ``session_ids`` apply the same rule
    (``tests/fixtures/session_labels.json``).
    """
    service = labels.get(LABEL_SERVICE) or labels.get("com.docker.compose.service")
    if service and service != SERVICE_NAME:
        return False
    return bool(
        labels.get(LABEL_STACK) == STACK_ID
        or labels.get(LABEL_PROJECT)
        or labels.get(LABEL_HOST_DIR)
    )

find_running_containers

find_running_containers() -> list[dict]

Running toolbox sessions of every project: containers a launcher labelled.

Never matched by image or name: a scratch docker run of the image, a job's FreeSurfer sibling or a GPU probe would be offered for Recreate and stopped.

Source code in tit/launch.py

def find_running_containers() -> list[dict]:
    """Running toolbox sessions of every project: containers a launcher labelled.

    Never matched by image or name: a scratch ``docker run`` of the image, a job's
    FreeSurfer sibling or a GPU probe would be offered for Recreate and stopped.
    """
    result = _docker("ps", "--format", "{{.ID}}")
    containers = []
    for cid in result.stdout.split():
        info = json.loads(_docker("inspect", cid).stdout)[0]
        if is_session(info.get("Config", {}).get("Labels") or {}):
            containers.append(info)
    return containers

choose_existing

choose_existing(containers: list[dict], options: LaunchOptions) -> tuple[dict, str]

Require a deliberate session and action; never infer consent from matching config.

Source code in tit/launch.py

def choose_existing(containers: list[dict], options: LaunchOptions) -> tuple[dict, str]:
    """Require a deliberate session and action; never infer consent from matching config."""
    options.echo("\nRunning TI-Toolbox containers")
    for index, info in enumerate(containers, 1):
        options.echo(f"  {index}. {info.get('Config', {}).get('Image', 'TI-Toolbox')}")
    selected = None
    if options.container:
        selected = next(
            (
                info
                for info in containers
                if options.container in (info["Id"], info["Name"].lstrip("/"))
            ),
            None,
        )
        if selected is None:
            raise LaunchError(
                "--container does not identify a running TI-Toolbox container"
            )
    elif len(containers) == 1:
        selected = containers[0]
    elif sys.stdin.isatty():
        try:
            index = int(input("Select container number: "))
            if 1 <= index <= len(containers):
                selected = containers[index - 1]
        except (ValueError, EOFError):
            pass
    if selected is None:
        raise LaunchError(
            "select a running session with --container NAME (multiple containers found)"
        )
    action = options.existing
    if action is None and sys.stdin.isatty():
        options.echo(
            "\nAvailable actions\n-----------------\n  1. Recreate (default)\n  2. Attach"
        )
        options.echo("\nRecreate stops this container and its jobs.")
        try:
            answer = input("Choose [1]: ").strip().lower()
            action = {
                "": "recreate",
                "1": "recreate",
                "r": "recreate",
                "recreate": "recreate",
                "2": "attach",
                "a": "attach",
                "attach": "attach",
            }.get(answer)
        except EOFError:
            action = None
    if action not in ("attach", "recreate"):
        raise LaunchError(
            "existing container left unchanged; choose --existing attach or "
            "--existing recreate (stops jobs and removes the selected container)"
        )
    return selected, action

container_credentials

container_credentials(info: dict) -> tuple[str, str]

(origin, token) read out of a running container's own environment.

The token is never written to the host filesystem — on attach it comes back from the container that holds it, which is why there is no state file here.

Source code in tit/launch.py

def container_credentials(info: dict) -> tuple[str, str]:
    """``(origin, token)`` read out of a running container's own environment.

    The token is never written to the host filesystem — on attach it comes back
    from the container that holds it, which is why there is no state file here.
    """
    env = dict(
        pair.split("=", 1) for pair in info["Config"].get("Env", []) if "=" in pair
    )
    token = env.get("TIT_SERVER_TOKEN", "")
    port = env.get("TIT_SERVER_PORT", str(DEFAULT_PORT))
    if not token:
        raise LaunchError(
            f"container {info['Name'].lstrip('/')} has no TIT_SERVER_TOKEN; "
            "stop it with `tit launch --stop` and start a fresh one."
        )
    return f"http://127.0.0.1:{port}", token

ensure_image

ensure_image(image: str, *, refresh: bool = True, echo=print) -> None

Refresh the mutable release image; retain cached images for offline/dev use.

Source code in tit/launch.py

def ensure_image(image: str, *, refresh: bool = True, echo=print) -> None:
    """Refresh the mutable release image; retain cached images for offline/dev use."""
    cached = _docker("image", "inspect", image, check=False).returncode == 0
    if cached and not (refresh and image == "idossha/ti-toolbox:v3.1.3"):
        echo(f"image {image} is already present")
        return
    echo(f"checking for updates to {image}…" if cached else f"downloading {image}…")
    result = subprocess.run(["docker", "pull", "--platform", PLATFORM, image])
    if result.returncode != 0:
        if cached:
            echo(f"Warning: could not refresh {image}; using the cached image.")
            return
        raise LaunchError(
            f"could not download {image}.\n"
            "  - If you are on a pre-release checkout, no such tag is published yet: build it\n"
            "    with `container/blueprint/build.sh --tag "
            + image
            + "` (30-60+ minutes), or\n"
            "    pass --image with a tag you already have (`docker images idossha/ti-toolbox`).\n"
            "  - Otherwise check your internet connection and registry access."
        )

wait_for_health

wait_for_health(origin: str, timeout: float = 180.0, *, echo=print) -> None

Poll /api/health until it answers {"status": "ok"}.

The generous default matches the Electron app's: a cold amd64 start under emulation is slow, and declaring a live stack dead costs the user the whole session.

Source code in tit/launch.py

def wait_for_health(origin: str, timeout: float = 180.0, *, echo=print) -> None:
    """Poll ``/api/health`` until it answers ``{"status": "ok"}``.

    The generous default matches the Electron app's: a cold amd64 start under
    emulation is slow, and declaring a live stack dead costs the user the whole
    session.
    """
    deadline = time.monotonic() + timeout
    last = "no response"
    while time.monotonic() < deadline:
        try:
            with urllib.request.urlopen(f"{origin}/api/health", timeout=2) as response:
                body = json.loads(response.read().decode("utf-8"))
            if body.get("status") == "ok":
                return
            last = f"health status {body.get('status')!r}"
        except (
            urllib.error.URLError,
            OSError,
            ValueError,
            json.JSONDecodeError,
        ) as err:
            last = str(err)
        time.sleep(0.5)
    raise LaunchError(
        f"the container started but {origin}/api/health did not answer within "
        f"{int(timeout)}s ({last}). Run `tit launch --logs` to see why."
    )

is_wsl

is_wsl() -> bool

True inside a WSL2 distribution: the Linux side of a Windows host.

WSL puts WSL_DISTRO_NAME (and, with interop on, WSL_INTEROP) in every process, so those two variables are the whole test. The kernel string is deliberately not consulted: a Docker Desktop container runs on that same "microsoft" kernel and must never be mistaken for the host. Being environment-only also lets a plain Linux run stand in for WSL in tests, and a WSL host stand in for plain Linux (tests/conftest.py clears the variables). loader.sh's is_wsl applies the identical rule.

Source code in tit/launch.py

def is_wsl() -> bool:
    """True inside a WSL2 distribution: the Linux side of a Windows host.

    WSL puts ``WSL_DISTRO_NAME`` (and, with interop on, ``WSL_INTEROP``) in every
    process, so those two variables are the whole test. The kernel string is deliberately
    not consulted: a Docker Desktop container runs on that same "microsoft" kernel and must
    never be mistaken for the host. Being environment-only also lets a plain Linux run stand
    in for WSL in tests, and a WSL host stand in for plain Linux (``tests/conftest.py``
    clears the variables). ``loader.sh``'s ``is_wsl`` applies the identical rule.
    """
    return bool(os.environ.get("WSL_DISTRO_NAME") or os.environ.get("WSL_INTEROP"))

translate_project_path

translate_project_path(raw: str) -> str

On WSL2, C:\Users\me\project is /mnt/c/Users/me/project; elsewhere unchanged.

Docker Desktop is handed the Linux spelling, and /mnt/c/... already is one, so it passes straight through. loader.sh applies the identical rule before it validates the directory, which keeps --print-config byte-identical on WSL too.

Source code in tit/launch.py

def translate_project_path(raw: str) -> str:
    """On WSL2, ``C:\\Users\\me\\project`` is ``/mnt/c/Users/me/project``; elsewhere unchanged.

    Docker Desktop is handed the Linux spelling, and ``/mnt/c/...`` already is one, so it
    passes straight through. ``loader.sh`` applies the identical rule before it validates
    the directory, which keeps ``--print-config`` byte-identical on WSL too.
    """
    if not raw or not is_wsl():
        return raw
    match = _WINDOWS_DRIVE.match(raw)
    if not match:
        return raw
    drive, rest = match.groups()
    return f"/mnt/{drive.lower()}/" + rest.replace("\\", "/").lstrip("/")

resolve_project

resolve_project(raw: str | None) -> str

The absolute, real path of the project directory, or a message saying what is wrong.

Source code in tit/launch.py

def resolve_project(raw: str | None) -> str:
    """The absolute, real path of the project directory, or a message saying what is wrong."""
    if not raw:
        raise LaunchError(
            "no project directory given. Pass --project /path/to/bids-project (or set "
            "TIT_PROJECT_DIR); TI-Toolbox opens one project at a time and will not guess which."
        )
    path = Path(translate_project_path(raw)).expanduser()
    if not path.is_dir():
        raise LaunchError(f"project directory does not exist: {path}")
    resolved = os.path.realpath(path)
    check_project_path(resolved)
    return resolved

check_project_path

check_project_path(path: str) -> None

Refuse what loader.sh refuses: a colon or newline, or the filesystem root.

A Docker bind SOURCE:TARGET cannot carry a colon in its source (a Windows drive prefix aside), and / has no name to mount at /mnt/<name>.

Source code in tit/launch.py

def check_project_path(path: str) -> None:
    """Refuse what ``loader.sh`` refuses: a colon or newline, or the filesystem root.

    A Docker bind ``SOURCE:TARGET`` cannot carry a colon in its source (a Windows drive
    prefix aside), and ``/`` has no name to mount at ``/mnt/<name>``.
    """
    rest = path[2:] if os.name == "nt" and _DRIVE_PREFIX.match(path) else path
    if "\n" in path or ":" in rest:
        raise LaunchError("project path cannot contain newlines or colons")
    if os.path.dirname(path) == path:
        raise LaunchError("choose a project directory, not the filesystem root")

start

start(options: LaunchOptions) -> tuple[str, str]

Ask before reusing or replacing any running toolbox container; return (origin, token).

Source code in tit/launch.py

def start(options: LaunchOptions) -> tuple[str, str]:
    """Ask before reusing or replacing any running toolbox container; return ``(origin, token)``."""
    echo = options.echo
    # Every input is checked before the first Docker call: a recreate stops the selected
    # container long before the run argv is built.
    host_project_dir = resolve_project(options.project)
    parse_port(options.port)
    parse_timeout(options.timeout)
    require_docker()
    image = options.image or default_image()
    image_ready = False

    running = find_running_containers()
    if running:
        selected, action = choose_existing(running, options)
        if action == "attach":
            origin, token = container_credentials(selected)
            options.session_project = (selected["Config"].get("Labels") or {}).get(
                LABEL_HOST_DIR, ""
            )
            echo(
                f"attached to {selected['Name'].lstrip('/')} at {origin}; using its existing configuration"
            )
            wait_for_health(origin, timeout=60.0, echo=echo)
            return origin, token
        # --existing recreate with no --container picked the only running session without
        # asking: it may replace this project's own session, never another project's.
        recorded = (selected["Config"].get("Labels") or {}).get(LABEL_HOST_DIR, "")
        if (
            options.existing == "recreate"
            and not options.container
            and len(running) == 1
            and not (recorded and os.path.realpath(recorded) == host_project_dir)
        ):
            raise LaunchError(
                f"the running TI-Toolbox session belongs to {recorded or 'another project'}. "
                "Name it with --container to replace it, or use --existing attach; "
                "nothing was changed."
            )
        owner = find_container(host_project_dir)
        if (
            owner is not None
            and owner["State"]["Running"]
            and owner["Id"] != selected["Id"]
        ):
            raise LaunchError(
                "another running container owns the requested project; select it explicitly"
            )
        load_spec()  # Validate the requested YAML before stopping an existing session.
        ensure_image(image, refresh=not bool(options.repo_dir), echo=echo)
        image_ready = True
        _docker("stop", selected["Id"])
        _docker("rm", selected["Id"])
    elif options.container or options.existing == "attach":
        raise LaunchError(
            "no selected running TI-Toolbox container is available to attach"
        )
    existing = find_container(host_project_dir)
    if existing is not None:
        if existing["State"]["Running"]:
            raise LaunchError(
                "another running container owns the requested project; select it explicitly"
            )
        echo(f"removing the stopped container {existing['Name'].lstrip('/')}")
        _docker("rm", existing["Id"])

    if not image_ready:
        ensure_image(image, refresh=not bool(options.repo_dir), echo=echo)

    port = find_free_port(options.port)
    if port != options.port:
        echo(f"port {options.port} is taken; using {port}")
    token = secrets.token_urlsafe(32)
    env = build_env(
        host_project_dir=host_project_dir,
        port=port,
        token=token,
        user_config=str(user_config_dir()),
        image_tag=image.rsplit(":", 1)[-1],
        repo_dir=options.repo_dir,
        static_dir=options.static_dir,
        server_reload=options.server_reload,
    )
    argv = build_run_argv(
        load_spec(), env, host_project_dir=host_project_dir, image=image
    )
    if probe_container_gpu(image, echo):
        argv[2:2] = [
            "--gpus",
            "all",
            "--env",
            "NVIDIA_DRIVER_CAPABILITIES=compute,utility",
        ]
    echo(f"starting {container_name(host_project_dir)} on port {port}…")
    _docker(*argv[1:])
    options.session_project = host_project_dir

    origin = f"http://127.0.0.1:{port}"
    echo("waiting for the server to answer…")
    try:
        wait_for_health(origin, timeout=options.timeout, echo=echo)
    except LaunchError:
        logs = _docker(
            "logs", "--tail", "20", container_name(host_project_dir), check=False
        )
        raise LaunchError(
            f"the container did not become healthy. Last log lines:\n{logs.stdout}{logs.stderr}"
        ) from None
    return origin, token

stop

stop(project: str) -> list[str]

Stop and remove this project's container. Already gone is success, not failure.

Source code in tit/launch.py

def stop(project: str) -> list[str]:
    """Stop and remove this project's container. Already gone is success, not failure."""
    require_docker()
    host_project_dir = resolve_project(project)
    info = find_container(host_project_dir)
    if info is None:
        return []
    name = info["Name"].lstrip("/")
    _docker("rm", "--force", info["Id"])
    return [name]

status

status(project: str) -> dict | None

{name, state, image, origin, health} for this project's container, or None.

Source code in tit/launch.py

def status(project: str) -> dict | None:
    """``{name, state, image, origin, health}`` for this project's container, or ``None``."""
    require_docker()
    info = find_container(resolve_project(project))
    if info is None:
        return None
    env = dict(
        pair.split("=", 1) for pair in info["Config"].get("Env", []) if "=" in pair
    )
    return {
        "name": info["Name"].lstrip("/"),
        "state": info["State"]["Status"],
        "health": (info["State"].get("Health") or {}).get("Status", "n/a"),
        "image": info["Config"]["Image"],
        "origin": f"http://127.0.0.1:{env.get('TIT_SERVER_PORT', DEFAULT_PORT)}",
    }

logs

logs(project: str, *, follow: bool = False, tail: str = '200') -> int

Stream the container's logs to this terminal; returns docker's exit code.

Source code in tit/launch.py

def logs(project: str, *, follow: bool = False, tail: str = "200") -> int:
    """Stream the container's logs to this terminal; returns docker's exit code."""
    require_docker()
    info = find_container(resolve_project(project))
    if info is None:
        raise LaunchError(
            "no TI-Toolbox container is running for that project directory"
        )
    argv = ["docker", "logs", "--tail", tail]
    if follow:
        argv.append("--follow")
    argv.append(info["Id"])
    return subprocess.run(argv).returncode

windows_openers

windows_openers(url: str) -> list[list[str]]

Commands that open url in the Windows default browser from inside WSL, in order.

wslview (wslu) is WSL-aware and honours the user's own setup; PowerShell and cmd.exe ship with Windows and reach WSL through interop. Only URL characters reach those two, because each one is a shell of sorts; the session URL is made of nothing else.

Source code in tit/launch.py

def windows_openers(url: str) -> list[list[str]]:
    """Commands that open ``url`` in the Windows default browser from inside WSL, in order.

    ``wslview`` (wslu) is WSL-aware and honours the user's own setup; PowerShell and
    ``cmd.exe`` ship with Windows and reach WSL through interop. Only URL characters
    reach those two, because each one is a shell of sorts; the session URL is made of
    nothing else.
    """
    commands: list[list[str]] = []
    wslview = shutil.which("wslview")
    if wslview:
        commands.append([wslview, url])
    if _URL_SAFE.match(url):
        powershell = shutil.which("powershell.exe")
        if powershell:
            commands.append(
                [
                    powershell,
                    "-NoProfile",
                    "-NonInteractive",
                    "-Command",
                    f"Start-Process -FilePath '{url}'",
                ]
            )
        cmd = shutil.which("cmd.exe")
        if cmd:
            commands.append([cmd, "/c", "start", "", url])
    return commands

open_in_browser

open_in_browser(url: str, *, echo=print) -> None

Open the session URL, saying so either way — a headless host has no browser to open.

On WSL the browser lives on the Windows side: webbrowser would hand the URL to xdg-open and report success whether or not anything showed it, so the Windows openers are tried instead and judged by their exit status.

Source code in tit/launch.py

def open_in_browser(url: str, *, echo=print) -> None:
    """Open the session URL, saying so either way — a headless host has no browser to open.

    On WSL the browser lives on the Windows side: ``webbrowser`` would hand the URL to
    ``xdg-open`` and report success whether or not anything showed it, so the Windows
    openers are tried instead and judged by their exit status.
    """
    if is_wsl():
        # cmd.exe warns when its working directory is a Linux path; /mnt/c is always Windows.
        cwd = "/mnt/c" if os.path.isdir("/mnt/c") else None
        for command in windows_openers(url):
            try:
                result = subprocess.run(
                    command,
                    cwd=cwd,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    timeout=30,
                    check=False,
                )
            except (OSError, subprocess.TimeoutExpired):
                continue
            if result.returncode == 0:
                echo("opened your browser")
                return
        echo("could not open a browser automatically; paste the URL above into one")
        return
    if webbrowser.open(url):
        echo("opened your browser")
    else:
        echo("could not open a browser automatically; paste the URL above into one")