mirror

mirror

Host-side git gate (mirror) management and upstream comparison.

The git gate is a bare git repository stored on the host. Its role varies with how the caller configures it:

• Upstream set, gatekeeping mode — the gate is a mirror of upstream and is the only repository the container can access, enforcing human review before changes reach upstream.
• Upstream set, online mode — the gate mirrors upstream and serves as a read-only clone accelerator (faster than cloning over the network).
• No upstream — sync() initialises the gate as a remoteless bare repo that the container can still push to. Nothing is fetched because there is no remote; subsequent syncs are no-ops.

GitGate is the main service class — wraps git CLI operations for syncing, comparing, and querying the gate.

All constructor parameters are plain values (strings, paths) — no terok-specific types like ProjectConfig.

Value types returned by GitGate methods:

• GateSyncResult — full sync outcome (created, updated branches, errors; upstream_url is None for remoteless gates)
• BranchSyncResult — selective branch sync outcome
• CommitInfo — single commit metadata (hash, date, author, message)
• GateStalenessInfo — frozen comparison of gate HEAD vs upstream HEAD

logger = logging.getLogger(__name__) module-attribute

GateAuthNotConfigured(scope)

Bases: RuntimeError

Raised when a scope has no vault key and personal-SSH fallback is not opted in.

Callers (the gate-sync CLI dispatch) turn this into a two-door remediation hint:

• generate a terok-managed key with terok ssh-init <project> and register it upstream, or
• opt in to the user's own ~/.ssh keys with --use-personal-ssh (or ssh.use_personal: true in the project YAML).

Source code in src/terok_sandbox/gate/mirror.py

def __init__(self, scope: str) -> None:
    self.scope = scope
    super().__init__(
        f"No SSH key is assigned to scope {scope!r} and personal-SSH "
        "fallback is not enabled.  Either run `terok ssh-init` to "
        "generate one, or pass --use-personal-ssh."
    )

scope = scope instance-attribute

GateSyncResult

Bases: TypedDict

Result of a full gate sync operation.

upstream_url is None when the gate is initialised without a remote — a local-only mirror that the container can push to but that never fetches external commits.

path instance-attribute

upstream_url instance-attribute

created instance-attribute

success instance-attribute

updated_branches instance-attribute

errors instance-attribute

cache_refreshed instance-attribute

BranchSyncResult

Bases: TypedDict

Result of a branch sync operation.

success instance-attribute

updated_branches instance-attribute

errors instance-attribute

CommitInfo

Bases: TypedDict

Information about a single git commit.

commit_hash instance-attribute

commit_date instance-attribute

commit_message instance-attribute

commit_author instance-attribute

GateStalenessInfo(branch, gate_head, upstream_head, is_stale, commits_behind, commits_ahead, last_checked, error) dataclass

Result of comparing gate vs upstream.

branch instance-attribute

gate_head instance-attribute

upstream_head instance-attribute

is_stale instance-attribute

commits_behind instance-attribute

commits_ahead instance-attribute

last_checked instance-attribute

error instance-attribute

GitGate(*, scope, gate_path, upstream_url=None, default_branch=None, use_personal_ssh=False, validate_gate_fn=None, clone_cache_base=None)

Repository + Gateway for a host-side git gate mirror.

Manages the bare git mirror that containers clone from. Provides operations for initial creation, incremental sync from upstream, selective branch fetching, and staleness detection.

Constructor takes plain parameters — no terok-specific types.

Initialise with plain parameters.

Parameters

scope: Credential scope for this gate's owner. Used to locate the per-scope vault SSH-agent socket. gate_path: Path to the bare git mirror on the host. upstream_url: Git upstream URL to sync from. default_branch: Branch name used for staleness comparisons. use_personal_ssh: When True, skip the vault socket entirely and let git fall through to the user's ~/.ssh keys / loaded agent. Default False — "terok never touches your real keys" is the advertised property. Opt in per-invocation (--use-personal-ssh) or per-project (ssh.use_personal: true in project YAML). validate_gate_fn: Optional callback (scope) -> None that validates no other scope uses the same gate with a different upstream. Injected by the orchestration layer; omitted for standalone use. clone_cache_base: Base directory for non-bare clone caches. When set, sync refreshes a working-tree cache at clone_cache_base / scope after updating the bare mirror. The cache accelerates task startup by enabling a host-side file copy instead of a full git clone.

