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

feat(system): S89: full external access (AIPASS_HOME dual registry) + feedback channel (cross-project bridge) + init update idempotency

Browse Source 
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-11 14:49:16 -07:00
co-authored by @devpulse
parent 5b2d14a846
commit e8a9af4bc0
14 changed files with 584 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.aipass/aipass_global_prompt.md
+16
View File
@@ -78,6 +78,8 @@ drone @flow list open # List active plans
**FPLAN** = Flow Plan. Building and executing. Default is for single focused tasks. Master is for multi-phase projects that spawn sub-FPLANs per phase. DPLANs come first, FPLANs come when you're ready to build.
**Never create plan files manually.** Always use `drone @flow create`. Flow handles numbering (global 4-digit sequence), registry tracking, templates, and date stamps. Manual files break the registry and produce wrong numbering. This applies to DPLANs, FPLANs, and APLANs — in any project, inside or outside the AIPass repo.
## Dispatch — Send Task + Wake a Branch
```
@@ -98,6 +100,18 @@ drone @ai_mail dispatch wake --fresh @target
- `email @target` = just mail, no wake (FYI only — use only when explicitly requested)
- `--dispatch` flag on `email` = adds dispatch header but doesn't auto-wake
## Feedback — Cross-Project Communication
Send feedback to devpulse from any project. Messages accumulate silently — no wake, no notification. DevPulse reads on demand. Works from any AIPass project (requires `AIPASS_HOME` set).
```
drone @devpulse feedback send "Subject" "Body" # Send feedback (sender auto-detected)
drone @devpulse feedback inbox # List all messages (devpulse only)
drone @devpulse feedback view <id> # Read message + thread
drone @devpulse feedback reply <id> "message" # Reply (lands in sender's ai_mail)
drone @devpulse feedback clear <id> # Remove a message
```
**Always reply to dispatch emails.** When devpulse or another branch sends you work, they're waiting for a response. Complete the task, then email back with results. No silent completions — if someone dispatched you, they need to know what happened.
## How to Work
@@ -108,6 +122,8 @@ drone @ai_mail dispatch wake --fresh @target
**Check seedgo standards.** Before building: `drone @seedgo standards_query aipass_standards` to know what applies. During: check your work against standards as you go. After: `drone @seedgo audit aipass @{branch}` as a final gate before committing.
**Ask before spelunking.** When you need to know how another branch works — how it routes, what config it uses, what functions are available — dispatch the question to that branch instead of reading through their files yourself. A quick `drone @ai_mail dispatch @target "Question" "How does X work?"` gets you an expert answer faster than digging through 4-5 unfamiliar files. Save deep investigation for when you're explicitly asked to check something out or need more context on a specific issue.
## Logging & Debugging
Prax is the **only** logging system. Every branch uses:
.gitignore
+1
View File
@@ -28,6 +28,7 @@ AIPASS_REGISTRY.json
.trinity/
.ai_mail.local/
ai_mail.local/
.feedback.local/
DASHBOARD.local.json
dev.local.md
STATUS.local.md
src/aipass/cli/README.md
+4 -4
View File
@@ -5,8 +5,8 @@
**Purpose:** Display and output formatting service for AIPass modules. Provides consistent terminal output — headers, success/error/warning messages, section breaks, and operation templates — so every module looks the same without duplicating Rich formatting code.
**Module:** `aipass.cli`
**Seedgo:** 100%
**Tests:** 142 passing (6 files, 5/5 modules covered)
**Last Updated:** 2026-04-10
**Tests:** 151 passing (6 files, 5/5 modules covered)
**Last Updated:** 2026-04-11
## Usage
@@ -66,7 +66,7 @@ cli/
├── cli_json/ # Auto-created JSON output (three-file pattern)
├── dropbox/ # Inbound file drop
├── logs/ # Branch-level logs
├── tests/ # 142 tests across 6 files
├── tests/ # 151 tests across 6 files
│ ├── test_bootstrap.py # bootstrap.py handler tests
│ ├── test_json_handler.py # json_handler tests
│ ├── test_display.py # display module tests
@@ -120,7 +120,7 @@ aipass --help # Via console_scripts entry point
---
*Last Updated: 2026-04-10*
*Last Updated: 2026-04-11*
---
[← Back to AIPass](../../../README.md)
src/aipass/cli/apps/handlers/init/bootstrap.py
+119 -30
View File
@@ -238,22 +238,99 @@ def _global_prompt_md(name: str) -> str:
f"# {name} — Project Context\n"
"<!-- Injected every turn via hook. -->\n"
"\n"
"## What is AIPass\n"
"\n"
"AIPass is a multi-agent framework. Agents live in directories with\n"
"persistent identity, memory, and communication. All AIPass infrastructure\n"
"is available from any project via the `drone` command.\n"
"\n"
"## Terminology\n"
"\n"
"- **Project** — this directory. Contains a registry and agents.\n"
"- **Agent** — a citizen with identity (`.trinity/`), memory, mailbox, "
"and code (`apps/`).\n"
"- **Agent** — a citizen with identity (`.trinity/`), memory, mailbox,\n"
" and code (`apps/`).\n"
f"- **Registry** — `{name}_REGISTRY.json` tracks all agents.\n"
"\n"
"## Setup: if drone commands fail\n"
"\n"
"If `drone` cannot find the AIPass registry, set the env var:\n"
"```bash\n"
"export AIPASS_HOME=/path/to/AIPass # path to AIPass installation\n"
"```\n"
"Add to your shell profile (`~/.bashrc` or `~/.zshrc`) to make it permanent.\n"
"\n"
"## Commands\n"
"\n"
"### Agent Lifecycle\n"
"```\n"
"aipass init agent <name> # Create a new agent\n"
"aipass init agent <name> # Create a new agent in src/<name>/\n"
"drone @spawn create <name> # Create agent (alternative)\n"
"drone @seedgo audit <project> # Standards audit\n"
"drone @ai_mail inbox # Check mailbox\n"
"drone @flow create . \"Subject\" # Create a plan\n"
"drone systems # List infrastructure\n"
"drone @spawn list # List registered agents\n"
"```\n"
"\n"
"### Standards\n"
"```\n"
"drone @seedgo audit <project> # Run full standards audit\n"
"drone @seedgo checklist <file> # Check a single file\n"
"```\n"
"\n"
"### Dispatch — Send Task + Wake an Agent (DEFAULT)\n"
"```\n"
"drone @ai_mail dispatch @<agent> \"Subject\" \"Body\" # Send + wake (default)\n"
"drone @ai_mail dispatch @<agent> \"Subject\" \"Body\" --fresh # Send + wake fresh session\n"
"drone @ai_mail dispatch wake @<agent> # Wake without sending\n"
"drone @ai_mail dispatch wake --fresh @<agent> # Wake fresh\n"
"drone @ai_mail email @<agent> \"Subject\" \"Body\" # FYI only (no wake)\n"
"```\n"
"\n"
"Use `dispatch` by default. Use `email` only when you don't need the agent to act now.\n"
"\n"
"### Communication (ai_mail)\n"
"```\n"
"drone @ai_mail inbox # Check your mailbox\n"
"drone @ai_mail view <id> # Read a message\n"
"drone @ai_mail close <id> # Mark message read\n"
"```\n"
"\n"
"### Feedback\n"
"```\n"
"drone @devpulse feedback send \"Subject\" \"Body\" # Send feedback (cross-project)\n"
"```\n"
"\n"
"### Plans (flow)\n"
"```\n"
"drone @flow create . \"Subject\" dplan # Create DPLAN (design/thinking)\n"
"drone @flow create . \"Subject\" master # Create FPLAN master (execution)\n"
"drone @flow create . \"Subject\" aplan # Create APLAN (agent-level task)\n"
"drone @flow list open # List active plans\n"
"drone @flow list # List all plans\n"
"drone @flow close <id> # Close a plan\n"
"drone @flow info <id> # View plan details\n"
"```\n"
"\n"
"**DPLAN** = Dev Plan. Thinking, brainstorming, architecture decisions. "
"Use before building.\n"
"**FPLAN** = Flow Plan. Building and executing. Use when the plan is clear "
"and work is underway.\n"
"\n"
"### Memory\n"
"```\n"
"drone @memory archive # Archive memories to vector store\n"
"drone @memory search <query> # Search archived memories\n"
"```\n"
"\n"
"### Git Workflow\n"
"```\n"
"drone @git pr 'description' # Create a pull request\n"
"drone @git status # Git status (branch-scoped)\n"
"drone @git sync # Sync with main\n"
"drone @git lock / unlock # Lock/unlock the repo\n"
"```\n"
"\n"
"### Infrastructure\n"
"```\n"
"drone systems # List all available infrastructure\n"
"drone --help # Full drone command reference\n"
"```\n"
"\n"
"## Patterns\n"
@@ -262,6 +339,8 @@ def _global_prompt_md(name: str) -> str:
"- **Standards** — run `drone @seedgo audit` to check compliance.\n"
"- **Identity** — agents have `.trinity/passport.json`. "
"Projects use the registry.\n"
"- **Memory** — update `.trinity/local.json` at session end. "
"Memory is presence.\n"
)
@@ -524,6 +603,7 @@ def update_project(target: Path) -> dict:
name = registry_path.stem.replace("_REGISTRY", "")
updated: list[str] = []
already_current: list[str] = []
skipped: list[str] = []
# Managed directories — create if missing (graceful recovery).
@@ -533,39 +613,47 @@ def update_project(target: Path) -> dict:
claude_dir = target / ".claude"
claude_dir.mkdir(exist_ok=True)
# --- Managed files: always overwrite with latest templates ---
# --- Managed files: write only when content has changed ---
global_prompt_path = aipass_dir / "aipass_global_prompt.md"
global_prompt_path.write_text(
_with_source(_global_prompt_md(name), global_prompt_path),
encoding="utf-8",
)
updated.append(str(global_prompt_path))
generated = _with_source(_global_prompt_md(name), global_prompt_path)
if not global_prompt_path.exists() or global_prompt_path.read_text(encoding="utf-8") != generated:
global_prompt_path.write_text(generated, encoding="utf-8")
updated.append(str(global_prompt_path))
else:
already_current.append(str(global_prompt_path))
settings_path = claude_dir / "settings.json"
settings_path.write_text(_claude_settings(), encoding="utf-8")
updated.append(str(settings_path))
generated = _claude_settings()
if not settings_path.exists() or settings_path.read_text(encoding="utf-8") != generated:
settings_path.write_text(generated, encoding="utf-8")
updated.append(str(settings_path))
else:
already_current.append(str(settings_path))
claude_md_path = target / "CLAUDE.md"
claude_md_path.write_text(
_with_source(_claude_md(name), claude_md_path),
encoding="utf-8",
)
updated.append(str(claude_md_path))
generated = _with_source(_claude_md(name), claude_md_path)
if not claude_md_path.exists() or claude_md_path.read_text(encoding="utf-8") != generated:
claude_md_path.write_text(generated, encoding="utf-8")
updated.append(str(claude_md_path))
else:
already_current.append(str(claude_md_path))
agents_md_path = target / "AGENTS.md"
agents_md_path.write_text(
_with_source(_agents_md(name), agents_md_path),
encoding="utf-8",
)
updated.append(str(agents_md_path))
generated = _with_source(_agents_md(name), agents_md_path)
if not agents_md_path.exists() or agents_md_path.read_text(encoding="utf-8") != generated:
agents_md_path.write_text(generated, encoding="utf-8")
updated.append(str(agents_md_path))
else:
already_current.append(str(agents_md_path))
gemini_md_path = target / "GEMINI.md"
gemini_md_path.write_text(
_with_source(_gemini_md(name), gemini_md_path),
encoding="utf-8",
)
updated.append(str(gemini_md_path))
generated = _with_source(_gemini_md(name), gemini_md_path)
if not gemini_md_path.exists() or gemini_md_path.read_text(encoding="utf-8") != generated:
gemini_md_path.write_text(generated, encoding="utf-8")
updated.append(str(gemini_md_path))
else:
already_current.append(str(gemini_md_path))
# --- User-owned files: always skip ---
for skip_name in (
@@ -580,5 +668,6 @@ def update_project(target: Path) -> dict:
"project_name": name,
"target": str(target),
"updated_files": updated,
"already_current": already_current,
"skipped_files": skipped,
}
src/aipass/cli/apps/modules/init_project.py
+34 -17
View File
@@ -327,7 +327,7 @@ def _handle_init_update(args: List[str]) -> bool:
from rich.table import Table
from rich import box
if not args or args[0] in ("--help", "-h", "help"):
if args and args[0] in ("--help", "-h", "help"):
console.print()
header("aipass init update — Refresh Scaffold Files")
console.print("[dim]Updates managed prompt and config files with the latest templates[/dim]")
@@ -348,8 +348,7 @@ def _handle_init_update(args: List[str]) -> bool:
return True
caller_cwd = os.environ.get("AIPASS_CALLER_CWD", os.getcwd())
target_arg = args[0]
target = Path(target_arg)
target = Path(args[0]) if args else Path(caller_cwd)
if not target.is_absolute():
target = Path(caller_cwd) / target
@@ -368,28 +367,42 @@ def _handle_init_update(args: List[str]) -> bool:
console.print()
header("Project Updated")
already_current = result.get("already_current", [])
summary = (
f"[bold]{result['project_name']}[/bold]\n"
f"\n"
f" [yellow]Target:[/yellow] [dim]{result['target']}[/dim]\n"
f" [yellow]Updated:[/yellow] {len(result['updated_files'])} files\n"
f" [yellow]Skipped:[/yellow] {len(result['skipped_files'])} files"
f" [yellow]Target:[/yellow] [dim]{result['target']}[/dim]\n"
f" [yellow]Updated:[/yellow] {len(result['updated_files'])} files\n"
f" [yellow]Already current:[/yellow] {len(already_current)} files\n"
f" [yellow]User-owned:[/yellow] {len(result['skipped_files'])} files (skipped)"
)
console.print(Panel(summary, border_style="green", box=box.ROUNDED))
# Updated files table
updated_table = Table(
show_header=True, header_style="bold cyan", border_style="dim", title="Updated"
)
updated_table.add_column("#", style="green", width=3)
updated_table.add_column("File", style="yellow")
for i, f in enumerate(result["updated_files"], 1):
updated_table.add_row(str(i), f)
console.print(updated_table)
# Updated files table (only show if something changed)
if result["updated_files"]:
updated_table = Table(
show_header=True, header_style="bold cyan", border_style="dim", title="Updated"
)
updated_table.add_column("#", style="green", width=3)
updated_table.add_column("File", style="yellow")
for i, f in enumerate(result["updated_files"], 1):
updated_table.add_row(str(i), f)
console.print(updated_table)
# Already current table
if already_current:
current_table = Table(
show_header=True, header_style="bold cyan", border_style="dim", title="Already current"
)
current_table.add_column("#", style="dim", width=3)
current_table.add_column("File", style="dim")
for i, f in enumerate(already_current, 1):
current_table.add_row(str(i), f)
console.print(current_table)
# Skipped files table
skipped_table = Table(
show_header=True, header_style="bold cyan", border_style="dim", title="Skipped"
show_header=True, header_style="bold cyan", border_style="dim", title="User-owned (skipped)"
)
skipped_table.add_column("#", style="dim", width=3)
skipped_table.add_column("File", style="dim")
@@ -398,12 +411,16 @@ def _handle_init_update(args: List[str]) -> bool:
console.print(skipped_table)
console.print()
success(f"Updated {len(result['updated_files'])} files")
if result["updated_files"]:
success(f"Updated {len(result['updated_files'])} files")
else:
success("All files already up to date")
json_handler.log_operation("aipass_init_update", {
"project_name": result["project_name"],
"target": result["target"],
"files_updated": len(result["updated_files"]),
"files_already_current": len(already_current),
"files_skipped": len(result["skipped_files"]),
})
src/aipass/cli/tests/test_bootstrap.py
+140 -1
View File
@@ -11,7 +11,7 @@ from pathlib import Path
import pytest
from aipass.cli.apps.handlers.init.bootstrap import _sanitize_name, init_project
from aipass.cli.apps.handlers.init.bootstrap import _sanitize_name, init_project, update_project
# ---------------------------------------------------------------------------
@@ -415,3 +415,142 @@ def test_init_project_agents_md_no_trinity(tmp_path):
content = (target / "AGENTS.md").read_text(encoding="utf-8")
assert ".trinity/" not in content
assert "KEEP_REGISTRY.json" in content
# ---------------------------------------------------------------------------
# update_project tests
# ---------------------------------------------------------------------------
def test_update_project_raises_if_no_registry(tmp_path):
"""ValueError when target has no *_REGISTRY.json (not an AIPass project)."""
target = tmp_path / "bare"
target.mkdir()
with pytest.raises(ValueError, match="No AIPass project found"):
update_project(target)
def test_update_project_return_dict_structure(tmp_path):
"""Return dict contains all required keys."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="upd")
result = update_project(target)
assert set(result.keys()) == {
"project_name",
"target",
"updated_files",
"already_current",
"skipped_files",
}
assert result["project_name"] == "UPD"
assert result["target"] == str(target.resolve())
assert isinstance(result["updated_files"], list)
assert isinstance(result["already_current"], list)
assert isinstance(result["skipped_files"], list)
def test_update_project_already_current_after_init(tmp_path):
"""Running update immediately after init reports all managed files as already current."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="fresh")
result = update_project(target)
assert len(result["updated_files"]) == 0
assert len(result["already_current"]) == 5
def test_update_project_idempotent(tmp_path):
"""Running update twice in a row produces no changes on second run."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="idem")
result1 = update_project(target)
result2 = update_project(target)
# Both runs should be identical
assert result1["updated_files"] == result2["updated_files"]
assert result1["already_current"] == result2["already_current"]
def test_update_project_updates_modified_managed_file(tmp_path):
"""A managed file with altered content is re-written on update."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="mod")
# Corrupt a managed file
claude_md = target / "CLAUDE.md"
claude_md.write_text("# Corrupted\n", encoding="utf-8")
result = update_project(target)
# CLAUDE.md must appear in updated, not already_current
assert str(claude_md.resolve()) in result["updated_files"]
assert str(claude_md.resolve()) not in result["already_current"]
# Content is restored
restored = claude_md.read_text(encoding="utf-8")
assert "MOD" in restored
assert "## What is AIPass" in restored
def test_update_project_never_touches_user_owned_files(tmp_path):
"""Registry, README, STATUS, .gitignore are always in skipped_files."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="skip")
# Modify user-owned files
(target / "README.md").write_text("# My custom README\n", encoding="utf-8")
(target / "STATUS.local.md").write_text("# Custom status\n", encoding="utf-8")
(target / ".gitignore").write_text("# custom\n", encoding="utf-8")
result = update_project(target)
skipped = result["skipped_files"]
assert any("REGISTRY" in s for s in skipped)
assert any("README.md" in s for s in skipped)
assert any("STATUS.local.md" in s for s in skipped)
assert any(".gitignore" in s for s in skipped)
# User customisations are preserved
assert (target / "README.md").read_text(encoding="utf-8") == "# My custom README\n"
assert (target / "STATUS.local.md").read_text(encoding="utf-8") == "# Custom status\n"
def test_update_project_creates_missing_managed_dirs(tmp_path):
"""update_project recreates .aipass/ and .claude/ if they were deleted."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="recover")
# Delete only the managed subdirectories (not root files like CLAUDE.md)
import shutil
shutil.rmtree(target / ".aipass")
shutil.rmtree(target / ".claude")
result = update_project(target)
assert (target / ".aipass" / "aipass_global_prompt.md").exists()
assert (target / ".claude" / "settings.json").exists()
# The 2 files inside deleted dirs are re-written; root files still match
assert len(result["updated_files"]) == 2
assert len(result["already_current"]) == 3
def test_update_project_skipped_files_count(tmp_path):
"""update_project always skips exactly 4 user-owned files."""
target = tmp_path / "proj"
target.mkdir()
init_project(target, project_name="count")
result = update_project(target)
assert len(result["skipped_files"]) == 4
src/aipass/devpulse/.aipass/aipass_local_prompt.md
+6 -2
View File
@@ -64,9 +64,13 @@ drone systems # All branches
drone, seedgo, prax, cli, ai_mail, api, flow, spawn, trigger, memory, devpulse (you — no apps/, coordinates via dispatch + agents)
## Your Project
## Your Projects
You have a personal project: **Compass** at `~/Projects/compass/`. It's a vector-based thinking engine for growing autonomous decision-making. The judgment library at `~/Projects/decisions.md` (27 entries) is the prototype data. Compass vectorizes these decisions into ChromaDB fragments so you can query past judgment patterns, feedback, and outcomes when facing new decisions. Own venv, own registry, own orchestration AI (eventually). Work on this when you have autonomy time. Compass is part of the larger Nexus vision. See decisions.md and key_learning nexus_is_the_answer.
Two personal projects, both part of the Nexus vision. Work on these during autonomy time.
**Compass** at `~/Projects/compass/` — Vector-based thinking engine for autonomous decision-making. 130 fragments (decisions + observations + learnings). Query before big choices. Stop building features, start using it (#033). Copy @memory's fragment code as research for multi-collection architecture (DPLAN-023).
**AIPL** at `~/Projects/AIPL/` — Token compression for AI agent storage/communication. ~45% savings proven. Phase 1 COMPLETE (style guide + 6 examples in docs/). DPLAN-0115. Polyglot agent builds Phase 2 (compression engine). Hand to Polyglot when ready.
## Working Habits
src/aipass/devpulse/apps/devpulse.py
+22 -7
View File
@@ -42,14 +42,29 @@ def discover_modules() -> list[Any]:
if file_path.name.startswith("_"):
continue
module_name = f"apps.modules.{file_path.stem}"
# Try package import first, fall back to relative import
module_names = [
f"aipass.devpulse.apps.modules.{file_path.stem}",
f"apps.modules.{file_path.stem}",
]
try:
module = importlib.import_module(module_name)
if hasattr(module, "handle_command"):
modules.append(module)
except Exception as e:
logger.error(f"[DEVPULSE] Failed to load module {module_name}: {e}")
loaded = False
for module_name in module_names:
try:
module = importlib.import_module(module_name)
if hasattr(module, "handle_command"):
modules.append(module)
loaded = True
break
except (ImportError, ModuleNotFoundError):
continue
except Exception as e:
logger.error(f"[DEVPULSE] Failed to load module {module_name}: {e}")
loaded = True
break
if not loaded:
logger.error(f"[DEVPULSE] Could not import module {file_path.stem}")
return modules
src/aipass/devpulse/pytest.ini
+1 -5
View File
@@ -8,11 +8,7 @@ python_functions = test_*
python_classes = Test*
# Command-line options (always applied)
addopts =
-v # Verbose output
--tb=short # Short traceback format
--strict-markers # Raise error on unknown markers
-ra # Show summary of all test outcomes
addopts = -v --tb=short --strict-markers -ra
# Test markers (for categorizing tests)
markers =
src/aipass/drone/.seedgo/bypass.json
+6
View File
@@ -14,6 +14,12 @@
],
"reason": "Raw passthrough of module subprocess stdout/stderr — console.print() breaks routed output"
},
{
"file": "apps/drone.py",
"standard": "encapsulation",
"lines": [29],
"reason": "drone.py entry point imports get_all_branches directly for systems display — single top-level consumer of branch listing, not a cross-branch import. Consistent with json_handler and file_handler exemptions."
},
{
"file": "apps/handlers/scanning/formatters.py",
"standard": "cli",
src/aipass/drone/apps/drone.py
+41 -8
View File
@@ -14,6 +14,7 @@ Standard branch entry point (apps/drone.py pattern).
"""
import importlib
import os
import sys
from pathlib import Path
from typing import List
@@ -25,6 +26,7 @@ from aipass.cli.apps.modules import console, err_console
from aipass.drone.apps.modules import BranchNotFoundError, CommandExecutionError, RegistryError
from aipass.drone.apps.modules.discovery import get_help
from aipass.drone.apps.modules.resolver import list_branches
from aipass.drone.apps.handlers.registry_handler import get_all_branches
from aipass.drone.apps.modules.router import route_command
from aipass.drone.apps.modules.module_registry import (
is_module,
@@ -153,29 +155,52 @@ def _handle_systems() -> int:
console.print("No registry found in current directory tree.")
return 0
branches = list_branches()
all_branches = get_all_branches()
modules = list_modules()
# Split registry branches into services (profile=library) and project branches
services = [b for b in all_branches if b.get("profile") == "library"]
project_branches = [b for b in all_branches if b.get("profile") != "library"]
# Infrastructure section — drone is the router, not a routable module
console.print("Infrastructure:")
console.print(f" @{'drone':<18} Command routing and module discovery (v{VERSION})")
console.print()
if modules:
console.print(f"Modules ({len(modules)}):")
# AIPass Services — internal modules + registry services (deduplicated)
# Exclude registry services that duplicate internal modules or drone itself
exclude = set(modules) | {"drone"}
deduped_services = [s for s in services if s.get("name", "").lower() not in exclude]
service_count = len(modules) + len(deduped_services)
if service_count:
console.print(f"AIPass Services ({service_count}):")
for name in modules:
info = get_module_info(name)
if info:
console.print(f" @{name:<18} {info.description}")
else:
console.print(f" @{name:<18} (not available)")
if branches:
for svc in sorted(deduped_services, key=lambda b: b.get("name", "").lower()):
name = svc.get("name", "").lower()
desc = svc.get("description", "")
console.print(f" @{name:<18} {desc}")
if project_branches:
console.print()
if branches:
console.print(f"Branches ({len(branches)}):")
for name in sorted(branches):
console.print(f" {name}")
if project_branches:
console.print(f"Branches ({len(project_branches)}):")
for branch in sorted(project_branches, key=lambda b: b.get("name", "").lower()):
console.print(f" @{branch.get('name', '').lower()}")
# Hint for external projects missing AIPass core branches.
# If AIPASS_HOME is not set AND 'drone' isn't among the local branches,
# we're in an external project that can't see the core branch set.
if not os.environ.get("AIPASS_HOME"):
branch_names = {b.get("name", "").lower() for b in all_branches}
if "drone" not in branch_names:
console.print()
console.print("[dim]Only local registry found. To access AIPass core branches:[/dim]")
console.print("[dim] export AIPASS_HOME=/path/to/AIPass[/dim]")
return 0
@@ -307,6 +332,8 @@ def _handle_custom_command(args: list[str]) -> int:
return _handle_module(module_name, [command] + cmd_args)
logger.warning("Custom command failed for target %s: %s", target, exc)
err_console.print(f"drone: {exc}")
if isinstance(exc, BranchNotFoundError) and not os.environ.get("AIPASS_HOME"):
err_console.print(" Tip: set AIPASS_HOME=/path/to/AIPass to access core branches.")
return 1
if result.stdout:
@@ -343,6 +370,8 @@ def _handle_target(args: List[str]) -> int:
return _handle_module(module_name, rest)
logger.warning("Introspection failed for %s: %s", target, exc)
err_console.print(f"drone: {exc}")
if isinstance(exc, BranchNotFoundError) and not os.environ.get("AIPASS_HOME"):
err_console.print(" Tip: set AIPASS_HOME=/path/to/AIPass to access core branches.")
return 1
if result.stdout:
console.print(result.stdout, end="", highlight=False)
@@ -364,6 +393,8 @@ def _handle_target(args: List[str]) -> int:
return _handle_module(module_name, rest)
logger.warning("Help lookup failed for %s: %s", target, exc)
err_console.print(f"drone: {exc}")
if isinstance(exc, BranchNotFoundError) and not os.environ.get("AIPASS_HOME"):
err_console.print(" Tip: set AIPASS_HOME=/path/to/AIPass to access core branches.")
return 1
return 0
@@ -386,6 +417,8 @@ def _handle_target(args: List[str]) -> int:
return _handle_module(module_name, rest)
logger.warning("Command routing failed for %s %s: %s", target, command, exc)
err_console.print(f"drone: {exc}")
if isinstance(exc, BranchNotFoundError) and not os.environ.get("AIPASS_HOME"):
err_console.print(" Tip: set AIPASS_HOME=/path/to/AIPass to access core branches.")
return 1
if result.stdout:
src/aipass/drone/apps/handlers/registry_handler.py
+95 -24
View File
@@ -51,9 +51,10 @@ def find_registry() -> Path:
Search order:
1. Explicitly set path via set_registry_path()
2. AIPASS_REGISTRY environment variable
3. Walk up from drone package location
4. Walk up from cwd
5. Default: package-relative path
3. Walk up from cwd
4. AIPASS_HOME env var — for external projects where CWD walk finds nothing
5. Walk up from drone package location
6. Default: package-relative path
The first directory that contains any *_REGISTRY.json is treated
as the project boundary. If that directory holds more than one
@@ -66,6 +67,13 @@ def find_registry() -> Path:
if hit is not None:
return hit
# AIPASS_HOME fallback — for external projects where CWD walk finds nothing
aipass_home = os.environ.get("AIPASS_HOME")
if aipass_home:
hit = _first_registry_in(Path(aipass_home))
if hit is not None:
return hit
# Walk up from this file (fallback for pip editable installs)
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
@@ -162,19 +170,18 @@ def reset_registry_path() -> None:
# Registry loading and querying
# ---------------------------------------------------------------------------
def load_registry() -> Dict[str, Any]:
"""Load the branch registry from disk.
def _load_registry_data(registry_path: Path) -> Dict[str, Any]:
"""Read, parse, and normalize a registry file.
Returns:
Registry dictionary with branches (normalized to dict format)
Performs file I/O and branch normalization (list → dict). Does NOT
run credential verification or log the operation — callers that need
those steps (i.e. load_registry) are responsible.
Raises:
RegistryNotFoundError: If registry file doesn't exist
RegistryCorruptError: If registry file is invalid JSON
RegistryCorruptError: If registry file is invalid JSON or malformed
RegistryPermissionError: If registry file cannot be read
"""
registry_path = get_registry_path()
if not registry_path.exists():
raise RegistryNotFoundError(
f"Registry not found at {registry_path}. "
@@ -200,7 +207,7 @@ def load_registry() -> Dict[str, Any]:
# Normalize: AIPASS_REGISTRY uses list format, convert to dict keyed by name
branches_raw = data["branches"]
if isinstance(branches_raw, list):
branches_dict = {}
branches_dict: Dict[str, Any] = {}
registry_dir = registry_path.parent
for branch in branches_raw:
name = branch.get("name", "").lower()
@@ -219,6 +226,31 @@ def load_registry() -> Dict[str, Any]:
elif not isinstance(branches_raw, dict):
raise RegistryCorruptError("Registry 'branches' must be a list or dict")
return data
def _get_aipass_home_registry_path() -> Optional[Path]:
"""Return the AIPass home registry path from AIPASS_HOME env var, or None."""
aipass_home = os.environ.get("AIPASS_HOME")
if not aipass_home:
return None
return _first_registry_in(Path(aipass_home))
def load_registry() -> Dict[str, Any]:
"""Load the branch registry from disk.
Returns:
Registry dictionary with branches (normalized to dict format)
Raises:
RegistryNotFoundError: If registry file doesn't exist
RegistryCorruptError: If registry file is invalid JSON
RegistryPermissionError: If registry file cannot be read
"""
registry_path = get_registry_path()
data = _load_registry_data(registry_path)
_verify_registry_credential(registry_path, data)
branch_count = len(data.get("branches", {}))
@@ -231,17 +263,36 @@ def get_all_branches(
branch_type: Optional[str] = None,
status: str = "active",
) -> List[Dict[str, Any]]:
"""Get all branches from the registry, optionally filtered."""
try:
registry = load_registry()
except RegistryNotFoundError:
logger.warning("get_all_branches: registry not found, returning empty list")
return []
"""Get all branches from the registry, optionally filtered.
branches = registry.get("branches", {}).values()
Merges branches from both the primary (local/project) registry and the
AIPass home registry (from AIPASS_HOME env var). Local branches take
precedence when names collide.
"""
merged: Dict[str, Any] = {}
# --- Primary registry ---
try:
primary = load_registry()
for name, branch in primary.get("branches", {}).items():
merged[name] = branch
except (RegistryNotFoundError, RegistryCorruptError, RegistryPermissionError) as exc:
logger.warning("get_all_branches: primary registry unavailable: %s", exc)
# --- AIPass home registry (if different from primary) ---
home_path = _get_aipass_home_registry_path()
primary_path = get_registry_path()
if home_path is not None and home_path != primary_path:
try:
home_data = _load_registry_data(home_path)
for name, branch in home_data.get("branches", {}).items():
if name not in merged:
merged[name] = branch
except (RegistryNotFoundError, RegistryCorruptError, RegistryPermissionError) as exc:
logger.warning("get_all_branches: AIPass home registry unavailable: %s", exc)
filtered = []
for branch in branches:
for branch in merged.values():
if status and branch.get("status") != status:
continue
if branch_type and branch.get("type") != branch_type:
@@ -252,11 +303,31 @@ def get_all_branches(
def get_branch_by_name(name: str) -> Optional[Dict[str, Any]]:
"""Get a single branch by name (case-insensitive)."""
"""Get a single branch by name (case-insensitive).
Checks the primary (local/project) registry first. If not found, falls
back to the AIPass home registry (AIPASS_HOME env var) when it points to
a different location.
"""
lower_name = name.lower()
# --- Primary registry ---
try:
registry = load_registry()
except RegistryNotFoundError:
logger.warning("get_branch_by_name: registry not found for lookup of '%s'", name)
return None
branch = registry.get("branches", {}).get(lower_name)
if branch is not None:
return branch
except (RegistryNotFoundError, RegistryCorruptError, RegistryPermissionError) as exc:
logger.warning("get_branch_by_name: primary registry unavailable for '%s': %s", name, exc)
return registry.get("branches", {}).get(name.lower())
# --- AIPass home registry fallback ---
home_path = _get_aipass_home_registry_path()
primary_path = get_registry_path()
if home_path is not None and home_path != primary_path:
try:
home_data = _load_registry_data(home_path)
return home_data.get("branches", {}).get(lower_name)
except (RegistryNotFoundError, RegistryCorruptError, RegistryPermissionError) as exc:
logger.warning("get_branch_by_name: AIPass home registry unavailable for '%s': %s", name, exc)
return None
src/aipass/drone/apps/modules/resolver.py
+4 -2
View File
@@ -135,6 +135,9 @@ def normalize_branch_arg(target: str) -> str:
def resolve_branch(symbolic_name: str) -> str:
"""Resolve a symbolic branch name to its absolute path.
Checks primary (local) registry first, then falls back to AIPASS_HOME
registry for cross-project resolution.
Args:
symbolic_name: Branch name with @ prefix (e.g. "@seedgo")
@@ -149,10 +152,9 @@ def resolve_branch(symbolic_name: str) -> str:
raise BranchNotFoundError(
f"Branch name must use @ prefix: '@{symbolic_name}' (got '{symbolic_name}')"
)
registry = load_registry()
name = normalize_branch_name(symbolic_name).lower()
branch = registry.get("branches", {}).get(name)
branch = get_branch_by_name(name)
if branch is None:
raise BranchNotFoundError(
src/aipass/prax/apps/handlers/monitoring/branch_detector.py
+95 -1
View File
@@ -109,12 +109,106 @@ class BranchDetector:
self.known_branches.update(fallback)
logger.info(f"Using fallback branches: {fallback}")
def _detect_external_project(self, decoded_path: str) -> Optional[str]:
"""Detect branch from an external AIPass project directory.
Looks for {PROJECT}_REGISTRY.json in the decoded path.
Args:
decoded_path: Actual filesystem path decoded from Claude project folder
Returns:
Project name in uppercase if registry found (e.g., 'VERA-STUDIO', 'AIPL')
"""
try:
project_dir = Path(decoded_path)
if not project_dir.exists():
return None
# Walk the directory tree looking for *_REGISTRY.json
for item in project_dir.iterdir():
if item.is_file() and item.name.endswith('_REGISTRY.json'):
# Extract project name from filename
# VERA-STUDIO_REGISTRY.json -> VERA-STUDIO
project_name = item.stem.replace('_REGISTRY', '')
return project_name
# If no registry found, use directory name as fallback
# /home/patrick/Projects/Vera-Studio -> VERA-STUDIO
return project_dir.name.upper().replace('-', '-')
except Exception:
return None
def _decode_claude_project_path(self, encoded_folder: str) -> Optional[str]:
"""Decode Claude Code project folder name to actual filesystem path.
Claude encodes paths by replacing "/" with "-" and prepending "-".
But directory names with hyphens make simple decoding ambiguous.
Strategy: Try all possible decodings and return the project root
(closest parent with _REGISTRY.json).
Args:
encoded_folder: Folder name like "-home-patrick-Projects-Vera-Studio"
Returns:
Project root directory path or None if no valid external project found
"""
if not encoded_folder.startswith('-'):
return None
# Remove leading dash
name = encoded_folder[1:]
segments = name.split('-')
# Try progressively joining segments from right to left
# This handles "Vera-Studio" which shouldn't be split
for i in range(len(segments)):
# Try path with i segments from left kept separate, rest joined
if i == 0:
path_parts = segments
else:
path_parts = segments[:i] + ['-'.join(segments[i:])]
project_path = '/' + '/'.join(path_parts)
project_dir = Path(project_path)
# Check if this path or any parent has a registry
if project_dir.exists():
# Walk up from this path looking for _REGISTRY.json
current = project_dir
while current != current.parent:
try:
for item in current.iterdir():
if item.is_file() and item.name.endswith('_REGISTRY.json'):
return str(current)
except (OSError, PermissionError):
pass
current = current.parent
# Stop at /home or root to avoid traversing too far
if current.name == 'home' or str(current) == '/':
break
return None
def _detect_from_claude_project(self, path_str: str) -> Optional[str]:
"""Detect branch from Claude Code project path encoding."""
projects_idx = path_str.index('.claude/projects/') + len('.claude/projects/')
remaining = path_str[projects_idx:]
project_folder = remaining.split('/')[0]
project_path = '/' + project_folder.replace('-', '/')
# Try external project detection first (covers Vera-Studio, AIPL, etc.)
decoded_path = self._decode_claude_project_path(project_folder)
if decoded_path:
external_result = self._detect_external_project(decoded_path)
if external_result:
return external_result
# Fallback: simple decoding for internal AIPass projects
if project_folder.startswith('-'):
project_path = '/' + project_folder[1:].replace('-', '/')
else:
project_path = '/' + project_folder.replace('-', '/')
for registered_path, branch_name in self.branch_map.items():
registered_normalized = registered_path.replace('_', '/')