git_gate

git_gate

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

The git gate is a bare mirror of the upstream repository stored on the host. In gatekeeping mode, it is the only repository the container can access, enforcing human review before changes reach upstream. In online mode, it serves as a read-only clone accelerator (faster than cloning over the network).

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

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

Value types returned by GitGate methods:

• :class:GateSyncResult — full sync outcome (created, updated branches, errors)
• :class:BranchSyncResult — selective branch sync outcome
• :class:CommitInfo — single commit metadata (hash, date, author, message)
• :class:GateStalenessInfo — frozen comparison of gate HEAD vs upstream HEAD

GateSyncResult

Bases: TypedDict

Result of a full gate sync operation.

BranchSyncResult

Bases: TypedDict

Result of a branch sync operation.

CommitInfo

Bases: TypedDict

Information about a single git commit.

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

Result of comparing gate vs upstream.

GitGate(*, project_id, gate_path, upstream_url=None, default_branch=None, ssh_host_dir=None, ssh_key_name=None, validate_gate_fn=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

project_id: Identifier for this gate's owner. 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. ssh_host_dir: Explicit SSH directory for git operations. When None, falls back to SandboxConfig().ssh_keys_dir / project_id. ssh_key_name: Explicit SSH key filename. validate_gate_fn: Optional callback (project_id) -> None that validates no other project uses the same gate with a different upstream. Injected by the orchestration layer; omitted for standalone use.

Source code in src/terok_sandbox/git_gate.py

def __init__(
    self,
    *,
    project_id: str,
    gate_path: Path | str,
    upstream_url: str | None = None,
    default_branch: str | None = None,
    ssh_host_dir: Path | str | None = None,
    ssh_key_name: str | None = None,
    validate_gate_fn: Callable[[str], None] | None = None,
) -> None:
    """Initialise with plain parameters.

    Parameters
    ----------
    project_id:
        Identifier for this gate's owner.
    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.
    ssh_host_dir:
        Explicit SSH directory for git operations.  When ``None``,
        falls back to ``SandboxConfig().ssh_keys_dir / project_id``.
    ssh_key_name:
        Explicit SSH key filename.
    validate_gate_fn:
        Optional callback ``(project_id) -> None`` that validates no other
        project uses the same gate with a different upstream.  Injected by
        the orchestration layer; omitted for standalone use.
    """
    self._project_id = project_id
    self._gate_path = Path(gate_path)
    self._upstream_url = upstream_url
    self._default_branch = default_branch
    self._ssh_host_dir = Path(ssh_host_dir) if ssh_host_dir else None
    self._ssh_key_name = ssh_key_name
    self._validate_gate_fn = validate_gate_fn

sync(branches=None, force_reinit=False)

Sync the host-side git mirror gate.

• Uses SSH configuration via GIT_SSH_COMMAND.
• If gate doesn't exist (or force_reinit), performs a fresh git clone --mirror.
• Always runs the sync logic afterward for consistent side effects.

Returns:

Type	Description
GateSyncResult	Dict with keys: path, upstream_url, created (bool), success,
GateSyncResult	updated_branches, errors.

Source code in src/terok_sandbox/git_gate.py

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

    - Uses SSH configuration via GIT_SSH_COMMAND.
    - If gate doesn't exist (or *force_reinit*), performs a fresh ``git clone --mirror``.
    - Always runs the sync logic afterward for consistent side effects.

    Returns:
        Dict with keys: path, upstream_url, created (bool), success,
        updated_branches, errors.
    """
    if not self._upstream_url:
        raise SystemExit("Project has no git.upstream_url configured")

    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:
        _clone_gate_mirror(self._upstream_url, gate_dir, env)
        created = True

    sync_result = self.sync_branches(branches)
    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"],
    }

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/git_gate.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)

        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/git_gate.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/git_gate.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)
        if result.returncode != 0 and self._default_branch:
            cmd[5] = "HEAD"
            result = subprocess.run(cmd, capture_output=True, text=True, env=env)
        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