Filesharing Clone Git Remote Task

task api name: git_clone

The git_clone task shallow-fetches a registered remote's branch and makes its checked-out files available to NorFab workers at the configured nf:// mount.

Inputs

Parameter	Required	Description
name	Yes	Name previously registered by create_remote_git

Output

The result contains the name, synchronization status, and UTC attempt time:

{
  "name": "automation-assets",
  "status": "cloned",
  "last_sync_attempt": "2026-08-29T10:30:00+00:00"
}

Status is cloned when content is made available and unchanged when the fetched commit matches the previous local HEAD and the mount exists. Failures return failed and preserve the previous shared snapshot.

Examples

CLIPython

nf# filesharing git clone-remote name automation-assets

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()
    result = client.run_job(
        service="filesharing",
        task="git_clone",
        workers="filesharing-worker-1",
        kwargs={"name": "automation-assets"},
    )
    print(result)
finally:
    nf.destroy()

Notes

• Create or configure the remote in inventory before cloning it.
• Public HTTPS remotes need no credentials.
• Authenticated remotes use the username and password/token registered at creation.
• A per-remote lock prevents clone, create, and delete operations from overlapping.
• .git and symbolic links are not made available.

Task Command Shell Reference

nf#man tree filesharing.git.clone-remote

R - required field, M - supports multiline input, D - dynamic key

root
└── filesharing:    File sharing service
    └── git:    Manage Git remotes
        └── clone-remote:    Synchronize a Git remote and make it available
            ├── timeout:    Job timeout
            ├── workers:    Filter workers to target, default 'all'
            ├── verbose-result:    Control output details, default 'False'
            ├── nowait:    Do not wait for job to complete, default 'False'
            └── name (R):    Configured remote name
nf#

Python API Reference

Synchronize content from an initialized Git remote and make it available.

The task shallow-fetches the configured branch, checks out the fetched commit, and compares it with the current local Git HEAD. Changed working-tree content is copied into a staging snapshot with .git and symbolic links excluded, then atomically made available beneath the File Sharing base directory. Public and HTTP Basic authenticated remotes are supported.

Parameters:

Name	Type	Description	Default
job	Job	Active worker job. Scheduler calls pass None and skip the operation when another task holds the remote lock.	required
name	str	Name of a Git remote initialized by create_remote_git.	required

Returns:

Type	Description
Result	A result containing the updated synchronization state. Successful
Result	results use cloned when a snapshot is made available and
Result	unchanged when Git reports the same commit and the mount exists.

Source code in norfab\workers\filesharing_worker\git_tasks.py

