Gate server lifecycle management.
Manages the terok-gate HTTP server that wraps git http-backend with
per-task token authentication. Containers reach the gate via http:// URLs
through host.containers.internal — standard HTTP protocol, no bind-mount
escape vector.
Networking across Podman versions:
- terok-shield handles container networking via its OCI hook. The gate
server port is passed as
loopback_ports in :class:ShieldConfig so that
shield's nftables rules allow containers to reach host loopback on that port.
Deployment modes (ordered by preference):
-
Systemd socket activation — zero-idle-cost, crash resilience, and no
PID management. Recommended for any Linux host with systemctl --user.
-
Managed terok-gate daemon process — best-effort fallback. Works
correctly but has a simpler lifecycle (manual start/stop, PID file).
No auto-start. Task creation checks reachability and fails with an
actionable error rather than silently starting a daemon.
GateServerStatus(mode, running, port)
dataclass
Current state of the gate server.
mode
instance-attribute
"systemd", "daemon", or "none".
running
instance-attribute
Whether the server is currently reachable.
is_systemd_available()
Check whether systemctl --user is usable.
Uses is-system-running which returns well-defined exit codes:
0 = running, 1 = degraded/starting/stopping — both mean systemd is
present. Any other code (or missing binary) means unavailable.
Source code in src/terok_sandbox/gate_server.py
| def is_systemd_available() -> bool:
"""Check whether ``systemctl --user`` is usable.
Uses ``is-system-running`` which returns well-defined exit codes:
0 = running, 1 = degraded/starting/stopping — both mean systemd is
present. Any other code (or missing binary) means unavailable.
"""
try:
result = subprocess.run(
["systemctl", "--user", "is-system-running"],
capture_output=True,
text=True,
timeout=5,
)
# "running" (0), "degraded" (1), "starting" (1), "stopping" (1)
# all indicate a usable user session.
return result.returncode in (0, 1)
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
|
is_socket_installed()
Check whether the terok-gate.socket unit file exists.
Source code in src/terok_sandbox/gate_server.py
| def is_socket_installed() -> bool:
"""Check whether the ``terok-gate.socket`` unit file exists."""
unit_dir = _systemd_unit_dir()
return (unit_dir / _SOCKET_UNIT).is_file()
|
is_socket_active()
Check whether the terok-gate.socket unit is active (listening).
Source code in src/terok_sandbox/gate_server.py
| def is_socket_active() -> bool:
"""Check whether the ``terok-gate.socket`` unit is active (listening)."""
try:
result = subprocess.run(
["systemctl", "--user", "is-active", _SOCKET_UNIT],
capture_output=True,
text=True,
timeout=5,
)
return result.stdout.strip() == "active"
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
|
install_systemd_units(cfg=None)
Render and install systemd socket+service units, then enable+start the socket.
Source code in src/terok_sandbox/gate_server.py
| def install_systemd_units(cfg: SandboxConfig | None = None) -> None:
"""Render and install systemd socket+service units, then enable+start the socket."""
import shutil
import terok_sandbox.gate
from ._util import render_template
from .gate_tokens import token_file_path
gate_bin = shutil.which("terok-gate")
if not gate_bin:
raise SystemExit(
"Cannot find 'terok-gate' on PATH.\n"
"Ensure terok-sandbox is installed (pip/pipx/poetry) and the binary is accessible."
)
unit_dir = _systemd_unit_dir()
unit_dir.mkdir(parents=True, exist_ok=True)
resource_dir = Path(terok_sandbox.gate.__file__).resolve().parent / "resources" / "systemd"
variables = {
"PORT": str(_get_port(cfg)),
"GATE_BASE_PATH": str(_get_gate_base_path(cfg)),
"TOKEN_FILE": str(token_file_path(cfg)),
"UNIT_VERSION": str(_UNIT_VERSION),
"TEROK_GATE_BIN": gate_bin,
}
for template_name in (_SOCKET_UNIT, "terok-gate@.service"):
template_path = resource_dir / template_name
if not template_path.is_file():
raise SystemExit(f"Missing systemd template: {template_path}")
content = render_template(template_path, variables)
(unit_dir / template_name).write_text(content, encoding="utf-8")
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True, timeout=10)
subprocess.run(
["systemctl", "--user", "enable", "--now", _SOCKET_UNIT],
check=True,
timeout=10,
)
|
uninstall_systemd_units(cfg=None)
Disable+stop the socket and remove unit files.
Source code in src/terok_sandbox/gate_server.py
| def uninstall_systemd_units(cfg: SandboxConfig | None = None) -> None: # noqa: ARG001
"""Disable+stop the socket and remove unit files."""
unit_dir = _systemd_unit_dir()
subprocess.run(
["systemctl", "--user", "disable", "--now", _SOCKET_UNIT],
check=False,
timeout=10,
)
subprocess.run(["systemctl", "--user", "daemon-reload"], check=False, timeout=10)
for name in (_SOCKET_UNIT, "terok-gate@.service"):
unit_file = unit_dir / name
if unit_file.is_file():
unit_file.unlink()
subprocess.run(["systemctl", "--user", "daemon-reload"], check=False, timeout=10)
|
start_daemon(port=None, cfg=None)
Start a terok-gate daemon process (non-systemd fallback).
Writes a PID file to runtime_root() / "gate-server.pid".
If TEROK_GATE_ADMIN_TOKEN is set in the environment, it is
forwarded to the daemon for host-level access to all repos.
Source code in src/terok_sandbox/gate_server.py
| def start_daemon(port: int | None = None, cfg: SandboxConfig | None = None) -> None:
"""Start a ``terok-gate`` daemon process (non-systemd fallback).
Writes a PID file to ``runtime_root() / "gate-server.pid"``.
If ``TEROK_GATE_ADMIN_TOKEN`` is set in the environment, it is
forwarded to the daemon for host-level access to all repos.
"""
from .gate_tokens import token_file_path
effective_port = port or _get_port(cfg)
gate_base = _get_gate_base_path(cfg)
gate_base.mkdir(parents=True, exist_ok=True)
pidfile = _pid_file(cfg)
pidfile.parent.mkdir(parents=True, exist_ok=True)
cmd = [
"terok-gate",
f"--base-path={gate_base}",
f"--token-file={token_file_path()}",
f"--port={effective_port}",
"--detach",
f"--pid-file={pidfile}",
]
admin_token = os.environ.get("TEROK_GATE_ADMIN_TOKEN")
if admin_token:
cmd.append(f"--admin-token={admin_token}")
bind_addr = os.environ.get("TEROK_GATE_BIND")
if bind_addr:
cmd.append(f"--bind={bind_addr}")
subprocess.run(cmd, check=True, timeout=10)
|
stop_daemon(cfg=None)
Stop the managed daemon by reading the PID file and sending SIGTERM.
Source code in src/terok_sandbox/gate_server.py
| def stop_daemon(cfg: SandboxConfig | None = None) -> None:
"""Stop the managed daemon by reading the PID file and sending SIGTERM."""
pidfile = _pid_file(cfg)
if not pidfile.is_file():
return
try:
pid = int(pidfile.read_text().strip())
if _is_managed_server(pid):
os.kill(pid, signal.SIGTERM)
except (ValueError, ProcessLookupError, PermissionError):
pass
finally:
if pidfile.is_file():
pidfile.unlink()
|
is_daemon_running(cfg=None)
Check whether the managed daemon process is alive via its PID file.
Source code in src/terok_sandbox/gate_server.py
| def is_daemon_running(cfg: SandboxConfig | None = None) -> bool:
"""Check whether the managed daemon process is alive via its PID file."""
pidfile = _pid_file(cfg)
if not pidfile.is_file():
return False
try:
pid = int(pidfile.read_text().strip())
if not _is_managed_server(pid):
return False
os.kill(pid, 0) # signal 0 = existence check
return True
except (ValueError, ProcessLookupError, PermissionError):
return False
|
get_server_status(cfg=None)
Return the current gate server status.
Source code in src/terok_sandbox/gate_server.py
| def get_server_status(cfg: SandboxConfig | None = None) -> GateServerStatus:
"""Return the current gate server status."""
port = _get_port(cfg)
if is_socket_installed():
if is_socket_active():
return GateServerStatus(mode="systemd", running=True, port=port)
# Socket installed but inactive — check if the daemon fallback is running
if is_daemon_running(cfg):
return GateServerStatus(mode="daemon", running=True, port=port)
return GateServerStatus(mode="systemd", running=False, port=port)
if is_daemon_running(cfg):
return GateServerStatus(mode="daemon", running=True, port=port)
return GateServerStatus(mode="none", running=False, port=port)
|
check_units_outdated(cfg=None)
Return a warning string if installed systemd units are stale, else None.
Checks both the unit version stamp and the baked --base-path against
the current configuration. Useful for gate-server status and
sickbay to surface upgrade hints without blocking task creation
(that's ensure_server_reachable's job).
Source code in src/terok_sandbox/gate_server.py
| def check_units_outdated(cfg: SandboxConfig | None = None) -> str | None:
"""Return a warning string if installed systemd units are stale, else ``None``.
Checks both the unit version stamp and the baked ``--base-path`` against
the current configuration. Useful for ``gate-server status`` and
``sickbay`` to surface upgrade hints without blocking task creation
(that's ``ensure_server_reachable``'s job).
"""
if not is_socket_installed():
return None
installed = _installed_unit_version()
if installed is None or installed < _UNIT_VERSION:
installed_label = "unversioned" if installed is None else f"v{installed}"
return (
f"Systemd units are outdated (installed {installed_label}, expected v{_UNIT_VERSION})."
)
return _base_path_diverged(cfg)
|
get_gate_base_path(cfg=None)
Return the gate base path (public API).
Source code in src/terok_sandbox/gate_server.py
| def get_gate_base_path(cfg: SandboxConfig | None = None) -> Path:
"""Return the gate base path (public API)."""
return _get_gate_base_path(cfg)
|
get_gate_server_port(cfg=None)
Return the configured gate server port.
Source code in src/terok_sandbox/gate_server.py
| def get_gate_server_port(cfg: SandboxConfig | None = None) -> int:
"""Return the configured gate server port."""
return _get_port(cfg)
|
ensure_server_reachable(cfg=None)
Verify the gate server is running and configured correctly.
Raises SystemExit if the server is down, systemd units are outdated,
or the installed base path diverges from the current configuration.
Called before task creation to fail early with an actionable message.
Source code in src/terok_sandbox/gate_server.py
| def ensure_server_reachable(cfg: SandboxConfig | None = None) -> None:
"""Verify the gate server is running and configured correctly.
Raises ``SystemExit`` if the server is down, systemd units are outdated,
or the installed base path diverges from the current configuration.
Called before task creation to fail early with an actionable message.
"""
server_status = get_server_status(cfg)
if server_status.running:
if server_status.mode == "systemd":
installed = _installed_unit_version()
if installed is None or installed < _UNIT_VERSION:
installed_label = "unversioned" if installed is None else f"v{installed}"
raise SystemExit(
"Gate server systemd units are outdated "
f"(installed {installed_label}, expected v{_UNIT_VERSION})."
)
path_warning = _base_path_diverged(cfg)
if path_warning:
raise SystemExit(path_warning)
return
msg = (
"Gate server is not running.\n"
"\n"
"The gate server serves git repos to task containers over the network,\n"
"replacing the previous volume-mount approach.\n"
"\n"
)
if is_systemd_available():
msg += "Recommended: install and start the systemd socket.\n"
else:
msg += "Start the gate daemon.\n"
raise SystemExit(msg)
|