Worktree sync - vcspull._internal.worktree_sync

Warning

Be careful with these! Internal APIs are not covered by version policies. They can break or be removed between minor versions!

If you need an internal API stabilized please file an issue.

Core worktree synchronization logic for vcspull.

class vcspull._internal.worktree_sync.WorktreeAction

class

Bases: Enum

Actions that can be taken on a worktree during sync.

CREATE = 'create'

attribute

Worktree doesn’t exist, will be created.

UPDATE = 'update'

attribute

Branch worktree exists, will pull latest.

UNCHANGED = 'unchanged'

attribute

Tag/commit worktree exists, already at target.

BLOCKED = 'blocked'

attribute

Worktree has uncommitted changes (safety).

ERROR = 'error'

attribute

Operation failed (ref not found, permission, etc.).

class vcspull._internal.worktree_sync.WorktreeCheck

class

Bases: object

A single check performed during worktree planning.

name: str

attribute

Check name: ‘validate_config’, ‘ref_exists’, ‘worktree_exists’, ‘is_dirty’.

passed: bool

attribute

Whether the check passed.

detail: str

attribute

Human-readable description of the check result.

exception: WorktreeError | None = None

attribute

Typed exception if the check failed.

__init__(name, passed, detail, exception=None)

method

Parameters:

• name (str)

• passed (bool)

• detail (str)

• exception (WorktreeError | None)

Return type:

None

class vcspull._internal.worktree_sync.WorktreePlanEntry

class

Bases: object

Planning information for a single worktree operation.

worktree_path: Path

attribute

Absolute path where the worktree will be/is located.

ref_type: str

attribute

Type of reference: ‘tag’, ‘branch’, or ‘commit’.

ref_value: str

attribute

The actual tag name, branch name, or commit SHA.

action: WorktreeAction

attribute

What action will be/was taken.

detail: str | None = None

attribute

Human-readable explanation of the action.

error: str | None = None

attribute

Error message if action is ERROR.

exists: bool = False

attribute

Whether the worktree currently exists.

is_dirty: bool = False

attribute

Whether the worktree has uncommitted changes.

current_ref: str | None = None

attribute

Current HEAD reference if worktree exists.

checks: list[WorktreeCheck]

attribute

Ordered audit trail of checks performed during planning.

__init__(worktree_path, ref_type, ref_value, action, detail=None, error=None, exists=False, is_dirty=False, current_ref=None, checks=<factory>)

method

Parameters:

• worktree_path (Path)

• ref_type (str)

• ref_value (str)

• action (WorktreeAction)

• detail (str | None)

• error (str | None)

• exists (bool)

• is_dirty (bool)

• current_ref (str | None)

• checks (list[WorktreeCheck])

Return type:

None

class vcspull._internal.worktree_sync.WorktreeSyncResult

class

Bases: object

Result of a worktree sync operation.

__init__(entries=<factory>, created=0, updated=0, unchanged=0, blocked=0, errors=0)

method

Parameters:

• entries (list[WorktreePlanEntry])

• created (int)

• updated (int)

• unchanged (int)

• blocked (int)

• errors (int)

Return type:

None

entries: list[WorktreePlanEntry]

attribute

List of worktree plan entries.

created: int = 0

attribute

Number of worktrees created.

updated: int = 0

attribute

Number of worktrees updated.

unchanged: int = 0

attribute

Number of worktrees left unchanged.

blocked: int = 0

attribute

Number of worktrees blocked due to dirty state.

errors: int = 0

attribute

Number of worktrees that encountered errors.

vcspull._internal.worktree_sync._get_ref_type_and_value(wt_config)

function

Extract the reference type and value from worktree config.

Returns:

Tuple of (ref_type, ref_value) or None if invalid config.

Return type:

tuple[str, str] | None

Parameters:

wt_config (WorktreeConfigDict)

Examples

>>> _get_ref_type_and_value({"dir": "../v1", "tag": "v1.0.0"})
('tag', 'v1.0.0')
>>> _get_ref_type_and_value({"dir": "../dev", "branch": "develop"})
('branch', 'develop')
>>> _get_ref_type_and_value({"dir": "../abc", "commit": "abc123"})
('commit', 'abc123')
>>> _get_ref_type_and_value({"dir": "../empty"}) is None
True
>>> multi = {"dir": "../multi", "tag": "v1", "branch": "main"}
>>> _get_ref_type_and_value(multi) is None
True
>>> _get_ref_type_and_value({"dir": "../wt", "tag": ""}) is None
True

