slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(spawn): dry-run-default + path-based update engine — kill #636 branch-scrambler (TDPLAN-0006 P0+P1)

Browse Source 
#636: drone @spawn update would scramble every branch's identity/memory in
one command. On a branch created seconds earlier, --dry-run proposed 30 renames
rotating identity dirs (apps->.trinity->.seedgo->.claude->.archive->.aipass),
README->DASHBOARD, and deep-merged stale template into live .trinity/. Root
cause: the CREATE path regenerated template-registry IDs in filesystem-walk
order (!= the master's hand-crafted IDs), so content-hash + rename-detection
saw a mismatch on a pristine branch. update --all would have destroyed all 13
citizens at once.

P0 — safety by default:
- update + repair are now dry-run by default; --apply required to write.
  Forgotten flag = safe preview-only no-op. --dry-run kept as alias.
- doctor_fix.py repair suggestions emit the matching --apply form
  (+ aipass test_doctor_fix updated to the new contract).

P1 — engine rebuild (update_ops.py v2.0):
- Path-based named-managed-files model replaces whole-tree hash-diff +
  rename-detection. ID divergence is moot — IDs are no longer used.
- .trinity/*, DASHBOARD.local.json, artifacts/birth_certificate.json,
  .seedgo/bypass.json = delivered on CREATE only, NEVER touched on update.
- Old ID engine (change_detection.py, reconcile.py) + orphaned tests deleted.

Verified on fresh sandbox: update --dry-run = 0 renames / 0 updates / 0
additions (create==update invariant); no-flag run = dry-run preview, filesystem
byte-identical; 313 spawn tests green; seedgo 100% (all 36 standards).

Closes #636. P2/P3/P4 (shared lib, seam, .recovery relocate) to follow.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-06 22:54:09 -07:00
co-authored by Claude Opus 4.8
parent 1ef1e2e89c
commit ccc8d6a97b
16 changed files with 334 additions and 1189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -125,4 +125,5 @@ branch_audits/
claude_4_7_transition_notes.md
README_ORIGINAL_DISABLED.md
*.bak
test/
test/
sandbox_test/
CHANGELOG.md
+22
View File
@@ -10,6 +10,28 @@ and this project uses [Calendar Versioning](https://calver.org/) in the format
## [2026.W23] - 2026-06-02
### Fixed
- **`drone @spawn update` no longer scrambles branches (#636, critical — TDPLAN-0006
P0+P1).** The update engine compared a freshly-created branch against the class
template by *content hash* with rename-detection, and because the CREATE path
regenerated template-registry IDs in filesystem-walk order (≠ the master's
hand-crafted IDs), a branch created seconds earlier produced **30 proposed renames**
that rotated identity/memory dirs into each other
(`apps→.trinity→.seedgo→.claude→.archive→.aipass`), turned `README` into
`DASHBOARD`, and deep-merged stale template into live `.trinity/` memory —
`update <class> --all` would have destroyed every citizen in one command. Rebuilt
`update_ops.py` (v2.0) on an explicit **named-managed-files + path-based** model:
`.trinity/*`, `DASHBOARD.local.json`, `artifacts/birth_certificate.json` and
`.seedgo/bypass.json` are delivered on **create only** and never touched on update;
the create==update invariant now yields **0 renames / 0 merges** on a fresh branch.
The old ID-based engine (`change_detection.py`, `reconcile.py`) is deleted.
- **Destructive spawn ops are now dry-run by default (TDPLAN-0006 P0).** `drone @spawn
update` and `drone @spawn repair` preview by default and require an explicit
`--apply` to write — forgetting a flag is now a safe no-op instead of irreversible
damage (`--dry-run` kept as an alias). `aipass doctor` repair suggestions emit the
matching `--apply` form.
### Added
- **Memory-pool auto-processing (TDPLAN-0005)** — dropped files in
src/aipass/aipass/apps/modules/doctor_fix.py
+7 -6
View File
@@ -83,7 +83,7 @@ def _build_pollution_items(agents: list, project: str) -> List[RemediationItem]:
f"Registry pollution: {len(hit.locations)} copies of "
f"{hit.agent_name} share registry_id {hit.registry_id}"
),
fix_command=f"drone @spawn repair @{project} --clean-pollution",
fix_command=f"drone @spawn repair @{project} --clean-pollution --apply",
)
)
return items
@@ -104,7 +104,7 @@ def _build_placement_items(agents: list, project_root: Path, project: str) -> Li
severity="warning",
category="placement",
description=f"Misplaced agent: {issue.agent_name} at {rel_path}",
fix_command=f"drone @spawn repair @{project} --relocate {rel_path} {suggested}",
fix_command=f"drone @spawn repair @{project} --relocate {rel_path} {suggested} --apply",
)
)
return items
@@ -122,7 +122,7 @@ def _build_registry_items(project_root: Path, agents: list, project: str) -> Lis
severity="warning",
category="registry",
description=f"Registry {issue.problem}: {issue.branch_name} at {issue.registered_path}",
fix_command=f"drone @spawn repair @{project} --dedup-registry",
fix_command=f"drone @spawn repair @{project} --dedup-registry --apply",
)
)
return items
@@ -145,7 +145,7 @@ def generate_remediation(project_root: Path) -> List[RemediationItem]:
severity="info",
category="pyproject",
description="Missing pyproject.toml",
fix_command=f"drone @spawn repair @{project} --add-pyproject",
fix_command=f"drone @spawn repair @{project} --add-pyproject --apply",
)
)
@@ -156,7 +156,7 @@ def generate_remediation(project_root: Path) -> List[RemediationItem]:
severity=severity,
category="root_artifact",
description=f"{hit.description}: {hit.name}/",
fix_command=f"drone @spawn repair @{project} --relocate-root {hit.name}",
fix_command=f"drone @spawn repair @{project} --relocate-root {hit.name} --apply",
)
)
@@ -184,7 +184,8 @@ def format_text_report(items: List[RemediationItem], project_name: str) -> str:
lines.append(f"[{item.severity.upper()}] {item.description}")
lines.append(f" Fix: {item.fix_command}")
lines.append("")
lines.append(f"Preview all fixes: drone @spawn repair @{project_name} --dry-run")
lines.append(f"Preview all fixes: drone @spawn repair @{project_name}")
lines.append(f"Apply all fixes: drone @spawn repair @{project_name} --apply")
return "\n".join(lines)
src/aipass/aipass/tests/test_doctor_fix.py
+5 -2
View File
@@ -248,10 +248,13 @@ class TestFormatTextReport:
assert "drone @spawn repair @test --relocate a b" in result
def test_dry_run_hint(self) -> None:
"""Report ends with dry-run suggestion."""
"""Report shows preview (dry-run default) and explicit --apply hints."""
items = [RemediationItem("info", "pyproject", "missing", "fix")]
result = format_text_report(items, "myproj")
assert "drone @spawn repair @myproj --dry-run" in result
# Repair is dry-run by default now: preview form has no flag, apply form is explicit
assert "Preview all fixes:" in result
assert "drone @spawn repair @myproj" in result
assert "drone @spawn repair @myproj --apply" in result
def test_critical_sorted_first(self) -> None:
"""Critical items appear before warning and info."""
src/aipass/spawn/.seedgo/bypass.json
+15 -35
View File
@@ -25,16 +25,6 @@
"standard": "json_structure",
"reason": "Pure string transformation — placeholder replacement engine, no file I/O"
},
{
"file": "apps/handlers/change_detection.py",
"standard": "json_structure",
"reason": "Pure data comparison — detects changes between registries, no writes"
},
{
"file": "apps/handlers/reconcile.py",
"standard": "json_structure",
"reason": "Pure analysis — compares filesystem vs metadata, no writes"
},
{
"file": "log_structure",
"standard": "log_structure",
@@ -65,11 +55,6 @@
"standard": "deep_nesting",
"reason": "sync_templates() depth 4 — linear workflow loading owners, iterating managed files, checking hashes. Minimal branching, straightforward responsibility."
},
{
"file": "apps/handlers/reconcile.py",
"standard": "deep_nesting",
"reason": "reconcile_branch_state() depth 6 — three independent checks (tracked files, tracked dirs, untracked files) with iterdir loops and hash computation. Structure reflects algorithm necessity."
},
{
"file": "apps/handlers/sync_registry_ops.py",
"standard": "deep_nesting",
@@ -125,16 +110,6 @@
"standard": "naming",
"reason": "dest, source are local variables inside helper functions, not module-level constants. Path objects for template/branch comparison."
},
{
"file": "apps/handlers/reconcile.py",
"standard": "naming",
"reason": "full_path, entry are local variables inside reconcile_branch_state(), not module-level constants. Loop iteration variables."
},
{
"file": "apps/handlers/change_detection.py",
"standard": "naming",
"reason": "common_ids is a local variable inside detect_changes(), not a module-level constant. Set intersection of template and branch IDs."
},
{
"file": "apps/handlers/passport_ops.py",
"standard": "naming",
@@ -145,16 +120,6 @@
"standard": "documentation",
"reason": "delete_branch() has a docstring at line 45 — checker doesn't recognize it due to multi-line function signature spanning lines 40-44."
},
{
"file": "apps/handlers/reconcile.py",
"standard": "documentation",
"reason": "reconcile_branch_state() has a docstring at line 57 — checker doesn't recognize it due to multi-line function signature spanning lines 52-56."
},
{
"file": "apps/handlers/change_detection.py",
"standard": "documentation",
"reason": "detect_changes() has a docstring at line 29 — checker doesn't recognize it due to multi-line function signature spanning lines 24-28."
},
{
"file": "apps/handlers/passport_ops.py",
"standard": "documentation",
@@ -241,11 +206,26 @@
"standard": "encapsulation",
"reason": "Test file — tests must import handlers directly to test them in isolation. Handler imports inside test methods are intentional."
},
{
"file": "tests/test_update.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "tests/test_update.py",
"standard": "encapsulation",
"reason": "Test file — tests must import handlers directly to test them in isolation. Handler imports inside test methods are intentional."
},
{
"file": "apps/handlers/repair_ops.py",
"standard": "windows_compat",
"lines": [89],
"reason": "fcntl import guarded by sys.platform != 'win32' check — lock_fd is None on Windows, so this line never executes."
},
{
"file": "apps/handlers/meta_ops.py",
"standard": "unused_function",
"reason": "load_branch_meta() no longer called by P1 engine but has tests and may be reused by sync-registry or P2 shared library."
}
],
"notes": {
src/aipass/spawn/README.md
+12 -11
View File
@@ -56,12 +56,13 @@ drone @spawn passport @dirname --role "Observer" --purpose "Monitoring"
### Update
Update is **preview-only by default** — `--apply` required to execute changes.
```bash
drone @spawn update @branch_name # Single branch (uses passport class)
drone @spawn update builder --all # All builder-class branches
drone @spawn update birthright --all # All birthright-class branches
drone @spawn update @branch_name --dry-run # Preview changes
drone @spawn update builder --all --dry-run # Preview batch update
drone @spawn update @branch_name # Preview changes (dry-run default)
drone @spawn update @branch_name --apply # Execute changes
drone @spawn update builder --all --apply # All builder-class branches
drone @spawn update @branch_name --dry-run # Explicit preview (same as default)
```
### Delete
@@ -79,12 +80,12 @@ drone @spawn sync-templates # Pull managed fi
drone @spawn regenerate-registry # Regenerate builder template hashes
drone @spawn regenerate-registry --all # All template classes
# Repair
drone @spawn repair <project_path> # Scan project for structural issues
drone @spawn repair <project_path> --dry-run # Preview only
drone @spawn repair --relocate @branch src/pkg/branch # Move branch to new location
drone @spawn repair --relocate @branch path --relocate-artifacts # Move branch + .chroma/ into it
drone @spawn repair <project_path> --clean-pollution # Archive + remove duplicate dirs
# Repair (preview-only by default — --apply required to execute)
drone @spawn repair <project_path> # Preview structural issues (dry-run default)
drone @spawn repair <project_path> --apply # Execute fixes
drone @spawn repair --relocate @branch src/pkg/branch --apply # Move branch to new location
drone @spawn repair --relocate @branch path --relocate-artifacts --apply # Move + .chroma/
drone @spawn repair <project_path> --clean-pollution --apply # Archive + remove duplicate dirs
```
### Introspection
src/aipass/spawn/apps/handlers/change_detection.py
-185
View File
@@ -1,185 +0,0 @@
# =================== AIPass ====================
# Name: change_detection.py
# Description: Template change detection — template vs branch metadata
# Version: 1.0.0
# Created: 2026-03-07
# Modified: 2026-03-10
# =============================================
"""Template change detection handler.
Compares the template registry (what the template currently defines) against
branch metadata (what the branch currently has). Uses ID-based detection to
identify additions, updates, renames, and pruned files.
"""
from pathlib import Path
from typing import Any
# =============================================================================
# CHANGE DETECTION
# =============================================================================
def detect_changes(
template_registry: dict,
branch_meta: dict,
branch_dir: Path,
old_branch_tracking: dict | None = None,
) -> dict:
"""Detect changes between template registry and branch metadata.
Uses ID-based matching:
- ID exists in both template and branch -> compare paths (rename?) and hashes (update?)
- ID only in template -> addition (new template file)
- ID only in branch -> pruned (removed from template)
Args:
template_registry: Template registry dict (from .template_registry.json).
branch_meta: Branch metadata dict (from .branch_meta.json).
branch_dir: Path to the branch directory (for existence checks).
old_branch_tracking: Snapshot of file_tracking from BEFORE metadata
regeneration (Phase 0 pattern). Used for accurate rename detection.
When None, falls back to current branch_meta tracking.
Returns:
Dict with change categories::
{
"additions": [{"file_id": ..., "template_path": ..., "hash": ...}, ...],
"updates": [{"file_id": ..., "template_path": ..., "branch_path": ...,
"template_hash": ..., "branch_hash": ...}, ...],
"renames": [{"file_id": ..., "template_path": ..., "branch_path": ...,
"old_name": ..., "new_name": ...}, ...],
"pruned": [{"file_id": ..., "branch_path": ..., "name": ...}, ...],
}
"""
branch_dir = Path(branch_dir)
additions: list[dict[str, Any]] = []
updates: list[dict[str, Any]] = []
renames: list[dict[str, Any]] = []
pruned: list[dict[str, Any]] = []
# Extract tracking data from branch metadata
branch_file_tracking = branch_meta.get("file_tracking", {})
branch_dir_tracking = branch_meta.get("directory_tracking", {})
# Merge file and directory IDs from branch meta into a single set for lookup
branch_ids = set(branch_file_tracking.keys()) | set(branch_dir_tracking.keys())
# Build template ID sets
template_files = template_registry.get("files", {})
template_dirs = template_registry.get("directories", {})
template_ids = set(template_files.keys()) | set(template_dirs.keys())
# -----------------------------------------------------------------
# 1. Detect ADDITIONS: IDs in template but not in branch
# -----------------------------------------------------------------
for file_id in template_ids - branch_ids:
# Get template info (could be file or directory)
t_info = template_files.get(file_id) or template_dirs.get(file_id)
if not t_info:
continue
template_path = t_info.get("path", t_info.get("current_name", ""))
entry: dict[str, Any] = {
"file_id": file_id,
"template_path": template_path,
}
# Include hash for files (directories don't have hashes)
if file_id in template_files:
entry["hash"] = t_info.get("content_hash", "")
additions.append(entry)
# -----------------------------------------------------------------
# 2. Detect PRUNED: IDs in branch but not in template
# -----------------------------------------------------------------
for file_id in branch_ids - template_ids:
b_info = branch_file_tracking.get(file_id) or branch_dir_tracking.get(file_id)
if not b_info:
continue
branch_path = b_info.get("current_path", "")
name = b_info.get("current_name", "")
pruned.append(
{
"file_id": file_id,
"branch_path": branch_path,
"name": name,
}
)
# -----------------------------------------------------------------
# 3. Detect RENAMES and UPDATES: IDs in both template and branch
# -----------------------------------------------------------------
common_ids = template_ids & branch_ids
for file_id in common_ids:
# Get info from both sides
t_info = template_files.get(file_id) or template_dirs.get(file_id)
b_info = branch_file_tracking.get(file_id) or branch_dir_tracking.get(file_id)
if not t_info or not b_info:
continue
template_path = t_info.get("path", t_info.get("current_name", ""))
branch_path = b_info.get("current_path", "")
# Check for RENAME: same ID but path changed in template
# Use old tracking snapshot for comparison if available (Phase 0 pattern)
if old_branch_tracking and file_id in old_branch_tracking:
old_info = old_branch_tracking[file_id]
old_template_name = old_info.get("template_name", "")
if old_template_name and template_path != old_template_name:
renames.append(
{
"file_id": file_id,
"template_path": template_path,
"branch_path": branch_path,
"old_name": old_template_name,
"new_name": template_path,
}
)
else:
# Fallback: compare against current branch meta
branch_template_name = b_info.get("template_name", "")
if branch_template_name and template_path != branch_template_name:
renames.append(
{
"file_id": file_id,
"template_path": template_path,
"branch_path": branch_path,
"old_name": branch_template_name,
"new_name": template_path,
}
)
# Check for UPDATE: same ID, same location, content hash changed
# Only applies to files (directories don't have content hashes)
if file_id in template_files and file_id in branch_file_tracking:
template_hash = t_info.get("content_hash", "")
branch_hash = b_info.get("content_hash", "")
if template_hash and branch_hash and template_hash != branch_hash:
updates.append(
{
"file_id": file_id,
"template_path": template_path,
"branch_path": branch_path,
"template_hash": template_hash,
"branch_hash": branch_hash,
}
)
return {
"additions": additions,
"updates": updates,
"renames": renames,
"pruned": pruned,
}
src/aipass/spawn/apps/handlers/reconcile.py
-220
View File
@@ -1,220 +0,0 @@
# =================== AIPass ====================
# Name: reconcile.py
# Description: Branch state reconciliation — filesystem vs metadata
# Version: 1.0.0
# Created: 2026-03-07
# Modified: 2026-03-10
# =============================================
"""Branch state reconciliation handler.
Compares what .branch_meta.json says should exist vs what actually exists
on the filesystem. Identifies missing files, untracked files, hash
mismatches, and missing directories.
"""
import hashlib
from pathlib import Path
from typing import Any
from aipass.prax.apps.modules.logger import system_logger as logger
# Directories/files to skip during filesystem scans
_SKIP_NAMES = {"__pycache__", ".git", ".branch_meta.json", ".template_registry.json"}
# =============================================================================
# HELPERS
# =============================================================================
def _compute_hash(file_path: Path) -> str:
"""Compute SHA-256 hash, first 12 hex chars."""
try:
sha256 = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256.update(chunk)
return sha256.hexdigest()[:12]
except (IOError, PermissionError) as e:
logger.warning("Failed to compute hash for %s: %s", file_path, e)
return ""
def _should_skip(name: str) -> bool:
"""Check if a file or directory name should be skipped during scan."""
return name in _SKIP_NAMES
# =============================================================================
# RECONCILIATION
# =============================================================================
def reconcile_branch_state(
branch_dir: Path,
branch_meta: dict,
trace: bool = False,
) -> dict:
"""Reconcile branch metadata with actual filesystem state.
Compares tracked files/directories in .branch_meta.json against what
exists on disk within the branch directory.
Args:
branch_dir: Path to the branch directory.
branch_meta: Loaded branch metadata dict (from .branch_meta.json).
trace: If True, enable verbose logging via prax.
Returns:
Dict with reconciliation results::
{
"missing_files": [{"file_id": ..., "path": ..., "template_name": ...}, ...],
"untracked_files": [{"path": ..., "name": ...}, ...],
"hash_mismatches": [{"file_id": ..., "path": ..., "tracked_hash": ..., "current_hash": ...}, ...],
"missing_dirs": [{"dir_id": ..., "path": ..., "template_name": ...}, ...],
"needs_update": bool,
}
"""
branch_dir = Path(branch_dir)
file_tracking = branch_meta.get("file_tracking", {})
dir_tracking = branch_meta.get("directory_tracking", {})
missing_files: list[dict[str, Any]] = []
untracked_files: list[dict[str, Any]] = []
hash_mismatches: list[dict[str, Any]] = []
missing_dirs: list[dict[str, Any]] = []
if trace:
logger.info(f"[reconcile] Starting reconciliation for {branch_dir.name}")
logger.info(f"[reconcile] Tracked files: {len(file_tracking)}, dirs: {len(dir_tracking)}")
# -------------------------------------------------------------------------
# Check 1: Tracked files — do they still exist on disk?
# -------------------------------------------------------------------------
for file_id, file_info in file_tracking.items():
tracked_path = file_info.get("current_path", "")
if not tracked_path:
continue
full_path = branch_dir / tracked_path
if not full_path.exists():
entry = {
"file_id": file_id,
"path": tracked_path,
"template_name": file_info.get("template_name", ""),
}
missing_files.append(entry)
if trace:
logger.info(f"[reconcile] MISSING: {file_id} -> {tracked_path}")
elif full_path.is_file():
# Check hash mismatch
tracked_hash = file_info.get("content_hash", "")
if tracked_hash:
current_hash = _compute_hash(full_path)
if current_hash and current_hash != tracked_hash:
entry = {
"file_id": file_id,
"path": tracked_path,
"tracked_hash": tracked_hash,
"current_hash": current_hash,
}
hash_mismatches.append(entry)
if trace:
logger.info(
f"[reconcile] HASH MISMATCH: {file_id} tracked={tracked_hash} current={current_hash}"
)
# -------------------------------------------------------------------------
# Check 2: Tracked directories — do they still exist?
# -------------------------------------------------------------------------
for dir_id, dir_info in dir_tracking.items():
tracked_path = dir_info.get("current_path", "")
if not tracked_path:
continue
full_path = branch_dir / tracked_path
if not full_path.is_dir():
entry = {
"dir_id": dir_id,
"path": tracked_path,
"template_name": dir_info.get("template_name", ""),
}
missing_dirs.append(entry)
if trace:
logger.info(f"[reconcile] MISSING DIR: {dir_id} -> {tracked_path}")
# -------------------------------------------------------------------------
# Check 3: Untracked files — on disk but not in metadata
# Only check within template-managed directories (tracked dir paths).
# -------------------------------------------------------------------------
tracked_paths = {info.get("current_path", "") for info in file_tracking.values()}
tracked_dir_paths = {info.get("current_path", "") for info in dir_tracking.values()}
# Scan within tracked directories for untracked files
for dir_path_str in tracked_dir_paths:
if not dir_path_str:
continue
scan_dir = branch_dir / dir_path_str
if not scan_dir.is_dir():
continue
try:
for item in scan_dir.iterdir():
if _should_skip(item.name):
continue
if not item.is_file():
continue
rel_path = str(item.relative_to(branch_dir))
if rel_path not in tracked_paths:
entry = {
"path": rel_path,
"name": item.name,
}
untracked_files.append(entry)
if trace:
logger.info(f"[reconcile] UNTRACKED: {rel_path}")
except (PermissionError, OSError) as exc:
if trace:
logger.warning(f"[reconcile] Scan error in {dir_path_str}: {exc}")
# Also check root-level files
try:
for item in branch_dir.iterdir():
if _should_skip(item.name):
continue
if not item.is_file():
continue
rel_path = str(item.relative_to(branch_dir))
if rel_path not in tracked_paths:
entry = {
"path": rel_path,
"name": item.name,
}
untracked_files.append(entry)
if trace:
logger.info(f"[reconcile] UNTRACKED (root): {rel_path}")
except (PermissionError, OSError) as exc:
if trace:
logger.warning(f"[reconcile] Root scan error: {exc}")
needs_update = bool(missing_files or untracked_files or hash_mismatches or missing_dirs)
if trace:
logger.info(
f"[reconcile] Complete: missing={len(missing_files)}, "
f"untracked={len(untracked_files)}, "
f"hash_mismatches={len(hash_mismatches)}, "
f"missing_dirs={len(missing_dirs)}, "
f"needs_update={needs_update}"
)
return {
"missing_files": missing_files,
"untracked_files": untracked_files,
"hash_mismatches": hash_mismatches,
"missing_dirs": missing_dirs,
"needs_update": needs_update,
}
src/aipass/spawn/apps/handlers/update_ops.py
+164 -384
View File
@@ -1,19 +1,18 @@
# =================== AIPass ====================
# Name: update_ops.py
# Description: Update handler — implementation logic for branch updates
# Version: 1.0.0
# Description: Update handler — path-based template sync engine (P1 rewrite, TDPLAN-0006)
# Version: 2.0.0
# Created: 2026-03-07
# Modified: 2026-03-10
# Modified: 2026-06-06
# =============================================
"""Update handler implementation for branch lifecycle management.
"""Update handler — path-based template sync engine.
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.
P1 rewrite (TDPLAN-0006, issue #636): replaces the broken ID-based
change-detection engine with explicit template walking. No renames,
no pruning, never touches identity/memory files.
"""
import copy
import json
import shutil
from datetime import datetime
@@ -25,17 +24,39 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.spawn.apps.handlers.meta_ops import (
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, branches_as_list
from aipass.spawn.apps.handlers.json import json_handler
_NEVER_UPDATE_PREFIXES = (".trinity/",)
_NEVER_UPDATE_FILES = frozenset(
{
"DASHBOARD.local.json",
"artifacts/birth_certificate.json",
".seedgo/bypass.json",
}
)
_SKIP_TRACKING = frozenset(
{
".spawn/.template_registry.json",
".spawn/.branch_meta.json",
}
)
def _is_never_update(resolved_path: str) -> bool:
"""Check if a resolved path is in the create-only set."""
if resolved_path in _NEVER_UPDATE_FILES:
return True
for prefix in _NEVER_UPDATE_PREFIXES:
if resolved_path.startswith(prefix):
return True
return False
# =============================================================================
# PUBLIC API
@@ -43,220 +64,133 @@ from aipass.spawn.apps.handlers.json import json_handler
def update_branch(branch_name: str, dry_run: bool = False, trace: bool = False) -> dict:
"""Update a single branch from template.
"""Update a single branch from its class 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.
Path-based engine: walks the template directory, resolves placeholder
paths, and for each file decides add/merge/skip. No renames, no pruning.
Identity files (.trinity/*, DASHBOARD, birth_certificate, bypass.json)
are never touched — create-only.
"""
errors: list[str] = []
counts = {
"additions": 0,
"renames": 0,
"updates": 0,
"pruned": 0,
"skipped_py": 0,
}
counts = {"additions": 0, "renames": 0, "updates": 0, "pruned": 0, "skipped_py": 0}
additions_detail: list[dict] = []
updates_detail: list[dict] = []
# ------------------------------------------------------------------
# 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}")
logger.info("[update] Resolved %s -> %s", branch_name, branch_dir)
# ------------------------------------------------------------------
# 1b. Read citizen_class from passport
# ------------------------------------------------------------------
citizen_class = _read_citizen_class(branch_dir)
if trace:
logger.info(f"[update] Citizen class: {citizen_class}")
# ------------------------------------------------------------------
# 2. Load template registry
# ------------------------------------------------------------------
template_dir = get_template_dir(citizen_class)
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 not template_dir.is_dir():
return _result(branch_name, False, counts, [f"Template directory not found: {template_dir}"], 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")
logger.info("[update] Citizen class: %s, template: %s", citizen_class, template_dir)
# ------------------------------------------------------------------
# 3. Load or generate branch metadata (first-time adoption)
# ------------------------------------------------------------------
branch_meta = load_branch_meta(branch_dir)
first_time = branch_meta is None
# Phase 0: Snapshot old tracking BEFORE any metadata generation/sync
# This preserves the old template_name values for rename detection
old_branch_tracking: dict | None = None
if branch_meta is not None:
old_branch_tracking = copy.deepcopy(branch_meta.get("file_tracking", {}))
if first_time:
if trace:
logger.info("[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, old_branch_tracking=old_branch_tracking)
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)
# Walk template directories — create missing ones in branch
for template_subdir in sorted(template_dir.rglob("*")):
if not template_subdir.is_dir():
continue
if "__pycache__" in template_subdir.parts:
continue
rel_dir = template_subdir.relative_to(template_dir).as_posix()
resolved_dir = replace_placeholders(rel_dir, replacements)
if _is_never_update(resolved_dir + "/"):
continue
if resolved_dir in _SKIP_TRACKING:
continue
dest_dir = branch_dir / resolved_dir
if not dest_dir.exists():
if not dry_run:
dest_dir.mkdir(parents=True, exist_ok=True)
additions_detail.append({"template_path": resolved_dir, "type": "directory"})
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}")
if trace:
logger.info("[update] Added directory: %s", resolved_dir)
# 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)
# Walk template files — add/merge/skip per type
for template_file in sorted(template_dir.rglob("*")):
if not template_file.is_file():
continue
if "__pycache__" in template_file.parts:
continue
rel_path = template_file.relative_to(template_dir).as_posix()
if rel_path in _SKIP_TRACKING:
continue
resolved_path = replace_placeholders(rel_path, replacements)
if _is_never_update(resolved_path):
if trace:
logger.info("[update] SKIP (create-only): %s", resolved_path)
continue
dest = branch_dir / resolved_path
if not dest.exists():
# ADDITION — missing file from template
if not dry_run:
dest.parent.mkdir(parents=True, exist_ok=True)
try:
content = template_file.read_text(encoding="utf-8")
content = replace_placeholders(content, replacements)
dest.write_text(content, encoding="utf-8")
except (UnicodeDecodeError, UnicodeEncodeError) as enc_err:
logger.warning("[update] Binary file, copying directly: %s (%s)", resolved_path, enc_err)
shutil.copy2(template_file, dest)
additions_detail.append({"template_path": resolved_path, "type": "file"})
counts["additions"] += 1
if trace:
logger.info("[update] Added: %s", resolved_path)
elif dest.suffix == ".py":
counts["skipped_py"] += 1
updates_detail.append({"template_path": rel_path, "branch_path": resolved_path})
if trace:
logger.info("[update] SKIP .py: %s", resolved_path)
elif dest.suffix == ".json":
result = _merge_json(template_file, dest, replacements, dry_run, 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}")
updates_detail.append({"template_path": rel_path, "branch_path": resolved_path})
elif result == "error":
errors.append(f"JSON merge failed for {resolved_path}")
# 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}")
# Refresh branch metadata (informational tracking)
if not dry_run:
template_registry = load_template_registry(template_dir)
if template_registry:
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)
# ------------------------------------------------------------------
# 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
if success:
if success and not dry_run:
json_handler.log_operation("update_executed", data={"branch": branch_name})
return _result(branch_name, success, counts, errors, dry_run)
return _result(
branch_name,
success,
counts,
errors,
dry_run,
_additions_detail=additions_detail,
_updates_detail=updates_detail,
_renames_detail=[],
_pruned_detail=[],
)
def update_all(dry_run: bool = False, trace: bool = False, citizen_class: str | None = None) -> list[dict]:
@@ -280,30 +214,28 @@ def update_all(dry_run: bool = False, trace: bool = False, citizen_class: str |
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
# Check class filter if specified
if citizen_class:
branch_dir = _resolve_branch_path(lower_name)
if branch_dir and branch_dir.is_dir():
actual_class = _read_citizen_class(branch_dir)
if actual_class != citizen_class:
if trace:
logger.info(f"[update] Skipping {name} (class={actual_class}, filter={citizen_class})")
logger.info("[update] Skipping %s (class=%s, filter=%s)", name, actual_class, citizen_class)
continue
if trace:
logger.info(f"[update] Processing branch: {name}")
logger.info("[update] Processing branch: %s", 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}")
logger.error("[update] Error updating %s: %s", name, exc)
results.append(
{
"branch": lower_name,
@@ -322,16 +254,12 @@ def update_all(dry_run: bool = False, trace: bool = False, citizen_class: str |
# =============================================================================
# INTERNAL EXECUTION FUNCTIONS
# INTERNAL HELPERS
# =============================================================================
def _read_citizen_class(branch_dir: Path) -> str:
"""Read citizen_class from a branch's passport.json.
Falls back to "builder" if passport doesn't exist, doesn't have
the citizen_class field, or has an unknown class (e.g. "manager").
"""
"""Read citizen_class from a branch's passport.json."""
from aipass.spawn.apps.handlers.class_registry import validate_class
passport_path = branch_dir / ".trinity" / "passport.json"
@@ -342,21 +270,17 @@ def _read_citizen_class(branch_dir: Path) -> str:
citizen_class = data.get("identity", {}).get("citizen_class", "builder")
if not validate_class(citizen_class):
logger.warning(
f"[update] Unknown citizen_class '{citizen_class}' in {passport_path}, falling back to 'builder'"
"[update] Unknown citizen_class '%s' in %s, falling back to 'builder'", citizen_class, passport_path
)
return "builder"
return citizen_class
except (json.JSONDecodeError, IOError) as e:
logger.warning(f"[update] Failed to read citizen_class from passport {passport_path}: {e}")
logger.warning("[update] Failed to read citizen_class from passport %s: %s", passport_path, e)
return "builder"
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 registry parent (project root).
"""
"""Resolve a branch name to its absolute directory path via the registry."""
registry_path = find_registry()
project_root = registry_path.parent
registry = load_registry(registry_path)
@@ -371,188 +295,44 @@ def _resolve_branch_path(branch_name: str) -> Path | None:
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
# NEVER overwrite existing .py files — even during additions
if dest.exists() and dest.suffix == ".py":
if trace:
logger.info(f"[update] SKIPPED existing .py file: {dest_rel_str}")
return
# Create parent directories
dest.parent.mkdir(parents=True, exist_ok=True)
# Don't overwrite any existing file — additions are for MISSING files only
if dest.exists():
if trace:
logger.info(f"[update] SKIPPED existing file: {dest_rel_str}")
return
try:
content = source.read_text(encoding="utf-8")
content = replace_placeholders(content, replacements)
dest.write_text(content, encoding="utf-8")
except (UnicodeDecodeError, UnicodeEncodeError) as e:
# Binary file — copy directly
logger.warning(f"[update] Text read/write failed for {template_path}, falling back to binary copy: {e}")
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,
def _merge_json(
template_file: Path,
dest: Path,
replacements: dict,
dry_run: bool,
trace: bool,
) -> str:
"""Handle a file update from template.
"""Deep-merge a template JSON file into the branch copy.
Returns:
"updated" if file was merged, "skipped_py" if .py file, "skipped" otherwise.
Returns "updated", "unchanged", or "error".
"""
template_path = upd_info.get("template_path", "")
branch_path = upd_info.get("branch_path", "")
try:
template_content = template_file.read_text(encoding="utf-8")
template_content = replace_placeholders(template_content, replacements)
template_data = json.loads(template_content)
if not template_path or not branch_path:
return "skipped"
existing_text = dest.read_text(encoding="utf-8")
existing_data = json.loads(existing_text)
# 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"
merged = deep_merge(template_data, existing_data)
merged_text = json.dumps(merged, indent=2, ensure_ascii=False) + "\n"
# 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 merged_text == existing_text:
if trace:
logger.info(f"[update] Source or dest missing for JSON merge: {template_path}")
return "skipped"
logger.info("[update] JSON unchanged: %s", dest.name)
return "unchanged"
try:
# Backup existing file first
if not dry_run:
backup_json(dest)
dest.write_text(merged_text, encoding="utf-8")
# 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
logger.info("[update] JSON merged: %s", dest.name)
return "updated"
# 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}")
except (json.JSONDecodeError, IOError) as exc:
logger.error("[update] JSON merge failed for %s: %s", dest.name, exc)
return "error"
# =============================================================================
src/aipass/spawn/apps/modules/core.py
+4 -2
View File
@@ -71,7 +71,8 @@ def print_introspection():
console.print(" - metadata.py (get_branch_name, normalize_branch_name, detect_profile — branch identity)")
console.print(" - placeholders.py (build_replacements_dict, validate_no_placeholders — template substitution)")
console.print(
" - file_ops.py (copy_template, rename_placeholder_paths, regenerate_template_registry, ensure_directory — filesystem ops)"
" - file_ops.py (copy_template, rename_placeholder_paths,"
" regenerate_template_registry, ensure_directory — filesystem ops)"
)
console.print(
" - meta_ops.py (load_template_registry, generate_branch_meta, save_branch_meta — branch metadata)"
@@ -80,7 +81,8 @@ def print_introspection():
" - registry.py (find_registry, add_to_registry, get_next_citizen_number — AIPASS_REGISTRY management)"
)
console.print(
" - class_registry.py (validate_class, get_default_class, get_available_classes, get_template_dir — citizen class lookup)"
" - class_registry.py (validate_class, get_default_class,"
" get_available_classes, get_template_dir — citizen class lookup)"
)
console.print()
src/aipass/spawn/apps/modules/repair.py
+7 -5
View File
@@ -83,7 +83,8 @@ def handle_repair(args: list[str]) -> int:
_print_help()
return 1
dry_run = "--dry-run" in args
apply = "--apply" in args
dry_run = not apply
if "--relocate" in args:
return _handle_relocate(args, dry_run)
@@ -101,7 +102,7 @@ def handle_repair(args: list[str]) -> int:
def _handle_relocate(args, dry_run):
"""Handle --relocate @branch new/path [--relocate-artifacts]."""
flags = {"--dry-run", "--relocate", "--relocate-artifacts"}
flags = {"--dry-run", "--apply", "--relocate", "--relocate-artifacts"}
positional = [a for a in args if a not in flags]
relocate_artifacts = "--relocate-artifacts" in args
@@ -137,7 +138,7 @@ def _handle_relocate(args, dry_run):
def _handle_clean_pollution(args, dry_run):
"""Handle --clean-pollution <project_path>."""
flags = {"--dry-run", "--clean-pollution"}
flags = {"--dry-run", "--apply", "--clean-pollution"}
positional = [a for a in args if a not in flags]
if not positional:
@@ -159,7 +160,7 @@ def _handle_clean_pollution(args, dry_run):
def _handle_scan(args, dry_run):
"""Handle default scan mode — report structural issues."""
flags = {"--dry-run"}
flags = {"--dry-run", "--apply"}
positional = [a for a in args if a not in flags]
if not positional:
@@ -189,7 +190,8 @@ def _print_help():
warning("Usage: drone @spawn repair <project_path> [options]")
console.print()
console.print(" [green]<project_path>[/green] Path to project root")
console.print(" [green]--dry-run[/green] Preview changes without modifying")
console.print(" [green]--apply[/green] Execute changes (default is preview-only)")
console.print(" [green]--dry-run[/green] Preview changes without modifying [dim](default)[/dim]")
console.print(" [green]--relocate[/green] @branch path Move a branch to a new location")
console.print(
" [green]--relocate-artifacts[/green] Move .chroma/ into branch (with --relocate, single-branch only)"
src/aipass/spawn/apps/modules/update.py
+8 -6
View File
@@ -89,31 +89,33 @@ def handle_update(args: list[str]) -> int:
"""
# Intercept --help before processing (argparse has add_help=False)
if "--help" in args or "-h" in args:
warning("Usage: drone @spawn update <@branch|class --all> [--dry-run] [--trace]")
warning("Usage: drone @spawn update <@branch|class --all> [--apply] [--dry-run] [--trace]")
console.print()
console.print(" [green]@branch[/green] Update a single branch (uses its own class)")
console.print(" [green]builder --all[/green] Update all builder-class branches")
console.print(" [green]birthright --all[/green] Update all birthright-class branches")
console.print(" [green]--dry-run[/green] Preview changes without modifying files")
console.print(" [green]--apply[/green] Execute changes (default is preview-only)")
console.print(" [green]--dry-run[/green] Preview changes without modifying files [dim](default)[/dim]")
console.print(" [green]--trace[/green] Enable verbose logging")
return 0
if not args:
warning("Usage: drone @spawn update <@branch|class --all> [--dry-run] [--trace]")
warning("Usage: drone @spawn update <@branch|class --all> [--apply] [--dry-run] [--trace]")
console.print()
console.print(" [green]@branch[/green] Update a single branch (uses its own class)")
console.print(" [green]builder --all[/green] Update all builder-class branches")
console.print(" [green]birthright --all[/green] Update all birthright-class branches")
console.print(" [green]--dry-run[/green] Preview changes without modifying files")
console.print(" [green]--apply[/green] Execute changes (default is preview-only)")
console.print(" [green]--dry-run[/green] Preview changes without modifying files [dim](default)[/dim]")
console.print(" [green]--trace[/green] Enable verbose logging")
return 1
from aipass.spawn.apps.modules.core import validate_class, get_available_classes
dry_run = "--dry-run" in args
apply = "--apply" in args
dry_run = not apply
trace = "--trace" in args
# Filter out flags to find positional args
positional = [a for a in args if not a.startswith("--")]
# Detect citizen class argument
src/aipass/spawn/apps/spawn.py
+5 -3
View File
@@ -38,8 +38,9 @@ def print_help():
console.print()
console.print(" [green]create[/green] [class] <path> Create a new branch from template")
console.print(" [green]passport[/green] <@dirname> Grant birthright citizenship (minimal)")
console.print(" [green]update[/green] <@branch> Update single branch (uses its class)")
console.print(" [green]update[/green] <class> --all Update all branches of a class")
console.print(" [green]update[/green] <@branch> Update single branch (preview-only by default)")
console.print(" [green]update[/green] <@branch> --apply Update single branch (execute changes)")
console.print(" [green]update[/green] <class> --all --apply Update all branches of a class")
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")
@@ -60,7 +61,8 @@ def print_help():
warning("--purpose", details="Agent purpose (brief)")
warning("--template", details="Template class name (builder, birthright) or custom directory path")
warning("--registry", details="Path to AIPASS_REGISTRY.json")
warning("--dry-run", details="Preview changes without modifying files")
warning("--apply", details="Execute changes (update/repair are preview-only by default)")
warning("--dry-run", details="Preview changes without modifying files (default for update/repair)")
warning("--trace", details="Enable verbose logging")
console.print()
src/aipass/spawn/templates/builder/.spawn/.template_registry.json
+1 -1
View File
@@ -1,7 +1,7 @@
{
"metadata": {
"version": "1.0.0",
"last_updated": "2026-05-22",
"last_updated": "2026-06-06",
"description": "Template file tracking registry for ID-based updates"
},
"files": {
src/aipass/spawn/tests/test_handlers.py
+1 -298
View File
@@ -6,7 +6,7 @@
# Modified: 2026-03-07
# =============================================
"""Tests for spawn handler modules: meta_ops, reconcile, change_detection, json_ops."""
"""Tests for spawn handler modules: meta_ops, json_ops."""
import pytest
from pathlib import Path
@@ -313,303 +313,6 @@ class TestDeepMerge:
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"
# =============================================================================
# backup_json tests
# =============================================================================
src/aipass/spawn/tests/test_update.py
+81 -30
View File
@@ -133,6 +133,7 @@ def branch_dir(tmp_path):
(branch / "tests" / "__init__.py").write_text("")
(branch / ".archive").mkdir()
(branch / "docs").mkdir()
(branch / ".spawn").mkdir()
return branch
@@ -346,34 +347,13 @@ class TestUpdateBranch:
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."""
def test_extra_files_not_pruned(self, tmp_path, template_dir, branch_dir, mock_registry):
"""P1 engine never prunes — extra branch files are left untouched."""
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),
@@ -381,15 +361,86 @@ class TestUpdateBranch:
result = update_branch("test_branch")
assert result["success"] is True
assert result["pruned"] >= 1
assert result["pruned"] == 0
assert extra_file.exists()
# 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 TestNeverUpdateGuard:
"""Tests for create-only file protection (P1 engine, TDPLAN-0006)."""
def test_trinity_files_never_touched(self, tmp_path, template_dir, branch_dir, mock_registry):
"""Update must never modify .trinity/ files even when template has them."""
from aipass.spawn.apps.handlers.update_ops import update_branch
# Add .trinity/ to template
trinity_tpl = template_dir / ".trinity"
trinity_tpl.mkdir(exist_ok=True)
(trinity_tpl / "passport.json").write_text('{"identity": {"role": "template"}}')
(trinity_tpl / "local.json").write_text('{"sessions": []}')
# Add .trinity/ to branch with different content
trinity_branch = branch_dir / ".trinity"
trinity_branch.mkdir(exist_ok=True)
(trinity_branch / "passport.json").write_text('{"identity": {"role": "real_agent"}}')
(trinity_branch / "local.json").write_text('{"sessions": [{"id": 1}]}')
passport_before = (trinity_branch / "passport.json").read_text()
local_before = (trinity_branch / "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),
):
result = update_branch("test_branch")
assert result["success"] is True
assert (trinity_branch / "passport.json").read_text() == passport_before
assert (trinity_branch / "local.json").read_text() == local_before
def test_dashboard_never_touched(self, tmp_path, template_dir, branch_dir, mock_registry):
"""Update must never modify DASHBOARD.local.json even when template differs."""
from aipass.spawn.apps.handlers.update_ops import update_branch
dashboard = branch_dir / "DASHBOARD.local.json"
dashboard_before = dashboard.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),
):
result = update_branch("test_branch")
assert result["success"] is True
assert dashboard.read_text() == dashboard_before
def test_zero_renames_always(self, tmp_path, template_dir, branch_dir, mock_registry):
"""P1 engine never proposes renames."""
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),
):
result = update_branch("test_branch", dry_run=True)
assert result["renames"] == 0
assert result.get("_renames_detail", []) == []
def test_create_update_invariant(self, tmp_path, template_dir, branch_dir, mock_registry):
"""Fresh branch from template should show 0 changes on update."""
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),
):
result = update_branch("test_branch", dry_run=True)
assert result["success"] is True
assert result["additions"] == 0
assert result["renames"] == 0
assert result["updates"] == 0
assert result["pruned"] == 0
class TestUpdateAll: