scripts-worktree-manager

#!/usr/bin/env python3 """

title: "=============================================================================" component_type: script version: "1.0.0" audience: contributor status: stable summary: "CODITECT Worktree Manager - Isolated Submodule Development" keywords: ['api', 'backend', 'docker', 'frontend', 'git'] tokens: ~500 created: 2025-12-22 updated: 2025-12-22 script_name: "worktree-manager.py" language: python executable: true usage: "python3 scripts/worktree-manager.py [options]" python_version: "3.10+" dependencies: [] modifies_files: false network_access: false requires_auth: false

CODITECT Worktree Manager - Isolated Submodule Development

Provides complete isolation for submodule development with:

• Git worktrees for parallel branch work
• Docker container isolation
• Claude Code session scoping
• Multi-workspace document index integration

Usage: python3 scripts/worktree-manager.py create [--container] python3 scripts/worktree-manager.py list [--submodule NAME] python3 scripts/worktree-manager.py remove python3 scripts/worktree-manager.py enter python3 scripts/worktree-manager.py status python3 scripts/worktree-manager.py orchestrate [--submodules LIST] """

import argparse import hashlib import json import os import re import shutil import subprocess import sys from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple

=============================================================================

CONFIGURATION

=============================================================================

WORKTREE_BASE = "worktrees" # Relative to master repo root CONTAINER_BASE = "containers" CONFIG_FILE = ".coditect/worktree-config.json"

SUBMODULE_TYPES = { "core": {"base_image": "python:3.11-slim", "ports": [8000]}, "cloud": {"base_image": "python:3.11-slim", "ports": [8080, 8443]}, "frontend": {"base_image": "node:20-slim", "ports": [3000, 5173]}, "integrations": {"base_image": "python:3.11-slim", "ports": [9000]}, "tools": {"base_image": "python:3.11-slim", "ports": []}, "personas": {"base_image": "python:3.11-slim", "ports": []}, "templates": {"base_image": "node:20-slim", "ports": [3000]}, "default": {"base_image": "ubuntu:22.04", "ports": []}, }

=============================================================================

UTILITY FUNCTIONS

=============================================================================

def run_command(cmd: List[str], cwd: Optional[str] = None, capture: bool = True) -> Tuple[int, str, str]: """Run a command and return exit code, stdout, stderr.""" try: result = subprocess.run( cmd, cwd=cwd, capture_output=capture, text=True, timeout=120 ) return result.returncode, result.stdout, result.stderr except subprocess.TimeoutExpired: return 1, "", "Command timed out" except Exception as e: return 1, "", str(e)

def get_master_root() -> Path: """Get the root of the master repository.""" # Walk up until we find .git that's not a file (submodule marker) current = Path.cwd() while current != current.parent: git_path = current / ".git" if git_path.exists() and git_path.is_dir(): # Check if this is the master repo (has submodules/) if (current / "submodules").exists(): return current current = current.parent

# Fallback: look for coditect-rollout-master in path
cwd = Path.cwd()
for parent in [cwd] + list(cwd.parents):
    if "coditect-rollout-master" in parent.name:
        return parent

return Path.cwd()

def get_submodule_type(submodule_path: str) -> str: """Determine submodule type from path.""" parts = submodule_path.split("/") if len(parts) >= 2: category = parts[1] if parts[0] == "submodules" else parts[0] if category in SUBMODULE_TYPES: return category return "default"

def generate_worktree_id(submodule: str, branch: str) -> str: """Generate unique worktree identifier.""" content = f"{submodule}:{branch}" return hashlib.sha256(content.encode()).hexdigest()[:12]

def load_config(master_root: Path) -> Dict: """Load worktree configuration.""" config_path = master_root / CONFIG_FILE if config_path.exists(): with open(config_path) as f: return json.load(f) return { "worktrees": {}, "containers": {}, "created_at": datetime.now().isoformat() }

def save_config(master_root: Path, config: Dict): """Save worktree configuration.""" config_path = master_root / CONFIG_FILE config_path.parent.mkdir(parents=True, exist_ok=True) config["updated_at"] = datetime.now().isoformat() with open(config_path, "w") as f: json.dump(config, f, indent=2)

=============================================================================

WORKTREE OPERATIONS

=============================================================================

def list_submodules(master_root: Path) -> List[Dict]: """List all submodules in the master repo.""" gitmodules_path = master_root / ".gitmodules" submodules = []

if not gitmodules_path.exists():
    return submodules

content = gitmodules_path.read_text()
current = {}

for line in content.splitlines():
    line = line.strip()
    if line.startswith("[submodule"):
        if current:
            submodules.append(current)
        match = re.search(r'"([^"]+)"', line)
        current = {"name": match.group(1) if match else "unknown"}
    elif "=" in line and current:
        key, value = line.split("=", 1)
        current[key.strip()] = value.strip()

if current:
    submodules.append(current)

return submodules

def create_worktree( master_root: Path, submodule: str, branch: str, with_container: bool = False ) -> Dict: """Create an isolated worktree for a submodule branch."""

submodule_path = master_root / submodule
if not submodule_path.exists():
    return {"error": f"Submodule not found: {submodule}"}

# Generate worktree path
worktree_id = generate_worktree_id(submodule, branch)
safe_submodule = submodule.replace("/", "-")
safe_branch = branch.replace("/", "-")
worktree_name = f"{safe_submodule}--{safe_branch}"
worktree_path = master_root / WORKTREE_BASE / worktree_name

# Check if already exists
if worktree_path.exists():
    return {
        "error": f"Worktree already exists: {worktree_path}",
        "path": str(worktree_path)
    }

# Create worktree directory structure
worktree_path.parent.mkdir(parents=True, exist_ok=True)

# Check if branch exists, create if not
code, stdout, stderr = run_command(
    ["git", "branch", "--list", branch],
    cwd=str(submodule_path)
)

branch_exists = bool(stdout.strip())

# Create the worktree
if branch_exists:
    cmd = ["git", "worktree", "add", str(worktree_path), branch]
else:
    cmd = ["git", "worktree", "add", "-b", branch, str(worktree_path)]

code, stdout, stderr = run_command(cmd, cwd=str(submodule_path))

if code != 0:
    return {"error": f"Failed to create worktree: {stderr}"}

# Create CLAUDE.md for the worktree with isolation context
claude_md_content = f"""# Isolated Worktree - {submodule} @ {branch}