vcspull._internal.worktree_sync.validate_worktree_config(wt_config)

function

Validate a worktree configuration dictionary.

Parameters:

wt_config (WorktreeConfigDict) – The worktree configuration to validate.

Raises:

WorktreeConfigError – If the configuration is invalid.

Return type:

None

Examples

>>> validate_worktree_config({"dir": "../v1", "tag": "v1.0.0"})
>>> validate_worktree_config({"dir": "../dev", "branch": "develop"})
>>> validate_worktree_config({"tag": "v1.0.0"})  # Missing dir
Traceback (most recent call last):
    ...
vcspull.exc.WorktreeConfigError: Worktree config missing required 'dir' field
>>> validate_worktree_config({"dir": "../proj"})  # No ref
Traceback (most recent call last):
    ...
vcspull.exc.WorktreeConfigError: Worktree config must specify one of: ...

vcspull._internal.worktree_sync._is_worktree_dirty(worktree_path)

function

Check if a worktree has uncommitted changes.

Returns True if dirty or if the check fails (fail-safe). This prevents destructive operations when the dirty state is unknown.

Parameters:

worktree_path (pathlib.Path) – Path to the worktree directory.

Returns:

True if the worktree has uncommitted changes or if the check fails.

Return type:

bool

Examples

>>> import pathlib
>>> _is_worktree_dirty(pathlib.Path("/nonexistent/path"))
True

vcspull._internal.worktree_sync._ref_exists(repo_path, ref, ref_type)

function

Check if a reference exists in the repository.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• ref (str) – The reference to check.

• ref_type (str) – Type of reference: ‘tag’, ‘branch’, or ‘commit’.

Returns:

True if the reference exists.

Return type:

bool

Examples

>>> import pathlib
>>> _ref_exists(pathlib.Path("/nonexistent/repo"), "v1.0.0", "tag")
False
>>> _ref_exists(pathlib.Path("/nonexistent/repo"), "main", "branch")
False

vcspull._internal.worktree_sync._get_worktree_head(worktree_path)

function

Get the current HEAD reference of a worktree.

Parameters:

worktree_path (pathlib.Path) – Path to the worktree.

Returns:

The HEAD reference or None if unable to determine.

Return type:

str | None

Examples

>>> import pathlib
>>> _get_worktree_head(pathlib.Path("/nonexistent/worktree")) is None
True

vcspull._internal.worktree_sync._worktree_exists(repo_path, worktree_path)

function

Check if a worktree is registered in the repository.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• worktree_path (pathlib.Path) – Path to check for worktree.

Returns:

True if the worktree exists and is registered.

Return type:

bool

Examples

>>> import pathlib
>>> _worktree_exists(pathlib.Path("/repo"), pathlib.Path("/nonexistent"))
False
>>> repo_dir = tmp_path / "repo"
>>> repo_dir.mkdir()
>>> _worktree_exists(repo_dir, tmp_path / "missing")
False

vcspull._internal.worktree_sync._resolve_worktree_path(wt_config, workspace_root)

function

Resolve the worktree path from config.

Parameters:

• wt_config (WorktreeConfigDict) – Worktree configuration.

• workspace_root (pathlib.Path) – The workspace root directory.

Returns:

Absolute path for the worktree.

Return type:

pathlib.Path

Examples

>>> import pathlib
>>> workspace = pathlib.Path("/home/user/code")
>>> wt = {"dir": "../sibling", "tag": "v1.0.0"}
>>> _resolve_worktree_path(wt, workspace)
PosixPath('/home/user/sibling')
>>> wt_abs = {"dir": "/tmp/worktree", "tag": "v1.0.0"}
>>> _resolve_worktree_path(wt_abs, workspace)
PosixPath('/tmp/worktree')

vcspull._internal.worktree_sync.plan_worktree_sync(repo_path, worktrees_config, workspace_root)

function

Plan worktree sync operations without executing them.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• worktrees_config (list[WorktreeConfigDict]) – List of worktree configurations.

• workspace_root (pathlib.Path) – The workspace root directory for resolving relative paths.

Returns:

List of planned operations.

Return type:

list[WorktreePlanEntry]

Examples

>>> import pathlib
>>> entries = plan_worktree_sync(
...     pathlib.Path("/nonexistent/repo"),
...     [{"dir": "../wt", "tag": "v1.0.0"}],
...     pathlib.Path("/nonexistent"),
... )
>>> len(entries)
1
>>> entries[0].action == WorktreeAction.ERROR
True

