reset_graph

View module diagram

Reset a repository to its pre-graph-run state.

Usage::

python -m agentrelay.reset_graph graphs/demo.yaml
python -m agentrelay.reset_graph graphs/demo.yaml --yes

Reads .workflow/<graph>/run_info.json (written by :mod:run_graph) to determine the starting HEAD, then:

1. Closes open PRs on graph branches.
2. Resets main to the starting HEAD and force-pushes (if safe).
3. Deletes remote and local branches created by the graph run.
4. Removes worktree directories.
5. Removes the .workflow/<graph>/ directory.

Idempotent: re-running after a successful reset is safe. Out-of-order resets (resetting an older graph while a newer one's commits remain) are detected and the main-branch reset is skipped to avoid history corruption.

ResetPlan dataclass

Describes the operations a reset will perform.

Attributes:

Name	Type	Description
graph_name	str	Name of the graph being reset.
repo_path	Path	Path to the repository root.
start_head	str	SHA to reset main to.
started_at	str	ISO timestamp of the original run.
branch_prefix	str	Remote branch prefix to match.
worktree_dir	Path	Directory containing workstream worktrees.
workflow_dir	Path	Signal/run-info directory.
can_reset_main	bool	Whether the main-branch reset is safe.
open_prs	list[dict[str, Any]]	Open PRs that will be closed.
remote_branches	list[str]	Remote branches that will be deleted.
log	list[str]	Messages emitted during planning.

Source code in src/agentrelay/reset_graph.py

@dataclass
class ResetPlan:
    """Describes the operations a reset will perform.

    Attributes:
        graph_name: Name of the graph being reset.
        repo_path: Path to the repository root.
        start_head: SHA to reset main to.
        started_at: ISO timestamp of the original run.
        branch_prefix: Remote branch prefix to match.
        worktree_dir: Directory containing workstream worktrees.
        workflow_dir: Signal/run-info directory.
        can_reset_main: Whether the main-branch reset is safe.
        open_prs: Open PRs that will be closed.
        remote_branches: Remote branches that will be deleted.
        log: Messages emitted during planning.
    """

    graph_name: str
    repo_path: Path
    start_head: str
    started_at: str
    branch_prefix: str
    worktree_dir: Path
    workflow_dir: Path
    can_reset_main: bool
    open_prs: list[dict[str, Any]] = field(default_factory=list)
    remote_branches: list[str] = field(default_factory=list)
    log: list[str] = field(default_factory=list)

plan_reset(graph_name, repo_path)

Build a reset plan by inspecting repository and GitHub state.

Does not modify anything — only reads.

Parameters:

Name	Type	Description	Default
graph_name	str	Name of the graph to reset.	required
repo_path	Path	Path to the repository root.	required

Returns:

Type	Description
ResetPlan	ResetPlan describing what the reset will do.

Source code in src/agentrelay/reset_graph.py

def plan_reset(
    graph_name: str,
    repo_path: Path,
) -> ResetPlan:
    """Build a reset plan by inspecting repository and GitHub state.

    Does not modify anything — only reads.

    Args:
        graph_name: Name of the graph to reset.
        repo_path: Path to the repository root.

    Returns:
        ResetPlan describing what the reset will do.
    """
    workflow_dir = repo_path / ".workflow" / graph_name
    worktree_dir = repo_path / ".worktrees" / graph_name
    branch_prefix = f"{_BRANCH_PREFIX}/{graph_name}/"

    run_info = _load_run_info(workflow_dir)
    start_head = run_info["start_head"]
    started_at = run_info.get("started_at", "unknown")

    plan = ResetPlan(
        graph_name=graph_name,
        repo_path=repo_path,
        start_head=start_head,
        started_at=started_at,
        branch_prefix=branch_prefix,
        worktree_dir=worktree_dir,
        workflow_dir=workflow_dir,
        can_reset_main=True,
    )

    # Check ancestry for out-of-order reset detection.
    if not git.merge_base_is_ancestor(repo_path, start_head, "HEAD"):
        plan.can_reset_main = False
        plan.log.append(
            f"WARNING: start_head {start_head[:12]} is not an ancestor of HEAD. "
            "This likely means graphs were run in a different order. "
            "Main-branch reset will be skipped. Reset graphs in reverse "
            "run order (most-recently-run first)."
        )

    # Discover open PRs on graph branches.
    try:
        plan.open_prs = gh.pr_list(repo_path, head_prefix=branch_prefix)
    except subprocess.CalledProcessError:
        plan.log.append(
            "Could not list PRs (gh CLI error). PR closing will be skipped."
        )

    # Discover remote branches.
    plan.remote_branches = git.ls_remote_branches(
        repo_path, f"refs/heads/{branch_prefix}*"
    )

    return plan

execute_reset(plan)

Execute a reset plan, returning a log of actions taken.

Parameters:

Name	Type	Description	Default
plan	ResetPlan	A reset plan from :func:plan_reset.	required

Returns:

Type	Description
list[str]	List of human-readable log messages.

Source code in src/agentrelay/reset_graph.py

def execute_reset(plan: ResetPlan) -> list[str]:
    """Execute a reset plan, returning a log of actions taken.

    Args:
        plan: A reset plan from :func:`plan_reset`.

    Returns:
        List of human-readable log messages.
    """
    log: list[str] = list(plan.log)
    repo = plan.repo_path

    # Step 0: Force-remove Docker containers (best-effort).
    try:
        containers = docker_ops.ps_by_label(f"agentrelay.graph={plan.graph_name}")
        for name in containers:
            try:
                docker_ops.force_rm(name)
            except subprocess.CalledProcessError:
                pass
        if containers:
            log.append(f"Removed {len(containers)} Docker container(s)")
    except (subprocess.CalledProcessError, FileNotFoundError):
        pass  # Docker not available or no containers found

    # Step 0b: Remove Docker network (best-effort).
    network_name = f"agentrelay-{plan.graph_name}"
    try:
        if docker_ops.network_exists(network_name):
            docker_ops.network_remove(network_name)
            log.append(f"Removed Docker network {network_name}")
    except (subprocess.CalledProcessError, FileNotFoundError):
        pass

    # Step 1: Close open PRs.
    for pr in plan.open_prs:
        pr_num = int(pr["number"])
        branch = pr["headRefName"]
        try:
            gh.pr_close(repo, pr_num)
            log.append(f"Closed PR #{pr_num} ({branch})")
        except subprocess.CalledProcessError:
            log.append(f"Failed to close PR #{pr_num} ({branch}), skipping")

    # Step 2: Reset main and force-push.
    if plan.can_reset_main:
        git.fetch_branch(repo, "main")
        git.reset_hard(repo, plan.start_head)
        git.push_force_with_lease(repo, "main")
        log.append(f"Reset main to {plan.start_head[:12]} and force-pushed")
    else:
        log.append("Skipped main-branch reset (out-of-order)")

    # Step 3: Delete remote branches.
    for branch in plan.remote_branches:
        try:
            git.push_delete_branch(repo, branch)
            log.append(f"Deleted remote branch {branch}")
        except subprocess.CalledProcessError:
            log.append(f"Failed to delete remote branch {branch}, skipping")

    # Step 3b: Delete local branches matching graph prefix.
    try:
        local_branches = git.branch_list_local(repo, f"{plan.branch_prefix}*")
    except subprocess.CalledProcessError:
        local_branches = []
    for branch in local_branches:
        try:
            git.branch_delete(repo, branch)
            log.append(f"Deleted local branch {branch}")
        except subprocess.CalledProcessError:
            pass

    # Step 4: Remove worktree directory.
    if plan.worktree_dir.is_dir():
        try:
            shutil.rmtree(plan.worktree_dir)
            log.append(f"Removed worktree directory {plan.worktree_dir}")
        except PermissionError:
            # Container-created files may be owned by a different UID.
            # Try cleanup via a privileged Docker container.
            try:
                subprocess.run(
                    [
                        "docker",
                        "run",
                        "--rm",
                        "-v",
                        f"{plan.worktree_dir}:/cleanup",
                        "ubuntu:24.04",
                        "rm",
                        "-rf",
                        "/cleanup",
                    ],
                    check=True,
                    capture_output=True,
                )
                # Directory contents deleted; remove the now-empty mount point.
                if plan.worktree_dir.is_dir():
                    shutil.rmtree(plan.worktree_dir, ignore_errors=True)
                log.append(
                    f"Removed worktree directory {plan.worktree_dir} (via Docker)"
                )
            except (subprocess.CalledProcessError, FileNotFoundError):
                log.append(
                    f"WARNING: Cannot remove {plan.worktree_dir} — "
                    "files owned by container UID. "
                    f"Run: sudo rm -rf {plan.worktree_dir}"
                )

    # Step 4b: Prune stale worktree references.
    try:
        git.worktree_prune(repo)
        log.append("Pruned stale git worktree references")
    except subprocess.CalledProcessError:
        log.append("Failed to prune worktree references, skipping")

    # Step 5: Remove workflow directory.
    if plan.workflow_dir.is_dir():
        shutil.rmtree(plan.workflow_dir)
        log.append(f"Removed workflow directory {plan.workflow_dir}")

    return log

print_plan(plan)

Print a human-readable summary of a reset plan.

Parameters:

Name	Type	Description	Default
plan	ResetPlan	A reset plan from :func:plan_reset.	required

Source code in src/agentrelay/reset_graph.py

def print_plan(plan: ResetPlan) -> None:
    """Print a human-readable summary of a reset plan.

    Args:
        plan: A reset plan from :func:`plan_reset`.
    """
    print(f"[reset] Graph: {plan.graph_name}")
    print(f"[reset] Start HEAD: {plan.start_head[:12]}")
    print(f"[reset] Run started at: {plan.started_at}")
    print(f"[reset] Open PRs to close: {len(plan.open_prs)}")
    print(f"[reset] Remote branches to delete: {len(plan.remote_branches)}")
    print(
        f"[reset] Reset main: {'yes' if plan.can_reset_main else 'SKIPPED (out-of-order)'}"
    )
    if plan.worktree_dir.is_dir():
        print(f"[reset] Worktree dir to remove: {plan.worktree_dir}")
    if plan.workflow_dir.is_dir():
        print(f"[reset] Workflow dir to remove: {plan.workflow_dir}")
    for msg in plan.log:
        print(f"[reset] {msg}")

reset_graph(graph_name, repo_path, *, yes=False)

Plan and execute a graph reset.

Parameters:

Name	Type	Description	Default
graph_name	str	Name of the graph to reset.	required
repo_path	Path	Path to the repository root.	required
yes	bool	Skip interactive confirmation.	False

Returns:

Type	Description
list[str]	List of log messages describing actions taken.

Source code in src/agentrelay/reset_graph.py

def reset_graph(
    graph_name: str,
    repo_path: Path,
    *,
    yes: bool = False,
) -> list[str]:
    """Plan and execute a graph reset.

    Args:
        graph_name: Name of the graph to reset.
        repo_path: Path to the repository root.
        yes: Skip interactive confirmation.

    Returns:
        List of log messages describing actions taken.
    """
    plan = plan_reset(graph_name, repo_path)
    print_plan(plan)

    if not yes:
        response = input("\n[reset] Continue? [y/N] ")
        if response.strip().lower() != "y":
            print("[reset] Aborted.")
            return []

    log = execute_reset(plan)
    for msg in log:
        print(f"[reset] {msg}")
    print("[reset] Done.")
    return log

main()

CLI entry point for resetting a graph run.

Source code in src/agentrelay/reset_graph.py

def main() -> None:
    """CLI entry point for resetting a graph run."""
    parser = argparse.ArgumentParser(
        description="Reset a repository to its pre-graph-run state.",
    )
    parser.add_argument(
        "graph",
        help="Path to graph YAML file",
    )
    parser.add_argument(
        "--yes",
        "-y",
        action="store_true",
        help="Skip confirmation prompt",
    )
    args = parser.parse_args()

    graph_path = Path(args.graph).resolve()
    if not graph_path.is_file():
        print(f"Error: graph file not found: {graph_path}", file=sys.stderr)
        sys.exit(1)

    graph_name = _resolve_graph_name(graph_path)
    repo_path = Path.cwd()

    try:
        reset_graph(graph_name, repo_path, yes=args.yes)
    except FileNotFoundError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)