Worktree ID: {worktree_id} Created: {datetime.now().isoformat()} Isolation Level: Full (Git + Container + Claude Session)

Context

This is an isolated worktree for development on:

• Submodule: {submodule}
• Branch: {branch}
• Parent: coditect-rollout-master

Isolation Guarantees

1. Git Isolation - Changes here don't affect other worktrees or main
2. File System - Separate working directory from main checkout
3. Claude Session - Scoped to this worktree only

Workflow

# Work in this isolated environment
cd {worktree_path}

# Make changes, commit normally
git add . && git commit -m "feat: Your changes"

# Push when ready
git push origin {branch}

# Merge via PR (recommended) or orchestration

Return to Main

cd {master_root}
python3 scripts/worktree-manager.py status

---

Auto-generated by CODITECT Worktree Manager """

worktree_claude_md = worktree_path / "CLAUDE.md"
# Only create if doesn't exist (don't overwrite submodule's CLAUDE.md)
if not worktree_claude_md.exists():
    worktree_claude_md.write_text(claude_md_content)

# Create .worktree-info for tooling
worktree_info = {
    "id": worktree_id,
    "submodule": submodule,
    "branch": branch,
    "created_at": datetime.now().isoformat(),
    "master_root": str(master_root),
    "worktree_path": str(worktree_path),
    "container": None
}

info_path = worktree_path / ".worktree-info.json"
with open(info_path, "w") as f:
    json.dump(worktree_info, f, indent=2)

# Update config
config = load_config(master_root)
config["worktrees"][worktree_id] = worktree_info
save_config(master_root, config)

result = {
    "success": True,
    "id": worktree_id,
    "path": str(worktree_path),
    "submodule": submodule,
    "branch": branch,
    "branch_created": not branch_exists
}

# Create container if requested
if with_container:
    container_result = create_container(master_root, worktree_id, submodule, worktree_path)
    result["container"] = container_result

return result

def create_container( master_root: Path, worktree_id: str, submodule: str, worktree_path: Path ) -> Dict: """Create a Docker container for the worktree."""

submodule_type = get_submodule_type(submodule)
type_config = SUBMODULE_TYPES.get(submodule_type, SUBMODULE_TYPES["default"])

container_name = f"coditect-{worktree_id}"

# Create Dockerfile
dockerfile_content = f"""# Auto-generated Dockerfile for {submodule}