vcspull._internal.worktree_sync.sync_worktree(repo_path, wt_config, workspace_root, *, dry_run=False)

function

Synchronize a single worktree.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• wt_config (WorktreeConfigDict) – Worktree configuration.

• workspace_root (pathlib.Path) – The workspace root directory.

• dry_run (bool) – If True, only plan without executing.

Returns:

Result of the sync operation.

Return type:

WorktreePlanEntry

Examples

>>> import pathlib
>>> entry = sync_worktree(
...     pathlib.Path("/nonexistent/repo"),
...     {"dir": "../wt", "tag": "v1.0.0"},
...     pathlib.Path("/nonexistent"),
...     dry_run=True,
... )
>>> entry.action == WorktreeAction.ERROR
True

vcspull._internal.worktree_sync._create_worktree(repo_path, worktree_path, ref_type, ref_value, wt_config)

function

Create a new worktree.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• worktree_path (pathlib.Path) – Path for the new worktree.

• ref_type (str) – Type of reference: ‘tag’, ‘branch’, or ‘commit’.

• ref_value (str) – The reference value.

• wt_config (WorktreeConfigDict) – Full worktree configuration.

Raises:

• subprocess.CalledProcessError – If the git command fails (e.g., ref not found).

• FileNotFoundError – If the repository path does not exist.

Return type:

None

Examples

This function requires a valid git repository. When called with an invalid path, it raises FileNotFoundError:

>>> import pathlib
>>> _create_worktree(
...     pathlib.Path("/nonexistent"),
...     pathlib.Path("/tmp/wt"),
...     "tag",
...     "v1.0.0",
...     {"dir": "../wt", "tag": "v1.0.0"},
... )
Traceback (most recent call last):
    ...
FileNotFoundError: ...

vcspull._internal.worktree_sync._update_worktree(worktree_path, branch)

function

Update a branch worktree by pulling latest changes.

Verifies the worktree is on the expected branch before pulling. If the worktree is on a different branch, checks out the expected branch first.

Parameters:

• worktree_path (pathlib.Path) – Path to the worktree.

• branch (str) – The expected branch name.

Raises:

• subprocess.CalledProcessError – If the git command fails.

• FileNotFoundError – If the worktree path does not exist.

Return type:

None

Examples

This function requires a valid git worktree. When called with an invalid path, it raises FileNotFoundError:

>>> import pathlib
>>> _update_worktree(pathlib.Path("/nonexistent"), "main")
Traceback (most recent call last):
    ...
FileNotFoundError: ...

vcspull._internal.worktree_sync.sync_all_worktrees(repo_path, worktrees_config, workspace_root, *, dry_run=False)

function

Synchronize all worktrees for a repository.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• worktrees_config (list[WorktreeConfigDict]) – List of worktree configurations.

• workspace_root (pathlib.Path) – The workspace root directory.

• dry_run (bool) – If True, only plan without executing.

Returns:

Summary of all sync operations.

Return type:

WorktreeSyncResult

Examples

>>> import pathlib
>>> result = sync_all_worktrees(
...     pathlib.Path("/nonexistent/repo"),
...     [{"dir": "../wt", "tag": "v1.0.0"}],
...     pathlib.Path("/nonexistent"),
...     dry_run=True,
... )
>>> result.errors
1
>>> len(result.entries)
1

vcspull._internal.worktree_sync.list_existing_worktrees(repo_path)

function

List all existing worktrees for a repository.

Parameters:

repo_path (pathlib.Path) – Path to the main repository.

Returns:

List of worktree paths.

Return type:

list[pathlib.Path]

Examples

>>> import pathlib
>>> list_existing_worktrees(pathlib.Path("/nonexistent/repo"))
[]

vcspull._internal.worktree_sync.prune_worktrees(repo_path, config_worktrees, workspace_root, *, dry_run=False)

function

Remove worktrees that are not in the configuration.

Parameters:

• repo_path (pathlib.Path) – Path to the main repository.

• config_worktrees (list[WorktreeConfigDict]) – List of configured worktrees.

• workspace_root (pathlib.Path) – The workspace root directory.

• dry_run (bool) – If True, only report what would be pruned.

Returns:

List of worktree paths that were (or would be) pruned.

Return type:

list[pathlib.Path]

Examples

>>> import pathlib
>>> prune_worktrees(
...     pathlib.Path("/nonexistent/repo"),
...     [],
...     pathlib.Path("/nonexistent"),
...     dry_run=True,
... )
[]