Source code in src/terok_sandbox/gate/mirror.py

def __init__(
    self,
    *,
    scope: str,
    gate_path: Path | str,
    upstream_url: str | None = None,
    default_branch: str | None = None,
    use_personal_ssh: bool = False,
    validate_gate_fn: Callable[[str], None] | None = None,
    clone_cache_base: Path | str | None = None,
) -> None:
    """Initialise with plain parameters.

    Parameters
    ----------
    scope:
        Credential scope for this gate's owner.  Used to locate the
        per-scope vault SSH-agent socket.
    gate_path:
        Path to the bare git mirror on the host.
    upstream_url:
        Git upstream URL to sync from.
    default_branch:
        Branch name used for staleness comparisons.
    use_personal_ssh:
        When ``True``, skip the vault socket entirely and let git fall
        through to the user's ``~/.ssh`` keys / loaded agent.  Default
        ``False`` — "terok never touches your real keys" is the advertised
        property.  Opt in per-invocation (``--use-personal-ssh``) or
        per-project (``ssh.use_personal: true`` in project YAML).
    validate_gate_fn:
        Optional callback ``(scope) -> None`` that validates no other
        scope uses the same gate with a different upstream.  Injected by
        the orchestration layer; omitted for standalone use.
    clone_cache_base:
        Base directory for non-bare clone caches.  When set,
        [`sync`][terok_sandbox.gate.mirror.GitGate.sync] refreshes a working-tree cache at
        ``clone_cache_base / scope`` after updating the bare mirror.
        The cache accelerates task startup by enabling a host-side
        file copy instead of a full ``git clone``.
    """
    self._scope = scope
    self._gate_path = Path(gate_path)
    self._upstream_url = upstream_url
    self._default_branch = default_branch
    self._use_personal_ssh = use_personal_ssh
    self._validate_gate_fn = validate_gate_fn
    self._clone_cache_base = Path(clone_cache_base) if clone_cache_base else None
    self._signer: _EphemeralSigner | None = None

cache_path property

Clone cache directory for this scope, or None if caching is disabled.

close()

Stop the ephemeral signer this gate started, if any.

Idempotent. Long-lived processes (the TUI) should call this explicitly so the signer thread and temp socket don't outlive the gate's last use.

Source code in src/terok_sandbox/gate/mirror.py

def close(self) -> None:
    """Stop the ephemeral signer this gate started, if any.

    Idempotent.  Long-lived processes (the TUI) should call this
    explicitly so the signer thread and temp socket don't outlive
    the gate's last use.
    """
    if self._signer is not None:
        self._signer.stop()
        self._signer = None

__del__()

Best-effort signer teardown on GC.

Source code in src/terok_sandbox/gate/mirror.py

def __del__(self) -> None:
    """Best-effort signer teardown on GC."""
    with contextlib.suppress(Exception):  # __del__ never raises
        self.close()

sync(branches=None, force_reinit=False)

Sync the host-side git mirror gate from upstream.

With an upstream configured, clones (or fetches) from it using the project's SSH setup. Without one, initialises a bare repo in place and returns a no-op sync — the gate then serves as a local-only remote that the container can push to, giving the agent somewhere to stage commits even when there is nothing external to mirror.

A remoteless gate that already exists is a proper no-op: nothing re-initialises, and the returned branch list is empty.

Source code in src/terok_sandbox/gate/mirror.py

