Adding stacky recreate command to recreate a stack from a passed in stacky info

src/stacky/commands/recreate.py

@@ -0,0 +1,120 @@
+"""Recreate command - rebuild a stack from `stacky info` output."""
+
+import re
+import sys
+from typing import Iterable, List, Optional, Tuple
+
+from stacky.git.branch import get_all_branches
+from stacky.git.refs import get_merge_base, set_parent, set_parent_commit
+from stacky.stack.models import StackBranchSet
+from stacky.utils.logging import cout, die, info
+from stacky.utils.types import BranchName, Commit, STACK_BOTTOMS
+from stacky.utils.ui import confirm
+
+
+_TREE_CHARS = "│┌├└─"  # │ ┌ ├ └ ─
+_LINE_RE = re.compile(
+    r"^(?P<indent>[\s│┌├└─]*)"
+    r"(?:[!~*]+\s*)*"
+    r"(?P<name>[A-Za-z0-9_][\w./-]*)"
+    r"(?:\s+\(.*\))?\s*$"
+)
+
+
+def _indent_of(line: str) -> int:
+    """Visual indent of the branch on a line, measured by the leftmost tree
+    character. Lines without any tree character (e.g. `* main`) are roots and
+    get -1 so they sort below everything else.
+    """
+    for i, ch in enumerate(line):
+        if ch in "┌├└":  # ┌ ├ └
+            return i
+    return -1
+
+
+def parse_stack_info(text: str) -> List[Tuple[BranchName, Optional[BranchName]]]:
+    """Parse `stacky info` output into a list of (child, parent) edges.
+
+    The bottom-most branch (e.g. `main`) is returned with parent=None.
+    Lines are processed top-to-bottom; each branch's parent is the next line
+    *below* it with a strictly smaller indent (since stacky prints the tree
+    upside-down, so visually-lower = closer to the root).
+    """
+    parsed: List[Tuple[int, BranchName]] = []
+    for raw in text.splitlines():
+        if not raw.strip():
+            continue
+        m = _LINE_RE.match(raw)
+        if not m:
+            continue
+        name = BranchName(m.group("name"))
+        parsed.append((_indent_of(raw), name))
+
+    edges: List[Tuple[BranchName, Optional[BranchName]]] = []
+    for i, (indent, name) in enumerate(parsed):
+        parent: Optional[BranchName] = None
+        for j in range(i + 1, len(parsed)):
+            j_indent, j_name = parsed[j]
+            if j_indent < indent:
+                parent = j_name
+                break
+        edges.append((name, parent))
+    return edges
+
+
+def _validate_branches_exist(edges: Iterable[Tuple[BranchName, Optional[BranchName]]]):
+    existing = set(get_all_branches())
+    missing: List[BranchName] = []
+    for child, parent in edges:
+        if child not in existing:
+            missing.append(child)
+        if parent is not None and parent not in existing:
+            missing.append(parent)
+    if missing:
+        die("Branches not found in repo: {}", ", ".join(sorted(set(missing))))
+
+
+def cmd_recreate(stack: StackBranchSet, args):
+    """Rebuild a stack from `stacky info` output piped on stdin or via --file."""
+    if args.file:
+        with open(args.file) as f:
+            text = f.read()
+    else:
+        if sys.stdin.isatty():
+            die("No input provided. Pipe `stacky info` output or pass --file.")
+        text = sys.stdin.read()
+
+    edges = parse_stack_info(text)
+    if not edges:
+        die("No branches parsed from input")
+
+    _validate_branches_exist(edges)
+
+    to_apply = [(c, p) for c, p in edges if p is not None]
+    roots = [c for c, p in edges if p is None]
+
+    cout("Will recreate stack with {} edges:\n", len(to_apply), fg="green")
+    for child, parent in to_apply:
+        cout("  {} -> {}\n", child, parent)
+    if roots:
+        cout("Stack bottoms (unchanged): {}\n", ", ".join(roots), fg="cyan")
+
+    for root in roots:
+        if root not in STACK_BOTTOMS:
+            info(
+                "Note: {} is treated as a root but is not a known stack bottom",
+                root,
+            )
+
+    if not args.force:
+        confirm()
+
+    for child, parent in to_apply:
+        base = get_merge_base(child, parent)  # type: ignore[arg-type]
+        if base is None:
+            die("Could not find merge-base between {} and {}", child, parent)
+        set_parent(child, parent, set_origin=True)
+        set_parent_commit(child, Commit(base))
+        info("Set {} -> {} (parent commit {})", child, parent, base[:8])
+
+    cout("Done. Run `stacky info` to verify.\n", fg="green")

src/stacky/main.py

@@ -37,6 +37,7 @@
 from stacky.commands.fold import (
     cmd_fold, inner_do_fold, finish_merge_fold_operation
 )
+from stacky.commands.recreate import cmd_recreate
 
 
 def main():
@@ -115,6 +116,8 @@ def main():
         current_branch = get_current_branch_name()
         if args.command == "continue":
             _handle_continue(stack, current_branch)
+        elif args.command == "recreate":
+            args.func(stack, args)
         else:
             if current_branch not in stack.stack:
                 main_branch = get_real_stack_bottom()
@@ -336,3 +339,12 @@ def _setup_other_commands(subparsers):
     fold_parser = subparsers.add_parser("fold", help="Fold current branch into parent branch and delete current branch")
     fold_parser.add_argument("--allow-empty", action="store_true", help="Allow empty commits during cherry-pick")
     fold_parser.set_defaults(func=cmd_fold)
