port_registry

port_registry

Shared port registry for multi-user isolation.

Every port allocation — infrastructure services and container web ports — flows through PortRegistry. Claims are persisted as per-user JSON files in a shared directory (default /tmp/terok-ports/). All users' claim files are read at allocation time to avoid collisions; socket bind tests verify the port is actually free.

A module-level singleton (_default) provides the convenience API (claim_port, release_port, etc.) used by the rest of the stack.

SERVICE_GATE = 'gate' module-attribute

SERVICE_PROXY = 'proxy' module-attribute

SERVICE_SSH_AGENT = 'ssh_agent' module-attribute

PORT_RANGE = _default.port_range module-attribute

Contiguous range for all auto-allocated ports (infra + web).

ServicePorts(gate, proxy, ssh_agent) dataclass

Resolved infrastructure service ports for one terok session.

gate instance-attribute

proxy instance-attribute

ssh_agent instance-attribute

PortRegistry(registry_dir, port_range)

File-based shared port registry for multi-user isolation.

Each instance manages its own in-memory claim set and shared directory. Use the module-level singleton (_default) for production code; tests can construct isolated instances with a temporary directory.

Source code in src/terok_sandbox/port_registry.py

def __init__(self, registry_dir: Path, port_range: range) -> None:
    self.registry_dir = registry_dir
    self.port_range = port_range
    self._held: dict[str, int] = {}
    self._service_ports: ServicePorts | None = None
    self._dir_ensured = False
    # Guards ``_held`` and ``_service_ports`` so concurrent callers
    # (e.g. textual worker threads each constructing their own
    # ``SandboxConfig``) cannot race between the "already held?"
    # check and the subsequent claim/write.  Reentrant because
    # ``resolve_service_ports`` calls ``claim`` under the lock.
    self._lock = threading.RLock()

registry_dir = registry_dir instance-attribute

port_range = port_range instance-attribute

resolve_service_ports(gate_pref, proxy_pref, ssh_pref, *, gate_explicit=False, proxy_explicit=False, ssh_explicit=False, state_dir=None)

Resolve and claim infrastructure ports (cached after first call).

Each _pref is a preferred starting port or None for auto-allocation. When *_explicit is True the port is a hard pin (SystemExit if busy).

When state_dir is provided, port assignments are persisted across restarts. If a previously saved port cannot be reclaimed, the call fails with SystemExit so the user can resolve the conflict.

Source code in src/terok_sandbox/port_registry.py

def resolve_service_ports(
    self,
    gate_pref: int | None,
    proxy_pref: int | None,
    ssh_pref: int | None,
    *,
    gate_explicit: bool = False,
    proxy_explicit: bool = False,
    ssh_explicit: bool = False,
    state_dir: Path | None = None,
) -> ServicePorts:
    """Resolve and claim infrastructure ports (cached after first call).

    Each *_pref* is a preferred starting port or ``None`` for auto-allocation.
    When ``*_explicit`` is True the port is a hard pin (``SystemExit`` if busy).

    When *state_dir* is provided, port assignments are persisted across
    restarts.  If a previously saved port cannot be reclaimed, the call
    fails with ``SystemExit`` so the user can resolve the conflict.
    """
    with self._lock:
        return self._resolve_service_ports_locked(
            gate_pref,
            proxy_pref,
            ssh_pref,
            gate_explicit=gate_explicit,
            proxy_explicit=proxy_explicit,
            ssh_explicit=ssh_explicit,
            state_dir=state_dir,
        )

claim(service_key, preferred=None, *, explicit=False, trusted=False)

Claim one port via the shared file-based registry.

Reads all users' claim files to avoid collisions, then verifies via socket bind that the port is actually free. The claim is persisted to the shared directory so other users can see it.

When trusted is True the port originates from a prior allocation by this user — the saved claims file (port-claims.json). Our own service may be listening, so the _is_port_free bind check is skipped; other-user collision checks still apply. The trusted flag is set by resolve_service_ports during the preference resolution phase.

Source code in src/terok_sandbox/port_registry.py

def claim(
    self,
    service_key: str,
    preferred: int | None = None,
    *,
    explicit: bool = False,
    trusted: bool = False,
) -> int:
    """Claim one port via the shared file-based registry.

    Reads all users' claim files to avoid collisions, then verifies
    via socket bind that the port is actually free.  The claim is
    persisted to the shared directory so other users can see it.

    When *trusted* is True the port originates from a prior
    allocation by this user — the saved claims file
    (``port-claims.json``).  Our own service may be listening, so
    the ``_is_port_free`` bind check is skipped; other-user
    collision checks still apply.  The ``trusted`` flag is set by
    [`resolve_service_ports`][terok_sandbox.port_registry.resolve_service_ports] during the preference resolution
    phase.
    """
    with self._lock:
        return self._claim_locked(service_key, preferred, explicit=explicit, trusted=trusted)

release(service_key)

Release a previously claimed port and update the shared claim file.

Source code in src/terok_sandbox/port_registry.py

def release(self, service_key: str) -> None:
    """Release a previously claimed port and update the shared claim file."""
    with self._lock:
        if self._held.pop(service_key, None) is not None:
            self._write_shared_claims(remove={service_key})

reset()

Clear all in-memory state (for testing).

Source code in src/terok_sandbox/port_registry.py

def reset(self) -> None:
    """Clear all in-memory state (for testing)."""
    with self._lock:
        self._held.clear()
        self._service_ports = None
        self._dir_ensured = False

resolve_service_ports(gate_pref, proxy_pref, ssh_pref, *, gate_explicit=False, proxy_explicit=False, ssh_explicit=False, state_dir=None)

Resolve and claim infrastructure ports via the default registry.

Source code in src/terok_sandbox/port_registry.py

def resolve_service_ports(
    gate_pref: int | None,
    proxy_pref: int | None,
    ssh_pref: int | None,
    *,
    gate_explicit: bool = False,
    proxy_explicit: bool = False,
    ssh_explicit: bool = False,
    state_dir: Path | None = None,
) -> ServicePorts:
    """Resolve and claim infrastructure ports via the default registry."""
    return _default.resolve_service_ports(
        gate_pref,
        proxy_pref,
        ssh_pref,
        gate_explicit=gate_explicit,
        proxy_explicit=proxy_explicit,
        ssh_explicit=ssh_explicit,
        state_dir=state_dir,
    )

claim_port(service_key, preferred=None, *, explicit=False)

Claim one port via the default registry.

Source code in src/terok_sandbox/port_registry.py

def claim_port(
    service_key: str,
    preferred: int | None = None,
    *,
    explicit: bool = False,
) -> int:
    """Claim one port via the default registry."""
    return _default.claim(service_key, preferred, explicit=explicit)

release_port(service_key)

Release a previously claimed port via the default registry.

Source code in src/terok_sandbox/port_registry.py

def release_port(service_key: str) -> None:
    """Release a previously claimed port via the default registry."""
    _default.release(service_key)

reset_cache()

Clear all in-memory state on the default registry (for testing).

Source code in src/terok_sandbox/port_registry.py

def reset_cache() -> None:
    """Clear all in-memory state on the default registry (for testing)."""
    _default.reset()