def sync(
    self,
    branches: list[str] | None = None,
    force_reinit: bool = False,
) -> GateSyncResult:
    """Sync the host-side git mirror gate from upstream.

    With an upstream configured, clones (or fetches) from it using the
    project's SSH setup.  Without one, initialises a bare repo in place
    and returns a no-op sync — the gate then serves as a local-only
    remote that the container can push to, giving the agent somewhere
    to stage commits even when there is nothing external to mirror.

    A remoteless gate that already exists is a proper no-op: nothing
    re-initialises, and the returned branch list is empty.
    """
    self._validate_gate()

    gate_dir = self._gate_path
    gate_exists = gate_dir.exists()
    gate_dir.parent.mkdir(parents=True, exist_ok=True)

    env = self._ssh_env()
    created = False
    if force_reinit and gate_exists:
        try:
            if gate_dir.is_dir():
                shutil.rmtree(gate_dir)
        except Exception as exc:
            logger.warning(f"Failed to remove gate dir {gate_dir}: {exc}")
        gate_exists = False

    if not gate_exists:
        if self._upstream_url:
            _clone_gate_mirror(self._upstream_url, gate_dir, env)
        else:
            _init_remoteless_gate(gate_dir)
        created = True

    # A remoteless gate has nothing to fetch — skip ``git remote update``
    # (which would fail on a repo with no origin) and the clone-cache
    # refresh (there is no bare mirror to track).
    if not self._upstream_url:
        return {
            "path": str(gate_dir),
            "upstream_url": None,
            "created": created,
            "success": True,
            "updated_branches": [],
            "errors": [],
            "cache_refreshed": False,
        }

    sync_result = self.sync_branches(branches)

    # Refresh the non-bare clone cache from the bare mirror (best-effort).
    cache_refreshed = False
    if sync_result["success"] and self._clone_cache_base:
        cache_refreshed = self._refresh_clone_cache()

    return {
        "path": str(gate_dir),
        "upstream_url": self._upstream_url,
        "created": created,
        "success": sync_result["success"],
        "updated_branches": sync_result["updated_branches"],
        "errors": sync_result["errors"],
        "cache_refreshed": cache_refreshed,
    }

sync_branches(branches=None)

Sync specific branches in the gate from upstream.

Parameters:

Name	Type	Description	Default
branches	list[str] | None	List of branches to sync (default: all via remote update)	None

Returns:

Type	Description
BranchSyncResult	Dict with keys: success, updated_branches, errors

Source code in src/terok_sandbox/gate/mirror.py

def sync_branches(self, branches: list[str] | None = None) -> BranchSyncResult:
    """Sync specific branches in the gate from upstream.

    Args:
        branches: List of branches to sync (default: all via remote update)

    Returns:
        Dict with keys: success, updated_branches, errors
    """
    gate_dir = self._gate_path

    if not gate_dir.exists():
        return {"success": False, "updated_branches": [], "errors": ["Gate not initialized"]}

    self._validate_gate()

    env = self._ssh_env()
    errors: list[str] = []
    updated: list[str] = []

    try:
        cmd = ["git", "-C", str(gate_dir), "remote", "update", "--prune"]
        result = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=120)  # nosec B603 — argv is a fixed list controlled by this module

        if result.returncode != 0:
            errors.append(f"remote update failed: {result.stderr}")
        else:
            updated = branches if branches else ["all"]

    except subprocess.TimeoutExpired:
        errors.append("Sync timed out")
    except Exception as e:
        errors.append(str(e))

    return {"success": len(errors) == 0, "updated_branches": updated, "errors": errors}

compare_vs_upstream(branch=None)

Compare gate HEAD vs upstream HEAD for a branch.

Parameters:

Name	Type	Description	Default
branch	str | None	Branch to compare (default: configured default_branch)	None

Returns:

Type	Description
GateStalenessInfo	GateStalenessInfo with comparison results

Source code in src/terok_sandbox/gate/mirror.py