@Task(
    input=RemoteNameInput,
    output=GitCloneResult,
    fastapi={"methods": ["POST"]},
    agent={"enabled": False},
    mcp={
        "annotations": {
            "title": "Clone Git Remote",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def git_clone(self, job: Job, name: str) -> Result:
    """Synchronize content from an initialized Git remote and make it available.

    The task shallow-fetches the configured branch, checks out the fetched
    commit, and compares it with the current local Git HEAD. Changed
    working-tree content is copied into a staging snapshot with ``.git``
    and symbolic links excluded, then atomically made available beneath the File
    Sharing base directory. Public and HTTP Basic authenticated remotes are
    supported.

    Args:
        job: Active worker job. Scheduler calls pass ``None`` and skip the
            operation when another task holds the remote lock.
        name: Name of a Git remote initialized by ``create_remote_git``.

    Returns:
        A result containing the updated synchronization state. Successful
        results use ``cloned`` when a snapshot is made available and
        ``unchanged`` when Git reports the same commit and the mount exists.
    """
    msg = f"{self.name} - synchronizing Git remote '{name}'"
    log.info(msg)
    if job is not None:
        job.event(msg)

    if name not in self.remotes:
        msg = f"{self.name} - remote '{name}' is not configured"
        log.warning(msg)
        if job is not None:
            job.event(msg, severity="ERROR")
        return Result(failed=True, errors=[f"Remote '{name}' is not configured"])

    remote = self.remotes[name]
    if remote["type"] != "git":
        msg = f"{self.name} - remote '{name}' is not a git remote"
        log.warning(msg)
        if job is not None:
            job.event(msg, severity="ERROR")
        return Result(failed=True, errors=[f"Remote '{name}' is not a git remote"])

    lock = remote["lock"]
    if job is None:
        acquired = lock.acquire(blocking=False)
    else:
        acquired = lock.acquire(
            timeout=job.timeout if job.timeout is not None else -1
        )
    if not acquired:
        msg = f"{self.name} - remote '{name}' is busy"
        log.warning(msg)
        if job is not None:
            job.event(msg, severity="WARNING")
        return Result(failed=True, errors=[f"Remote '{name}' is busy"])

    remote_dir = os.path.dirname(remote["repository"])
    mount_dir = self._safe_path(remote["mount"])
    staging_dir = os.path.join(remote_dir, "staging")
    snapshot_dir = os.path.join(staging_dir, "snapshot")
    error = "Remote synchronization failed"

    try:
        # Staging is recreated for each sync so an incomplete snapshot is
        # never visible from the configured mount.
        shutil.rmtree(staging_dir, ignore_errors=True)
        if not os.path.isdir(os.path.join(remote["repository"], ".git")):
            raise ValueError("Local Git repository is not initialized")

        msg = (
            f"{self.name} - fetching branch '{remote['branch']}' for remote "
            f"'{name}'"
        )
        log.info(msg)
        if job is not None:
            job.event(msg)
        with Repo(remote["repository"]) as repository:
            previous_sha = (
                repository.head.commit.hexsha
                if repository.head.is_valid()
                else None
            )
            if remote["username"] is not None:
                credentials = base64.b64encode(
                    f"{remote['username']}:{remote['password']}".encode()
                ).decode()
                git_environment = {
                    "GIT_CONFIG_COUNT": "1",
                    "GIT_CONFIG_KEY_0": "http.extraHeader",
                    "GIT_CONFIG_VALUE_0": f"Authorization: Basic {credentials}",
                }
            else:
                git_environment = {}
            fetched = repository.remotes.origin.fetch(
                remote["branch"], depth=1, env=git_environment
            )
            sha = fetched[0].commit.hexsha
            repository.git.checkout("-B", remote["branch"], sha)

        changed = previous_sha != sha or not os.path.isdir(mount_dir)
        if changed:
            msg = f"{self.name} - making Git remote '{name}' available at {remote['mount']}"
            log.info(msg)
            if job is not None:
                job.event(msg)

            # Copy only regular repository files; the Git metadata and
            # symbolic links must not enter the shared namespace.
            os.makedirs(snapshot_dir, exist_ok=False)
            for root, directories, files in os.walk(
                remote["repository"], topdown=True
            ):
                relative_root = os.path.relpath(root, remote["repository"]).replace(
                    os.sep, "/"
                )
                if relative_root == ".":
                    relative_root = ""

                for directory in directories[:]:
                    path = os.path.join(root, directory)
                    if os.path.islink(path):
                        raise ValueError("Git repository contains a symbolic link")
                    if directory == ".git":
                        directories.remove(directory)

                for filename in files:
                    relative_path = "/".join(
                        part for part in [relative_root, filename] if part
                    )
                    source = os.path.join(root, filename)
                    if os.path.islink(source):
                        raise ValueError("Git repository contains a symbolic link")
                    destination = os.path.join(
                        snapshot_dir, *relative_path.split("/")
                    )
                    os.makedirs(os.path.dirname(destination), exist_ok=True)
                    shutil.copy2(source, destination)

            # Swap complete directories so readers see either the old or
            # the new snapshot, never a partially copied one.
            previous_dir = os.path.join(staging_dir, "previous")
            os.makedirs(os.path.dirname(mount_dir), exist_ok=True)
            if os.path.exists(mount_dir):
                os.replace(mount_dir, previous_dir)
            try:
                os.replace(snapshot_dir, mount_dir)
            except Exception:
                if os.path.exists(previous_dir):
                    os.replace(previous_dir, mount_dir)
                raise

        remote["status"] = "cloned" if changed else "unchanged"
    except Exception as exc:
        remote["status"] = "failed"
        error = f"{type(exc).__name__}: remote synchronization failed"
        msg = (
            f"{self.name} - Git remote '{name}' synchronization failed with "
            f"{type(exc).__name__}"
        )
        log.error(msg)
        if job is not None:
            job.event(msg, severity="ERROR")
    finally:
        remote["last_sync_attempt"] = datetime.now(timezone.utc).isoformat()
        remote["last_sync_timer"] = time.monotonic()
        shutil.rmtree(staging_dir, ignore_errors=True)
        lock.release()

    result = {
        "name": remote["name"],
        "status": remote["status"],
        "last_sync_attempt": remote["last_sync_attempt"],
    }
    if remote["status"] == "failed":
        return Result(
            result=result,
            failed=True,
            errors=[error],
        )
    msg = (
        f"{self.name} - Git remote '{name}' synchronization completed with status "
        f"'{remote['status']}'"
    )
    log.info(msg)
    if job is not None:
        job.event(msg)
    return Result(result=result)