+
+    # recreate
+    recreate_parser = subparsers.add_parser(
+        "recreate",
+        help="Rebuild a stack from `stacky info` output (read from stdin or --file)",
+    )
+    recreate_parser.add_argument("--file", help="Read stack info from a file instead of stdin")
+    recreate_parser.add_argument("--force", "-f", action="store_true", help="Bypass confirmation")
+    recreate_parser.set_defaults(func=cmd_recreate)

src/stacky/tests/test_commands/test_recreate.py

@@ -0,0 +1,110 @@
+"""Tests for stacky.commands.recreate."""
+
+import unittest
+
+from stacky.commands.recreate import parse_stack_info
+
+
+class TestParseStackInfo(unittest.TestCase):
+    def test_simple_linear_stack(self):
+        text = "\n".join([
+            " ┌── feat-c",
+            " ├── feat-b",
+            " ├── feat-a",
+            "* main",
+        ])
+        edges = parse_stack_info(text)
+        self.assertEqual(edges, [
+            ("feat-c", "main"),
+            ("feat-b", "main"),
+            ("feat-a", "main"),
+            ("main", None),
+        ])
+
+    def test_branched_stack(self):
+        # Two children of feat-a, plus a sibling chain rooted at main.
+        text = "\n".join([
+            " │   ┌── feat-a-c2",
+            " │   ┌── feat-a-c1",
+            " ├── feat-a",
+            " ├── feat-b",
+            "* main",
+        ])
+        edges = parse_stack_info(text)
+        self.assertEqual(edges, [
+            ("feat-a-c2", "feat-a"),
+            ("feat-a-c1", "feat-a"),
+            ("feat-a", "main"),
+            ("feat-b", "main"),
+            ("main", None),
+        ])
+
+    def test_strips_status_markers_and_pr_suffix(self):
+        text = "\n".join([
+            " ┌── ! feat-x",
+            " ├── !~ feat-y (#42 some title)",
+            "* main",
+        ])
+        edges = parse_stack_info(text)
+        self.assertEqual(edges, [
+            ("feat-x", "main"),
+            ("feat-y", "main"),
+            ("main", None),
+        ])
+
+    def test_full_frostdb_example(self):
+        # The exact tree the user reconstructed by hand.
+        text = "\n".join([
+            " ┌── ! ynannapaneni/adding-p-validation-for-hybrid",
+            " ├── ! ynannapaneni/FDBCORE-XX-simulator-for-log-system-v2",
+            " ├── !~ ynannapaneni/FDBCORE-XX-agentbox-git-ignore",
+            " │           ┌── ynannapaneni/FDBCORE-4496-dmi-uses-real-time-stream-manager",
+            " │       ┌── ynannapaneni/FDBCORE-45165-rsm-on-connection-closed",
+            " │   ┌── ynannapaneni/FDBCORE-45165-component-test-for-realtime-delivery-framework",
+            " ├── ! ynannapaneni/FDBCORE-44860-real-time-mutation-delivery-framework",
+            " ├── !~ ynannapaneni/FDBCORE-40432-fix-bw-lag-metric-spike-issues",
+            " │       ┌── ynannapaneni/FDBCORE-37707-adding-hybrid-worker-dd-queue",
+            " │   ┌── ynannapaneni/FDBCORE-33707-adding-dd-queue-unit-tests",
+            " ├── ! ynannapaneni/FDBCORE-33707-creating-relocate-data-for-shard",
+            " ├── !~ frostdb_4",
+            " ├── !~ frostdb_3",
+            " ├── !~ frostdb_2",
+            "* main",
+        ])
+        edges = dict(parse_stack_info(text))
+        self.assertEqual(edges["main"], None)
+        self.assertEqual(edges["frostdb_2"], "main")
+        self.assertEqual(edges["frostdb_3"], "main")
+        self.assertEqual(edges["frostdb_4"], "main")
+        self.assertEqual(
+            edges["ynannapaneni/FDBCORE-45165-component-test-for-realtime-delivery-framework"],
+            "ynannapaneni/FDBCORE-44860-real-time-mutation-delivery-framework",
+        )
+        self.assertEqual(
+            edges["ynannapaneni/FDBCORE-45165-rsm-on-connection-closed"],
+            "ynannapaneni/FDBCORE-45165-component-test-for-realtime-delivery-framework",
+        )
+        self.assertEqual(
+            edges["ynannapaneni/FDBCORE-4496-dmi-uses-real-time-stream-manager"],
+            "ynannapaneni/FDBCORE-45165-rsm-on-connection-closed",
+        )
+        self.assertEqual(
+            edges["ynannapaneni/FDBCORE-37707-adding-hybrid-worker-dd-queue"],
+            "ynannapaneni/FDBCORE-33707-adding-dd-queue-unit-tests",
+        )
+        self.assertEqual(
+            edges["ynannapaneni/FDBCORE-33707-adding-dd-queue-unit-tests"],
+            "ynannapaneni/FDBCORE-33707-creating-relocate-data-for-shard",
+        )
+        self.assertEqual(
+            edges["ynannapaneni/FDBCORE-33707-creating-relocate-data-for-shard"],
+            "main",
+        )
+
+    def test_empty_input(self):
+        self.assertEqual(parse_stack_info(""), [])
+        self.assertEqual(parse_stack_info("   \n  \n"), [])
+
+
+if __name__ == "__main__":
+    unittest.main()