def compare_vs_upstream(self, branch: str | None = None) -> GateStalenessInfo:
    """Compare gate HEAD vs upstream HEAD for a branch.

    Args:
        branch: Branch to compare (default: configured default_branch)

    Returns:
        GateStalenessInfo with comparison results
    """
    branch = branch or self._default_branch
    now = datetime.now().isoformat()

    if not branch:
        return GateStalenessInfo(
            branch=None,
            gate_head=None,
            upstream_head=None,
            is_stale=False,
            commits_behind=None,
            commits_ahead=None,
            last_checked=now,
            error="No branch configured",
        )

    env = self._ssh_env()

    # Get gate HEAD
    gate_head = _get_gate_branch_head(self._gate_path, branch, env)
    if gate_head is None:
        return GateStalenessInfo(
            branch=branch,
            gate_head=None,
            upstream_head=None,
            is_stale=False,
            commits_behind=None,
            commits_ahead=None,
            last_checked=now,
            error="Gate not initialized",
        )

    # Get upstream HEAD
    if not self._upstream_url:
        return GateStalenessInfo(
            branch=branch,
            gate_head=gate_head,
            upstream_head=None,
            is_stale=False,
            commits_behind=None,
            commits_ahead=None,
            last_checked=now,
            error="No upstream URL configured",
        )

    upstream_info = _get_upstream_head(self._upstream_url, branch, env)
    if upstream_info is None:
        return GateStalenessInfo(
            branch=branch,
            gate_head=gate_head,
            upstream_head=None,
            is_stale=False,
            commits_behind=None,
            commits_ahead=None,
            last_checked=now,
            error="Could not reach upstream",
        )

    upstream_head = upstream_info["commit_hash"]
    is_stale = gate_head != upstream_head

    commits_behind = None
    commits_ahead = None
    if is_stale:
        commits_behind = _count_commits_range(self._gate_path, gate_head, upstream_head, env)
        commits_ahead = _count_commits_range(self._gate_path, upstream_head, gate_head, env)

    return GateStalenessInfo(
        branch=branch,
        gate_head=gate_head,
        upstream_head=upstream_head,
        is_stale=is_stale,
        commits_behind=commits_behind if is_stale else 0,
        commits_ahead=commits_ahead if is_stale else 0,
        last_checked=now,
        error=None,
    )

last_commit()

Get information about the last commit on the configured branch.

Returns None if the gate doesn't exist or is not accessible.

Source code in src/terok_sandbox/gate/mirror.py

def last_commit(self) -> CommitInfo | None:
    """Get information about the last commit on the configured branch.

    Returns ``None`` if the gate doesn't exist or is not accessible.
    """
    try:
        gate_dir = self._gate_path

        if not gate_dir.exists() or not gate_dir.is_dir():
            return None

        env = self._ssh_env()

        rev = f"refs/heads/{self._default_branch}" if self._default_branch else "HEAD"
        cmd = [
            "git",
            "-C",
            str(gate_dir),
            "log",
            "-1",
            rev,
            "--pretty=format:%H%x00%ad%x00%an%x00%s",
            "--date=iso",
        ]

        result = subprocess.run(cmd, capture_output=True, text=True, env=env)  # nosec B603 — argv is a fixed list controlled by this module
        if result.returncode != 0 and self._default_branch:
            cmd[5] = "HEAD"
            result = subprocess.run(cmd, capture_output=True, text=True, env=env)  # nosec B603 — argv is a fixed list controlled by this module
        if result.returncode != 0:
            return None

        parts = result.stdout.strip().split("\x00", 3)
        if len(parts) == 4:
            return {
                "commit_hash": parts[0],
                "commit_date": parts[1],
                "commit_author": parts[2],
                "commit_message": parts[3],
            }
        return None

    except Exception:
        return None

is_ssh_url(url)

Return True for SSH-scheme git URLs.

Accepts the two forms git itself accepts:

• ssh://[user@]host[:port]/path — explicit URL scheme.
• [user@]host:path — scp-style shorthand. The user part is optional (git@github.com:foo.git, deploy@host:repo.git, bare github.com:foo.git).

Shared with terok-main: both the gate's env builder and callers that branch on "does this project use SSH?" (e.g. deploy-key prompts, gate-sync fallback hints) must agree on one definition.

Source code in src/terok_sandbox/gate/mirror.py

def is_ssh_url(url: str | None) -> bool:
    """Return ``True`` for SSH-scheme git URLs.

    Accepts the two forms git itself accepts:

    - ``ssh://[user@]host[:port]/path`` — explicit URL scheme.
    - ``[user@]host:path`` — scp-style shorthand.  The user part is
      optional (``git@github.com:foo.git``, ``deploy@host:repo.git``,
      bare ``github.com:foo.git``).

    Shared with terok-main: both the gate's env builder and callers that
    branch on "does this project use SSH?" (e.g. deploy-key prompts,
    gate-sync fallback hints) must agree on one definition.
    """
    if not url:
        return False
    candidate = url.strip()
    lowered = candidate.lower()
    if lowered.startswith("ssh://"):
        return True
    if "://" in candidate:
        return False
    return bool(_SCP_SSH_RE.match(candidate))