This commit is contained in:
Input-X
2026-03-07 10:26:35 +00:00
parent f64c01916a
commit 2b525cfa4d
20 changed files with 3381 additions and 115 deletions
+14 -4
View File
@@ -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)
```
---
@@ -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,
}
@@ -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,
}
@@ -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 ""
@@ -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
+97
View File
@@ -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()
@@ -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()
@@ -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()
+177
View File
@@ -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()
@@ -0,0 +1 @@
"""Spawn plugins package — extension point for spawn customization."""
+20 -8
View File
@@ -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",
}
View File
-97
View File
@@ -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 <command> [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 <command> [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 <path> 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 <target_path> [--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"
+5
View File
@@ -0,0 +1,5 @@
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
+1
View File
@@ -0,0 +1 @@
"""Shared test fixtures for spawn test suite."""
+6 -6
View File
@@ -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():
+786
View File
@@ -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")
+569
View File
@@ -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
+494
View File
@@ -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