FROM {type_config['base_image']}

LABEL org.coditect.worktree-id="{worktree_id}" LABEL org.coditect.submodule="{submodule}"

Install common tools

RUN apt-get update && apt-get install -y \ git \ curl \ vim \ && rm -rf /var/lib/apt/lists/*

Set working directory

WORKDIR /workspace

Copy worktree contents

COPY . /workspace/

Default command

CMD ["bash"] """

dockerfile_path = worktree_path / "Dockerfile.worktree"
dockerfile_path.write_text(dockerfile_content)

# Create docker-compose for the worktree
ports_config = "\n".join([f'      - "{p}:{p}"' for p in type_config["ports"]])

compose_content = f"""# Auto-generated docker-compose for {submodule} worktree

version: '3.8'

services: {container_name}: build: context: . dockerfile: Dockerfile.worktree container_name: {container_name} volumes: - .:/workspace working_dir: /workspace environment: - CODITECT_WORKTREE_ID={worktree_id} - CODITECT_SUBMODULE={submodule} ports: {ports_config if ports_config else ' - "8000:8000"'} stdin_open: true tty: true """

compose_path = worktree_path / "docker-compose.worktree.yml"
compose_path.write_text(compose_content)

return {
    "name": container_name,
    "dockerfile": str(dockerfile_path),
    "compose": str(compose_path),
    "ports": type_config["ports"],
    "start_command": f"docker-compose -f {compose_path} up -d"
}

def list_worktrees(master_root: Path, submodule_filter: Optional[str] = None) -> List[Dict]: """List all worktrees.""" config = load_config(master_root) worktrees = []

for wt_id, wt_info in config.get("worktrees", {}).items():
    if submodule_filter and wt_info.get("submodule") != submodule_filter:
        continue

    # Check if worktree still exists
    wt_path = Path(wt_info.get("worktree_path", ""))
    wt_info["exists"] = wt_path.exists()

    # Get git status if exists
    if wt_info["exists"]:
        code, stdout, stderr = run_command(
            ["git", "status", "--porcelain"],
            cwd=str(wt_path)
        )
        wt_info["has_changes"] = bool(stdout.strip())
        wt_info["change_count"] = len(stdout.strip().splitlines()) if stdout.strip() else 0

    worktrees.append(wt_info)

return worktrees

def remove_worktree(master_root: Path, worktree_path: str, force: bool = False) -> Dict: """Remove a worktree.""" wt_path = Path(worktree_path)

if not wt_path.exists():
    return {"error": f"Worktree not found: {worktree_path}"}

# Load worktree info
info_path = wt_path / ".worktree-info.json"
if info_path.exists():
    with open(info_path) as f:
        wt_info = json.load(f)
    submodule_path = master_root / wt_info.get("submodule", "")
else:
    # Try to find submodule from git
    code, stdout, stderr = run_command(
        ["git", "worktree", "list", "--porcelain"],
        cwd=str(wt_path)
    )
    submodule_path = wt_path  # Fallback

# Check for uncommitted changes
code, stdout, stderr = run_command(
    ["git", "status", "--porcelain"],
    cwd=str(wt_path)
)

if stdout.strip() and not force:
    return {
        "error": "Worktree has uncommitted changes. Use --force to remove anyway.",
        "changes": stdout.strip().splitlines()
    }

# Remove worktree via git
cmd = ["git", "worktree", "remove", str(wt_path)]
if force:
    cmd.append("--force")

code, stdout, stderr = run_command(cmd, cwd=str(submodule_path))

if code != 0:
    # Try direct removal as fallback
    try:
        shutil.rmtree(wt_path)
        run_command(["git", "worktree", "prune"], cwd=str(submodule_path))
    except Exception as e:
        return {"error": f"Failed to remove worktree: {e}"}

# Update config
config = load_config(master_root)
worktrees = config.get("worktrees", {})

# Find and remove from config
to_remove = None
for wt_id, wt_info in worktrees.items():
    if wt_info.get("worktree_path") == str(wt_path):
        to_remove = wt_id
        break

if to_remove:
    del worktrees[to_remove]
    config["worktrees"] = worktrees
    save_config(master_root, config)

return {"success": True, "removed": str(wt_path)}

def get_worktree_status(master_root: Path) -> Dict: """Get comprehensive status of all worktrees.""" config = load_config(master_root) worktrees = list_worktrees(master_root) submodules = list_submodules(master_root)

# Count by status
active = [w for w in worktrees if w.get("exists")]
with_changes = [w for w in active if w.get("has_changes")]

# Group by submodule
by_submodule = {}
for wt in worktrees:
    sub = wt.get("submodule", "unknown")
    if sub not in by_submodule:
        by_submodule[sub] = []
    by_submodule[sub].append(wt)

return {
    "total_submodules": len(submodules),
    "total_worktrees": len(worktrees),
    "active_worktrees": len(active),
    "worktrees_with_changes": len(with_changes),
    "by_submodule": by_submodule,
    "worktrees": worktrees
}

=============================================================================

ORCHESTRATION

=============================================================================

def orchestrate_command( master_root: Path, command: str, submodules: Optional[List[str]] = None, branches: Optional[List[str]] = None ) -> Dict: """Run a command across multiple worktrees."""

worktrees = list_worktrees(master_root)

# Filter by submodules if specified
if submodules:
    worktrees = [w for w in worktrees if w.get("submodule") in submodules]

# Filter by branches if specified
if branches:
    worktrees = [w for w in worktrees if w.get("branch") in branches]

results = []

for wt in worktrees:
    if not wt.get("exists"):
        continue

    wt_path = wt.get("worktree_path")
    code, stdout, stderr = run_command(
        ["bash", "-c", command],
        cwd=wt_path
    )

    results.append({
        "worktree": wt.get("id"),
        "submodule": wt.get("submodule"),
        "branch": wt.get("branch"),
        "exit_code": code,
        "stdout": stdout[:500] if stdout else "",
        "stderr": stderr[:500] if stderr else "",
        "success": code == 0
    })

return {
    "command": command,
    "total": len(results),
    "successful": sum(1 for r in results if r["success"]),
    "failed": sum(1 for r in results if not r["success"]),
    "results": results
}

=============================================================================

DOCUMENT INDEX INTEGRATION

=============================================================================

def register_worktree_workspace(master_root: Path, worktree_id: str) -> Dict: """Register a worktree as a workspace in the document index.""" config = load_config(master_root) wt_info = config.get("worktrees", {}).get(worktree_id)

if not wt_info:
    return {"error": f"Worktree not found: {worktree_id}"}

wt_path = wt_info.get("worktree_path")

# Find context-db.py
context_db_paths = [
    master_root / "submodules/core/coditect-core/scripts/context-db.py",
    master_root / ".coditect/scripts/context-db.py",
    Path.cwd() / "scripts/context-db.py"
]

context_db = None
for p in context_db_paths:
    if p.exists():
        context_db = p
        break

if not context_db:
    return {"error": "context-db.py not found"}

# Register the worktree as a workspace
code, stdout, stderr = run_command([
    "python3", str(context_db),
    "--register-workspace", wt_path,
    "--workspace-name", f"worktree-{worktree_id}",
    "--no-auto-discover"  # Worktrees don't have submodules
])

if code != 0:
    return {"error": f"Failed to register workspace: {stderr}"}

return {
    "success": True,
    "workspace_name": f"worktree-{worktree_id}",
    "path": wt_path
}

=============================================================================

CLI

=============================================================================

def main(): parser = argparse.ArgumentParser( description="CODITECT Worktree Manager - Isolated Submodule Development", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples:

Create isolated worktree for feature work

python3 scripts/worktree-manager.py create submodules/core/coditect-core feature/new-api

Create with Docker container

python3 scripts/worktree-manager.py create submodules/cloud/backend hotfix --container

List all worktrees

python3 scripts/worktree-manager.py list

List worktrees for specific submodule

python3 scripts/worktree-manager.py list --submodule submodules/core/coditect-core

Get status overview

python3 scripts/worktree-manager.py status

Remove a worktree

python3 scripts/worktree-manager.py remove worktrees/submodules-core-coditect-core--feature-new-api

Run command across all worktrees

python3 scripts/worktree-manager.py orchestrate "git status"

Run command on specific submodules

python3 scripts/worktree-manager.py orchestrate "npm test" --submodules frontend,templates """ )

subparsers = parser.add_subparsers(dest="command", help="Commands")

# Create command
create_parser = subparsers.add_parser("create", help="Create isolated worktree")
create_parser.add_argument("submodule", help="Submodule path (e.g., submodules/core/coditect-core)")
create_parser.add_argument("branch", help="Branch name to create/checkout")
create_parser.add_argument("--container", "-c", action="store_true", help="Also create Docker container")
create_parser.add_argument("--register", "-r", action="store_true", help="Register as doc workspace")

# List command
list_parser = subparsers.add_parser("list", help="List worktrees")
list_parser.add_argument("--submodule", "-s", help="Filter by submodule")
list_parser.add_argument("--json", action="store_true", help="Output as JSON")

# Remove command
remove_parser = subparsers.add_parser("remove", help="Remove worktree")
remove_parser.add_argument("path", help="Worktree path to remove")
remove_parser.add_argument("--force", "-f", action="store_true", help="Force removal even with changes")

# Status command
status_parser = subparsers.add_parser("status", help="Show worktree status overview")
status_parser.add_argument("--json", action="store_true", help="Output as JSON")

# Orchestrate command
orch_parser = subparsers.add_parser("orchestrate", help="Run command across worktrees")
orch_parser.add_argument("cmd", help="Command to run")
orch_parser.add_argument("--submodules", help="Comma-separated list of submodules")
orch_parser.add_argument("--branches", help="Comma-separated list of branches")
orch_parser.add_argument("--json", action="store_true", help="Output as JSON")

# Enter command (prints cd command)
enter_parser = subparsers.add_parser("enter", help="Get path to enter worktree")
enter_parser.add_argument("submodule", help="Submodule name")
enter_parser.add_argument("branch", help="Branch name")

args = parser.parse_args()

if not args.command:
    parser.print_help()
    return

master_root = get_master_root()

if args.command == "create":
    result = create_worktree(
        master_root,
        args.submodule,
        args.branch,
        with_container=args.container
    )

    if "error" in result:
        print(f"Error: {result['error']}")
        sys.exit(1)

    print(f"\n{'='*60}")
    print("Isolated Worktree Created")
    print(f"{'='*60}\n")
    print(f"  Worktree ID:  {result['id']}")
    print(f"  Path:         {result['path']}")
    print(f"  Submodule:    {result['submodule']}")
    print(f"  Branch:       {result['branch']}")
    if result.get('branch_created'):
        print(f"  Note:         New branch created")

    if result.get('container'):
        print(f"\n  Container:")
        print(f"    Name:       {result['container']['name']}")
        print(f"    Start:      {result['container']['start_command']}")

    print(f"\n  To enter:")
    print(f"    cd {result['path']}")
    print()

    # Register as doc workspace if requested
    if args.register:
        reg_result = register_worktree_workspace(master_root, result['id'])
        if reg_result.get('success'):
            print(f"  Registered as doc workspace: {reg_result['workspace_name']}")

elif args.command == "list":
    worktrees = list_worktrees(master_root, args.submodule)

    if args.json:
        print(json.dumps(worktrees, indent=2))
    else:
        if not worktrees:
            print("No worktrees found.")
            return

        print(f"\n{'='*80}")
        print("CODITECT Worktrees")
        print(f"{'='*80}\n")

        for wt in worktrees:
            status = "✓" if wt.get("exists") else "✗"
            changes = f" ({wt.get('change_count', 0)} changes)" if wt.get("has_changes") else ""
            print(f"  {status} {wt.get('id', 'unknown')[:12]}")
            print(f"    Submodule: {wt.get('submodule')}")
            print(f"    Branch:    {wt.get('branch')}{changes}")
            print(f"    Path:      {wt.get('worktree_path')}")
            print()

elif args.command == "remove":
    result = remove_worktree(master_root, args.path, force=args.force)

    if "error" in result:
        print(f"Error: {result['error']}")
        if result.get("changes"):
            print("Uncommitted changes:")
            for change in result["changes"][:10]:
                print(f"  {change}")
        sys.exit(1)

    print(f"Removed: {result['removed']}")

elif args.command == "status":
    status = get_worktree_status(master_root)

    if args.json:
        print(json.dumps(status, indent=2))
    else:
        print(f"\n{'='*60}")
        print("CODITECT Worktree Status")
        print(f"{'='*60}\n")

        print(f"  Total Submodules:        {status['total_submodules']}")
        print(f"  Total Worktrees:         {status['total_worktrees']}")
        print(f"  Active Worktrees:        {status['active_worktrees']}")
        print(f"  With Uncommitted Changes: {status['worktrees_with_changes']}")

        if status['by_submodule']:
            print(f"\n  By Submodule:")
            for sub, wts in status['by_submodule'].items():
                print(f"    {sub}: {len(wts)} worktree(s)")
        print()

elif args.command == "orchestrate":
    submodules = args.submodules.split(",") if args.submodules else None
    branches = args.branches.split(",") if args.branches else None

    result = orchestrate_command(master_root, args.cmd, submodules, branches)

    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print(f"\n{'='*60}")
        print(f"Orchestration: {result['command']}")
        print(f"{'='*60}\n")

        print(f"  Total: {result['total']} | Success: {result['successful']} | Failed: {result['failed']}")
        print()

        for r in result['results']:
            status = "✓" if r['success'] else "✗"
            print(f"  {status} {r['submodule']} @ {r['branch']}")
            if r['stdout']:
                for line in r['stdout'].splitlines()[:3]:
                    print(f"      {line}")
            if r['stderr'] and not r['success']:
                print(f"      ERROR: {r['stderr'][:100]}")
        print()

elif args.command == "enter":
    worktrees = list_worktrees(master_root)

    for wt in worktrees:
        if wt.get("submodule") == args.submodule and wt.get("branch") == args.branch:
            if wt.get("exists"):
                print(wt.get("worktree_path"))
                return
            else:
                print(f"Worktree exists in config but not on disk: {wt.get('worktree_path')}")
                sys.exit(1)

    print(f"No worktree found for {args.submodule} @ {args.branch}")
    print(f"Create with: python3 scripts/worktree-manager.py create {args.submodule} {args.branch}")
    sys.exit(1)

if name == "main": main()