Skip to content

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 upstreamsync() 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.

Ref model

The gate keeps three ref namespaces, and the split is what makes sync safe:

  • refs/heads/* — the container-facing view. Sync on its own only ever creates branches or fast-forwards them; every other change (delete, non-fast-forward move) is returned as a pending op and applied only via apply_pending_ops after the operator confirmed it. Agent work pushed to the gate but not yet upstream therefore survives any number of syncs.
  • refs/terok/upstream/* — a private snapshot of upstream's heads, force-updated and pruned freely on every fetch. It is the "what did upstream have last time" memory that classifies each branch: a gate head equal to its snapshot entry has no gate-local work (a destructive op on it is lossless), anything else diverged locally (lossy).
  • refs/terok/backup/<branch>/<stamp>-<sha12> — the old tip of every destructively changed branch, written before the change so nothing ever becomes unreachable. Expired per the retention policy by prune_backups.
  • refs/terok/attic/<branch> — the last upstream tip a branch had before it went pending (deleted upstream, or rewritten while the gate head stayed behind). The pruning, force-updating snapshot forgets that tip immediately, but it is exactly what proves a pending op lossless — so the attic keeps it until the branch is resolved (op applied, or gate and upstream agree again). First writer wins: across repeated upstream rewrites the attic still names the last tip the gate was actually in sync with.

The whole refs/terok namespace is hidden (transfer.hideRefs), so containers can neither see nor overwrite the snapshot, attic, or backups — a container that could push the snapshot could forge "lossless".

Value types returned by GitGate methods:

  • GateSyncResult — full sync outcome: applied ops, pending destructive ops, kept gate-only branches (upstream_url is None for remoteless gates)
  • AppliedOp / PendingOp — one branch-level ref change, performed or awaiting confirmation
  • ApplyPendingResult — outcome of applying confirmed pending ops
  • BackupRef — one parsed backup ref
  • CommitInfo — single commit metadata (hash, date, author, message)
  • GateStalenessInfo — frozen comparison of gate HEAD vs upstream HEAD

logger = logging.getLogger(__name__) module-attribute

OpKind = Literal['create', 'fast_forward', 'force_update', 'delete'] module-attribute

Branch-level ref change kinds. create/fast_forward are the only kinds sync applies on its own; force_update/delete only ever appear as pending ops.

PendingReason = Literal['upstream_rewrite', 'upstream_delete', 'unknown_provenance'] module-attribute

Why a destructive op is proposed. unknown_provenance marks branches found during migration from a pre-snapshot mirror gate — upstream doesn't have them now, but whether it ever did is unknowable, so they are surfaced once and otherwise left alone.

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

AppliedOp

Bases: TypedDict

One branch-level ref change sync (or an approved apply) performed.

branch instance-attribute

kind instance-attribute

old_sha instance-attribute

new_sha instance-attribute

PendingOp

Bases: TypedDict

A destructive branch change awaiting operator confirmation.

lossless is the heart of the safety story: True means the gate tip equals what upstream last advertised for this branch — no agent commits would be discarded. gate_only_commits quantifies the lossy case (None when provenance is unknown and the count would be meaningless). gate_sha doubles as the compare-and-swap guard in apply_pending_ops: an op is refused if the branch moved since it was proposed.

branch instance-attribute

kind instance-attribute

reason instance-attribute

gate_sha instance-attribute

upstream_sha instance-attribute

old_snapshot_sha instance-attribute

lossless instance-attribute

gate_only_commits instance-attribute

GateSyncResult

Bases: TypedDict

Result of a 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.

pending ops are proposals, not failures — a sync that fetched cleanly and applied its safe ops reports success: True regardless of how much destructive work awaits confirmation. notes carries non-fatal observations (moved tags, expired backups, foreign refs).

The clone-cache refresh is best-effort: cache_error carries the failure description when the refresh was attempted and failed, so callers can report it instead of silently claiming a clean sync. cache_refreshed stays False both on failure and when no cache is configured; cache_error distinguishes the two.

path instance-attribute

upstream_url instance-attribute

created instance-attribute

migrated instance-attribute

success instance-attribute

errors instance-attribute

notes instance-attribute

applied instance-attribute

pending instance-attribute

gate_only_branches instance-attribute

cache_refreshed instance-attribute

cache_error instance-attribute

ApplyPendingResult

Bases: TypedDict

Result of applying operator-confirmed pending ops.

Ops whose branch moved between proposal and apply land in errors and leave the branch untouched — the rest are still applied, so a single race never voids a whole confirmation. backups maps each changed branch to the backup ref holding its previous tip (empty when backups are disabled).

success instance-attribute

applied instance-attribute

backups instance-attribute

errors instance-attribute

BackupRef

Bases: TypedDict

One parsed refs/terok/backup/… entry.

ref instance-attribute

branch instance-attribute

saved_at instance-attribute

sha 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, backups_enabled=True, backup_retention_days=30)

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. backups_enabled: When True (default), every destructive branch change applied via apply_pending_ops first saves the old tip under refs/terok/backup/. Opt out per project when the reflog alone is protection enough. backup_retention_days: Backups older than this are expired by prune_backups (which a successful sync runs automatically). 0 keeps backups forever.

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,
    backups_enabled: bool = True,
    backup_retention_days: int = 30,
) -> 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``.
    backups_enabled:
        When ``True`` (default), every destructive branch change applied
        via [`apply_pending_ops`][terok_sandbox.gate.mirror.GitGate.apply_pending_ops]
        first saves the old tip under ``refs/terok/backup/``.  Opt out
        per project when the reflog alone is protection enough.
    backup_retention_days:
        Backups older than this are expired by
        [`prune_backups`][terok_sandbox.gate.mirror.GitGate.prune_backups]
        (which a successful sync runs automatically).  ``0`` keeps
        backups forever.
    """
    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._backups_enabled = backups_enabled
    self._backup_retention_days = backup_retention_days
    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 gate from upstream, destroying nothing.

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 op lists are empty.

Only safe branch changes are applied here — creates and fast-forwards. Deletes and non-fast-forward moves come back as pending proposals for apply_pending_ops; branches that exist only on the gate (agent work not yet upstream) are listed in gate_only_branches and never touched.

branches restricts the sync to the named branches (the auto-sync allowlist); the default full sync covers everything upstream has.

force_reinit recreates the whole local footprint — the bare mirror and the clone cache — so a hopeless state can always be recovered with one flag. Deletion failures propagate: rebuilding over stale or partial data would silently defeat the point of a from-scratch recovery. This is the one path that still discards gate-local work, which is exactly why it hides behind an explicit operator flag and never runs implicitly.

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 gate from upstream, destroying nothing.

    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 op lists are empty.

    Only safe branch changes are applied here — creates and
    fast-forwards.  Deletes and non-fast-forward moves come back as
    ``pending`` proposals for
    [`apply_pending_ops`][terok_sandbox.gate.mirror.GitGate.apply_pending_ops];
    branches that exist only on the gate (agent work not yet upstream)
    are listed in ``gate_only_branches`` and never touched.

    *branches* restricts the sync to the named branches (the auto-sync
    allowlist); the default full sync covers everything upstream has.

    ``force_reinit`` recreates the whole local footprint — the bare
    mirror *and* the clone cache — so a hopeless state can always be
    recovered with one flag.  Deletion failures propagate: rebuilding
    over stale or partial data would silently defeat the point of a
    from-scratch recovery.  This is the one path that still discards
    gate-local work, which is exactly why it hides behind an explicit
    operator flag and never runs implicitly.
    """
    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:
        if gate_exists:
            shutil.rmtree(gate_dir)
            gate_exists = False
        if self.cache_path is not None and self.cache_path.exists():
            shutil.rmtree(self.cache_path)

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

    # A remoteless gate has nothing to fetch — skip the upstream fetch
    # (which would fail on a repo with no origin) and the clone-cache
    # refresh (there is no upstream state to track).
    if not self._upstream_url:
        return _empty_sync_result(str(gate_dir), created=created)

    migrated, errors, notes, applied, pending, gate_only = self._sync_from_upstream(
        env, branches, freshly_created=created
    )
    success = not errors

    # A dangling gate HEAD breaks every fresh clone of the gate, so a
    # failed heal is a sync failure, not a footnote.
    if success and (head_error := self._align_gate_head(env)):
        errors.append(head_error)
        success = False

    if success and (expired := self.prune_backups()):
        notes.append(f"expired {len(expired)} backup ref(s) past retention")

    # Refresh the non-bare clone cache from the bare mirror (best-effort).
    cache_error: str | None = None
    cache_refreshed = False
    if success and self._clone_cache_base:
        cache_error = self._refresh_clone_cache()
        cache_refreshed = cache_error is None

    return {
        "path": str(gate_dir),
        "upstream_url": self._upstream_url,
        "created": created,
        "migrated": migrated,
        "success": success,
        "errors": errors,
        "notes": notes,
        "applied": applied,
        "pending": pending,
        "gate_only_branches": gate_only,
        "cache_refreshed": cache_refreshed,
        "cache_error": cache_error,
    }

pending_ops()

Recompute the pending destructive ops without touching the network.

Compares the current gate heads against the snapshot and attic — state the last sync left behind — so TUI badges and confirmation dialogs can refresh cheaply (no fetch, no SSH signer). The one thing this cannot see is the one-shot unknown_provenance batch a migration sync reports.

Source code in src/terok_sandbox/gate/mirror.py
def pending_ops(self) -> list[PendingOp]:
    """Recompute the pending destructive ops without touching the network.

    Compares the current gate heads against the snapshot and attic —
    state the last sync left behind — so TUI badges and confirmation
    dialogs can refresh cheaply (no fetch, no SSH signer).  The one
    thing this cannot see is the one-shot ``unknown_provenance`` batch
    a migration sync reports.
    """
    if not self._gate_path.exists():
        return []
    snapshot = _read_refs(self._gate_path, _SNAPSHOT_PREFIX)
    attic = _read_refs(self._gate_path, _ATTIC_PREFIX)
    heads = _read_refs(self._gate_path, _HEADS_PREFIX)

    pending: list[PendingOp] = []
    for branch, gate_sha in sorted(heads.items()):
        if (upstream_sha := snapshot.get(branch)) is not None:
            if gate_sha != upstream_sha and not self._is_ancestor(gate_sha, upstream_sha):
                pending.append(
                    self._pending(
                        branch,
                        "force_update",
                        "upstream_rewrite",
                        gate_sha=gate_sha,
                        upstream_sha=upstream_sha,
                        old_sha=attic.get(branch),
                    )
                )
        elif (attic_sha := attic.get(branch)) is not None:
            pending.append(
                self._pending(
                    branch,
                    "delete",
                    "upstream_delete",
                    gate_sha=gate_sha,
                    upstream_sha=None,
                    old_sha=attic_sha,
                )
            )
    return pending

apply_pending_ops(ops)

Apply operator-confirmed destructive ops, backing up every old tip.

Each op is guarded by compare-and-swap on the gate_sha it was proposed against: a branch that moved since (an agent pushed) fails that op only and stays untouched — confirmations never apply to state the operator didn't see. Unless backups are disabled, the old tip is first saved under refs/terok/backup/ so even an approved mistake is one update-ref away from recovery.

Finishes with the same HEAD-alignment and clone-cache refresh a sync performs — deleting the branch HEAD points at is precisely when healing matters.

Source code in src/terok_sandbox/gate/mirror.py
def apply_pending_ops(self, ops: list[PendingOp]) -> ApplyPendingResult:
    """Apply operator-confirmed destructive ops, backing up every old tip.

    Each op is guarded by compare-and-swap on the ``gate_sha`` it was
    proposed against: a branch that moved since (an agent pushed) fails
    *that op only* and stays untouched — confirmations never apply to
    state the operator didn't see.  Unless backups are disabled, the
    old tip is first saved under ``refs/terok/backup/`` so even an
    approved mistake is one ``update-ref`` away from recovery.

    Finishes with the same HEAD-alignment and clone-cache refresh a
    sync performs — deleting the branch HEAD points at is precisely
    when healing matters.
    """
    applied: list[AppliedOp] = []
    backups: dict[str, str] = {}
    errors: list[str] = []

    for op in ops:
        backup_ref, error = self._apply_one_pending_op(op)
        if error is not None:
            errors.append(error)
            continue
        if backup_ref is not None:
            backups[op["branch"]] = backup_ref
        applied.append(
            {
                "branch": op["branch"],
                "kind": op["kind"],
                "old_sha": op["gate_sha"],
                "new_sha": op["upstream_sha"],
            }
        )

    if applied:
        env = self._ssh_env()
        if head_error := self._align_gate_head(env):
            errors.append(head_error)
        elif self._clone_cache_base and (cache_error := self._refresh_clone_cache()):
            errors.append(f"clone cache refresh failed: {cache_error}")

    return {"success": not errors, "applied": applied, "backups": backups, "errors": errors}

list_backups()

Return all backup refs, newest first, parsed from their names.

Source code in src/terok_sandbox/gate/mirror.py
def list_backups(self) -> list[BackupRef]:
    """Return all backup refs, newest first, parsed from their names."""
    entries: list[BackupRef] = []
    for path, sha in _read_refs(self._gate_path, _BACKUP_PREFIX).items():
        branch, _, leaf = path.rpartition("/")
        if not branch or not (m := _BACKUP_STAMP_RE.match(leaf)):
            continue
        saved_at = datetime.strptime(m.group(1), _BACKUP_STAMP_FORMAT).replace(tzinfo=UTC)
        entries.append(
            {
                "ref": f"{_BACKUP_PREFIX}{path}",
                "branch": branch,
                "saved_at": saved_at.isoformat(),
                "sha": sha,
            }
        )
    return sorted(entries, key=lambda e: e["saved_at"], reverse=True)

prune_backups(older_than_days=None)

Delete backup refs older than the retention window; return them.

Age is the ref-name timestamp — when the backup was taken, the only clock that matters for "have I had a chance to notice". With retention 0 (or a nonexistent gate) nothing is ever expired.

Source code in src/terok_sandbox/gate/mirror.py
def prune_backups(self, older_than_days: int | None = None) -> list[str]:
    """Delete backup refs older than the retention window; return them.

    Age is the ref-name timestamp — when the backup was *taken*, the
    only clock that matters for "have I had a chance to notice".  With
    retention ``0`` (or a nonexistent gate) nothing is ever expired.
    """
    days = self._backup_retention_days if older_than_days is None else older_than_days
    if days <= 0 or not self._gate_path.exists():
        return []
    cutoff = datetime.now(UTC) - timedelta(days=days)
    expired = [
        entry["ref"]
        for entry in self.list_backups()
        if datetime.fromisoformat(entry["saved_at"]) < cutoff
    ]
    for ref in expired:
        _git(self._gate_path, "update-ref", "-d", ref)
    return expired

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))