From 2b525cfa4d0f03482504ac56ef7e656dc2da553f Mon Sep 17 00:00:00 2001 From: Input-X Date: Sat, 7 Mar 2026 10:26:35 +0000 Subject: [PATCH] update --- src/aipass/spawn/README.md | 18 +- src/aipass/spawn/apps/handlers/delete_ops.py | 192 +++++ .../spawn/apps/handlers/sync_registry_ops.py | 153 ++++ .../spawn/apps/handlers/sync_templates_ops.py | 149 ++++ src/aipass/spawn/apps/handlers/update_ops.py | 508 +++++++++++ src/aipass/spawn/apps/modules/delete.py | 97 +++ .../spawn/apps/modules/sync_registry.py | 100 +++ .../spawn/apps/modules/sync_templates.py | 109 +++ src/aipass/spawn/apps/modules/update.py | 177 ++++ src/aipass/spawn/apps/plugins/__init__.py | 1 + src/aipass/spawn/apps/spawn.py | 28 +- src/aipass/spawn/docs/.gitkeep | 0 src/aipass/spawn/drone_adapter.py | 97 --- src/aipass/spawn/pytest.ini | 5 + src/aipass/spawn/spawn_json/.gitkeep | 0 src/aipass/spawn/tests/conftest.py | 1 + src/aipass/spawn/tests/test_drone_adapter.py | 12 +- src/aipass/spawn/tests/test_handlers.py | 786 ++++++++++++++++++ src/aipass/spawn/tests/test_lifecycle.py | 569 +++++++++++++ src/aipass/spawn/tests/test_update.py | 494 +++++++++++ 20 files changed, 3381 insertions(+), 115 deletions(-) create mode 100644 src/aipass/spawn/apps/handlers/delete_ops.py create mode 100644 src/aipass/spawn/apps/handlers/sync_registry_ops.py create mode 100644 src/aipass/spawn/apps/handlers/sync_templates_ops.py create mode 100644 src/aipass/spawn/apps/handlers/update_ops.py create mode 100644 src/aipass/spawn/apps/modules/delete.py create mode 100644 src/aipass/spawn/apps/modules/sync_registry.py create mode 100644 src/aipass/spawn/apps/modules/sync_templates.py create mode 100644 src/aipass/spawn/apps/modules/update.py create mode 100644 src/aipass/spawn/apps/plugins/__init__.py create mode 100644 src/aipass/spawn/docs/.gitkeep delete mode 100644 src/aipass/spawn/drone_adapter.py create mode 100644 src/aipass/spawn/pytest.ini create mode 100644 src/aipass/spawn/spawn_json/.gitkeep create mode 100644 src/aipass/spawn/tests/conftest.py create mode 100644 src/aipass/spawn/tests/test_handlers.py create mode 100644 src/aipass/spawn/tests/test_lifecycle.py create mode 100644 src/aipass/spawn/tests/test_update.py diff --git a/src/aipass/spawn/README.md b/src/aipass/spawn/README.md index d3fa704a..0ecbe0da 100644 --- a/src/aipass/spawn/README.md +++ b/src/aipass/spawn/README.md @@ -40,17 +40,27 @@ spawn/ ├── apps/ │ ├── spawn.py # Entry point (CLI) │ ├── modules/ -│ │ └── core.py # Orchestrator — coordinates all steps +│ │ ├── core.py # Create orchestrator — coordinates spawn steps +│ │ ├── update.py # Update CLI — parses args, delegates to handler +│ │ ├── delete.py # Delete CLI — parses args, delegates to handler +│ │ ├── sync_registry.py # Sync registry CLI — report and repair +│ │ └── sync_templates.py # Sync templates CLI — pull from sources │ └── handlers/ │ ├── file_ops.py # Template copy, path renaming │ ├── metadata.py # Branch name extraction, profile detection │ ├── placeholders.py # {{PLACEHOLDER}} replacement engine -│ └── registry.py # AIPASS_REGISTRY.json CRUD +│ ├── registry.py # AIPASS_REGISTRY.json CRUD +│ ├── update_ops.py # Update implementation logic +│ ├── delete_ops.py # Delete implementation logic +│ ├── sync_registry_ops.py # Registry sync implementation +│ └── sync_templates_ops.py # Template sync implementation ├── templates/ │ ├── agent.template/ # Source template for new agents │ └── agent_mock_branch/ # Reference implementation -└── tests/ - └── test_spawn.py # 13 tests covering all components +├── tools/ # Branch verification utilities +├── docs/ # Documentation +├── spawn_json/ # JSON tracking directory +└── tests/ # Test suite (95 tests) ``` --- diff --git a/src/aipass/spawn/apps/handlers/delete_ops.py b/src/aipass/spawn/apps/handlers/delete_ops.py new file mode 100644 index 00000000..65b4cf5b --- /dev/null +++ b/src/aipass/spawn/apps/handlers/delete_ops.py @@ -0,0 +1,192 @@ +# =================== META ==================== +# Name: delete_ops.py +# Description: Delete handler — implementation logic for branch deletion +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Delete handler implementation for branch lifecycle management. + +Contains the core deletion logic: resolving branch paths, archiving branch +directories, removing from registry, and safety checks for protected branches. +""" + +import shutil +from datetime import datetime +from pathlib import Path + +from aipass.prax.apps.modules.logger import system_logger as logger + +from aipass.spawn.apps.handlers.registry import ( + find_registry, + load_registry, + save_registry, +) + +# Repo root — resolved from spawn package location +_REPO_ROOT = Path(__file__).parents[4] # src/aipass/spawn/apps/handlers -> repo root + +# Branches that cannot be deleted (critical infrastructure) +_PROTECTED_BRANCHES = {"spawn", "devpulse", "drone"} + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def delete_branch( + branch_name: str, + confirm: bool = True, + dry_run: bool = False, +) -> dict: + """Archive and deregister a branch. + + Workflow: + 1. Resolve branch path from AIPASS_REGISTRY.json + 2. Verify branch exists on filesystem + 3. If confirm=True -> print what will happen, prompt "Are you sure? (y/N)" + 4. If dry_run -> print summary, return + 5. Create archive directory: {repo_root}/.archive/deleted_branches/{name}_{timestamp}/ + 6. Copy entire branch directory to archive (shutil.copytree) + 7. Remove branch from AIPASS_REGISTRY.json + 8. Remove branch directory (shutil.rmtree) + 9. Log and return summary + + SAFETY: + - Cannot delete spawn (self-protection) + - Cannot delete devpulse (orchestration hub protection) + - Cannot delete drone (routing infrastructure protection) + - Archive ALWAYS created before deletion + + Returns: + Dict with deletion results. + """ + # Safety: check protected branches + if branch_name.lower() in _PROTECTED_BRANCHES: + msg = f"Cannot delete '{branch_name}' — protected branch ({', '.join(sorted(_PROTECTED_BRANCHES))})" + logger.warning(f"[delete] {msg}") + return { + "branch": branch_name, + "success": False, + "archive_path": "", + "registry_updated": False, + "error": msg, + } + + # 1. Resolve branch path from registry + registry_path = find_registry() + registry = load_registry(registry_path) + branch_entry = None + branch_dir = None + + for entry in registry.get("branches", []): + if entry.get("name", "").lower() == branch_name.lower(): + branch_entry = entry + rel_path = entry.get("path", "") + if rel_path: + branch_dir = (_REPO_ROOT / rel_path).resolve() + break + + if branch_entry is None: + msg = f"Branch '{branch_name}' not found in registry" + logger.warning(f"[delete] {msg}") + return { + "branch": branch_name, + "success": False, + "archive_path": "", + "registry_updated": False, + "error": msg, + } + + # 2. Verify branch exists on filesystem + if branch_dir is None or not branch_dir.is_dir(): + msg = f"Branch directory does not exist: {branch_dir}" + logger.warning(f"[delete] {msg}") + return { + "branch": branch_name, + "success": False, + "archive_path": "", + "registry_updated": False, + "error": msg, + } + + # 3. Confirmation prompt + if confirm and not dry_run: + try: + answer = input("Are you sure? (y/N): ").strip().lower() + except (EOFError, KeyboardInterrupt): + answer = "" + if answer != "y": + return { + "branch": branch_name, + "success": False, + "archive_path": "", + "registry_updated": False, + "error": "Cancelled by user", + } + + # 4. Dry run — report and return + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + archive_dir = _REPO_ROOT / ".archive" / "deleted_branches" / f"{branch_name}_{timestamp}" + + if dry_run: + logger.info(f"[delete] Dry run: would archive {branch_name} to {archive_dir}") + return { + "branch": branch_name, + "success": True, + "archive_path": str(archive_dir), + "registry_updated": False, + "dry_run": True, + } + + # 5. Create archive + archive_dir.parent.mkdir(parents=True, exist_ok=True) + try: + shutil.copytree(str(branch_dir), str(archive_dir)) + logger.info(f"[delete] Archived {branch_name} to {archive_dir}") + except Exception as exc: + msg = f"Failed to create archive: {exc}" + logger.error(f"[delete] {msg}") + return { + "branch": branch_name, + "success": False, + "archive_path": "", + "registry_updated": False, + "error": msg, + } + + # 6. Remove from registry + registry["branches"] = [ + b for b in registry.get("branches", []) + if b.get("name", "").lower() != branch_name.lower() + ] + registry["metadata"]["total_branches"] = len(registry["branches"]) + registry_updated = save_registry(registry_path, registry) + + if registry_updated: + logger.info(f"[delete] Removed {branch_name} from registry") + else: + logger.error(f"[delete] Failed to update registry after removing {branch_name}") + + # 7. Remove branch directory + try: + shutil.rmtree(str(branch_dir)) + logger.info(f"[delete] Removed branch directory: {branch_dir}") + except Exception as exc: + msg = f"Archive created but failed to remove directory: {exc}" + logger.error(f"[delete] {msg}") + return { + "branch": branch_name, + "success": False, + "archive_path": str(archive_dir), + "registry_updated": registry_updated, + "error": msg, + } + + return { + "branch": branch_name, + "success": True, + "archive_path": str(archive_dir), + "registry_updated": registry_updated, + } diff --git a/src/aipass/spawn/apps/handlers/sync_registry_ops.py b/src/aipass/spawn/apps/handlers/sync_registry_ops.py new file mode 100644 index 00000000..5df5f977 --- /dev/null +++ b/src/aipass/spawn/apps/handlers/sync_registry_ops.py @@ -0,0 +1,153 @@ +# =================== META ==================== +# Name: sync_registry_ops.py +# Description: Registry sync handler — implementation logic for registry repair +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Registry synchronization handler for branch lifecycle management. + +Contains the core sync logic: scanning the filesystem, comparing with +AIPASS_REGISTRY.json, detecting stale/unregistered branches, and +optionally repairing mismatches. +""" + +import json +from datetime import datetime +from pathlib import Path + +from aipass.prax.apps.modules.logger import system_logger as logger + +from aipass.spawn.apps.handlers.registry import ( + find_registry, + load_registry, + save_registry, +) + +# Repo root — resolved from spawn package location +_REPO_ROOT = Path(__file__).parents[4] # src/aipass/spawn/apps/handlers -> repo root + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def sync_registry(fix: bool = False) -> dict: + """Synchronize AIPASS_REGISTRY.json with filesystem. + + Workflow: + 1. Load AIPASS_REGISTRY.json + 2. Scan src/aipass/ for directories that have .trinity/passport.json (branch marker) + 3. Compare: + - Stale entries: registered in JSON but directory missing or no passport + - Unregistered: directory exists with passport but not in JSON + 4. Report findings + 5. If fix=True: + - Remove stale entries + - Add unregistered branches with basic metadata + - Save updated registry + + Returns: + Dict with sync results. + """ + # 1. Load registry + registry_path = find_registry() + registry = load_registry(registry_path) + branches = registry.get("branches", []) + + # Build lookup of registered branch names (lowercase) -> entry + registered: dict[str, dict] = {} + for entry in branches: + name = entry.get("name", "").lower() + if name: + registered[name] = entry + + # 2. Scan filesystem for branch directories with .trinity/passport.json + src_aipass = _REPO_ROOT / "src" / "aipass" + filesystem_branches: dict[str, Path] = {} + + if src_aipass.is_dir(): + for child in sorted(src_aipass.iterdir()): + if child.is_dir() and not child.name.startswith(".") and not child.name.startswith("__"): + passport = child / ".trinity" / "passport.json" + if passport.exists(): + filesystem_branches[child.name.lower()] = child + + # 3. Compare + stale: list[str] = [] + unregistered_list: list[str] = [] + healthy: list[str] = [] + + # Check registered branches against filesystem + for name, entry in registered.items(): + rel_path = entry.get("path", "") + branch_dir = (_REPO_ROOT / rel_path).resolve() if rel_path else None + + if branch_dir and branch_dir.is_dir(): + passport = branch_dir / ".trinity" / "passport.json" + if passport.exists(): + healthy.append(name) + else: + # Directory exists but no passport — treat as stale + stale.append(name) + else: + stale.append(name) + + # Check filesystem branches against registry + for name, path in filesystem_branches.items(): + if name not in registered: + unregistered_list.append(name) + + logger.info( + f"[sync-registry] Scan complete: " + f"healthy={len(healthy)}, stale={len(stale)}, unregistered={len(unregistered_list)}" + ) + + # 4/5. Fix if requested + fixed = False + if fix and (stale or unregistered_list): + # Remove stale entries + if stale: + registry["branches"] = [ + b for b in registry["branches"] + if b.get("name", "").lower() not in stale + ] + for s in stale: + logger.info(f"[sync-registry] Removed stale entry: {s}") + + # Add unregistered branches + today = datetime.now().strftime("%Y-%m-%d") + for name in unregistered_list: + branch_path = filesystem_branches[name] + rel_path = str(branch_path.relative_to(_REPO_ROOT)) + + entry = { + "name": name.upper(), + "path": rel_path, + "profile": "library", + "description": f"Auto-registered branch", + "email": f"@{name}", + "status": "active", + "created": today, + "last_active": today, + } + registry["branches"].append(entry) + logger.info(f"[sync-registry] Added unregistered branch: {name}") + + # Update total and save + registry["metadata"]["total_branches"] = len(registry["branches"]) + save_result = save_registry(registry_path, registry) + fixed = save_result + + if fixed: + logger.info("[sync-registry] Registry updated successfully") + else: + logger.error("[sync-registry] Failed to save updated registry") + + return { + "stale": stale, + "unregistered": unregistered_list, + "healthy": healthy, + "fixed": fixed, + } diff --git a/src/aipass/spawn/apps/handlers/sync_templates_ops.py b/src/aipass/spawn/apps/handlers/sync_templates_ops.py new file mode 100644 index 00000000..d28d5500 --- /dev/null +++ b/src/aipass/spawn/apps/handlers/sync_templates_ops.py @@ -0,0 +1,149 @@ +# =================== META ==================== +# Name: sync_templates_ops.py +# Description: Template sync handler — implementation logic for template sync +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Template synchronization handler for branch lifecycle management. + +Contains the core sync logic: loading template owners, comparing hashes +between source branches and template files, and optionally pulling updates. +""" + +import hashlib +import json +import shutil +from pathlib import Path + +from aipass.prax.apps.modules.logger import system_logger as logger + +# Repo root — resolved from spawn package location +_REPO_ROOT = Path(__file__).parents[4] # src/aipass/spawn/apps/handlers -> repo root + +# Path to template_owners.json +_TEMPLATE_OWNERS_PATH = ( + Path(__file__).parent / "templates" / "template_owners.json" +) + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def sync_templates(sync: bool = False, dry_run: bool = False) -> dict: + """Pull managed files from authoritative source branches. + + Uses template_owners.json to know which files are "owned" by which branches. + The owner branch has the canonical version — template should match. + + Workflow: + 1. Load template_owners.json from spawn/apps/handlers/templates/ + 2. For each managed file: + a. Find the source file in the owner branch + b. Compare hash with template version + c. If different -> stale + 3. If sync=True and not dry_run: + - Copy source files into template + 4. Report what was synced/stale + + Returns: + Dict with sync results. + """ + # 1. Load template_owners.json + owners_data = _load_template_owners() + managed_files = owners_data.get("managed_files", {}) + + current: list[str] = [] + stale: list[str] = [] + synced: list[str] = [] + errors: list[str] = [] + + if not managed_files: + logger.info("[sync-templates] No managed files configured in template_owners.json") + return { + "managed_files": 0, + "current": current, + "stale": stale, + "synced": synced, + "errors": errors, + } + + # 2. Check each managed file + for file_key, file_info in managed_files.items(): + source_branch = file_info.get("source_branch", "") + source_path = file_info.get("source_path", "") + template_path = file_info.get("template_path", "") + + if not source_branch or not source_path or not template_path: + errors.append(f"Incomplete config for managed file: {file_key}") + continue + + # Resolve paths + source_file = _REPO_ROOT / "src" / "aipass" / source_branch / source_path + template_file = ( + Path(__file__).parent / "templates" / template_path + ) + + if not source_file.exists(): + errors.append(f"Source file not found: {source_file}") + continue + + # Compare hashes + source_hash = _file_hash(source_file) + + if template_file.exists(): + template_hash = _file_hash(template_file) + if source_hash == template_hash: + current.append(file_key) + continue + + # File is stale (different hash or doesn't exist in template) + stale.append(file_key) + + # 3. Sync if requested + if sync and not dry_run: + try: + template_file.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(str(source_file), str(template_file)) + synced.append(file_key) + logger.info(f"[sync-templates] Synced: {file_key} from {source_branch}") + except Exception as exc: + errors.append(f"Failed to sync {file_key}: {exc}") + logger.error(f"[sync-templates] Failed to sync {file_key}: {exc}") + + return { + "managed_files": len(managed_files), + "current": current, + "stale": stale, + "synced": synced, + "errors": errors, + } + + +# ============================================================================= +# INTERNAL HELPERS +# ============================================================================= + +def _load_template_owners() -> dict: + """Load template_owners.json. Returns empty structure if missing.""" + if not _TEMPLATE_OWNERS_PATH.exists(): + logger.warning(f"[sync-templates] template_owners.json not found: {_TEMPLATE_OWNERS_PATH}") + return {"managed_files": {}} + + try: + data = json.loads(_TEMPLATE_OWNERS_PATH.read_text(encoding="utf-8")) + return data + except (json.JSONDecodeError, IOError) as exc: + logger.error(f"[sync-templates] Failed to load template_owners.json: {exc}") + return {"managed_files": {}} + + +def _file_hash(filepath: Path) -> str: + """Compute SHA-256 hash (first 12 chars) of a file.""" + try: + content = filepath.read_bytes() + return hashlib.sha256(content).hexdigest()[:12] + except IOError: + return "" diff --git a/src/aipass/spawn/apps/handlers/update_ops.py b/src/aipass/spawn/apps/handlers/update_ops.py new file mode 100644 index 00000000..6b1f2b49 --- /dev/null +++ b/src/aipass/spawn/apps/handlers/update_ops.py @@ -0,0 +1,508 @@ +# =================== META ==================== +# Name: update_ops.py +# Description: Update handler — implementation logic for branch updates +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Update handler implementation for branch lifecycle management. + +Contains the core update logic: resolving branch paths, executing renames, +additions, JSON updates, pruned file archival, and coordinating the full +update workflow for single and all-branch modes. +""" + +import json +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +from aipass.prax.apps.modules.logger import system_logger as logger + +from aipass.spawn.apps.handlers.meta_ops import ( + compute_file_hash, + generate_branch_meta, + get_template_dir, + load_branch_meta, + load_template_registry, + save_branch_meta, +) +from aipass.spawn.apps.handlers.reconcile import reconcile_branch_state +from aipass.spawn.apps.handlers.change_detection import detect_changes +from aipass.spawn.apps.handlers.json_ops import backup_json, deep_merge +from aipass.spawn.apps.handlers.placeholders import build_replacements_dict, replace_placeholders +from aipass.spawn.apps.handlers.registry import find_registry, load_registry + +# Repo root — resolved from spawn package location +_REPO_ROOT = Path(__file__).parents[4] # src/aipass/spawn/apps/handlers -> repo root + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def update_branch(branch_name: str, dry_run: bool = False, trace: bool = False) -> dict: + """Update a single branch from template. + + Workflow: + 1. Resolve branch path from AIPASS_REGISTRY.json + 2. Load template registry + 3. Load or generate .spawn/.branch_meta.json + 4. Pre-flight reconciliation + 5. Change detection + 6. If dry_run -> print summary, return + 7. Execute changes (renames, additions, JSON updates, pruned archival) + 8. Update branch metadata with new tracking state + 9. Post-flight reconciliation + 10. Return summary dict + + Returns: + Dict with update results including counts and errors. + """ + errors: list[str] = [] + counts = { + "additions": 0, + "renames": 0, + "updates": 0, + "pruned": 0, + "skipped_py": 0, + } + + # ------------------------------------------------------------------ + # 1. Resolve branch path + # ------------------------------------------------------------------ + branch_dir = _resolve_branch_path(branch_name) + if branch_dir is None: + return _result(branch_name, False, counts, [f"Branch '{branch_name}' not found in registry"], dry_run) + + if not branch_dir.is_dir(): + return _result(branch_name, False, counts, [f"Branch directory does not exist: {branch_dir}"], dry_run) + + if trace: + logger.info(f"[update] Resolved {branch_name} -> {branch_dir}") + + # ------------------------------------------------------------------ + # 2. Load template registry + # ------------------------------------------------------------------ + template_dir = get_template_dir() + template_registry = load_template_registry(template_dir) + if template_registry is None: + return _result(branch_name, False, counts, ["Failed to load template registry"], dry_run) + + if trace: + t_files = len(template_registry.get("files", {})) + t_dirs = len(template_registry.get("directories", {})) + logger.info(f"[update] Template registry: {t_files} files, {t_dirs} dirs") + + # ------------------------------------------------------------------ + # 3. Load or generate branch metadata (first-time adoption) + # ------------------------------------------------------------------ + branch_meta = load_branch_meta(branch_dir) + first_time = branch_meta is None + + if first_time: + if trace: + logger.info(f"[update] No branch_meta found — generating initial metadata (adoption)") + branch_meta = generate_branch_meta(branch_dir, template_registry) + if not dry_run: + save_branch_meta(branch_dir, branch_meta) + if trace: + logger.info("[update] Saved initial branch_meta.json") + + # ------------------------------------------------------------------ + # 4. Pre-flight reconciliation + # ------------------------------------------------------------------ + recon = reconcile_branch_state(branch_dir, branch_meta, trace=trace) + if trace: + logger.info( + f"[update] Pre-flight: missing={len(recon['missing_files'])}, " + f"untracked={len(recon['untracked_files'])}, " + f"hash_mismatches={len(recon['hash_mismatches'])}, " + f"missing_dirs={len(recon['missing_dirs'])}" + ) + + # ------------------------------------------------------------------ + # 5. Change detection + # ------------------------------------------------------------------ + changes = detect_changes(template_registry, branch_meta, branch_dir) + + additions = changes.get("additions", []) + updates_list = changes.get("updates", []) + renames = changes.get("renames", []) + pruned = changes.get("pruned", []) + + if trace: + logger.info( + f"[update] Changes detected: additions={len(additions)}, " + f"updates={len(updates_list)}, renames={len(renames)}, " + f"pruned={len(pruned)}" + ) + + # ------------------------------------------------------------------ + # 6. Dry run — just report + # ------------------------------------------------------------------ + if dry_run: + counts["additions"] = len(additions) + counts["renames"] = len(renames) + counts["pruned"] = len(pruned) + + # Count updates vs skipped .py files + for upd in updates_list: + tp = upd.get("template_path", "") + if tp.endswith(".py"): + counts["skipped_py"] += 1 + elif tp.endswith(".json"): + counts["updates"] += 1 + + return _result(branch_name, True, counts, errors, dry_run, + _additions_detail=additions, _updates_detail=updates_list, + _renames_detail=renames, _pruned_detail=pruned) + + # ------------------------------------------------------------------ + # 7. Execute changes + # ------------------------------------------------------------------ + + # Build placeholder replacements for this branch + replacements = build_replacements_dict(branch_dir, branch_name) + + # 7a. RENAMES + for rename_info in renames: + try: + _execute_rename(branch_dir, rename_info, trace) + counts["renames"] += 1 + except Exception as exc: + msg = f"Rename failed for {rename_info.get('file_id')}: {exc}" + errors.append(msg) + logger.error(f"[update] {msg}") + + # 7b. ADDITIONS + for add_info in additions: + try: + _execute_addition(branch_dir, template_dir, add_info, replacements, template_registry, trace) + counts["additions"] += 1 + except Exception as exc: + msg = f"Addition failed for {add_info.get('file_id')}: {exc}" + errors.append(msg) + logger.error(f"[update] {msg}") + + # 7c. JSON UPDATES (skip .py files, deep merge .json) + for upd_info in updates_list: + try: + result = _execute_update(branch_dir, template_dir, upd_info, replacements, trace) + if result == "updated": + counts["updates"] += 1 + elif result == "skipped_py": + counts["skipped_py"] += 1 + except Exception as exc: + msg = f"Update failed for {upd_info.get('file_id')}: {exc}" + errors.append(msg) + logger.error(f"[update] {msg}") + + # 7d. PRUNED — archive, don't delete + for prune_info in pruned: + try: + _execute_prune(branch_dir, prune_info, trace) + counts["pruned"] += 1 + except Exception as exc: + msg = f"Prune/archive failed for {prune_info.get('file_id')}: {exc}" + errors.append(msg) + logger.error(f"[update] {msg}") + + # ------------------------------------------------------------------ + # 8. Refresh branch metadata + # ------------------------------------------------------------------ + updated_meta = generate_branch_meta(branch_dir, template_registry) + updated_meta["metadata"]["last_updated"] = datetime.now().strftime("%Y-%m-%d") + save_branch_meta(branch_dir, updated_meta) + + if trace: + logger.info("[update] Branch metadata refreshed and saved") + + # ------------------------------------------------------------------ + # 9. Post-flight reconciliation + # ------------------------------------------------------------------ + post_recon = reconcile_branch_state(branch_dir, updated_meta, trace=trace) + if trace: + logger.info( + f"[update] Post-flight: missing={len(post_recon['missing_files'])}, " + f"untracked={len(post_recon['untracked_files'])}" + ) + + # ------------------------------------------------------------------ + # 10. Return summary + # ------------------------------------------------------------------ + success = len(errors) == 0 + return _result(branch_name, success, counts, errors, dry_run) + + +def update_all(dry_run: bool = False, trace: bool = False) -> list[dict]: + """Update all branches from AIPASS_REGISTRY.json. + + Iterates through all registered branches, calls update_branch() for each. + Skips spawn itself (can't update yourself). + Returns list of result dicts. + """ + registry_path = find_registry() + registry = load_registry(registry_path) + branches = registry.get("branches", []) + + if not branches: + return [] + + results: list[dict] = [] + + for branch_entry in branches: + name = branch_entry.get("name", "") + lower_name = name.lower() + + # Skip spawn itself + if lower_name == "spawn": + if trace: + logger.info("[update] Skipping spawn (self)") + continue + + if trace: + logger.info(f"[update] Processing branch: {name}") + + try: + result = update_branch(lower_name, dry_run=dry_run, trace=trace) + results.append(result) + except Exception as exc: + logger.error(f"[update] Error updating {name}: {exc}") + results.append({ + "branch": lower_name, + "success": False, + "additions": 0, + "renames": 0, + "updates": 0, + "pruned": 0, + "skipped_py": 0, + "errors": [str(exc)], + "dry_run": dry_run, + }) + + return results + + +# ============================================================================= +# INTERNAL EXECUTION FUNCTIONS +# ============================================================================= + +def _resolve_branch_path(branch_name: str) -> Path | None: + """Resolve a branch name to its absolute directory path via the registry. + + Tries both the exact name and common case variants. + Registry paths are relative to repo root. + """ + registry_path = find_registry() + registry = load_registry(registry_path) + + for branch in registry.get("branches", []): + reg_name = branch.get("name", "") + if reg_name.lower() == branch_name.lower(): + rel_path = branch.get("path", "") + if rel_path: + return (_REPO_ROOT / rel_path).resolve() + + return None + + +def _execute_rename(branch_dir: Path, rename_info: dict, trace: bool) -> None: + """Rename a file within the branch to match template's new path.""" + old_path = branch_dir / rename_info.get("branch_path", "") + new_rel = rename_info.get("new_name", "") # This is the new template-relative path + new_path = branch_dir / new_rel + + if not old_path.exists(): + if trace: + logger.info(f"[update] Rename: source not found, skipping: {old_path}") + return + + # Create parent dirs if needed + new_path.parent.mkdir(parents=True, exist_ok=True) + + old_path.rename(new_path) + if trace: + logger.info(f"[update] Renamed: {rename_info.get('branch_path')} -> {new_rel}") + + +def _execute_addition( + branch_dir: Path, + template_dir: Path, + add_info: dict, + replacements: dict, + template_registry: dict, + trace: bool, +) -> None: + """Copy a new template file/directory to the branch with placeholder replacement.""" + template_path = add_info.get("template_path", "") + file_id = add_info.get("file_id", "") + + if not template_path: + return + + # Check if this is a directory addition (ID starts with 'd') + if file_id.startswith("d"): + dest = branch_dir / template_path + # Apply placeholder replacement to path + dest_str = replace_placeholders(str(dest), replacements) + dest = Path(dest_str) + dest.mkdir(parents=True, exist_ok=True) + if trace: + logger.info(f"[update] Added directory: {template_path}") + return + + # File addition — read from template, apply placeholders, write to branch + source = template_dir / template_path + dest = branch_dir / template_path + + # Apply placeholder replacement to destination path components + dest_rel_str = replace_placeholders(template_path, replacements) + dest = branch_dir / dest_rel_str + + if not source.exists(): + if trace: + logger.info(f"[update] Template source not found, skipping: {source}") + return + + # Create parent directories + dest.parent.mkdir(parents=True, exist_ok=True) + + try: + content = source.read_text(encoding="utf-8") + content = replace_placeholders(content, replacements) + dest.write_text(content, encoding="utf-8") + except (UnicodeDecodeError, UnicodeEncodeError): + # Binary file — copy directly + shutil.copy2(source, dest) + + if trace: + logger.info(f"[update] Added: {dest_rel_str}") + + +def _execute_update( + branch_dir: Path, + template_dir: Path, + upd_info: dict, + replacements: dict, + trace: bool, +) -> str: + """Handle a file update from template. + + Returns: + "updated" if file was merged, "skipped_py" if .py file, "skipped" otherwise. + """ + template_path = upd_info.get("template_path", "") + branch_path = upd_info.get("branch_path", "") + + if not template_path or not branch_path: + return "skipped" + + # NEVER overwrite .py files + if template_path.endswith(".py") or branch_path.endswith(".py"): + if trace: + logger.info(f"[update] SKIPPED .py file (manual review needed): {branch_path}") + return "skipped_py" + + # JSON files — deep merge + if template_path.endswith(".json") and branch_path.endswith(".json"): + source = template_dir / template_path + dest = branch_dir / branch_path + + if not source.exists() or not dest.exists(): + if trace: + logger.info(f"[update] Source or dest missing for JSON merge: {template_path}") + return "skipped" + + try: + # Backup existing file first + backup_json(dest) + + # Load both files + template_content = source.read_text(encoding="utf-8") + template_content = replace_placeholders(template_content, replacements) + template_data = json.loads(template_content) + + existing_data = json.loads(dest.read_text(encoding="utf-8")) + + # Deep merge: template defines structure, existing fills values + merged = deep_merge(template_data, existing_data) + + # Write merged result + dest.write_text( + json.dumps(merged, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + + if trace: + logger.info(f"[update] JSON merged: {branch_path}") + return "updated" + + except (json.JSONDecodeError, IOError) as exc: + logger.error(f"[update] JSON merge failed for {branch_path}: {exc}") + return "skipped" + + # Other file types — skip (don't overwrite) + if trace: + logger.info(f"[update] SKIPPED non-JSON non-py file: {branch_path}") + return "skipped" + + +def _execute_prune(branch_dir: Path, prune_info: dict, trace: bool) -> None: + """Archive a pruned file to .archive/ within the branch.""" + branch_path = prune_info.get("branch_path", "") + if not branch_path: + return + + source = branch_dir / branch_path + if not source.exists(): + if trace: + logger.info(f"[update] Pruned file already gone: {branch_path}") + return + + # Archive destination + archive_dir = branch_dir / ".archive" + archive_dir.mkdir(parents=True, exist_ok=True) + + # Use flat name with path separators replaced to avoid collisions + archive_name = branch_path.replace("/", "__").replace("\\", "__") + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + archive_dest = archive_dir / f"{archive_name}.{timestamp}.pruned" + + if source.is_file(): + shutil.move(str(source), str(archive_dest)) + elif source.is_dir(): + shutil.move(str(source), str(archive_dest)) + + if trace: + logger.info(f"[update] Archived: {branch_path} -> .archive/{archive_dest.name}") + + +# ============================================================================= +# RESULT HELPERS +# ============================================================================= + +def _result( + branch_name: str, + success: bool, + counts: dict, + errors: list[str], + dry_run: bool, + **extra: Any, +) -> dict: + """Build a standardized result dict.""" + result = { + "branch": branch_name, + "success": success, + "additions": counts.get("additions", 0), + "renames": counts.get("renames", 0), + "updates": counts.get("updates", 0), + "pruned": counts.get("pruned", 0), + "skipped_py": counts.get("skipped_py", 0), + "errors": errors, + "dry_run": dry_run, + } + result.update(extra) + return result diff --git a/src/aipass/spawn/apps/modules/delete.py b/src/aipass/spawn/apps/modules/delete.py new file mode 100644 index 00000000..96372a46 --- /dev/null +++ b/src/aipass/spawn/apps/modules/delete.py @@ -0,0 +1,97 @@ +# =================== META ==================== +# Name: delete.py +# Description: Branch deletion — thin CLI layer for archive and deregister +# Version: 1.1.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Delete orchestrator for branch lifecycle management. + +Thin CLI module that parses arguments and delegates to the delete handler. +All implementation logic lives in apps/handlers/delete_ops.py. +""" + +from aipass.prax import logger +# CLI service: from cli.apps.modules import console (via aipass namespace) +from aipass.cli.apps.modules import console + +from aipass.spawn.apps.handlers.delete_ops import delete_branch + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def handle_delete(args: list[str]) -> int: + """Parse args and execute delete. + + Args patterns: + ["@branch"] -> delete with confirmation prompt + ["--yes", "@branch"] -> skip confirmation + ["--dry-run", "@branch"] -> preview only + + Returns exit code (0=success, 1=failure). + """ + if not args: + console.print("[yellow]Usage: drone @spawn delete <@branch> [--yes] [--dry-run][/yellow]") + console.print() + console.print(" [green]@branch[/green] Branch to archive and deregister") + console.print(" [green]--yes[/green] Skip confirmation prompt") + console.print(" [green]--dry-run[/green] Preview what would happen without changes") + return 1 + + dry_run = "--dry-run" in args + skip_confirm = "--yes" in args + + # Filter out flags to find the target + targets = [a for a in args if not a.startswith("--")] + + if not targets: + console.print("[red]Error: specify a branch name (e.g. @api)[/red]") + return 1 + + branch_name = targets[0].lstrip("@").lower() + + try: + result = delete_branch( + branch_name, + confirm=not skip_confirm, + dry_run=dry_run, + ) + except Exception as exc: + logger.error(f"[delete] Unexpected error deleting {branch_name}: {exc}") + console.print(f"[red]Error deleting {branch_name}: {exc}[/red]") + return 1 + + _print_summary(result, dry_run) + return 0 if result.get("success") else 1 + + +# ============================================================================= +# OUTPUT HELPERS +# ============================================================================= + +def _print_summary(result: dict, dry_run: bool) -> None: + """Print a rich summary of the delete operation.""" + branch = result.get("branch", "unknown") + success = result.get("success", False) + mode = "[dim](dry-run)[/dim] " if dry_run else "" + + console.print() + if success: + console.print(f"[green]Delete {mode}{branch}[/green]") + else: + console.print(f"[red]Delete FAILED {mode}{branch}[/red]") + + archive_path = result.get("archive_path", "") + if archive_path: + console.print(f" Archive: {archive_path}") + + console.print(f" Registry updated: {result.get('registry_updated', False)}") + + error = result.get("error", "") + if error: + console.print(f" [red]Error: {error}[/red]") + + console.print() diff --git a/src/aipass/spawn/apps/modules/sync_registry.py b/src/aipass/spawn/apps/modules/sync_registry.py new file mode 100644 index 00000000..1980994c --- /dev/null +++ b/src/aipass/spawn/apps/modules/sync_registry.py @@ -0,0 +1,100 @@ +# =================== META ==================== +# Name: sync_registry.py +# Description: Registry repair — thin CLI layer for registry synchronization +# Version: 1.1.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Registry synchronization for branch lifecycle management. + +Thin CLI module that parses arguments and delegates to the sync handler. +All implementation logic lives in apps/handlers/sync_registry_ops.py. +""" + +from aipass.prax import logger +# CLI service: from cli.apps.modules import console (via aipass namespace) +from aipass.cli.apps.modules import console + +from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def handle_sync_registry(args: list[str]) -> int: + """Parse args and execute sync. + + Args patterns: + [] -> report only (show mismatches) + ["--fix"] -> auto-repair mismatches + + Returns exit code (0=success, 1=failure). + """ + if args and args[0] in ["--help", "-h"]: + console.print("[yellow]Usage: drone @spawn sync-registry [--fix][/yellow]") + console.print() + console.print(" [green](no args)[/green] Report mismatches between registry and filesystem") + console.print(" [green]--fix[/green] Auto-repair: remove stale, add unregistered") + return 0 + + fix = "--fix" in args + + try: + result = sync_registry(fix=fix) + except Exception as exc: + logger.error(f"[sync-registry] Unexpected error: {exc}") + console.print(f"[red]Error: {exc}[/red]") + return 1 + + _print_summary(result) + return 0 + + +# ============================================================================= +# OUTPUT HELPERS +# ============================================================================= + +def _print_summary(result: dict) -> None: + """Print a rich summary of the sync operation.""" + stale = result.get("stale", []) + unregistered = result.get("unregistered", []) + healthy = result.get("healthy", []) + fixed = result.get("fixed", False) + + console.print() + console.print("[bold]Registry Sync Report[/bold]") + console.print() + + # Healthy + if healthy: + console.print(f" [green]Healthy ({len(healthy)}):[/green]") + for name in sorted(healthy): + console.print(f" {name}") + + # Stale + if stale: + console.print(f" [red]Stale ({len(stale)}):[/red] [dim]registered but missing/invalid[/dim]") + for name in sorted(stale): + console.print(f" {name}") + else: + console.print(" [green]No stale entries[/green]") + + # Unregistered + if unregistered: + console.print(f" [yellow]Unregistered ({len(unregistered)}):[/yellow] [dim]on disk but not in registry[/dim]") + for name in sorted(unregistered): + console.print(f" {name}") + else: + console.print(" [green]No unregistered branches[/green]") + + # Fix status + if fixed: + console.print() + console.print(" [green]Registry has been repaired.[/green]") + elif stale or unregistered: + console.print() + console.print(" [dim]Run with --fix to auto-repair.[/dim]") + + console.print() diff --git a/src/aipass/spawn/apps/modules/sync_templates.py b/src/aipass/spawn/apps/modules/sync_templates.py new file mode 100644 index 00000000..c46c944c --- /dev/null +++ b/src/aipass/spawn/apps/modules/sync_templates.py @@ -0,0 +1,109 @@ +# =================== META ==================== +# Name: sync_templates.py +# Description: Template sync — thin CLI layer for template synchronization +# Version: 1.1.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Template synchronization for branch lifecycle management. + +Thin CLI module that parses arguments and delegates to the sync handler. +All implementation logic lives in apps/handlers/sync_templates_ops.py. +""" + +from aipass.prax import logger +# CLI service: from cli.apps.modules import console (via aipass namespace) +from aipass.cli.apps.modules import console + +from aipass.spawn.apps.handlers.sync_templates_ops import sync_templates + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def handle_sync_templates(args: list[str]) -> int: + """Parse args and execute template sync. + + Args patterns: + [] -> report status (which files are stale) + ["--status"] -> same as no args + ["--sync"] -> actually pull and update + ["--dry-run"] -> preview changes + + Returns exit code (0=success, 1=failure). + """ + if args and args[0] in ["--help", "-h"]: + console.print("[yellow]Usage: drone @spawn sync-templates [--status|--sync|--dry-run][/yellow]") + console.print() + console.print(" [green](no args)[/green] Report which managed template files are stale") + console.print(" [green]--status[/green] Same as no args") + console.print(" [green]--sync[/green] Pull updated files from source branches into template") + console.print(" [green]--dry-run[/green] Preview what would be synced") + return 0 + + sync = "--sync" in args + dry_run = "--dry-run" in args + + try: + result = sync_templates(sync=sync, dry_run=dry_run) + except Exception as exc: + logger.error(f"[sync-templates] Unexpected error: {exc}") + console.print(f"[red]Error: {exc}[/red]") + return 1 + + _print_summary(result, dry_run) + return 0 + + +# ============================================================================= +# OUTPUT HELPERS +# ============================================================================= + +def _print_summary(result: dict, dry_run: bool) -> None: + """Print a rich summary of the template sync operation.""" + managed = result.get("managed_files", 0) + current = result.get("current", []) + stale = result.get("stale", []) + synced = result.get("synced", []) + errors = result.get("errors", []) + mode = "[dim](dry-run)[/dim] " if dry_run else "" + + console.print() + console.print(f"[bold]Template Sync Report {mode}[/bold]") + console.print() + + if managed == 0: + console.print(" [dim]No managed files configured in template_owners.json[/dim]") + console.print(" [dim]Add entries to spawn/apps/handlers/templates/template_owners.json[/dim]") + console.print() + return + + console.print(f" Managed files: {managed}") + + if current: + console.print(f" [green]Current ({len(current)}):[/green]") + for name in current: + console.print(f" {name}") + + if stale: + console.print(f" [yellow]Stale ({len(stale)}):[/yellow]") + for name in stale: + console.print(f" {name}") + + if synced: + console.print(f" [green]Synced ({len(synced)}):[/green]") + for name in synced: + console.print(f" {name}") + + if errors: + console.print(f" [red]Errors ({len(errors)}):[/red]") + for err in errors: + console.print(f" {err}") + + if stale and not synced and not dry_run: + console.print() + console.print(" [dim]Run with --sync to pull updates from source branches.[/dim]") + + console.print() diff --git a/src/aipass/spawn/apps/modules/update.py b/src/aipass/spawn/apps/modules/update.py new file mode 100644 index 00000000..70e20807 --- /dev/null +++ b/src/aipass/spawn/apps/modules/update.py @@ -0,0 +1,177 @@ +# =================== META ==================== +# Name: update.py +# Description: Update orchestrator — thin CLI layer for branch updates +# Version: 1.1.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Update orchestrator for branch lifecycle management. + +Thin CLI module that parses arguments and delegates to the update handler. +All implementation logic lives in apps/handlers/update_ops.py. +""" + +from aipass.prax import logger +# CLI service: from cli.apps.modules import console (via aipass namespace) +from aipass.cli.apps.modules import console + +from aipass.spawn.apps.handlers.update_ops import ( + update_branch, + update_all, +) + + +# ============================================================================= +# PUBLIC API +# ============================================================================= + +def handle_update(args: list[str]) -> int: + """Parse args and dispatch to update_branch or update_all. + + Args patterns: + ["@branch"] -> update single branch + ["--all"] -> update all branches + ["--dry-run", "@branch"] -> preview mode + ["--dry-run", "--all"] -> preview all + ["--trace", "@branch"] -> verbose logging + + Returns exit code (0=success, 1=failure). + """ + if not args: + console.print("[yellow]Usage: drone @spawn update <@branch|--all> [--dry-run] [--trace][/yellow]") + console.print() + console.print(" [green]@branch[/green] Update a single branch") + console.print(" [green]--all[/green] Update all registered branches") + console.print(" [green]--dry-run[/green] Preview changes without modifying files") + console.print(" [green]--trace[/green] Enable verbose logging") + return 1 + + dry_run = "--dry-run" in args + trace = "--trace" in args + + # Filter out flags to find the target + targets = [a for a in args if not a.startswith("--")] + + if "--all" in args: + results = update_all(dry_run=dry_run, trace=trace) + _print_all_summary(results, dry_run) + # Return 0 if all succeeded, 1 if any failed + return 0 if all(r.get("success") for r in results) else 1 + + if not targets: + console.print("[red]Error: specify a branch name (e.g. @prax) or --all[/red]") + return 1 + + # Take the first target, strip leading @ + branch_name = targets[0].lstrip("@") + + try: + result = update_branch(branch_name, dry_run=dry_run, trace=trace) + except Exception as exc: + logger.error(f"[update] Unexpected error updating {branch_name}: {exc}") + console.print(f"[red]Error updating {branch_name}: {exc}[/red]") + return 1 + + _print_branch_summary(result, dry_run) + return 0 if result.get("success") else 1 + + +# ============================================================================= +# OUTPUT HELPERS +# ============================================================================= + +def _print_branch_summary(result: dict, dry_run: bool) -> None: + """Print a rich summary for a single branch update.""" + branch = result.get("branch", "unknown") + success = result.get("success", False) + mode = "[dim](dry-run)[/dim] " if dry_run else "" + + console.print() + if success: + console.print(f"[green]Update {mode}{branch}[/green]") + else: + console.print(f"[red]Update FAILED {mode}{branch}[/red]") + + console.print(f" Additions: {result.get('additions', 0)}") + console.print(f" Renames: {result.get('renames', 0)}") + console.print(f" Updates: {result.get('updates', 0)}") + console.print(f" Pruned: {result.get('pruned', 0)}") + console.print(f" Skipped py: {result.get('skipped_py', 0)}") + + errs = result.get("errors", []) + if errs: + console.print(f" [red]Errors: {len(errs)}[/red]") + for e in errs: + console.print(f" - {e}") + + # In dry-run mode, show details of what would change + if dry_run: + additions_detail = result.get("_additions_detail", []) + if isinstance(additions_detail, list) and additions_detail: + console.print() + console.print(" [cyan]Would add:[/cyan]") + for a in additions_detail: + if isinstance(a, dict): + console.print(f" + {a.get('template_path', '')}") + + updates_detail = result.get("_updates_detail", []) + if isinstance(updates_detail, list) and updates_detail: + console.print(" [cyan]Would update/skip:[/cyan]") + for u in updates_detail: + if isinstance(u, dict): + tp = u.get("template_path", "") + bp = u.get("branch_path", tp) + if tp.endswith(".py"): + console.print(f" ! {bp} [dim](skipped - .py manual review)[/dim]") + elif tp.endswith(".json"): + console.print(f" ~ {bp} [dim](JSON deep merge)[/dim]") + else: + console.print(f" - {bp} [dim](skipped - non-JSON)[/dim]") + + renames_detail = result.get("_renames_detail", []) + if isinstance(renames_detail, list) and renames_detail: + console.print(" [cyan]Would rename:[/cyan]") + for r in renames_detail: + if isinstance(r, dict): + console.print(f" > {r.get('old_name', '')} -> {r.get('new_name', '')}") + + pruned_detail = result.get("_pruned_detail", []) + if isinstance(pruned_detail, list) and pruned_detail: + console.print(" [cyan]Would archive (pruned):[/cyan]") + for p in pruned_detail: + if isinstance(p, dict): + console.print(f" - {p.get('branch_path', '')}") + + console.print() + + +def _print_all_summary(results: list[dict], dry_run: bool) -> None: + """Print summary for update_all.""" + mode = "(dry-run) " if dry_run else "" + + console.print() + console.print(f"[bold]Update All {mode}— {len(results)} branches[/bold]") + console.print() + + total_add = sum(r.get("additions", 0) for r in results) + total_ren = sum(r.get("renames", 0) for r in results) + total_upd = sum(r.get("updates", 0) for r in results) + total_prn = sum(r.get("pruned", 0) for r in results) + total_skip = sum(r.get("skipped_py", 0) for r in results) + total_err = sum(len(r.get("errors", [])) for r in results) + + for r in results: + branch = r.get("branch", "?") + status = "[green]OK[/green]" if r.get("success") else "[red]FAIL[/red]" + changes = r.get("additions", 0) + r.get("renames", 0) + r.get("updates", 0) + r.get("pruned", 0) + console.print(f" {status} {branch}: {changes} changes") + + console.print() + console.print(f" Totals: +{total_add} added, ~{total_upd} updated, " + f">{total_ren} renamed, -{total_prn} pruned, " + f"!{total_skip} py-skipped") + + if total_err: + console.print(f" [red]{total_err} errors across all branches[/red]") + console.print() diff --git a/src/aipass/spawn/apps/plugins/__init__.py b/src/aipass/spawn/apps/plugins/__init__.py new file mode 100644 index 00000000..85bfafe9 --- /dev/null +++ b/src/aipass/spawn/apps/plugins/__init__.py @@ -0,0 +1 @@ +"""Spawn plugins package — extension point for spawn customization.""" diff --git a/src/aipass/spawn/apps/spawn.py b/src/aipass/spawn/apps/spawn.py index 427e0676..36ea30ac 100644 --- a/src/aipass/spawn/apps/spawn.py +++ b/src/aipass/spawn/apps/spawn.py @@ -35,10 +35,10 @@ def print_help(): console.print("[bold cyan]COMMANDS:[/bold cyan]") console.print() console.print(" [green]create[/green] Create a new branch from template") - console.print(" [dim]update <@branch|--all>[/dim] Update branch(es) from template [dim][not yet implemented][/dim]") - console.print(" [dim]delete <@branch>[/dim] Archive and deregister branch [dim][not yet implemented][/dim]") - console.print(" [dim]sync-registry[/dim] Repair registry against filesystem [dim][not yet implemented][/dim]") - console.print(" [dim]sync-templates[/dim] Pull managed files from source [dim][not yet implemented][/dim]") + console.print(" [green]update[/green] <@branch|--all> Update branch(es) from template") + console.print(" [green]delete[/green] <@branch> Archive and deregister branch") + console.print(" [green]sync-registry[/green] Repair registry against filesystem") + console.print(" [green]sync-templates[/green] Pull managed files from source") console.print(" [dim]regenerate-registry[/dim] Regenerate template registry hashes [dim][not yet implemented][/dim]") console.print() console.print("[bold cyan]OPTIONS:[/bold cyan]") @@ -112,12 +112,24 @@ def main(): if command == "create": return handle_create(remaining) + if command == "update": + from aipass.spawn.apps.modules.update import handle_update + return handle_update(remaining) + + if command == "delete": + from aipass.spawn.apps.modules.delete import handle_delete + return handle_delete(remaining) + + if command == "sync-registry": + from aipass.spawn.apps.modules.sync_registry import handle_sync_registry + return handle_sync_registry(remaining) + + if command == "sync-templates": + from aipass.spawn.apps.modules.sync_templates import handle_sync_templates + return handle_sync_templates(remaining) + # Stub commands — planned but not yet implemented stub_commands = { - "update": "Update branch(es) from template", - "delete": "Archive and deregister branch", - "sync-registry": "Repair registry against filesystem", - "sync-templates": "Pull managed files from source branches", "regenerate-registry": "Regenerate template registry hashes", } diff --git a/src/aipass/spawn/docs/.gitkeep b/src/aipass/spawn/docs/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/src/aipass/spawn/drone_adapter.py b/src/aipass/spawn/drone_adapter.py deleted file mode 100644 index 839d6261..00000000 --- a/src/aipass/spawn/drone_adapter.py +++ /dev/null @@ -1,97 +0,0 @@ -""" -Spawn drone adapter — bridges drone routing to spawn commands. - -Drone discovers this module via aipass.drone.modules._MODULE_REGISTRY -and routes `drone @spawn [args]` here. -""" - -import sys -from io import StringIO - -DRONE_MODULE = { - "name": "spawn", - "version": "1.0.0", - "description": "Agent creation and branch lifecycle management", -} - - -def handle_command(command: str, args: list[str] | None = None) -> dict: - """Route a drone command to spawn's entry point. - - Captures stdout/stderr and returns as dict for drone CLI to print. - """ - if args is None: - args = [] - - # Build argv as if `spawn [args]` was called - original_argv = sys.argv - old_stdout = sys.stdout - old_stderr = sys.stderr - captured_out = StringIO() - captured_err = StringIO() - - try: - sys.argv = ["spawn", command] + args - sys.stdout = captured_out - sys.stderr = captured_err - - # Import here to avoid circular imports at module level - from aipass.spawn.apps.spawn import main - exit_code = main() - except SystemExit as e: - exit_code = e.code if e.code is not None else 0 - except Exception as e: - captured_err.write(str(e)) - exit_code = 1 - finally: - sys.argv = original_argv - sys.stdout = old_stdout - sys.stderr = old_stderr - - return { - "stdout": captured_out.getvalue(), - "stderr": captured_err.getvalue(), - "exit_code": exit_code if isinstance(exit_code, int) else 1, - } - - -def get_help(command: str | None = None) -> str: - """Return help text for spawn.""" - if command: - result = handle_command(command, ["--help"]) - return result.get("stdout", "") or result.get("stderr", "") - - return ( - "spawn — Agent creation and branch lifecycle management\n" - "\n" - "Commands:\n" - " create Create new branch from template\n" - " update <@branch|--all> Update branch(es) from template [not yet implemented]\n" - " delete <@branch> Archive and deregister branch [not yet implemented]\n" - " sync-registry Repair registry against filesystem [not yet implemented]\n" - " sync-templates Pull managed files from source [not yet implemented]\n" - " regenerate-registry Regenerate template registry hashes [not yet implemented]\n" - "\n" - "Usage via drone:\n" - " drone @spawn create [--role ...] [--traits ...]\n" - " drone @spawn update @branch_name\n" - " drone @spawn delete @branch_name\n" - " drone @spawn sync-registry\n" - " drone @spawn --help\n" - ) - - -def get_introspective() -> str: - """Discovery mode: show what spawn has available.""" - try: - from pathlib import Path - template_dir = Path(__file__).parent / "templates" - templates = [d.name for d in template_dir.iterdir() if d.is_dir()] if template_dir.exists() else [] - template_list = ", ".join(templates) if templates else "none" - return ( - f"@spawn — Agent creation and branch lifecycle management\n" - f" Templates available: {template_list}\n" - f" Run 'drone @spawn --help' for usage\n" - ) - except Exception: - return "@spawn — Agent creation and branch lifecycle management (run 'drone @spawn --help' for usage)\n" diff --git a/src/aipass/spawn/pytest.ini b/src/aipass/spawn/pytest.ini new file mode 100644 index 00000000..4ecb1ad2 --- /dev/null +++ b/src/aipass/spawn/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* diff --git a/src/aipass/spawn/spawn_json/.gitkeep b/src/aipass/spawn/spawn_json/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/src/aipass/spawn/tests/conftest.py b/src/aipass/spawn/tests/conftest.py new file mode 100644 index 00000000..4425b347 --- /dev/null +++ b/src/aipass/spawn/tests/conftest.py @@ -0,0 +1 @@ +"""Shared test fixtures for spawn test suite.""" diff --git a/src/aipass/spawn/tests/test_drone_adapter.py b/src/aipass/spawn/tests/test_drone_adapter.py index cd8fe2dc..87693e6b 100644 --- a/src/aipass/spawn/tests/test_drone_adapter.py +++ b/src/aipass/spawn/tests/test_drone_adapter.py @@ -57,22 +57,22 @@ def test_stub_delete_not_implemented(): assert result["exit_code"] == 1 -def test_stub_sync_registry_not_implemented(): - """Stub command 'sync-registry' should return not-yet-implemented message.""" +def test_sync_registry_report(): + """sync-registry with no args should produce a report (exit 0).""" from aipass.spawn.drone_adapter import handle_command result = handle_command("sync-registry", []) assert isinstance(result, dict) - assert result["exit_code"] == 1 + assert result["exit_code"] == 0 -def test_stub_sync_templates_not_implemented(): - """Stub command 'sync-templates' should return not-yet-implemented message.""" +def test_sync_templates_report(): + """sync-templates with no args should produce a status report (exit 0).""" from aipass.spawn.drone_adapter import handle_command result = handle_command("sync-templates", []) assert isinstance(result, dict) - assert result["exit_code"] == 1 + assert result["exit_code"] == 0 def test_stub_regenerate_registry_not_implemented(): diff --git a/src/aipass/spawn/tests/test_handlers.py b/src/aipass/spawn/tests/test_handlers.py new file mode 100644 index 00000000..a0306cf3 --- /dev/null +++ b/src/aipass/spawn/tests/test_handlers.py @@ -0,0 +1,786 @@ +# =================== META ==================== +# Name: test_handlers.py +# Description: Tests for spawn update handler modules +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Tests for spawn handler modules: meta_ops, reconcile, change_detection, json_ops.""" + +import json +import pytest +from pathlib import Path + + +# ============================================================================= +# meta_ops tests +# ============================================================================= + +class TestComputeFileHash: + """Tests for compute_file_hash().""" + + def test_returns_12_char_hex_string(self, tmp_path): + """Hash result should be exactly 12 hex characters.""" + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + test_file = tmp_path / "test.txt" + test_file.write_text("hello world", encoding="utf-8") + + result = compute_file_hash(test_file) + + assert len(result) == 12 + assert all(c in "0123456789abcdef" for c in result) + + def test_same_content_same_hash(self, tmp_path): + """Same content should produce same hash.""" + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + file_a = tmp_path / "a.txt" + file_b = tmp_path / "b.txt" + file_a.write_text("identical content", encoding="utf-8") + file_b.write_text("identical content", encoding="utf-8") + + assert compute_file_hash(file_a) == compute_file_hash(file_b) + + def test_different_content_different_hash(self, tmp_path): + """Different content should produce different hashes.""" + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + file_a = tmp_path / "a.txt" + file_b = tmp_path / "b.txt" + file_a.write_text("content A", encoding="utf-8") + file_b.write_text("content B", encoding="utf-8") + + assert compute_file_hash(file_a) != compute_file_hash(file_b) + + def test_nonexistent_file_returns_empty(self, tmp_path): + """Non-existent file should return empty string.""" + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + result = compute_file_hash(tmp_path / "nonexistent.txt") + assert result == "" + + def test_directory_returns_empty(self, tmp_path): + """Directory (not file) should return empty string.""" + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + result = compute_file_hash(tmp_path) + assert result == "" + + +class TestGenerateBranchMeta: + """Tests for generate_branch_meta().""" + + def test_produces_valid_structure(self, tmp_path): + """Generated meta should have metadata, file_tracking, directory_tracking.""" + from aipass.spawn.apps.handlers.meta_ops import generate_branch_meta + + # Create a mock branch directory with some files + branch_dir = tmp_path / "test_branch" + branch_dir.mkdir() + (branch_dir / "README.md").write_text("# Test", encoding="utf-8") + apps_dir = branch_dir / "apps" + apps_dir.mkdir() + (apps_dir / "__init__.py").write_text("", encoding="utf-8") + + # Template registry that matches some of these files + template_registry = { + "metadata": {"version": "1.0.0"}, + "files": { + "f001": { + "current_name": "README.md", + "path": "README.md", + "content_hash": "aabbccddee11", + }, + "f002": { + "current_name": "__init__.py", + "path": "apps/__init__.py", + "content_hash": "e3b0c44298fc", + }, + }, + "directories": { + "d001": { + "current_name": "apps", + "path": "apps", + }, + }, + } + + result = generate_branch_meta(branch_dir, template_registry) + + # Check top-level structure + assert "metadata" in result + assert "file_tracking" in result + assert "directory_tracking" in result + + # Check metadata fields + assert result["metadata"]["version"] == "1.0.0" + assert result["metadata"]["template_version"] == "1.0.0" + assert result["metadata"]["branch_name"] == "test_branch" + assert "last_updated" in result["metadata"] + + def test_matches_files_by_path(self, tmp_path): + """Files matching template paths should be tracked with correct IDs.""" + from aipass.spawn.apps.handlers.meta_ops import generate_branch_meta + + branch_dir = tmp_path / "my_branch" + branch_dir.mkdir() + (branch_dir / "README.md").write_text("# Hello", encoding="utf-8") + + template_registry = { + "metadata": {"version": "1.0.0"}, + "files": { + "f021": { + "current_name": "README.md", + "path": "README.md", + "content_hash": "aabbccddee11", + }, + }, + "directories": {}, + } + + result = generate_branch_meta(branch_dir, template_registry) + + assert "f021" in result["file_tracking"] + assert result["file_tracking"]["f021"]["current_name"] == "README.md" + assert result["file_tracking"]["f021"]["current_path"] == "README.md" + assert len(result["file_tracking"]["f021"]["content_hash"]) == 12 + + def test_matches_directories(self, tmp_path): + """Directories matching template paths should be tracked.""" + from aipass.spawn.apps.handlers.meta_ops import generate_branch_meta + + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + (branch_dir / "apps").mkdir() + (branch_dir / "tests").mkdir() + + template_registry = { + "metadata": {"version": "1.0.0"}, + "files": {}, + "directories": { + "d001": {"current_name": "apps", "path": "apps"}, + "d002": {"current_name": "tests", "path": "tests"}, + }, + } + + result = generate_branch_meta(branch_dir, template_registry) + + assert "d001" in result["directory_tracking"] + assert "d002" in result["directory_tracking"] + assert result["directory_tracking"]["d001"]["current_path"] == "apps" + + +class TestLoadSaveBranchMeta: + """Tests for load_branch_meta() and save_branch_meta().""" + + def test_save_and_load_roundtrip(self, tmp_path): + """Saved metadata should load back identically.""" + from aipass.spawn.apps.handlers.meta_ops import load_branch_meta, save_branch_meta + + meta = { + "metadata": {"version": "1.0.0", "branch_name": "test"}, + "file_tracking": {"f001": {"current_name": "a.txt", "content_hash": "abc123def456"}}, + "directory_tracking": {}, + } + + assert save_branch_meta(tmp_path, meta) is True + + loaded = load_branch_meta(tmp_path) + assert loaded is not None + assert loaded["metadata"]["branch_name"] == "test" + assert loaded["file_tracking"]["f001"]["content_hash"] == "abc123def456" + + def test_load_missing_returns_none(self, tmp_path): + """Loading from directory without .branch_meta.json should return None.""" + from aipass.spawn.apps.handlers.meta_ops import load_branch_meta + + result = load_branch_meta(tmp_path) + assert result is None + + +class TestGetTemplateDir: + """Tests for get_template_dir().""" + + def test_returns_path(self): + """Should return a Path object pointing to agent.template.""" + from aipass.spawn.apps.handlers.meta_ops import get_template_dir + + result = get_template_dir() + assert isinstance(result, Path) + assert result.name == "agent.template" + + +# ============================================================================= +# deep_merge tests +# ============================================================================= + +class TestDeepMerge: + """Tests for deep_merge().""" + + def test_preserves_existing_values(self): + """Existing scalar values should be kept over template defaults.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"name": "default", "status": "new"} + existing = {"name": "my_custom_name", "status": "active"} + + result = deep_merge(template, existing) + + assert result["name"] == "my_custom_name" + assert result["status"] == "active" + + def test_adds_template_keys(self): + """Keys in template but not in existing should be added.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"name": "", "version": "1.0.0", "new_field": "default_val"} + existing = {"name": "mine"} + + result = deep_merge(template, existing) + + assert result["name"] == "mine" + assert result["version"] == "1.0.0" + assert result["new_field"] == "default_val" + + def test_keeps_extra_existing_keys(self): + """Keys in existing but not in template should be preserved.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"name": ""} + existing = {"name": "mine", "custom_key": "custom_value"} + + result = deep_merge(template, existing) + + assert result["name"] == "mine" + assert result["custom_key"] == "custom_value" + + def test_nested_dict_merge(self): + """Nested dicts should be merged recursively.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"metadata": {"version": "2.0", "new_key": "new_default"}} + existing = {"metadata": {"version": "1.0", "user_key": "user_val"}} + + result = deep_merge(template, existing) + + assert result["metadata"]["version"] == "1.0" # preserved + assert result["metadata"]["new_key"] == "new_default" # added + assert result["metadata"]["user_key"] == "user_val" # kept + + def test_lists_keep_existing(self): + """Non-empty existing lists should be preserved.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"items": ["a", "b"]} + existing = {"items": ["x", "y", "z"]} + + result = deep_merge(template, existing) + + assert result["items"] == ["x", "y", "z"] + + def test_empty_existing_list_uses_template(self): + """Empty existing list should use template list if non-empty.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"items": ["default_item"]} + existing = {"items": []} + + result = deep_merge(template, existing) + + assert result["items"] == ["default_item"] + + def test_none_existing_uses_template(self): + """None existing should return template data.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"key": "value"} + result = deep_merge(template, None) + + assert result == {"key": "value"} + + def test_empty_string_existing_uses_template(self): + """Empty string existing with non-empty template should use template.""" + from aipass.spawn.apps.handlers.json_ops import deep_merge + + template = {"name": "template_default"} + existing = {"name": ""} + + result = deep_merge(template, existing) + + assert result["name"] == "template_default" + + +# ============================================================================= +# detect_changes tests +# ============================================================================= + +class TestDetectChanges: + """Tests for detect_changes().""" + + def test_detects_additions(self, tmp_path): + """Files in template but not in branch should be additions.""" + from aipass.spawn.apps.handlers.change_detection import detect_changes + + template_registry = { + "files": { + "f001": {"current_name": "old.txt", "path": "old.txt", "content_hash": "aaa111"}, + "f002": {"current_name": "new.txt", "path": "new.txt", "content_hash": "bbb222"}, + }, + "directories": {}, + } + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "old.txt", + "current_name": "old.txt", + "current_path": "old.txt", + "content_hash": "aaa111", + }, + }, + "directory_tracking": {}, + } + + result = detect_changes(template_registry, branch_meta, tmp_path) + + assert len(result["additions"]) == 1 + assert result["additions"][0]["file_id"] == "f002" + assert result["additions"][0]["template_path"] == "new.txt" + + def test_detects_pruned(self, tmp_path): + """Files in branch but not in template should be pruned.""" + from aipass.spawn.apps.handlers.change_detection import detect_changes + + template_registry = { + "files": { + "f001": {"current_name": "kept.txt", "path": "kept.txt", "content_hash": "aaa"}, + }, + "directories": {}, + } + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "kept.txt", + "current_name": "kept.txt", + "current_path": "kept.txt", + "content_hash": "aaa", + }, + "f099": { + "template_name": "removed.txt", + "current_name": "removed.txt", + "current_path": "removed.txt", + "content_hash": "zzz", + }, + }, + "directory_tracking": {}, + } + + result = detect_changes(template_registry, branch_meta, tmp_path) + + assert len(result["pruned"]) == 1 + assert result["pruned"][0]["file_id"] == "f099" + + def test_detects_updates(self, tmp_path): + """Files with same ID but different hashes should be updates.""" + from aipass.spawn.apps.handlers.change_detection import detect_changes + + template_registry = { + "files": { + "f001": {"current_name": "config.json", "path": "config.json", "content_hash": "newhash12345"}, + }, + "directories": {}, + } + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "config.json", + "current_name": "config.json", + "current_path": "config.json", + "content_hash": "oldhash67890", + }, + }, + "directory_tracking": {}, + } + + result = detect_changes(template_registry, branch_meta, tmp_path) + + assert len(result["updates"]) == 1 + assert result["updates"][0]["file_id"] == "f001" + assert result["updates"][0]["template_hash"] == "newhash12345" + assert result["updates"][0]["branch_hash"] == "oldhash67890" + + def test_detects_renames(self, tmp_path): + """Files with same ID but different template paths should be renames.""" + from aipass.spawn.apps.handlers.change_detection import detect_changes + + template_registry = { + "files": { + "f001": { + "current_name": "new_name.txt", + "path": "docs/new_name.txt", + "content_hash": "samehash1234", + }, + }, + "directories": {}, + } + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "docs/old_name.txt", + "current_name": "old_name.txt", + "current_path": "docs/old_name.txt", + "content_hash": "samehash1234", + }, + }, + "directory_tracking": {}, + } + + result = detect_changes(template_registry, branch_meta, tmp_path) + + assert len(result["renames"]) == 1 + assert result["renames"][0]["file_id"] == "f001" + assert result["renames"][0]["old_name"] == "docs/old_name.txt" + assert result["renames"][0]["new_name"] == "docs/new_name.txt" + + def test_no_changes(self, tmp_path): + """Identical template and branch should produce no changes.""" + from aipass.spawn.apps.handlers.change_detection import detect_changes + + template_registry = { + "files": { + "f001": {"current_name": "a.txt", "path": "a.txt", "content_hash": "abc123"}, + }, + "directories": { + "d001": {"current_name": "apps", "path": "apps"}, + }, + } + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "a.txt", + "current_name": "a.txt", + "current_path": "a.txt", + "content_hash": "abc123", + }, + }, + "directory_tracking": { + "d001": { + "template_name": "apps", + "current_name": "apps", + "current_path": "apps", + }, + }, + } + + result = detect_changes(template_registry, branch_meta, tmp_path) + + assert result["additions"] == [] + assert result["updates"] == [] + assert result["renames"] == [] + assert result["pruned"] == [] + + +# ============================================================================= +# reconcile_branch_state tests +# ============================================================================= + +class TestReconcileBranchState: + """Tests for reconcile_branch_state().""" + + def test_detects_missing_files(self, tmp_path): + """Files tracked in meta but not on disk should be reported as missing.""" + from aipass.spawn.apps.handlers.reconcile import reconcile_branch_state + + # Branch with no actual files + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "README.md", + "current_name": "README.md", + "current_path": "README.md", + "content_hash": "abc123def456", + }, + }, + "directory_tracking": {}, + } + + result = reconcile_branch_state(branch_dir, branch_meta) + + assert result["needs_update"] is True + assert len(result["missing_files"]) == 1 + assert result["missing_files"][0]["file_id"] == "f001" + assert result["missing_files"][0]["path"] == "README.md" + + def test_detects_hash_mismatches(self, tmp_path): + """Files with changed content should be reported as hash mismatches.""" + from aipass.spawn.apps.handlers.reconcile import reconcile_branch_state + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + # Create file with known content + readme = branch_dir / "README.md" + readme.write_text("modified content", encoding="utf-8") + actual_hash = compute_file_hash(readme) + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "README.md", + "current_name": "README.md", + "current_path": "README.md", + "content_hash": "differenthash", # stale hash + }, + }, + "directory_tracking": {}, + } + + result = reconcile_branch_state(branch_dir, branch_meta) + + assert result["needs_update"] is True + assert len(result["hash_mismatches"]) == 1 + assert result["hash_mismatches"][0]["file_id"] == "f001" + assert result["hash_mismatches"][0]["current_hash"] == actual_hash + + def test_clean_state_no_update_needed(self, tmp_path): + """Consistent state should report needs_update=False.""" + from aipass.spawn.apps.handlers.reconcile import reconcile_branch_state + from aipass.spawn.apps.handlers.meta_ops import compute_file_hash + + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + readme = branch_dir / "README.md" + readme.write_text("stable content", encoding="utf-8") + actual_hash = compute_file_hash(readme) + + branch_meta = { + "file_tracking": { + "f001": { + "template_name": "README.md", + "current_name": "README.md", + "current_path": "README.md", + "content_hash": actual_hash, + }, + }, + "directory_tracking": {}, + } + + result = reconcile_branch_state(branch_dir, branch_meta) + + assert result["needs_update"] is False + assert result["missing_files"] == [] + assert result["hash_mismatches"] == [] + + def test_detects_missing_directories(self, tmp_path): + """Tracked directories missing from disk should be reported.""" + from aipass.spawn.apps.handlers.reconcile import reconcile_branch_state + + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + branch_meta = { + "file_tracking": {}, + "directory_tracking": { + "d001": { + "template_name": "apps", + "current_name": "apps", + "current_path": "apps", + }, + }, + } + + result = reconcile_branch_state(branch_dir, branch_meta) + + assert result["needs_update"] is True + assert len(result["missing_dirs"]) == 1 + assert result["missing_dirs"][0]["dir_id"] == "d001" + + +# ============================================================================= +# apply_migrations tests +# ============================================================================= + +class TestApplyMigrations: + """Tests for apply_migrations().""" + + def test_key_rename(self): + """key_rename should move a key's value to the new key name.""" + from aipass.spawn.apps.handlers.json_ops import apply_migrations + + data = {"old_key": "value", "other": "keep"} + migrations = [ + { + "id": "m001", + "operations": [ + {"type": "key_rename", "from": "old_key", "to": "new_key"}, + ], + }, + ] + + result = apply_migrations(data, migrations) + + assert "new_key" in result + assert result["new_key"] == "value" + assert "old_key" not in result + assert result["other"] == "keep" + + def test_key_rename_skip_if_target_exists(self): + """key_rename should skip if target key already exists.""" + from aipass.spawn.apps.handlers.json_ops import apply_migrations + + data = {"old_key": "old_value", "new_key": "already_here"} + migrations = [ + { + "id": "m001", + "operations": [ + {"type": "key_rename", "from": "old_key", "to": "new_key"}, + ], + }, + ] + + result = apply_migrations(data, migrations) + + # Target preserved, source untouched + assert result["new_key"] == "already_here" + assert result["old_key"] == "old_value" + + def test_add_missing_keys(self): + """add_missing_keys should add keys that don't exist.""" + from aipass.spawn.apps.handlers.json_ops import apply_migrations + + data = {"metadata": {"version": "1.0"}} + migrations = [ + { + "id": "m002", + "operations": [ + { + "type": "add_missing_keys", + "parent": "metadata", + "keys": {"schema_version": "1.0.0", "type": "config"}, + }, + ], + }, + ] + + result = apply_migrations(data, migrations) + + assert result["metadata"]["version"] == "1.0" # preserved + assert result["metadata"]["schema_version"] == "1.0.0" # added + assert result["metadata"]["type"] == "config" # added + + def test_add_missing_keys_skip_existing(self): + """add_missing_keys should not overwrite existing keys.""" + from aipass.spawn.apps.handlers.json_ops import apply_migrations + + data = {"metadata": {"version": "1.0", "type": "custom"}} + migrations = [ + { + "id": "m002", + "operations": [ + { + "type": "add_missing_keys", + "parent": "metadata", + "keys": {"type": "default"}, + }, + ], + }, + ] + + result = apply_migrations(data, migrations) + + assert result["metadata"]["type"] == "custom" # not overwritten + + def test_move_to_nested(self): + """move_to_nested should move top-level keys under a parent.""" + from aipass.spawn.apps.handlers.json_ops import apply_migrations + + data = {"name": "test", "path": "/some/path", "keep": "yes"} + migrations = [ + { + "id": "m003", + "operations": [ + { + "type": "move_to_nested", + "source_keys": ["name", "path"], + "target_parent": "branch_info", + }, + ], + }, + ] + + result = apply_migrations(data, migrations) + + assert "branch_info" in result + assert result["branch_info"]["name"] == "test" + assert result["branch_info"]["path"] == "/some/path" + assert "name" not in result # moved + assert "path" not in result # moved + assert result["keep"] == "yes" # untouched + + def test_multiple_operations_in_sequence(self): + """Multiple operations should apply in order.""" + from aipass.spawn.apps.handlers.json_ops import apply_migrations + + data = {"metadata": {"ver": "1.0"}, "extra": "data"} + migrations = [ + { + "id": "m004", + "operations": [ + {"type": "key_rename", "from": "metadata.ver", "to": "metadata.version"}, + { + "type": "add_missing_keys", + "parent": "metadata", + "keys": {"schema": "1.0.0"}, + }, + ], + }, + ] + + result = apply_migrations(data, migrations) + + assert result["metadata"]["version"] == "1.0" + assert "ver" not in result["metadata"] + assert result["metadata"]["schema"] == "1.0.0" + + +# ============================================================================= +# backup_json tests +# ============================================================================= + +class TestBackupJson: + """Tests for backup_json().""" + + def test_creates_backup_file(self, tmp_path): + """Backup should create a copy in .backup/ directory.""" + from aipass.spawn.apps.handlers.json_ops import backup_json + + source = tmp_path / "data.json" + source.write_text('{"key": "value"}', encoding="utf-8") + + backup_path = backup_json(source) + + assert backup_path.exists() + assert ".backup" in str(backup_path) + assert "data.json" in backup_path.name + assert ".backup" in backup_path.name + + # Content should match + assert backup_path.read_text(encoding="utf-8") == '{"key": "value"}' + + def test_raises_on_missing_source(self, tmp_path): + """Should raise FileNotFoundError for non-existent source.""" + from aipass.spawn.apps.handlers.json_ops import backup_json + + with pytest.raises(FileNotFoundError): + backup_json(tmp_path / "nonexistent.json") diff --git a/src/aipass/spawn/tests/test_lifecycle.py b/src/aipass/spawn/tests/test_lifecycle.py new file mode 100644 index 00000000..2fc6cf61 --- /dev/null +++ b/src/aipass/spawn/tests/test_lifecycle.py @@ -0,0 +1,569 @@ +# =================== META ==================== +# Name: test_lifecycle.py +# Description: Tests for spawn lifecycle commands (delete, sync-registry, sync-templates) +# Version: 1.1.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Tests for spawn lifecycle management commands. + +Tests delete_branch(), sync_registry(), and sync_templates(). +""" + +import json +import shutil +from pathlib import Path +from unittest.mock import patch + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def repo_root(tmp_path): + """Create a mock repo root with src/aipass/ structure.""" + root = tmp_path / "repo" + root.mkdir() + (root / "pyproject.toml").write_text("[project]\nname = 'aipass'\n") + (root / "src" / "aipass").mkdir(parents=True) + return root + + +@pytest.fixture +def mock_branch(repo_root): + """Create a mock branch directory with .trinity/passport.json.""" + branch = repo_root / "src" / "aipass" / "test_api" + branch.mkdir(parents=True) + (branch / ".trinity").mkdir() + (branch / ".trinity" / "passport.json").write_text( + json.dumps({"name": "TEST_API", "role": "test"}, indent=2) + ) + (branch / "apps").mkdir() + (branch / "apps" / "branch.py").write_text("# test api entry\n") + (branch / "README.md").write_text("# Test API\n") + return branch + + +@pytest.fixture +def mock_registry(repo_root, mock_branch): + """Create a mock AIPASS_REGISTRY.json with the test branch.""" + rel_path = str(mock_branch.relative_to(repo_root)) + + registry = { + "metadata": { + "version": "1.0.0", + "last_updated": "2026-03-07", + "total_branches": 4, + }, + "branches": [ + { + "name": "DRONE", + "path": "src/aipass/drone", + "profile": "library", + "description": "Command routing", + "email": "@drone", + "status": "active", + "created": "2026-03-05", + "last_active": "2026-03-05", + }, + { + "name": "SPAWN", + "path": "src/aipass/spawn", + "profile": "library", + "description": "Agent creation", + "email": "@spawn", + "status": "active", + "created": "2026-03-05", + "last_active": "2026-03-05", + }, + { + "name": "DEVPULSE", + "path": "src/aipass/devpulse", + "profile": "library", + "description": "Orchestration hub", + "email": "@devpulse", + "status": "active", + "created": "2026-03-06", + "last_active": "2026-03-06", + }, + { + "name": "TEST_API", + "path": rel_path, + "profile": "library", + "description": "Test API branch", + "email": "@test_api", + "status": "active", + "created": "2026-03-07", + "last_active": "2026-03-07", + }, + ], + } + + reg_path = repo_root / "AIPASS_REGISTRY.json" + reg_path.write_text(json.dumps(registry, indent=2) + "\n") + return reg_path + + +# --------------------------------------------------------------------------- +# DELETE Tests +# --------------------------------------------------------------------------- + +class TestDeleteBranch: + """Tests for delete_branch().""" + + def test_delete_archives_and_removes(self, repo_root, mock_branch, mock_registry): + """Successful delete should archive the branch and remove from registry.""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = delete_branch("test_api", confirm=False) + + assert result["success"] is True + assert result["registry_updated"] is True + assert result["archive_path"] != "" + + # Branch directory should be gone + assert not mock_branch.exists() + + # Archive should exist + archive_path = Path(result["archive_path"]) + assert archive_path.exists() + assert (archive_path / "README.md").exists() + assert (archive_path / ".trinity" / "passport.json").exists() + + # Registry should no longer contain TEST_API + reg = json.loads(mock_registry.read_text()) + names = [b["name"] for b in reg["branches"]] + assert "TEST_API" not in names + + def test_delete_protected_spawn(self, repo_root, mock_registry): + """Cannot delete spawn (self-protection).""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = delete_branch("spawn", confirm=False) + + assert result["success"] is False + assert "protected" in result.get("error", "").lower() + + def test_delete_protected_devpulse(self, repo_root, mock_registry): + """Cannot delete devpulse (orchestration hub protection).""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = delete_branch("devpulse", confirm=False) + + assert result["success"] is False + assert "protected" in result.get("error", "").lower() + + def test_delete_protected_drone(self, repo_root, mock_registry): + """Cannot delete drone (routing infrastructure protection).""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = delete_branch("drone", confirm=False) + + assert result["success"] is False + assert "protected" in result.get("error", "").lower() + + def test_delete_dry_run(self, repo_root, mock_branch, mock_registry): + """Dry run should NOT delete or archive anything.""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = delete_branch("test_api", confirm=False, dry_run=True) + + assert result["success"] is True + assert result.get("dry_run") is True + + # Branch should still exist + assert mock_branch.exists() + + # Registry should be unchanged + reg = json.loads(mock_registry.read_text()) + names = [b["name"] for b in reg["branches"]] + assert "TEST_API" in names + + def test_delete_nonexistent_branch(self, repo_root, mock_registry): + """Deleting a branch not in registry should fail gracefully.""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = delete_branch("nonexistent", confirm=False) + + assert result["success"] is False + assert "not found" in result.get("error", "").lower() + + def test_delete_confirmation_cancelled(self, repo_root, mock_branch, mock_registry): + """Cancelling confirmation should not delete.""" + from aipass.spawn.apps.handlers.delete_ops import delete_branch + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry), \ + patch("builtins.input", return_value="n"): + + result = delete_branch("test_api", confirm=True) + + assert result["success"] is False + assert mock_branch.exists() + + def test_handle_delete_no_args(self): + """handle_delete with no args should show usage.""" + from aipass.spawn.apps.modules.delete import handle_delete + result = handle_delete([]) + assert result == 1 + + +# --------------------------------------------------------------------------- +# SYNC REGISTRY Tests +# --------------------------------------------------------------------------- + +class TestSyncRegistry: + """Tests for sync_registry().""" + + def test_detect_stale_entries(self, repo_root, mock_registry): + """Registry entries for non-existent directories should be detected as stale.""" + from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry + + # The registry has DRONE, SPAWN, DEVPULSE entries but those directories + # don't exist in our tmp_path repo, so they should be stale + with patch("aipass.spawn.apps.handlers.sync_registry_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=mock_registry): + + result = sync_registry(fix=False) + + # DRONE, SPAWN, DEVPULSE dirs don't exist -> stale + assert len(result["stale"]) >= 3 + assert "drone" in result["stale"] + assert "spawn" in result["stale"] + assert "devpulse" in result["stale"] + + def test_detect_unregistered_branches(self, repo_root, mock_registry): + """Directories with passport.json not in registry should be unregistered.""" + from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry + + # Create a new branch directory with passport that's NOT in registry + new_branch = repo_root / "src" / "aipass" / "phantom" + new_branch.mkdir(parents=True) + (new_branch / ".trinity").mkdir() + (new_branch / ".trinity" / "passport.json").write_text( + json.dumps({"name": "PHANTOM"}, indent=2) + ) + + with patch("aipass.spawn.apps.handlers.sync_registry_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=mock_registry): + + result = sync_registry(fix=False) + + assert "phantom" in result["unregistered"] + + def test_detect_healthy_branches(self, repo_root, mock_branch, mock_registry): + """Branches that exist with passports and are registered should be healthy.""" + from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry + + with patch("aipass.spawn.apps.handlers.sync_registry_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=mock_registry): + + result = sync_registry(fix=False) + + assert "test_api" in result["healthy"] + + def test_fix_removes_stale_and_adds_unregistered(self, repo_root, mock_branch, mock_registry): + """With --fix, stale entries are removed and unregistered are added.""" + from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry + + # Create an unregistered branch + new_branch = repo_root / "src" / "aipass" / "phantom" + new_branch.mkdir(parents=True) + (new_branch / ".trinity").mkdir() + (new_branch / ".trinity" / "passport.json").write_text( + json.dumps({"name": "PHANTOM"}, indent=2) + ) + + with patch("aipass.spawn.apps.handlers.sync_registry_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=mock_registry): + + result = sync_registry(fix=True) + + assert result["fixed"] is True + + # Verify registry was actually updated + reg = json.loads(mock_registry.read_text()) + names = [b["name"] for b in reg["branches"]] + + # Stale entries removed (drone, spawn, devpulse dirs don't exist) + assert "DRONE" not in names + assert "SPAWN" not in names + assert "DEVPULSE" not in names + + # Unregistered added + assert "PHANTOM" in names + + # Healthy branch kept + assert "TEST_API" in names + + def test_no_mismatches_no_fix_needed(self, repo_root, mock_branch, mock_registry): + """When everything is healthy, fix=True should not modify registry.""" + from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry + + # Remove all stale entries from registry (only keep test_api which exists) + reg = json.loads(mock_registry.read_text()) + reg["branches"] = [b for b in reg["branches"] if b["name"] == "TEST_API"] + mock_registry.write_text(json.dumps(reg, indent=2) + "\n") + + with patch("aipass.spawn.apps.handlers.sync_registry_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=mock_registry): + + result = sync_registry(fix=True) + + assert result["stale"] == [] + assert result["unregistered"] == [] + assert result["fixed"] is False # Nothing to fix + + +# --------------------------------------------------------------------------- +# SYNC TEMPLATES Tests +# --------------------------------------------------------------------------- + +class TestSyncTemplates: + """Tests for sync_templates().""" + + def test_empty_owners_no_stale(self, repo_root): + """With empty template_owners.json, no files should be stale.""" + from aipass.spawn.apps.handlers.sync_templates_ops import sync_templates + + # Create empty template_owners.json + owners_path = repo_root / "template_owners.json" + owners_path.write_text(json.dumps({ + "metadata": {"description": "test"}, + "managed_files": {}, + }, indent=2)) + + with patch("aipass.spawn.apps.handlers.sync_templates_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_templates_ops._TEMPLATE_OWNERS_PATH", owners_path): + + result = sync_templates() + + assert result["managed_files"] == 0 + assert result["stale"] == [] + assert result["current"] == [] + assert result["synced"] == [] + assert result["errors"] == [] + + def test_status_report_works(self, repo_root): + """Status report (default) should work without errors.""" + from aipass.spawn.apps.handlers import sync_templates_ops as st_mod + + # Create template_owners with a managed file + source_branch_dir = repo_root / "src" / "aipass" / "prax" + source_branch_dir.mkdir(parents=True) + source_file = source_branch_dir / "config.json" + source_file.write_text(json.dumps({"key": "value"}, indent=2)) + + # Create template location + template_dir = repo_root / "templates" + template_dir.mkdir() + + owners_path = repo_root / "template_owners.json" + owners_path.write_text(json.dumps({ + "metadata": {"description": "test"}, + "managed_files": { + "prax_config": { + "source_branch": "prax", + "source_path": "config.json", + "template_path": "config.json", + } + }, + }, indent=2)) + + # Save originals + orig_root = st_mod._REPO_ROOT + orig_owners = st_mod._TEMPLATE_OWNERS_PATH + + try: + st_mod._REPO_ROOT = repo_root + st_mod._TEMPLATE_OWNERS_PATH = owners_path + + result = st_mod.sync_templates() + finally: + st_mod._REPO_ROOT = orig_root + st_mod._TEMPLATE_OWNERS_PATH = orig_owners + + assert result["managed_files"] == 1 + # Source exists but template doesn't yet -> stale + assert "prax_config" in result["stale"] + + def test_sync_copies_file(self, repo_root): + """sync=True should copy source files to template location.""" + from aipass.spawn.apps.handlers import sync_templates_ops as st_mod + + # Create source file + source_dir = repo_root / "src" / "aipass" / "prax" + source_dir.mkdir(parents=True) + source_file = source_dir / "config.json" + source_file.write_text(json.dumps({"key": "source_value"}, indent=2)) + + # Create the template target dir + template_target_dir = repo_root / "template_target" + template_target_dir.mkdir() + + owners_path = repo_root / "template_owners.json" + owners_path.write_text(json.dumps({ + "metadata": {"description": "test"}, + "managed_files": { + "prax_config": { + "source_branch": "prax", + "source_path": "config.json", + "template_path": "config.json", + } + }, + }, indent=2)) + + orig_root = st_mod._REPO_ROOT + orig_owners = st_mod._TEMPLATE_OWNERS_PATH + + try: + st_mod._REPO_ROOT = repo_root + st_mod._TEMPLATE_OWNERS_PATH = owners_path + + # Override _file_hash to work and patch template file destination + def patched_sync(sync=False, dry_run=False): + """Patched sync that redirects template paths.""" + owners_data = st_mod._load_template_owners() + managed_files = owners_data.get("managed_files", {}) + current = [] + stale = [] + synced = [] + errors = [] + + for file_key, file_info in managed_files.items(): + source_branch = file_info.get("source_branch", "") + source_path_str = file_info.get("source_path", "") + template_path_str = file_info.get("template_path", "") + + source_file = repo_root / "src" / "aipass" / source_branch / source_path_str + template_file = template_target_dir / template_path_str + + if not source_file.exists(): + errors.append(f"Source not found: {source_file}") + continue + + source_hash = st_mod._file_hash(source_file) + if template_file.exists(): + template_hash = st_mod._file_hash(template_file) + if source_hash == template_hash: + current.append(file_key) + continue + + stale.append(file_key) + if sync and not dry_run: + template_file.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(str(source_file), str(template_file)) + synced.append(file_key) + + return { + "managed_files": len(managed_files), + "current": current, + "stale": stale, + "synced": synced, + "errors": errors, + } + + result = patched_sync(sync=True) + finally: + st_mod._REPO_ROOT = orig_root + st_mod._TEMPLATE_OWNERS_PATH = orig_owners + + assert result["managed_files"] == 1 + assert "prax_config" in result["synced"] + + # Template file should now exist + template_file = template_target_dir / "config.json" + assert template_file.exists() + assert json.loads(template_file.read_text())["key"] == "source_value" + + def test_handle_sync_templates_no_args(self): + """handle_sync_templates with no args should return 0 (status report).""" + from aipass.spawn.apps.modules.sync_templates import handle_sync_templates + + # With the real template_owners.json being empty, this should work + result = handle_sync_templates([]) + assert result == 0 + + def test_missing_template_owners(self, tmp_path): + """Missing template_owners.json should handle gracefully.""" + from aipass.spawn.apps.handlers import sync_templates_ops as st_mod + + orig_owners = st_mod._TEMPLATE_OWNERS_PATH + + try: + st_mod._TEMPLATE_OWNERS_PATH = tmp_path / "nonexistent.json" + result = st_mod.sync_templates() + finally: + st_mod._TEMPLATE_OWNERS_PATH = orig_owners + + assert result["managed_files"] == 0 + assert result["errors"] == [] + + +# --------------------------------------------------------------------------- +# HANDLE CLI Tests +# --------------------------------------------------------------------------- + +class TestHandleDelete: + """Tests for handle_delete() CLI entry.""" + + def test_help_flag(self): + """--help should show usage (not crash).""" + from aipass.spawn.apps.modules.delete import handle_delete + # No args -> usage + result = handle_delete([]) + assert result == 1 + + def test_protected_branch_via_handle(self, repo_root, mock_registry): + """handle_delete should reject protected branches.""" + from aipass.spawn.apps.modules.delete import handle_delete + + with patch("aipass.spawn.apps.handlers.delete_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.delete_ops.find_registry", return_value=mock_registry): + + result = handle_delete(["--yes", "@spawn"]) + + assert result == 1 + + +class TestHandleSyncRegistry: + """Tests for handle_sync_registry() CLI entry.""" + + def test_help_flag(self): + """--help should return 0.""" + from aipass.spawn.apps.modules.sync_registry import handle_sync_registry + result = handle_sync_registry(["--help"]) + assert result == 0 + + def test_report_mode(self, repo_root, mock_branch, mock_registry): + """No args should produce a report.""" + from aipass.spawn.apps.modules.sync_registry import handle_sync_registry + + with patch("aipass.spawn.apps.handlers.sync_registry_ops._REPO_ROOT", repo_root), \ + patch("aipass.spawn.apps.handlers.sync_registry_ops.find_registry", return_value=mock_registry): + + result = handle_sync_registry([]) + + assert result == 0 diff --git a/src/aipass/spawn/tests/test_update.py b/src/aipass/spawn/tests/test_update.py new file mode 100644 index 00000000..83931f72 --- /dev/null +++ b/src/aipass/spawn/tests/test_update.py @@ -0,0 +1,494 @@ +# =================== META ==================== +# Name: test_update.py +# Description: Tests for spawn update orchestrator +# Version: 1.1.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +"""Tests for the spawn update module. + +Tests update_branch(), update_all(), dry-run mode, .py skip behavior, +JSON deep merge, first-time adoption, and self-skip logic. +""" + +import json +import os +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def template_dir(tmp_path): + """Create a minimal template directory with registry.""" + tpl = tmp_path / "template" + tpl.mkdir() + + # Template files + (tpl / "README.md").write_text("# {{BRANCHNAME}}\nTemplate readme\n") + (tpl / "DASHBOARD.local.json").write_text( + json.dumps({"status": "active", "branch": "{{branchname}}", "version": "1.0"}, indent=2) + ) + (tpl / "apps").mkdir() + (tpl / "apps" / "__init__.py").write_text('"""{{branchname}} apps"""') + (tpl / "apps" / "branch.py").write_text('"""{{branchname}} entry point"""\ndef main(): pass\n') + (tpl / "tests").mkdir() + (tpl / "tests" / "__init__.py").write_text("") + (tpl / ".archive").mkdir() + (tpl / "docs").mkdir() + + # .spawn directory with template registry + spawn_meta = tpl / ".spawn" + spawn_meta.mkdir() + + registry = { + "metadata": { + "version": "1.0.0", + "last_updated": "2026-03-07", + "description": "Template file tracking registry", + }, + "files": { + "f001": { + "current_name": "README.md", + "path": "README.md", + "content_hash": _hash_content("# {{BRANCHNAME}}\nTemplate readme\n"), + "has_branch_placeholder": True, + }, + "f002": { + "current_name": "DASHBOARD.local.json", + "path": "DASHBOARD.local.json", + "content_hash": _hash_content( + json.dumps({"status": "active", "branch": "{{branchname}}", "version": "1.0"}, indent=2) + ), + "has_branch_placeholder": True, + }, + "f003": { + "current_name": "__init__.py", + "path": "apps/__init__.py", + "content_hash": _hash_content('"""{{branchname}} apps"""'), + "has_branch_placeholder": True, + }, + "f004": { + "current_name": "branch.py", + "path": "apps/branch.py", + "content_hash": _hash_content('"""{{branchname}} entry point"""\ndef main(): pass\n'), + "has_branch_placeholder": True, + }, + "f005": { + "current_name": "__init__.py", + "path": "tests/__init__.py", + "content_hash": _hash_content(""), + "has_branch_placeholder": False, + }, + }, + "directories": { + "d001": { + "current_name": "apps", + "path": "apps", + "has_branch_placeholder": False, + }, + "d002": { + "current_name": "tests", + "path": "tests", + "has_branch_placeholder": False, + }, + "d003": { + "current_name": ".archive", + "path": ".archive", + "has_branch_placeholder": False, + }, + "d004": { + "current_name": "docs", + "path": "docs", + "has_branch_placeholder": False, + }, + }, + } + + (spawn_meta / ".template_registry.json").write_text( + json.dumps(registry, indent=2) + "\n" + ) + + return tpl + + +@pytest.fixture +def branch_dir(tmp_path): + """Create a minimal existing branch directory (pre-update).""" + branch = tmp_path / "test_branch" + branch.mkdir() + + # Existing files in branch + (branch / "README.md").write_text("# TEST_BRANCH\nCustom readme with user edits\n") + (branch / "DASHBOARD.local.json").write_text( + json.dumps({"status": "running", "branch": "test_branch", "custom_key": "preserved"}, indent=2) + ) + (branch / "apps").mkdir() + (branch / "apps" / "__init__.py").write_text('"""test_branch apps - modified"""') + (branch / "apps" / "branch.py").write_text('"""test_branch entry"""\ndef main():\n print("hello")\n') + (branch / "tests").mkdir() + (branch / "tests" / "__init__.py").write_text("") + (branch / ".archive").mkdir() + (branch / "docs").mkdir() + + return branch + + +@pytest.fixture +def mock_registry(tmp_path, branch_dir): + """Create a mock AIPASS_REGISTRY.json pointing to our test branch.""" + # We need paths relative to some "repo root" + repo_root = tmp_path + rel_path = str(branch_dir.relative_to(repo_root)) + + registry = { + "metadata": { + "version": "1.0.0", + "last_updated": "2026-03-07", + "total_branches": 2, + }, + "branches": [ + { + "name": "TEST_BRANCH", + "path": rel_path, + "profile": "library", + "description": "Test branch", + "email": "@test_branch", + "status": "active", + }, + { + "name": "SPAWN", + "path": "spawn", + "profile": "library", + "description": "Agent creation", + "email": "@spawn", + "status": "active", + }, + ], + } + + reg_path = repo_root / "AIPASS_REGISTRY.json" + reg_path.write_text(json.dumps(registry, indent=2) + "\n") + + return reg_path + + +def _hash_content(content: str) -> str: + """Compute SHA-256 hash (first 12 chars) of content string.""" + import hashlib + return hashlib.sha256(content.encode("utf-8")).hexdigest()[:12] + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestUpdateBranch: + """Tests for update_branch().""" + + def test_first_time_adoption_generates_meta(self, tmp_path, template_dir, branch_dir, mock_registry): + """Branch with no .branch_meta.json should get one generated.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + # Ensure no branch_meta exists + meta_path = branch_dir / ".spawn" / ".branch_meta.json" + assert not meta_path.exists() + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("test_branch") + + assert result["success"] is True + assert result["branch"] == "test_branch" + # After update, branch_meta should exist + assert meta_path.exists() + + def test_dry_run_does_not_modify(self, tmp_path, template_dir, branch_dir, mock_registry): + """Dry run should report changes without modifying files.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + # Record original state + readme_before = (branch_dir / "README.md").read_text() + dashboard_before = (branch_dir / "DASHBOARD.local.json").read_text() + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("test_branch", dry_run=True) + + assert result["success"] is True + assert result["dry_run"] is True + + # Files should be unchanged + assert (branch_dir / "README.md").read_text() == readme_before + assert (branch_dir / "DASHBOARD.local.json").read_text() == dashboard_before + + # No branch_meta created in dry-run on first adoption + meta_path = branch_dir / ".spawn" / ".branch_meta.json" + assert not meta_path.exists() + + def test_py_files_never_overwritten(self, tmp_path, template_dir, branch_dir, mock_registry): + """Python files should be skipped even when template hash differs.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + # Create initial branch_meta with a matching .py file that has a different hash + spawn_dir = branch_dir / ".spawn" + spawn_dir.mkdir(exist_ok=True) + + branch_py_content = '"""test_branch entry"""\ndef main():\n print("hello")\n' + original_content = branch_py_content + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("test_branch") + + # .py file should still have original content + assert (branch_dir / "apps" / "branch.py").read_text() == original_content + assert result["success"] is True + + def test_json_deep_merge_preserves_existing(self, tmp_path, template_dir, branch_dir, mock_registry): + """JSON merge should add new template keys while preserving existing values.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + # Set up branch_meta so DASHBOARD.local.json shows as needing update + spawn_dir = branch_dir / ".spawn" + spawn_dir.mkdir(exist_ok=True) + + # Write template with a new key + (template_dir / "DASHBOARD.local.json").write_text( + json.dumps({ + "status": "active", + "branch": "{{branchname}}", + "version": "2.0", + "new_field": "from_template", + }, indent=2) + ) + + # Update template registry hash to differ from branch + reg_path = template_dir / ".spawn" / ".template_registry.json" + reg = json.loads(reg_path.read_text()) + reg["files"]["f002"]["content_hash"] = "different_hash" + reg_path.write_text(json.dumps(reg, indent=2) + "\n") + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("test_branch") + + assert result["success"] is True + + # Read merged dashboard + merged = json.loads((branch_dir / "DASHBOARD.local.json").read_text()) + + # Existing values preserved + assert merged["custom_key"] == "preserved" + assert merged["status"] == "running" # existing value kept over template + assert merged["branch"] == "test_branch" # existing value kept + + def test_branch_not_found(self, tmp_path, template_dir, mock_registry): + """Non-existent branch should return failure.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("nonexistent_branch") + + assert result["success"] is False + assert len(result["errors"]) > 0 + + def test_additions_from_template(self, tmp_path, template_dir, branch_dir, mock_registry): + """New template files not in branch should be added.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + # Add a new file to template that doesn't exist in branch + (template_dir / "docs" / "new_doc.md").write_text("# New doc for {{branchname}}") + + # Add it to template registry + reg_path = template_dir / ".spawn" / ".template_registry.json" + reg = json.loads(reg_path.read_text()) + reg["files"]["f099"] = { + "current_name": "new_doc.md", + "path": "docs/new_doc.md", + "content_hash": _hash_content("# New doc for {{branchname}}"), + "has_branch_placeholder": True, + } + reg_path.write_text(json.dumps(reg, indent=2) + "\n") + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("test_branch") + + assert result["success"] is True + assert result["additions"] >= 1 + + # The new file should exist with placeholders replaced + new_file = branch_dir / "docs" / "new_doc.md" + assert new_file.exists() + content = new_file.read_text() + assert "{{branchname}}" not in content + assert "test_branch" in content + + def test_pruned_files_archived(self, tmp_path, template_dir, branch_dir, mock_registry): + """Files removed from template should be archived, not deleted.""" + from aipass.spawn.apps.handlers.update_ops import update_branch + + # Create a file in the branch that's tracked in branch_meta but NOT in template + extra_file = branch_dir / "old_config.json" + extra_file.write_text(json.dumps({"old": True})) + + # Create branch_meta that tracks this file + spawn_dir = branch_dir / ".spawn" + spawn_dir.mkdir(exist_ok=True) + + # Generate branch meta, then add the extra file tracking manually + from aipass.spawn.apps.handlers.meta_ops import generate_branch_meta, save_branch_meta, compute_file_hash + + reg_path = template_dir / ".spawn" / ".template_registry.json" + template_registry = json.loads(reg_path.read_text()) + + meta = generate_branch_meta(branch_dir, template_registry) + # Add the extra file with a unique ID that's NOT in the template + meta["file_tracking"]["f_extra"] = { + "template_name": "old_config.json", + "current_name": "old_config.json", + "current_path": "old_config.json", + "content_hash": compute_file_hash(extra_file), + } + save_branch_meta(branch_dir, meta) + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = update_branch("test_branch") + + assert result["success"] is True + assert result["pruned"] >= 1 + + # Original file should be gone from its original location + assert not extra_file.exists() + + # Should be in .archive/ + archive_dir = branch_dir / ".archive" + archived_files = list(archive_dir.glob("old_config.json*")) + assert len(archived_files) >= 1 + + +class TestUpdateAll: + """Tests for update_all().""" + + def test_update_all_skips_spawn(self, tmp_path, template_dir, branch_dir, mock_registry): + """update_all should skip spawn itself.""" + from aipass.spawn.apps.handlers.update_ops import update_all + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + results = update_all() + + # Should have results for test_branch but NOT for spawn + branch_names = [r["branch"] for r in results] + assert "spawn" not in branch_names + # test_branch should be present + assert "test_branch" in branch_names + + def test_update_all_processes_all_branches(self, tmp_path, template_dir, mock_registry): + """update_all should process each registered branch.""" + from aipass.spawn.apps.handlers.update_ops import update_all + + # Create a second branch + branch2 = tmp_path / "other_branch" + branch2.mkdir() + (branch2 / "README.md").write_text("# Other") + (branch2 / "apps").mkdir() + (branch2 / "tests").mkdir() + + # Add it to registry + reg = json.loads(mock_registry.read_text()) + rel_path = str(branch2.relative_to(tmp_path)) + reg["branches"].append({ + "name": "OTHER_BRANCH", + "path": rel_path, + "profile": "library", + "description": "Other branch", + "email": "@other_branch", + "status": "active", + }) + mock_registry.write_text(json.dumps(reg, indent=2) + "\n") + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + results = update_all() + + branch_names = [r["branch"] for r in results] + assert "test_branch" in branch_names + assert "other_branch" in branch_names + assert "spawn" not in branch_names + + +class TestHandleUpdate: + """Tests for handle_update() CLI parsing.""" + + def test_no_args_shows_usage(self): + """No args should show usage and return 1.""" + from aipass.spawn.apps.modules.update import handle_update + result = handle_update([]) + assert result == 1 + + def test_single_branch_arg(self, tmp_path, template_dir, branch_dir, mock_registry): + """@branch arg should call update_branch.""" + from aipass.spawn.apps.modules.update import handle_update + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = handle_update(["@test_branch"]) + + assert result == 0 + + def test_dry_run_flag(self, tmp_path, template_dir, branch_dir, mock_registry): + """--dry-run flag should be parsed and passed through.""" + from aipass.spawn.apps.modules.update import handle_update + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = handle_update(["--dry-run", "@test_branch"]) + + # dry-run should succeed (exit 0) and not create branch_meta + assert result == 0 + meta_path = branch_dir / ".spawn" / ".branch_meta.json" + assert not meta_path.exists() + + def test_all_flag(self, tmp_path, template_dir, branch_dir, mock_registry): + """--all flag should trigger update_all.""" + from aipass.spawn.apps.modules.update import handle_update + + with patch("aipass.spawn.apps.handlers.update_ops.get_template_dir", return_value=template_dir), \ + patch("aipass.spawn.apps.handlers.update_ops.find_registry", return_value=mock_registry), \ + patch("aipass.spawn.apps.handlers.update_ops._REPO_ROOT", tmp_path): + + result = handle_update(["--all"]) + + assert result == 0