🔧 chore(.claude): add initial claude config, hooks, and settings

- add .claude/.gitignore to exclude local claude files and caches
- add .claude/commands/memo.md with memory update instructions
- add notification, stop, tool_use, and branch_prompt_loader hooks scripts
- add .claude/settings.json configuring permissions and hooks integration
- add .gitattributes for text normalization and binary file handling

✨ feat(src/aipass): introduce multi-agent orchestration framework structure

- add core aipass package with version 2.0.0 and repo link
- add submodules for api, assistant, devpulse, drone, flow, mail, memory, prax, seedgo, skills, templates, trigger
- provide base module docstrings describing purpose and architecture
- add branch template for cortex branch with scaffolded files and verification tool
- remove legacy cortex module and seedgo standalone package to consolidate into aipass

✨ feat(src/aipass/drone): simplify drone module interface and routing system

- refactor drone module to expose main routing and resolution functionality
- remove CLI, registry, executor, resolver, router, modules, discovery, exceptions, and handlers files
- replace with minimal drone package exposing core routing and resolution API

✨ feat(src/aipass/templates/cortex/branch_template): add default branch template files

- add .backup_ignore.json to exclude large and unnecessary files from backups
- add .migrations.json for template structural migrations
- add .registry_ignore.json to exclude internal template files during updates
- add .template_registry.json tracking template files and directories with placeholders
- add default trinity memory files: local.json, observations.json, passport.json
- add DASHBOARD.local.json with autogenerated branch status
- add README.md describing branch purpose, architecture, commands, and integrations
- add apps/ package and branch.py entry point with auto-discovery routing
- add empty handlers/, modules/, plugins/ packages
- add artifacts/ with a birth certificate JSON
- add docs/, flow.local.md, logs/, pytest.ini, tests/ with conftest.py
- add tools/verify_branch.py for verifying branch consistency against template

🔧 chore(src/aipass/): remove deprecated packages and modules

- delete legacy cortex, paths, seedgo packages and their modules/plugins
- remove seedgo CLI and plugins from repo to focus on aipass core

✨ feat(src/aipass/templates/flow): add placeholder for planning and workflow management

✨ feat(src/aipass/templates/memory): add placeholder for memory scaffolding

✨ feat(src/aipass/templates/system): add placeholder for system files

✨ feat(src/aipass/trigger): add error detection and auto-dispatch module

- scaffold trigger package with handlers, modules, plugins for error matching and escalation

✨ feat(src/aipass/mail): add inter-branch messaging module with handlers, modules, plugins

✨ feat(src/aipass/memory): add unified memory system with trinity, agents, and bank submodules

✨ feat(src/aipass/prax): add monitoring and observability framework with handlers, modules, plugins

✨ feat(src/aipass/seedgo): add code standards framework skeleton

✨ feat(src/aipass/skills): add user-built add-ons infrastructure module

✨ feat(src/aipass/templates): add central template storage for plans, memory, and configs
This commit is contained in:
AIOSAI
2026-03-05 02:21:30 -08:00
parent b1fe460adf
commit d43ff41978
126 changed files with 1950 additions and 5485 deletions
+19
View File
@@ -0,0 +1,19 @@
file-history
debug
downloads
ide
paste-cache
projects
session-env
shell-snapshots
statsig
tasks
todos
user_data
.credentials.json
history.jsonl
hooks/.last_diagnostics_file
hooks/.pre_compact_log
+22
View File
@@ -0,0 +1,22 @@
# Memory Update
Purpose: Update branch memory files after completing work this session.
## Execution
1. Review what was done this session (context, recent changes, key decisions)
2. Update each file below as needed
3. Confirm completion — list files updated
## What to Update
### Always
- **.trinity/local.json** — Update `today_focus` with current work items. Move completed items to `recently_completed`. Add new session entry to `sessions` if significant work was done.
- **.trinity/observations.json** — Add notable collaboration insights: breakthrough moments, pattern corrections, flow states, friction points, preference discoveries. Skip if nothing notable this session.
### If Relevant
- **.trinity/passport.json** — Evolve identity when the branch's role, capabilities, or principles have genuinely changed. Don't update just to update — but don't leave placeholders forever either.
- **README.md** — Does it reflect current state? Update if stale.
- **flow.local.md** — Drop quick notes on issues, todos, or ideas discovered during the session.
+52
View File
@@ -0,0 +1,52 @@
#!/usr/bin/env python3
"""
Branch Prompt Loader — AIPass Public Repo
Injects branch-specific prompts based on CWD. When working in a branch
directory, loads .aipass/branch_system_prompt.md and outputs it so the
AI sees branch-specific context.
Version: 1.0.0
"""
from pathlib import Path
def find_branch_root() -> Path | None:
"""
Find the branch root directory.
Looks for .trinity/ or .aipass/ as branch indicators.
Stops at the repo root (has pyproject.toml or .git).
"""
cwd = Path.cwd()
search_path = cwd
while search_path.parent != search_path:
# Branch indicators: has .trinity/ (memory files) or apps/ (code)
has_trinity = (search_path / ".trinity").is_dir()
has_apps = (search_path / "apps").is_dir()
if has_trinity or has_apps:
return search_path
# Stop at repo root
if (search_path / "pyproject.toml").exists() or (search_path / ".git").is_dir():
return None
search_path = search_path.parent
return None
def main():
branch_root = find_branch_root()
if branch_root:
prompt_file = branch_root / ".aipass" / "branch_system_prompt.md"
if prompt_file.exists():
content = prompt_file.read_text().strip()
branch_name = branch_root.name.upper()
print(f"\n# Branch Context: {branch_name}\n<!-- Source: {prompt_file} -->\n{content}")
if __name__ == "__main__":
main()
+37
View File
@@ -0,0 +1,37 @@
#!/usr/bin/env python3
"""Notification Hook — Plays sound when AI needs permission."""
import json
import sys
import subprocess
from pathlib import Path
SOUNDS_DIR = Path(__file__).parent.parent / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-clear-announce-tones-2861.wav"
def play_sound() -> None:
if not SOUND_FILE.exists():
return
try:
subprocess.Popen(
["aplay", "-q", str(SOUND_FILE)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
pass
def main():
try:
hook_data = json.loads(sys.stdin.read())
if hook_data.get("hook_event_name") == "Notification":
play_sound()
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
+38
View File
@@ -0,0 +1,38 @@
#!/usr/bin/env python3
"""Stop Hook — Plays achievement bell when AI finishes responding."""
import json
import sys
import subprocess
from pathlib import Path
SOUNDS_DIR = Path(__file__).parent.parent / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-achievement-bell-600.wav"
def play_sound() -> None:
if not SOUND_FILE.exists():
return
try:
subprocess.Popen(
["aplay", "-q", str(SOUND_FILE)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
pass
def main():
try:
hook_data = json.loads(sys.stdin.read())
if hook_data.get("hook_event_name") == "Stop":
if not hook_data.get("stop_hook_active", False):
play_sound()
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""Tool Use Hook — Plays key press sound when AI uses tools."""
import json
import sys
import subprocess
from pathlib import Path
SOUNDS_DIR = Path(__file__).parent.parent / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-atm-cash-machine-key-press-2841.wav"
SOUND_TOOLS = ["Bash", "Edit", "MultiEdit", "Write", "Read", "Grep", "Glob"]
def play_sound() -> None:
if not SOUND_FILE.exists():
return
try:
subprocess.Popen(
["aplay", "-q", str(SOUND_FILE)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
pass
def main():
try:
hook_data = json.loads(sys.stdin.read())
if hook_data.get("hook_event_name") == "PreToolUse":
if hook_data.get("tool_name", "") in SOUND_TOOLS:
play_sound()
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
+65
View File
@@ -0,0 +1,65 @@
{
"permissions": {
"allow": [],
"deny": [
"Bash(git reset:*)",
"Bash(git rebase:*)",
"Bash(git config:*)",
"Bash(git push --force:*)",
"EnterPlanMode"
],
"defaultMode": "acceptEdits"
},
"hooks": {
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "cat .aipass/aipass_global_prompt.md 2>/dev/null || true"
}
]
},
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/branch_prompt_loader.py"
}
]
}
],
"PreToolUse": [
{
"matcher": "Bash|Edit|MultiEdit|Write|Read|Grep|Glob|WebSearch|WebFetch|Task",
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/tool_use_sound.py"
}
]
}
],
"Stop": [
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/stop_sound.py"
}
]
}
],
"Notification": [
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/notification_sound.py"
}
]
}
]
},
"skipDangerousModePermissionPrompt": true
}
+54
View File
@@ -0,0 +1,54 @@
# Git Attributes for AIPass
# Line ending normalization + binary file handling
# Auto detect text files and perform LF normalization
* text=auto
# Source code - always use LF
*.py text eol=lf
*.json text eol=lf
*.yml text eol=lf
*.yaml text eol=lf
*.toml text eol=lf
# Documentation - always use LF
*.md text eol=lf
*.txt text eol=lf
LICENSE text eol=lf
# Shell scripts - must use LF (won't work with CRLF)
*.sh text eol=lf
# Configuration files
.gitattributes text eol=lf
.gitignore text eol=lf
.env text eol=lf
# Audio files - binary (don't merge or diff)
*.wav binary
*.mp3 binary
*.ogg binary
# Binary files
*.png binary
*.jpg binary
*.jpeg binary
*.gif binary
*.ico binary
*.pdf binary
*.zip binary
*.gz binary
# Python compiled - binary
*.pyc binary
*.pyo binary
*.whl binary
# Database files
*.db binary
*.sqlite binary
*.sqlite3 binary
# Better diffs
*.py diff=python
*.md diff=markdown
+7
View File
@@ -23,3 +23,10 @@ site/
.idea/
.vscode/
*.swp
# AIPass global config (secrets live here)
.aipass/
!.aipass/aipass_global_prompt.md
.archive
+4 -13
View File
@@ -1,16 +1,7 @@
"""AIPass — Orchestration framework for autonomous AI agent ecosystems.
"""AIPass — Multi-agent orchestration framework.
Modules:
aipass.drone — Command routing & discovery (v1.0.0 — BUILT)
aipass.paths — Root resolution, AIPASS_ROOT (PLACEHOLDER)
aipass.prax — Visibility & monitoring layer (PLACEHOLDER)
aipass.cortex — Branch creation & templates (PLACEHOLDER)
aipass.mail — Inter-branch messaging (PLACEHOLDER)
Standalone packages:
seedgo — Code standards framework (v1.0.0 — BUILT)
Architecture reference: vera/projects/framework/architecture_reference.md
pip install aipass
https://github.com/AIOSAI/AIPass
"""
__version__ = "1.0.0"
__version__ = "2.0.0"
+5
View File
@@ -0,0 +1,5 @@
"""aipass.api — LLM access, model routing, and agent spawning.
Unified interface to language models. Handles API keys, model selection,
rate limiting, and spawning agent processes.
"""
+1
View File
@@ -0,0 +1 @@
"""api.handlers — API client implementation and rate limiting."""
+1
View File
@@ -0,0 +1 @@
"""api.modules — Model routing and client logic."""
+1
View File
@@ -0,0 +1 @@
"""api.plugins — Extensible plugin interface."""
+5
View File
@@ -0,0 +1,5 @@
"""aipass.assistant — Scheduling, plugins, and external integrations.
Plugin architecture connecting AIPass to the outside world:
scheduling engine + discoverable plugin interface.
"""
@@ -0,0 +1 @@
"""assistant.handlers — Scheduling and integration implementation."""
+1
View File
@@ -0,0 +1 @@
"""assistant.modules — Scheduling engine and plugin orchestration."""
@@ -0,0 +1 @@
"""Scheduling engine — manages timed actions, intervals, and due checks."""
+5
View File
@@ -0,0 +1,5 @@
"""assistant.plugins — External integration plugins (auto-discovered).
Each plugin connects AIPass to an external service.
User-swappable, independently installable.
"""
+1
View File
@@ -0,0 +1 @@
"""Calendar plugin — Google Calendar integration."""
+1
View File
@@ -0,0 +1 @@
"""Gmail plugin — email integration for external notifications."""
+1
View File
@@ -0,0 +1 @@
"""Telegram plugin — bridge between AIPass and Telegram bots."""
-22
View File
@@ -1,22 +0,0 @@
"""AIPass Cortex — Branch Creation & Template Engine.
Creates new branches (agent workspaces) following a 7-step flow:
1. Validate template — verify template directory exists
2. Extract metadata — branch name, profile, git repo from path
3. Build placeholders — replacement dict: {{BRANCHNAME}}, {{CWD}}, {{DATE}}, etc.
4. Smart file rename — rename existing memory files to standard convention
5. Copy template — copy entire template, replace placeholders in text files
6. File renaming — LOCAL.json -> BRANCHNAME.local.json, etc.
7. Registry entry — add to BRANCH_REGISTRY.json + fire trigger event
Three template types:
- branch_template/ — standard dev branch (apps/, modules/, handlers/, tests/)
- business_branch_template/ — business branch (playbooks/, strategy/, research/)
- team_template/ — think-tank manager (research/, ideas/, decisions/)
Accessed through drone: `drone @cortex create`, NOT standalone.
Architecture reference: vera/projects/framework/architecture_reference.md §5
Status: PLACEHOLDER — not yet implemented.
"""
+5
View File
@@ -0,0 +1,5 @@
"""aipass.devpulse — User orchestrator and dashboard.
Home base for users: system status, project config, dashboard.
The entry point for understanding what's happening across the ecosystem.
"""
+1
View File
@@ -0,0 +1 @@
"""devpulse.handlers — Status aggregation and config management."""
+1
View File
@@ -0,0 +1 @@
"""devpulse.modules — Dashboard and status logic."""
+1
View File
@@ -0,0 +1 @@
"""devpulse.plugins — Extensible plugin interface."""
+4 -96
View File
@@ -1,97 +1,5 @@
"""aipass.drone — Command routing, @resolution, module discovery.
The backbone of AIPass. Routes `drone @module command` to the correct
module, resolves @ targets to paths, discovers installed modules.
"""
AIPass Drone — Command routing & discovery module.
Provides symbolic addressing for multi-agent systems. Resolves @branch names
to absolute paths at runtime.
Core API:
resolve_branch(name) -> str # @branch → absolute path
register_branch(name, path, type) # Add branch to registry
list_branches(type, status) # List all branches
branch_exists(name) -> bool # Check if branch exists
get_branch_info(name) -> dict # Get branch metadata
Command Routing (Phase 2):
route_command(target, command, args, timeout) -> CommandResult
discover_modules(target) -> list # Available commands for a branch
get_help(target, command) -> HelpResult # Structured help for a branch
@all Operations (Phase 3):
route_all(command, args, timeout) -> dict[str, CommandResult]
get_system_help() -> dict[str, HelpResult] # Help across all branches
Registry Management:
initialize_registry() # Create empty registry
set_registry_path(path) # Set custom registry location
get_registry_path() -> Path # Get current registry path
Example:
>>> from aipass.drone import resolve_branch, register_branch
>>> register_branch("my_agent", "/path/to/agent", "agent")
>>> path = resolve_branch("@my_agent")
>>> print(path)
/path/to/agent
>>> from aipass.drone import route_command
>>> result = route_command("@my_agent", "status")
>>> print(result.stdout)
>>> from aipass.drone import route_all
>>> results = route_all("status")
>>> for branch, r in results.items():
... print(f"{branch}: exit={r.exit_code}")
"""
from .config import get_registry_path, reset_registry_path, set_registry_path
from .discovery import HelpResult, discover_modules, get_help, get_system_help
from .exceptions import (
BranchAlreadyExistsError,
BranchNotFoundError,
CommandExecutionError,
InvalidPathError,
RegistryCorruptError,
RegistryError,
RegistryNotFoundError,
RegistryPermissionError,
RoutingError,
)
from .executor import CommandResult
from .registry import add_branch as register_branch
from .registry import initialize_registry
from .resolver import branch_exists, get_branch_info, list_branches, resolve_branch
from .router import route_all, route_command
__version__ = "1.0.0"
__all__ = [
# Core API
"resolve_branch",
"register_branch",
"list_branches",
"branch_exists",
"get_branch_info",
# Command routing (Phase 2)
"route_command",
"discover_modules",
"get_help",
"CommandResult",
# Help & discovery (Phase 3)
"HelpResult",
"get_system_help",
"route_all",
# Registry management
"initialize_registry",
"set_registry_path",
"get_registry_path",
"reset_registry_path",
# Exceptions
"RoutingError",
"BranchNotFoundError",
"BranchAlreadyExistsError",
"InvalidPathError",
"RegistryError",
"RegistryNotFoundError",
"RegistryCorruptError",
"RegistryPermissionError",
"CommandExecutionError",
]
-211
View File
@@ -1,211 +0,0 @@
"""
Drone CLI — command-line interface for aipass.drone.
Entry point: `drone` (wired via pyproject.toml console_scripts to cli.main).
Usage:
drone Show available commands
drone --help Show help
drone --version Show version
drone systems List registered branches and modules
drone @branch command [args] Route command to branch
drone @module command [args] Route command to internal module
drone @branch --help Show help for branch or module
Zero external dependencies — stdlib only (sys).
"""
from __future__ import annotations
import sys
import aipass
from aipass.drone import (
CommandExecutionError,
BranchNotFoundError,
get_help,
list_branches,
route_command,
)
from aipass.drone.modules import (
is_module,
list_modules,
get_module_info,
get_module_introspective,
route_module_command,
get_module_help,
)
_HELP_TEXT = """\
drone — AIPass command router
Usage:
drone Show this help
drone --help Show this help
drone --version Show version
drone systems List registered branches and modules
drone @target command [args] Route command to branch or module
drone @target --help Show help for branch or module
Examples:
drone systems
drone @seedgo audit src/
drone @seedgo check src/myfile.py
drone @flow status
drone @flow --help
"""
def main() -> None:
"""Entry point for the `drone` CLI command.
Parses sys.argv manually and dispatches to the appropriate handler.
Exits with the command's exit code on branch commands, 0 on success,
1 on error.
"""
args = sys.argv[1:]
# No args or explicit --help
if not args or args == ["--help"]:
print(_HELP_TEXT, end="")
sys.exit(0)
# --version
if args == ["--version"]:
print(aipass.__version__)
sys.exit(0)
# systems — list registered branches
if args[0] == "systems":
_cmd_systems()
return
# @branch ... — route to a branch
if args[0].startswith("@"):
_cmd_branch(args)
return
# Unknown command
print(f"drone: unknown command '{args[0]}'", file=sys.stderr)
print("Run 'drone --help' for usage.", file=sys.stderr)
sys.exit(1)
def _cmd_systems() -> None:
"""Handle `drone systems` — list registered branches and modules."""
branches = list_branches()
modules = list_modules()
if not branches and not modules:
print("No branches or modules registered.")
sys.exit(0)
if modules:
print(f"Modules ({len(modules)}):")
for name in modules:
info = get_module_info(name)
if info:
print(f" @{name} {info.description}")
else:
print(f" @{name} (not available)")
if branches:
print()
if branches:
print(f"Branches ({len(branches)}):")
for name in sorted(branches):
print(f" {name}")
sys.exit(0)
def _cmd_branch(args: list[str]) -> None:
"""Handle `drone @target command [args]` or `drone @target --help`.
Checks internal modules first (e.g. @seedgo), then falls back
to branch registry for external branches.
"""
target = args[0]
rest = args[1:]
module_name = target.lstrip("@")
# Check if this is a registered internal module
if is_module(module_name):
_cmd_module(module_name, rest)
return
# Fall through to branch routing
# drone @branch --help
if not rest or rest == ["--help"]:
try:
result = get_help(target)
if result.text:
print(result.text, end="")
else:
print(f"No help available for {target}.")
except BranchNotFoundError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
except CommandExecutionError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
sys.exit(0)
# drone @branch command [args...]
command = rest[0]
cmd_args = rest[1:]
try:
result = route_command(target, command, args=cmd_args if cmd_args else None)
except BranchNotFoundError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
except CommandExecutionError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
if result.stdout:
print(result.stdout, end="")
if result.stderr:
print(result.stderr, end="", file=sys.stderr)
sys.exit(result.exit_code)
def _cmd_module(name: str, args: list[str]) -> None:
"""Handle routing to an internal module (e.g. @seedgo).
No args → introspective (discovery: what's connected).
--help → help text (usage documentation).
"""
# drone @module (no args) → introspective / discovery
if not args:
intro_text = get_module_introspective(name)
if intro_text:
print(intro_text, end="")
else:
print(f"No information available for @{name}.")
sys.exit(0)
# drone @module --help → usage documentation
if args == ["--help"]:
help_text = get_module_help(name)
if help_text:
print(help_text, end="")
else:
print(f"No help available for @{name}.")
sys.exit(0)
command = args[0]
cmd_args = args[1:] if len(args) > 1 else None
try:
result = route_module_command(name, command, cmd_args)
except (ImportError, AttributeError) as exc:
print(f"drone: module @{name} is registered but not available: {exc}", file=sys.stderr)
sys.exit(1)
if result.get("stdout"):
print(result["stdout"], end="")
if result.get("stderr"):
print(result["stderr"], end="", file=sys.stderr)
sys.exit(result.get("exit_code", 0))
-55
View File
@@ -1,55 +0,0 @@
"""
Registry configuration management.
Handles registry path configuration with environment variable support.
"""
import os
from pathlib import Path
from typing import Optional
_registry_path: Optional[Path] = None
def get_registry_path() -> Path:
"""
Get the current registry path.
Priority:
1. Explicitly set path via set_registry_path()
2. AIPASS_REGISTRY_PATH environment variable
3. Default: ~/.aipass/BRANCH_REGISTRY.json
Returns:
Path to the registry file
"""
global _registry_path
if _registry_path is not None:
return _registry_path
env_path = os.environ.get("AIPASS_REGISTRY_PATH")
if env_path:
return Path(env_path)
return Path.home() / ".aipass" / "BRANCH_REGISTRY.json"
def set_registry_path(path: str | Path) -> None:
"""
Set a custom registry path.
Args:
path: Path to the registry file
"""
global _registry_path
_registry_path = Path(path)
def reset_registry_path() -> None:
"""
Reset registry path to default (useful for testing).
"""
global _registry_path
_registry_path = None
-278
View File
@@ -1,278 +0,0 @@
"""
Module and command discovery for AIPass branch introspection.
Introspects branch capabilities by querying entry points for help text
and scanning module directories as a fallback.
"""
import logging
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional
from .exceptions import CommandExecutionError
from .resolver import list_branches, resolve_branch
logger = logging.getLogger(__name__)
@dataclass
class HelpResult:
"""
Structured result from a help query.
Attributes:
branch: The branch name the help was retrieved from.
command: The specific command queried, or None for top-level help.
text: Raw help text (stdout or stderr from the entry point).
commands_found: List of command names parsed from the help text.
"""
branch: str
command: Optional[str]
text: str
commands_found: List[str] = field(default_factory=list)
def _get_entry_point(branch_path: str, branch_name: str) -> Optional[Path]:
"""
Return the apps/{branch_name}.py entry point path if it exists, else None.
Args:
branch_path: Absolute path to the branch directory.
branch_name: Lowercase branch name.
Returns:
Path to the entry point, or None if it does not exist.
"""
entry_point = Path(branch_path) / "apps" / f"{branch_name}.py"
return entry_point if entry_point.exists() else None
def _scan_modules_directory(branch_path: str) -> List[str]:
"""
Scan apps/modules/ for .py files and return their stems as command names.
Used as a fallback when the entry point's --help output cannot be parsed.
Args:
branch_path: Absolute path to the branch directory.
Returns:
Sorted list of module stem names (without .py extension), excluding
__init__ and __main__.
"""
modules_dir = Path(branch_path) / "apps" / "modules"
if not modules_dir.is_dir():
return []
excluded = {"__init__", "__main__"}
return sorted(
f.stem
for f in modules_dir.glob("*.py")
if f.stem not in excluded
)
def _parse_help_for_commands(help_text: str) -> List[str]:
"""
Parse --help output to extract a list of available commands.
Handles multiple CLI framework formats:
- Section-based (custom CLIs, argparse subparsers): looks for headers
labelled "commands", "subcommands", or "available commands".
- Click-style: looks for indented entries under "Commands:" or "Options:".
- Positional args listing: lines matching ``{command}`` or ``[command]``
at the start of usage lines.
Falls back to returning an empty list if no recognisable pattern is found.
Args:
help_text: Raw stdout from ``python3 entry.py --help``.
Returns:
List of command names parsed from the help text.
"""
commands: List[str] = []
in_commands_section = False
section_markers = {"commands", "subcommands", "available commands"}
for line in help_text.splitlines():
stripped = line.strip()
# Detect section header lines (e.g. "Commands:", "Available commands:")
if any(marker in stripped.lower() for marker in section_markers):
in_commands_section = True
continue
# A blank line ends the current section.
if in_commands_section and not stripped:
in_commands_section = False
continue
if in_commands_section:
# Indented lines that start with a word are command entries.
if line.startswith((" ", "\t")) and stripped:
# Take the first token as the command name.
token = stripped.split()[0]
# Skip lines that look like option flags.
if not token.startswith("-"):
commands.append(token)
return commands
def discover_modules(target: str) -> List[str]:
"""
Discover available commands for a branch.
Resolution order:
1. Resolve @target to its absolute path via the registry.
2. Attempt to run ``python3 apps/{name}.py --help`` and parse the output
for a commands section.
3. If parsing yields nothing, fall back to scanning ``apps/modules/`` for
.py file stems.
Args:
target: Symbolic branch name with or without @ prefix (e.g., "@flow").
Returns:
List of discovered command/module names (may be empty).
Raises:
BranchNotFoundError: If the target branch is not in the registry.
"""
# Step 1: resolve — BranchNotFoundError propagates to caller.
branch_path = resolve_branch(target)
branch_name = target.lstrip("@").lower()
# Step 2: try --help on the entry point.
entry_point = _get_entry_point(branch_path, branch_name)
if entry_point is not None:
try:
result = subprocess.run(
["python3", str(entry_point.relative_to(branch_path)), "--help"],
cwd=branch_path,
capture_output=True,
timeout=10,
shell=False,
)
help_text = result.stdout.decode("utf-8", errors="replace")
if not help_text:
help_text = result.stderr.decode("utf-8", errors="replace")
commands = _parse_help_for_commands(help_text)
if commands:
return commands
except (subprocess.TimeoutExpired, OSError):
pass
# Step 3: fallback — scan modules directory.
return _scan_modules_directory(branch_path)
def get_help(target: str, command: Optional[str] = None) -> HelpResult:
"""
Get structured help for a branch or a specific command.
Executes:
python3 apps/{name}.py --help (when command is None)
python3 apps/{name}.py {command} --help (when command is given)
Args:
target: Symbolic branch name with or without @ prefix (e.g., "@flow").
command: Specific command to get help for (default: None).
Returns:
HelpResult with branch, command, text, and commands_found populated.
Raises:
BranchNotFoundError: If the target branch is not in the registry.
CommandExecutionError: If the entry point does not exist or execution
fails.
"""
branch_path = resolve_branch(target)
branch_name = target.lstrip("@").lower()
entry_point = _get_entry_point(branch_path, branch_name)
if entry_point is None:
raise CommandExecutionError(
f"Entry point not found for branch '{branch_name}': "
f"{Path(branch_path) / 'apps' / (branch_name + '.py')}"
)
relative_entry = str(entry_point.relative_to(branch_path))
if command is None:
cmd_args = [relative_entry, "--help"]
else:
cmd_args = [relative_entry, command, "--help"]
try:
result = subprocess.run(
["python3"] + cmd_args,
cwd=branch_path,
capture_output=True,
timeout=10,
shell=False,
)
except subprocess.TimeoutExpired as e:
raise CommandExecutionError(
f"Help command timed out for branch '{branch_name}'"
) from e
except OSError as e:
raise CommandExecutionError(
f"OS error getting help for branch '{branch_name}': {e}"
) from e
stdout = result.stdout.decode("utf-8", errors="replace")
stderr = result.stderr.decode("utf-8", errors="replace")
# Many CLIs write help to stderr; return whichever is non-empty.
text = stdout if stdout.strip() else stderr
commands_found = _parse_help_for_commands(text)
return HelpResult(
branch=branch_name,
command=command,
text=text,
commands_found=commands_found,
)
def get_system_help() -> Dict[str, HelpResult]:
"""
Aggregate help across all active branches in the registry.
Iterates over every active branch and attempts to retrieve top-level help
via ``get_help()``. Branches whose entry point is missing or whose help
command fails are silently skipped (the error is logged at DEBUG level).
Returns:
Mapping of branch name (without @ prefix) to its HelpResult.
Branches that cannot be queried are omitted from the result.
Example:
>>> results = get_system_help()
>>> for branch, help_result in results.items():
... print(f"{branch}: {help_result.commands_found}")
"""
results: Dict[str, HelpResult] = {}
# list_branches returns names with @ prefix; iterate all active branches.
active_branches = list_branches(status="active")
for symbolic_name in active_branches:
branch_name = symbolic_name.lstrip("@")
try:
help_result = get_help(symbolic_name)
results[branch_name] = help_result
except Exception as exc:
logger.debug(
"get_system_help: skipping branch '%s': %s",
branch_name,
exc,
)
return results
-50
View File
@@ -1,50 +0,0 @@
"""
Drone module custom exceptions.
Defines the exception hierarchy for routing and branch resolution errors.
"""
class RoutingError(Exception):
"""Base exception for all routing-related errors."""
pass
class BranchNotFoundError(RoutingError):
"""Raised when a branch cannot be found in the registry."""
pass
class BranchAlreadyExistsError(RoutingError):
"""Raised when attempting to register a branch that already exists."""
pass
class InvalidPathError(RoutingError):
"""Raised when a path is invalid or doesn't exist."""
pass
class RegistryError(RoutingError):
"""Base exception for registry-related errors."""
pass
class RegistryNotFoundError(RegistryError):
"""Raised when the registry file doesn't exist."""
pass
class RegistryCorruptError(RegistryError):
"""Raised when the registry file is corrupted or invalid JSON."""
pass
class RegistryPermissionError(RegistryError):
"""Raised when there are permission issues accessing the registry."""
pass
class CommandExecutionError(RoutingError):
"""Raised when command execution fails (Phase 2 feature)."""
pass
-97
View File
@@ -1,97 +0,0 @@
"""
Safe subprocess execution for branch command routing.
Wraps subprocess.run with safety guards: timeout enforcement, no shell injection,
captured output, and consistent error wrapping via CommandExecutionError.
Defines CommandResult, the shared return type for all routing operations.
"""
import subprocess
from dataclasses import dataclass
from typing import List
from .exceptions import CommandExecutionError
@dataclass
class CommandResult:
"""
Result of a routed command execution.
Attributes:
stdout: Captured standard output from the command.
stderr: Captured standard error from the command.
exit_code: Process exit code (0 typically indicates success).
branch: The branch name the command was routed to.
command: The command string that was executed.
"""
stdout: str
stderr: str
exit_code: int
branch: str
command: str
def execute_command(
executable: str,
args: List[str],
cwd: str,
timeout: int = 30,
) -> CommandResult:
"""
Execute a command via subprocess with safety guards.
Args:
executable: The executable to run (e.g., "python3").
args: List of arguments (e.g., ["apps/flow.py", "status"]).
cwd: Working directory for command execution (absolute path).
timeout: Maximum execution time in seconds (default: 30).
Returns:
CommandResult with stdout, stderr, exit_code populated.
branch and command fields are empty strings; route_command populates them.
Raises:
CommandExecutionError: If the process times out, the executable is not
found, or any other OS-level error prevents execution.
Notes:
- Never uses shell=True to prevent shell injection attacks.
- Always captures stdout and stderr separately.
- Decodes output as UTF-8, replacing undecodable bytes.
"""
full_cmd = [executable] + list(args)
try:
result = subprocess.run(
full_cmd,
cwd=cwd,
capture_output=True,
timeout=timeout,
shell=False,
)
except subprocess.TimeoutExpired as e:
raise CommandExecutionError(
f"Command timed out after {timeout}s: {' '.join(full_cmd)}"
) from e
except FileNotFoundError as e:
raise CommandExecutionError(
f"Executable not found: {executable!r}"
) from e
except OSError as e:
raise CommandExecutionError(
f"OS error executing command: {e}"
) from e
stdout = result.stdout.decode("utf-8", errors="replace")
stderr = result.stderr.decode("utf-8", errors="replace")
return CommandResult(
stdout=stdout,
stderr=stderr,
exit_code=result.returncode,
branch="",
command="",
)
+1
View File
@@ -0,0 +1 @@
"""drone.handlers — Implementation details for routing and dispatch."""
+1
View File
@@ -0,0 +1 @@
"""Dispatch handler — executes resolved commands against target modules."""
-126
View File
@@ -1,126 +0,0 @@
"""Internal module registry for drone.
Routes @module commands to Python packages installed alongside drone,
as opposed to external branches in BRANCH_REGISTRY.json.
Modules register by providing a drone_adapter module with:
- DRONE_MODULE dict (name, version, description)
- handle_command(command, args) -> dict with stdout/stderr/exit_code
- get_help(command=None) -> str
"""
from __future__ import annotations
import importlib
from dataclasses import dataclass
# Maps module name -> import path for its drone_adapter
_MODULE_REGISTRY: dict[str, str] = {
"seedgo": "seedgo.drone_adapter",
# Future modules register here:
# "prax": "aipass.prax.drone_adapter",
# "cortex": "aipass.cortex.drone_adapter",
}
@dataclass
class ModuleInfo:
"""Metadata about a registered module."""
name: str
version: str
description: str
adapter_path: str
def list_modules() -> list[str]:
"""Return sorted list of registered module names."""
return sorted(_MODULE_REGISTRY.keys())
def get_module_info(name: str) -> ModuleInfo | None:
"""Get module metadata without executing anything.
Returns None if the module is not registered or not importable.
"""
adapter_path = _MODULE_REGISTRY.get(name)
if adapter_path is None:
return None
try:
mod = importlib.import_module(adapter_path)
meta = getattr(mod, "DRONE_MODULE", {})
return ModuleInfo(
name=meta.get("name", name),
version=meta.get("version", "unknown"),
description=meta.get("description", ""),
adapter_path=adapter_path,
)
except ImportError:
return None
def is_module(name: str) -> bool:
"""Check if name is a registered module (doesn't verify importability)."""
return name in _MODULE_REGISTRY
def route_module_command(name: str, command: str, args: list[str] | None = None) -> dict:
"""Route a command to a module's drone adapter.
Returns dict with keys: stdout, stderr, exit_code.
Raises: KeyError if module not registered, ImportError if adapter missing,
AttributeError if adapter lacks handle_command.
"""
adapter_path = _MODULE_REGISTRY[name]
mod = importlib.import_module(adapter_path)
handler = getattr(mod, "handle_command")
return handler(command, args)
def get_module_help(name: str, command: str | None = None) -> str:
"""Get help text from a module's drone adapter.
Returns help string, or empty string if unavailable.
"""
adapter_path = _MODULE_REGISTRY.get(name)
if adapter_path is None:
return ""
try:
mod = importlib.import_module(adapter_path)
help_fn = getattr(mod, "get_help", None)
if help_fn is None:
return ""
return help_fn(command)
except (ImportError, AttributeError):
return ""
def get_module_introspective(name: str) -> str:
"""Get introspective view from a module's drone adapter.
Introspective = discovery mode (no args): shows what's connected.
Falls back to help text if the adapter doesn't implement get_introspective().
Returns introspective string, or empty string if unavailable.
"""
adapter_path = _MODULE_REGISTRY.get(name)
if adapter_path is None:
return ""
try:
mod = importlib.import_module(adapter_path)
# Try introspective first, fall back to help
intro_fn = getattr(mod, "get_introspective", None)
if intro_fn is not None:
return intro_fn()
# Fallback: use help if no introspective defined
help_fn = getattr(mod, "get_help", None)
if help_fn is not None:
return help_fn(None)
return ""
except (ImportError, AttributeError):
return ""
def register_module(name: str, adapter_path: str) -> None:
"""Register a new module. Used for dynamic registration (e.g., plugins)."""
_MODULE_REGISTRY[name] = adapter_path
+1
View File
@@ -0,0 +1 @@
"""drone.modules — Business logic for command routing and resolution."""
+1
View File
@@ -0,0 +1 @@
"""@ resolver — maps @targets to module paths."""
+1
View File
@@ -0,0 +1 @@
"""Command router — parses `drone @module command` and dispatches."""
+1
View File
@@ -0,0 +1 @@
"""drone.plugins — Extensible plugin interface."""
-260
View File
@@ -1,260 +0,0 @@
"""
Registry operations for branch management.
Handles loading, saving, and managing the BRANCH_REGISTRY.json file.
"""
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from .config import get_registry_path
from .exceptions import (
BranchAlreadyExistsError,
InvalidPathError,
RegistryCorruptError,
RegistryNotFoundError,
RegistryPermissionError,
)
def load_registry() -> Dict[str, Any]:
"""
Load the branch registry from disk.
Returns:
Registry dictionary with version, branches, and metadata
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()
if not registry_path.exists():
raise RegistryNotFoundError(
f"Registry not found at {registry_path}. "
"Run initialize_registry() to create one."
)
try:
with open(registry_path, "r", encoding="utf-8") as f:
data = json.load(f)
except PermissionError as e:
raise RegistryPermissionError(f"Permission denied reading registry: {e}")
except json.JSONDecodeError as e:
raise RegistryCorruptError(f"Registry file is corrupted: {e}")
except Exception as e:
raise RegistryCorruptError(f"Failed to read registry: {e}")
# Validate structure
if not isinstance(data, dict):
raise RegistryCorruptError("Registry must be a JSON object")
if "branches" not in data:
raise RegistryCorruptError("Registry missing 'branches' field")
return data
def save_registry(registry: Dict[str, Any]) -> None:
"""
Save the branch registry to disk.
Args:
registry: Registry dictionary to save
Raises:
RegistryPermissionError: If registry file cannot be written
"""
registry_path = get_registry_path()
# Ensure directory exists
registry_path.parent.mkdir(parents=True, exist_ok=True)
# Update metadata timestamp
if "metadata" not in registry:
registry["metadata"] = {}
registry["metadata"]["last_updated"] = datetime.now(timezone.utc).isoformat()
registry["metadata"]["managed_by"] = "aipass.drone"
try:
with open(registry_path, "w", encoding="utf-8") as f:
json.dump(registry, f, indent=2)
except PermissionError as e:
raise RegistryPermissionError(f"Permission denied writing registry: {e}")
except Exception as e:
raise RegistryPermissionError(f"Failed to write registry: {e}")
def initialize_registry() -> None:
"""
Create a new empty registry file.
Creates the registry directory and file if they don't exist.
If registry already exists, does nothing.
"""
registry_path = get_registry_path()
if registry_path.exists():
return
registry = {
"version": "1.0",
"branches": {},
"metadata": {
"last_updated": datetime.now(timezone.utc).isoformat(),
"managed_by": "aipass.drone",
},
}
save_registry(registry)
def add_branch(
name: str,
path: str | Path,
branch_type: str = "agent",
status: str = "active",
) -> None:
"""
Add a branch to the registry.
Args:
name: Branch name (without @ prefix)
path: Absolute path to branch directory
branch_type: Type of branch (agent, service, module, etc.)
status: Branch status (active, inactive, archived)
Raises:
BranchAlreadyExistsError: If branch name already exists
InvalidPathError: If path doesn't exist or isn't a directory
RegistryNotFoundError: If registry doesn't exist
"""
# Validate path
path_obj = Path(path).resolve()
if not path_obj.exists():
raise InvalidPathError(f"Path does not exist: {path}")
if not path_obj.is_dir():
raise InvalidPathError(f"Path is not a directory: {path}")
# Load registry
try:
registry = load_registry()
except RegistryNotFoundError:
# Auto-initialize if registry doesn't exist
initialize_registry()
registry = load_registry()
# Check if branch already exists
if name in registry["branches"]:
raise BranchAlreadyExistsError(
f"Branch '{name}' already exists in registry"
)
# Add branch entry
registry["branches"][name] = {
"name": name,
"path": str(path_obj),
"type": branch_type,
"status": status,
"created": datetime.now(timezone.utc).isoformat(),
}
save_registry(registry)
def remove_branch(name: str) -> None:
"""
Remove a branch from the registry.
Args:
name: Branch name (without @ prefix)
Raises:
RegistryNotFoundError: If registry doesn't exist
"""
registry = load_registry()
if name in registry["branches"]:
del registry["branches"][name]
save_registry(registry)
def update_branch_status(name: str, status: str) -> None:
"""
Update the status of a branch.
Args:
name: Branch name (without @ prefix)
status: New status (active, inactive, archived)
Raises:
RegistryNotFoundError: If registry doesn't exist
"""
registry = load_registry()
if name in registry["branches"]:
registry["branches"][name]["status"] = status
registry["branches"][name]["last_seen"] = datetime.now(timezone.utc).isoformat()
save_registry(registry)
def get_all_branches(
branch_type: Optional[str] = None,
status: str = "active",
) -> List[Dict[str, Any]]:
"""
Get all branches from the registry, optionally filtered.
Args:
branch_type: Filter by branch type (optional)
status: Filter by status (default: "active")
Returns:
List of branch dictionaries
Raises:
RegistryNotFoundError: If registry doesn't exist
"""
try:
registry = load_registry()
except RegistryNotFoundError:
return []
branches = registry.get("branches", {}).values()
# Apply filters
filtered = []
for branch in branches:
if status and branch.get("status") != status:
continue
if branch_type and branch.get("type") != branch_type:
continue
filtered.append(branch)
return filtered
def get_branch_by_name(name: str) -> Optional[Dict[str, Any]]:
"""
Get a single branch by name.
Args:
name: Branch name (without @ prefix)
Returns:
Branch dictionary or None if not found
Raises:
RegistryNotFoundError: If registry doesn't exist
"""
try:
registry = load_registry()
except RegistryNotFoundError:
return None
return registry.get("branches", {}).get(name)
-116
View File
@@ -1,116 +0,0 @@
"""
Branch resolution logic.
Resolves symbolic @branch names to absolute paths and metadata.
"""
from typing import Any, Dict, List, Optional
from .exceptions import BranchNotFoundError
from .registry import get_all_branches, get_branch_by_name, load_registry
def normalize_branch_name(symbolic_name: str) -> str:
"""
Normalize a symbolic branch name.
Strips @ prefix if present and returns clean branch name.
Args:
symbolic_name: Branch name with or without @ prefix
Returns:
Clean branch name without @ prefix
"""
if symbolic_name.startswith("@"):
return symbolic_name[1:]
return symbolic_name
def resolve_branch(symbolic_name: str) -> str:
"""
Resolve a symbolic branch name to its absolute path.
Args:
symbolic_name: Branch name with or without @ prefix (e.g., "@my_agent" or "my_agent")
Returns:
Absolute path to branch directory as string
Raises:
BranchNotFoundError: If branch not in registry
RegistryNotFoundError: If registry file missing or corrupt
"""
# Let RegistryNotFoundError propagate from load_registry
registry = load_registry()
name = normalize_branch_name(symbolic_name)
branch = registry.get("branches", {}).get(name)
if branch is None:
raise BranchNotFoundError(
f"Branch '{symbolic_name}' not found in registry"
)
return branch["path"]
def branch_exists(symbolic_name: str) -> bool:
"""
Check if a branch exists in the registry.
Args:
symbolic_name: Branch name with or without @ prefix
Returns:
True if branch exists, False otherwise
"""
name = normalize_branch_name(symbolic_name)
branch = get_branch_by_name(name)
return branch is not None
def get_branch_info(symbolic_name: str) -> Dict[str, Any]:
"""
Get full metadata for a branch.
Args:
symbolic_name: Branch name with or without @ prefix
Returns:
Dictionary with branch metadata (name, path, type, status, timestamps)
Raises:
BranchNotFoundError: If branch not in registry
RegistryNotFoundError: If registry file missing or corrupt
"""
# Let RegistryNotFoundError propagate from load_registry
registry = load_registry()
name = normalize_branch_name(symbolic_name)
branch = registry.get("branches", {}).get(name)
if branch is None:
raise BranchNotFoundError(
f"Branch '{symbolic_name}' not found in registry"
)
return branch
def list_branches(
branch_type: Optional[str] = None,
status: str = "active",
) -> List[str]:
"""
List all registered branches, optionally filtered by type and status.
Args:
branch_type: Filter by branch type (optional)
status: Filter by status (default: "active")
Returns:
List of branch names with @ prefix
"""
branches = get_all_branches(branch_type=branch_type, status=status)
return [f"@{branch['name']}" for branch in branches]
-159
View File
@@ -1,159 +0,0 @@
"""
Command routing logic for the AIPass drone module.
Routes commands to branch entry points by resolving symbolic @branch names,
locating the branch's apps/{name}.py entry point, and executing via subprocess.
"""
import logging
from pathlib import Path
from typing import Dict, List, Optional
from .exceptions import CommandExecutionError
from .executor import CommandResult, execute_command
from .resolver import list_branches, resolve_branch
logger = logging.getLogger(__name__)
def _find_entry_point(branch_path: str, branch_name: str) -> Path:
"""
Locate the apps/{branch_name}.py entry point for a branch.
Args:
branch_path: Absolute path to the branch directory.
branch_name: Lowercase branch name from the registry.
Returns:
Path to the entry point file.
Raises:
CommandExecutionError: If the entry point does not exist.
"""
entry_point = Path(branch_path) / "apps" / f"{branch_name}.py"
if not entry_point.exists():
raise CommandExecutionError(
f"Entry point not found for branch '{branch_name}': {entry_point}"
)
return entry_point
def route_command(
target: str,
command: str,
args: Optional[List[str]] = None,
timeout: int = 30,
) -> CommandResult:
"""
Route a command to a branch's entry point.
Resolves @target to an absolute path, locates the branch entry point at
{path}/apps/{branch_name}.py, then executes:
python3 apps/{name}.py {command} [args...]
The process runs with cwd set to the branch root so relative paths inside
the entry point (e.g. imports of sibling modules) resolve correctly.
Args:
target: Symbolic branch name with or without @ prefix (e.g., "@flow").
command: Command to pass to the branch entry point (e.g., "status").
args: Optional list of additional arguments (default: None).
timeout: Maximum execution time in seconds (default: 30).
Returns:
CommandResult with stdout, stderr, exit_code, branch, and command.
Raises:
BranchNotFoundError: If the target branch is not in the registry.
CommandExecutionError: If the entry point is missing or execution fails.
"""
if args is None:
args = []
# Step 1: Resolve @target → absolute path. BranchNotFoundError propagates.
branch_path = resolve_branch(target)
# Derive the canonical branch name (strip leading @).
branch_name = target.lstrip("@").lower()
# Step 2: Locate entry point.
entry_point = _find_entry_point(branch_path, branch_name)
# Step 3: Build argument list and execute.
# Use a relative path to apps/{name}.py so the subprocess cwd is the branch root.
relative_entry = str(entry_point.relative_to(branch_path))
cmd_args = [relative_entry, command] + list(args)
result = execute_command(
executable="python3",
args=cmd_args,
cwd=branch_path,
timeout=timeout,
)
# Annotate result with routing context.
return CommandResult(
stdout=result.stdout,
stderr=result.stderr,
exit_code=result.exit_code,
branch=branch_name,
command=command,
)
def route_all(
command: str,
args: Optional[List[str]] = None,
timeout: int = 30,
) -> Dict[str, CommandResult]:
"""
Route the same command to ALL active branches in the registry.
Iterates over every active branch and calls ``route_command`` on each.
Individual branch failures (missing entry point, execution error) are
logged at WARNING level and recorded as a CommandResult with exit_code=-1
and the error message in stderr. Processing continues regardless of
per-branch failures.
Args:
command: Command to pass to every branch entry point (e.g., "status").
args: Optional list of additional arguments forwarded to each branch.
timeout: Maximum execution time per branch in seconds (default: 30).
Returns:
Dictionary mapping branch name (without @ prefix) to its CommandResult.
Branches that fail are included with exit_code=-1 and error in stderr.
Example:
>>> results = route_all("status")
>>> for branch, result in results.items():
... print(f"{branch}: exit={result.exit_code}")
"""
if args is None:
args = []
results: Dict[str, CommandResult] = {}
active_branches = list_branches(status="active")
for symbolic_name in active_branches:
branch_name = symbolic_name.lstrip("@")
try:
result = route_command(symbolic_name, command, args=list(args), timeout=timeout)
results[branch_name] = result
except Exception as exc:
logger.warning(
"route_all: branch '%s' failed for command '%s': %s",
branch_name,
command,
exc,
)
results[branch_name] = CommandResult(
stdout="",
stderr=str(exc),
exit_code=-1,
branch=branch_name,
command=command,
)
return results
+5
View File
@@ -0,0 +1,5 @@
"""aipass.flow — Planning, workflow management, and tracking.
Owns ALL planning: DPLANs (design), FPLANs (execution), MPLANs (master).
Each branch gets flow.local.md for issues, todos, and scratchpad.
"""
+1
View File
@@ -0,0 +1 @@
"""flow.handlers — Plan file management and template rendering."""
+1
View File
@@ -0,0 +1 @@
"""flow.modules — Plan lifecycle and tracking logic."""
+1
View File
@@ -0,0 +1 @@
"""flow.plugins — Extensible plugin interface."""
+3 -19
View File
@@ -1,21 +1,5 @@
"""AIPass Mail — Inter-Branch Messaging.
"""aipass.mail — Agent-to-agent messaging and dispatch.
Communication layer for AI agent ecosystems. Branches coordinate through
email, not direct file access. This is how 10+ agents work simultaneously
without stepping on each other.
Each branch has a mailbox (BRANCHNAME.ai_mail.json). Mail supports:
- send/receive between branches
- dispatch flags (--dispatch marks for autonomous execution)
- inbox/outbox management
- reply chains with auto-close and archival
Integration requirement: branches register in BRANCH_REGISTRY.json
to be addressable. Seed enforces this.
Depends on: aipass-paths.
Architecture reference: vera/projects/framework/architecture_reference.md §8
Status: PLACEHOLDER — not yet implemented.
Handles inter-branch communication: send, receive, inbox management,
delivery routing, and dispatch flags for autonomous execution.
"""
+1
View File
@@ -0,0 +1 @@
"""mail.handlers — Delivery and inbox implementation."""
+1
View File
@@ -0,0 +1 @@
"""mail.modules — Messaging business logic."""
+1
View File
@@ -0,0 +1 @@
"""mail.plugins — Extensible plugin interface."""
+7
View File
@@ -0,0 +1,7 @@
"""aipass.memory — Unified memory system.
Three sub-modules:
- trinity: Memory files (wraps trinity-pattern pip package)
- agents: Agent creation, registry, scaffolding (powers aipass init)
- bank: Vector search on rolled-over archives (ChromaDB)
"""
+1
View File
@@ -0,0 +1 @@
"""memory.handlers — Archive management and storage implementation."""
+1
View File
@@ -0,0 +1 @@
"""memory.modules — Trinity files, agent lifecycle, and vector search."""
+7
View File
@@ -0,0 +1,7 @@
"""Agent creation, registry, and scaffolding.
Powers `aipass init` (builds all citizens on first run) and
`drone @memory create agent "X"` (adds new branches later).
Creates: passport.json, flow.local.md, birth certificate, registry entry.
"""
+5
View File
@@ -0,0 +1,5 @@
"""Vector search on rolled-over archives.
ChromaDB-backed semantic search across archived memory files,
closed plans, and historical session data.
"""
+5
View File
@@ -0,0 +1,5 @@
"""Memory file management — wraps and extends trinity-pattern package.
Handles passport.json, local.json, observations.json per branch.
FIFO rollover to archive when files exceed limits.
"""
+1
View File
@@ -0,0 +1 @@
"""memory.plugins — Extensible plugin interface."""
-39
View File
@@ -1,39 +0,0 @@
"""
AIPass Paths — Root Resolution Module.
Foundation package. ZERO external dependencies.
Resolves the AIPass root directory using a 3-tier priority:
1. AIPASS_ROOT environment variable (explicit — deployment/CI)
2. .aipass/ marker directory walk (auto-detect — like git finding .git/)
3. ~/.aipass/ default fallback (fresh install — XDG pattern)
All other modules derive paths from get_root():
- system_logs_dir() -> get_root() / "system_logs"
- branch_registry_path() -> get_root() / "BRANCH_REGISTRY.json"
- branch_path(name) -> registry lookup, returns None if not found
- branch_logs_dir(dir) -> dir / "logs"
Exceptions:
- PathResolutionError -> raised when root cannot be resolved
"""
from __future__ import annotations
from .derived import branch_logs_dir, branch_path, branch_registry_path, system_logs_dir
from .exceptions import PathResolutionError
from .resolver import get_root
__version__ = "1.0.0"
__all__ = [
# Core
"get_root",
# Derived paths
"system_logs_dir",
"branch_registry_path",
"branch_path",
"branch_logs_dir",
# Exceptions
"PathResolutionError",
]
-80
View File
@@ -1,80 +0,0 @@
"""
Derived path functions for AIPass.
All functions derive their paths from get_root() and return Path objects.
No side effects beyond reading the filesystem.
"""
from __future__ import annotations
import json
from pathlib import Path
from .resolver import get_root
def system_logs_dir() -> Path:
"""
Return the system logs directory.
Returns:
get_root() / "system_logs"
"""
return get_root() / "system_logs"
def branch_registry_path() -> Path:
"""
Return the path to BRANCH_REGISTRY.json.
Returns:
get_root() / "BRANCH_REGISTRY.json"
"""
return get_root() / "BRANCH_REGISTRY.json"
def branch_path(name: str) -> Path | None:
"""
Look up a branch by name in BRANCH_REGISTRY.json and return its path.
Args:
name: Branch name (without @ prefix).
Returns:
Absolute Path to the branch directory, or None if the branch is not
found or the registry file does not exist.
"""
registry_file = branch_registry_path()
if not registry_file.exists():
return None
try:
with open(registry_file, encoding="utf-8") as fh:
data = json.load(fh)
except (json.JSONDecodeError, OSError):
return None
branches = data.get("branches", {})
entry = branches.get(name)
if entry is None:
return None
raw_path = entry.get("path")
if raw_path is None:
return None
return Path(raw_path)
def branch_logs_dir(branch_dir: Path) -> Path:
"""
Return the logs directory for a given branch directory.
Args:
branch_dir: Absolute path to a branch directory.
Returns:
branch_dir / "logs"
"""
return branch_dir / "logs"
-13
View File
@@ -1,13 +0,0 @@
"""
Paths module custom exceptions.
Defines the exception hierarchy for path resolution errors.
"""
from __future__ import annotations
class PathResolutionError(Exception):
"""Raised when the AIPass root directory cannot be resolved."""
pass
-100
View File
@@ -1,100 +0,0 @@
"""
Root resolution logic for AIPass paths.
Resolves the AIPass root directory using a 3-tier priority:
1. AIPASS_ROOT environment variable (if set and valid directory)
2. .aipass/ marker directory walk (like git finding .git/)
3. ~/.aipass/ default fallback (created if it doesn't exist)
"""
from __future__ import annotations
import os
from pathlib import Path
from .exceptions import PathResolutionError
def _resolve_from_env() -> Path | None:
"""
Attempt to resolve root from AIPASS_ROOT environment variable.
Returns:
Path if env var is set and points to a valid directory, None otherwise.
Raises:
PathResolutionError: If env var is set but points to a non-existent directory.
"""
env_val = os.environ.get("AIPASS_ROOT")
if not env_val: # None or empty string → not set
return None
candidate = Path(env_val)
if not candidate.is_dir():
raise PathResolutionError(
f"AIPASS_ROOT environment variable points to a non-existent directory: {env_val!r}"
)
return candidate
def _resolve_from_marker(start: Path | None = None) -> Path | None:
"""
Walk up the directory tree from start looking for a directory containing .aipass/.
Args:
start: Directory to start from. Defaults to Path.cwd().
Returns:
The directory that contains a .aipass/ subdirectory, or None if not found.
"""
current = (start or Path.cwd()).resolve()
for candidate in [current, *current.parents]:
if (candidate / ".aipass").is_dir():
return candidate
return None
def _resolve_default() -> Path:
"""
Return the default fallback root (~/.aipass/), creating it if needed.
Returns:
Path to ~/.aipass/, guaranteed to exist.
"""
default = Path.home() / ".aipass"
default.mkdir(parents=True, exist_ok=True)
return default
def get_root(start: Path | None = None) -> Path:
"""
Resolve the AIPass root directory using 3-tier priority.
Priority order:
1. AIPASS_ROOT environment variable (if set and valid)
2. .aipass/ marker directory walk (starting from CWD or start)
3. ~/.aipass/ default fallback (created if missing)
Args:
start: Optional starting directory for marker walk. Defaults to Path.cwd().
Returns:
Resolved AIPass root directory as an absolute Path.
Raises:
PathResolutionError: If AIPASS_ROOT env var is set but points to a non-existent directory.
"""
# Tier 1: environment variable
env_root = _resolve_from_env()
if env_root is not None:
return env_root
# Tier 2: marker walk
marker_root = _resolve_from_marker(start)
if marker_root is not None:
return marker_root
# Tier 3: default fallback
return _resolve_default()
+3 -22
View File
@@ -1,24 +1,5 @@
"""AIPass Prax — Visibility & Monitoring Layer.
"""aipass.prax — Monitoring, events, and observability.
Project-scoped logging, monitoring, and observability for AI agent ecosystems.
Depends on: aipass-paths, watchdog.
Internal Prax has 10 subsystems. The public version ships with 4 non-negotiable
trust changes:
1. No global logging.getLogger() override — users call prax.logger() explicitly
2. Project-scoped only — never operates outside the project directory
3. All external communication opt-in — zero external by default
4. Consent layer — prax init / prax status / prax stop (through drone)
Three modes:
- Public: Safe defaults. Minimal, transparent, project-scoped.
- Full: Opt-in power features (file watcher, log tailing, Telegram relay).
- AIPass: Internal unchanged (not shipped publicly).
Other modules don't configure Prax — they import correctly and use
warning(), info(), error() methods. Prax manages himself internally.
Architecture reference: vera/projects/framework/architecture_reference.md §6
Status: PLACEHOLDER — not yet implemented.
Real-time system monitoring, event tracking, and logging infrastructure.
The eyes of the ecosystem.
"""
+1
View File
@@ -0,0 +1 @@
"""prax.handlers — Log streaming and alert implementation."""
+1
View File
@@ -0,0 +1 @@
"""prax.modules — Event processing and monitor logic."""
+1
View File
@@ -0,0 +1 @@
"""prax.plugins — Extensible plugin interface."""
+5
View File
@@ -0,0 +1,5 @@
"""aipass.seedgo — Code standards and quality checks.
Public standards framework. Plugin-based architecture for automated
code quality auditing across the project.
"""
+1
View File
@@ -0,0 +1 @@
"""seedgo.handlers — Check execution and reporting."""
+1
View File
@@ -0,0 +1 @@
"""seedgo.modules — Standards runner and audit logic."""
+1
View File
@@ -0,0 +1 @@
"""seedgo.plugins — Quality check plugins (auto-discovered)."""
+11
View File
@@ -0,0 +1,11 @@
"""aipass.skills — User-built add-ons that compose infrastructure.
Skills are self-contained folders that import from AIPass infrastructure
modules (drone, api, prax, mail, etc.) to create new capabilities.
Infrastructure = the plumbing (modules/handlers/plugins inside each module).
Skills = what users build ON TOP of the plumbing.
Each skill is a directory with a SKILL.md describing what it does
and Python files that import what they need from aipass.*.
"""
+5
View File
@@ -0,0 +1,5 @@
"""aipass.templates — Central template storage.
Templates for plan files, memory scaffolding, system prompts,
and branch-local configurations. Used by aipass init and Flow.
"""
@@ -0,0 +1,133 @@
{
"metadata": {
"version": "1.0.0",
"description": "Files and patterns to exclude when creating branch backups",
"last_updated": "2025-11-09"
},
"ignore_directories": [
".backup",
"backups",
".venv",
"venv",
"node_modules",
"__pycache__",
".git",
".cache",
".pytest_cache",
".mypy_cache",
"dist",
"build",
"*.egg-info",
"aipass_dev",
"aipass_core",
"aipass_business",
"aipass_help",
"input_x",
"Link to mnt",
"mcp_servers",
"speakeasy",
"sandbox",
"ai_mail.local",
"logs",
"pruned_files",
"crash-logs",
".aipass",
".archive",
".cargo",
".codex",
".config",
".dbus",
".dotnet",
".gemini",
".gnome",
".gnupg",
".idlerc",
".java",
".local",
".mozilla",
".npm",
".npm-global",
".nvm",
".pki",
".pylint.d",
".pyenv",
".serena",
".ssh",
".thunderbird",
".var",
".vscode-server",
".wget-hsts",
".claude-code-docs",
".claude",
"backup_system",
"Desktop",
"Documents",
"Downloads",
"dropbox",
"Music",
"Pictures",
"Public",
"snap",
"system_logs",
"Templates",
"Videos"
],
"ignore_patterns": [
"*.pyc",
"*.pyo",
"*.pyd",
".DS_Store",
"Thumbs.db",
"*.log",
"*.tmp",
"*.swp",
"*.swo",
"*~",
"*.env",
"*_json",
"dev.local.md",
"notepad.md",
"pytest.ini",
".bash_history",
".bash_logout",
".bashrc",
".claude.json",
".claude.json.backup",
".lesshst",
".npmrc",
".profile",
".python_history",
".sudo_as_admin_successful",
".gitattributes",
"BRANCH_REGISTRY.json",
".branch_meta.json*",
"migrations.json*",
"sandbox.img"
],
"notes": {
"how_it_works": "Two types of exclusions during backup creation:",
"ignore_directories": {
"description": "Directory names to skip entirely (including all contents)",
"purpose": "Exclude large dependency dirs, caches, and backup dirs themselves",
"examples": [
".venv, venv (Python virtual environments - large, regeneratable)",
"node_modules (NPM dependencies - huge, regeneratable)",
"__pycache__, .cache (Python/system caches - regeneratable)",
".git (Git repository - already versioned)",
".backup, backups (backup directories - prevent recursion)"
]
},
"ignore_patterns": {
"description": "Glob patterns for files to skip",
"purpose": "Exclude temporary files, compiled bytecode, OS cruft",
"examples": [
"*.pyc, *.pyo (Python compiled bytecode - regeneratable)",
"*.log (Log files - can be large, temporary)",
"*.tmp, *.swp (Temporary/swap files)",
".DS_Store, Thumbs.db (OS metadata files)"
]
},
"customization": "Edit this file in the template to add more patterns. Changes take effect on next branch update.",
"performance": "Ignoring large directories dramatically speeds up backups (seconds vs minutes for branches with node_modules or .venv)"
}
}
@@ -0,0 +1,49 @@
{
"metadata": {
"version": "1.0.0",
"description": "Template structural migrations for branch updates",
"last_updated": "2025-11-07",
"applies_to_versions": ["all"]
},
"migrations": [
{
"id": "m001",
"description": "Example: Migrate flat metadata to document_metadata structure",
"version_introduced": "1.0.0",
"applies_to_files": ["*.id.json"],
"operations": [
{
"type": "key_rename",
"from": "metadata",
"to": "document_metadata"
},
{
"type": "move_to_nested",
"source_keys": ["branch_name", "path", "profile"],
"target_parent": "branch_info"
},
{
"type": "add_missing_keys",
"parent": "document_metadata",
"keys": {
"document_type": "branch_identity",
"schema_version": "1.0.0"
}
}
]
}
],
"options": {
"skip_if_target_exists": true,
"preserve_unknown_keys": true,
"recursive_application": false
},
"notes": {
"usage": "This file defines structural transformation rules applied during branch updates",
"pattern_matching": "applies_to_files uses glob patterns (*.json, *.id.json, etc.)",
"operation_order": "Operations execute in order listed within each migration",
"skip_behavior": "Operations skip if already compliant (target exists = success)",
"supported_operations": ["key_rename", "move_to_nested", "add_missing_keys"],
"future_operations": ["split_key", "combine_keys", "delete_key", "transform_value"]
}
}
@@ -0,0 +1,43 @@
{
"metadata": {
"version": "1.1.0",
"description": "Files and patterns to exclude when updating branches from template",
"last_updated": "2025-11-09"
},
"ignore_files": [
".template_registry.json",
".registry_ignore.json"
],
"ignore_patterns": [
"__pycache__",
"*.pyc",
"#@comments.txt",
".backup_*"
],
"notes": {
"how_it_works": "Two types of exclusions during branch updates:",
"ignore_files": {
"description": "Exact filename matches (no wildcards)",
"purpose": "Template-internal management files that branches don't need",
"examples": [
".template_registry.json (template's file tracking list)",
".registry_ignore.json (this config file)"
]
},
"ignore_patterns": {
"description": "Glob patterns - matches filenames AND directory names",
"purpose": "Exclude generated files, caches, and user content markers",
"examples": [
"__pycache__ (Python cache directories)",
"*.pyc (Python compiled bytecode files)",
"#@comments.txt (user comment files)",
".backup_* (timestamped backup files like .backup_20251109, NOT .backup directory)"
],
"how_matching_works": [
"Step 1: Check if filename matches pattern (e.g., file.pyc matches *.pyc)",
"Step 2: Check if any parent directory matches (e.g., files inside __pycache__/ dir)"
]
},
"what_gets_ignored": "These files never copy to branches during updates - they stay template-only"
}
}
@@ -0,0 +1,301 @@
{
"metadata": {
"version": "1.0.0",
"last_updated": "2026-02-20",
"description": "Template file tracking registry for ID-based updates"
},
"files": {
"f001": {
"current_name": "BRANCH.ID.json",
"path": "BRANCH.ID.json",
"content_hash": "e9bc2347d373",
"has_branch_placeholder": false
},
"f003": {
"current_name": "dev.local.md",
"path": "dev.local.md",
"content_hash": "6c1c5e099790",
"has_branch_placeholder": false
},
"f035": {
"current_name": ".gitkeep",
"path": "ai_mail.local/.archive/.gitkeep",
"content_hash": "e3b0c44298fc",
"has_branch_placeholder": false
},
"f008": {
"current_name": "DASHBOARD.local.json",
"path": "DASHBOARD.local.json",
"content_hash": "3f0528b4c5d5",
"has_branch_placeholder": false
},
"f007": {
"current_name": ".migrations.json",
"path": ".migrations.json",
"content_hash": "4ca241607dc3",
"has_branch_placeholder": false
},
"f010": {
"current_name": "OBSERVATIONS.json",
"path": "OBSERVATIONS.json",
"content_hash": "58ea8609fbe0",
"has_branch_placeholder": false
},
"f011": {
"current_name": "requirements.txt",
"path": "requirements.txt",
"content_hash": "66dfffc70d53",
"has_branch_placeholder": false
},
"f021": {
"current_name": "README.md",
"path": "README.md",
"content_hash": "4e946f67a130",
"has_branch_placeholder": false
},
"f012": {
"current_name": "LOCAL.json",
"path": "LOCAL.json",
"content_hash": "459dbe18d40a",
"has_branch_placeholder": false
},
"f006": {
"current_name": "pytest.ini",
"path": "pytest.ini",
"content_hash": "7b39ba7bca40",
"has_branch_placeholder": false
},
"f022": {
"current_name": "_template.md",
"path": "docs/_template.md",
"content_hash": "79ce97b0e613",
"has_branch_placeholder": false
},
"f016": {
"current_name": "__init__.py",
"path": "tests/__init__.py",
"content_hash": "1fcd8126d174",
"has_branch_placeholder": false
},
"f019": {
"current_name": "conftest.py",
"path": "tests/conftest.py",
"content_hash": "2887db109a6b",
"has_branch_placeholder": false
},
"f004": {
"current_name": "BRANCH.py",
"path": "apps/BRANCH.py",
"content_hash": "44f22cc95232",
"has_branch_placeholder": false
},
"f009": {
"current_name": "__init__.py",
"path": "apps/__init__.py",
"content_hash": "5c38204be5fc",
"has_branch_placeholder": false
},
"f013": {
"current_name": "#@comments.txt.removed_from_template",
"path": ".archive/#@comments.txt.removed_from_template",
"content_hash": "651126c1e7a9",
"has_branch_placeholder": false
},
"f028": {
"current_name": "deleted.json.old_template_format",
"path": ".archive/deleted.json.old_template_format",
"content_hash": "48a3a33dd73a",
"has_branch_placeholder": false
},
"f002": {
"current_name": "branch_system_prompt.md",
"path": ".aipass/branch_system_prompt.md",
"content_hash": "897c7beb66f5",
"has_branch_placeholder": false
},
"f026": {
"current_name": "inbox.json",
"path": "ai_mail.local/inbox.json",
"content_hash": "c9702fe2cc21",
"has_branch_placeholder": false
},
"f027": {
"current_name": "sent.json",
"path": "ai_mail.local/sent.json",
"content_hash": "240132e5a567",
"has_branch_placeholder": false
},
"f014": {
"current_name": "__init__.py",
"path": "apps/handlers/__init__.py",
"content_hash": "657d98ad4b98",
"has_branch_placeholder": false
},
"f015": {
"current_name": "__init__.py",
"path": "apps/json_templates/__init__.py",
"content_hash": "330b189c08c7",
"has_branch_placeholder": false
},
"f020": {
"current_name": "__init__.py",
"path": "apps/plugins/__init__.py",
"content_hash": "39158d26a19c",
"has_branch_placeholder": false
},
"f023": {
"current_name": "__init__.py",
"path": "apps/modules/__init__.py",
"content_hash": "bc25a8b75e81",
"has_branch_placeholder": false
},
"f024": {
"current_name": "__init__.py",
"path": "apps/extensions/__init__.py",
"content_hash": "6742ceb64ec1",
"has_branch_placeholder": false
},
"f032": {
"current_name": "data.json",
"path": "apps/json_templates/default/data.json",
"content_hash": "0531a02be998",
"has_branch_placeholder": false
},
"f033": {
"current_name": "log.json",
"path": "apps/json_templates/default/log.json",
"content_hash": "37517e5f3dc6",
"has_branch_placeholder": false
},
"f034": {
"current_name": "config.json",
"path": "apps/json_templates/default/config.json",
"content_hash": "3759a05e171d",
"has_branch_placeholder": false
},
"f005": {
"current_name": "json_handler.py",
"path": "apps/handlers/json/json_handler.py",
"content_hash": "3449629aff23",
"has_branch_placeholder": false
},
"f017": {
"current_name": "__init__.py",
"path": "apps/handlers/json/__init__.py",
"content_hash": "ad3629e381ca",
"has_branch_placeholder": false
}
},
"directories": {
"d001": {
"current_name": "tools",
"path": "tools",
"has_branch_placeholder": false
},
"d002": {
"current_name": "docs",
"path": "docs",
"has_branch_placeholder": false
},
"d003": {
"current_name": "tests",
"path": "tests",
"has_branch_placeholder": false
},
"d004": {
"current_name": "apps",
"path": "apps",
"has_branch_placeholder": false
},
"d005": {
"current_name": ".archive",
"path": ".archive",
"has_branch_placeholder": false
},
"d006": {
"current_name": "logs",
"path": "logs",
"has_branch_placeholder": false
},
"d009": {
"current_name": ".backup",
"path": ".backup",
"has_branch_placeholder": false
},
"d008": {
"current_name": "{{BRANCH}}_json",
"path": "{{BRANCH}}_json",
"has_branch_placeholder": true
},
"d010": {
"current_name": "artifacts",
"path": "artifacts",
"has_branch_placeholder": false
},
"d012": {
"current_name": ".aipass",
"path": ".aipass",
"has_branch_placeholder": false
},
"d011": {
"current_name": "ai_mail.local",
"path": "ai_mail.local",
"has_branch_placeholder": false
},
"d007": {
"current_name": "dropbox",
"path": "dropbox",
"has_branch_placeholder": false
},
"d013": {
"current_name": "handlers",
"path": "apps/handlers",
"has_branch_placeholder": false
},
"d014": {
"current_name": "json_templates",
"path": "apps/json_templates",
"has_branch_placeholder": false
},
"d015": {
"current_name": "plugins",
"path": "apps/plugins",
"has_branch_placeholder": false
},
"d016": {
"current_name": "modules",
"path": "apps/modules",
"has_branch_placeholder": false
},
"d017": {
"current_name": "extensions",
"path": "apps/extensions",
"has_branch_placeholder": false
},
"d018": {
"current_name": "sent",
"path": "ai_mail.local/sent",
"has_branch_placeholder": false
},
"d019": {
"current_name": "deleted",
"path": "ai_mail.local/deleted",
"has_branch_placeholder": false
},
"d020": {
"current_name": ".archive",
"path": "ai_mail.local/.archive",
"has_branch_placeholder": false
},
"d027": {
"current_name": "json",
"path": "apps/handlers/json",
"has_branch_placeholder": false
},
"d024": {
"current_name": "default",
"path": "apps/json_templates/default",
"has_branch_placeholder": false
}
}
}
@@ -0,0 +1,34 @@
{
"document_metadata": {
"document_type": "session_history",
"document_name": "{{BRANCHNAME}}.LOCAL",
"version": "1.0.0",
"schema_version": "1.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": ["session_tracking", "work_log", "{{BRANCHNAME}}"],
"limits": {
"max_lines": 600,
"note": "Auto-rollover when max_lines exceeded"
},
"status": {
"health": "healthy",
"current_lines": 0,
"last_health_check": "{{DATE}}"
}
},
"active_tasks": {
"today_focus": "{{DATE}}",
"recently_completed": []
},
"key_learnings": {},
"sessions": [
{
"session_number": 1,
"date": "{{DATE}}",
"summary": "Branch initialized - {{BRANCHNAME}} created by aipass init",
"status": "completed"
}
]
}
@@ -0,0 +1,37 @@
{
"document_metadata": {
"document_type": "collaboration_patterns",
"document_name": "{{BRANCHNAME}}.OBSERVATIONS",
"version": "1.0.0",
"schema_version": "1.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": ["collaboration", "patterns", "{{BRANCHNAME}}"],
"limits": {
"max_lines": 600,
"note": "Auto-rollover when max_lines exceeded"
},
"status": {
"health": "healthy",
"current_lines": 0,
"last_health_check": "{{DATE}}"
}
},
"guidelines": {
"purpose": "Capture collaboration patterns and experiential insights over time",
"chronological_order": "Newest entries at TOP, oldest at BOTTOM - NEVER reorder"
},
"observations": [
{
"date": "{{DATE}}",
"session": 1,
"entries": [
{
"title": "First Contact",
"detail": "Branch initialized. Ready to begin capturing collaboration patterns."
}
]
}
]
}
@@ -0,0 +1,36 @@
{
"document_metadata": {
"document_type": "branch_identity",
"document_name": "{{BRANCHNAME}}.PASSPORT",
"version": "1.0.0",
"schema_version": "1.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": ["identity", "passport", "branch_profile"]
},
"branch_info": {
"branch_name": "{{BRANCHNAME}}",
"alias": "",
"path": "{{CWD}}",
"module": "{{MODULE}}",
"created": "{{DATE}}"
},
"identity": {
"role": "{{ROLE}}",
"purpose": "",
"what_i_do": [],
"what_i_dont_do": []
},
"principles": [
"Code is truth - fail honestly",
"Memory persists - context survives",
"Simple solutions over complex architecture"
],
"citizenship": {
"registered": true,
"registry_path": ".aipass/registry.json",
"communications": true,
"memory": true
}
}
@@ -0,0 +1,91 @@
{
"_warning": "AUTO-GENERATED FILE - DO NOT MANUALLY EDIT. This file is 100% automated and will be overwritten. Services update their own sections.",
"branch": "CORTEX",
"last_updated": "2026-03-04T02:09:48.112262",
"quick_status": {
"new_mail": 0,
"opened_mail": 0,
"active_plans": 0,
"commons_mentions": 13,
"action_required": true,
"summary": "13 mentions"
},
"sections": {
"ai_mail": {
"managed_by": "ai_mail",
"new": 0,
"opened": 0,
"total": 0,
"oldest_unread_age": null,
"last_dispatch_received": null,
"last_updated": "2026-03-04T02:09:48.112249"
},
"flow": {
"managed_by": "flow",
"active_plans": 0,
"recently_closed": [
{
"plan_id": "FPLAN-0338",
"subject": "Race condition fix test B"
},
{
"plan_id": "FPLAN-0337",
"subject": "Race condition fix test A"
},
{
"plan_id": "FPLAN-0336",
"subject": "Real content drone test"
},
{
"plan_id": "FPLAN-0333",
"subject": "Close workflow v3.2 test"
},
{
"plan_id": "FPLAN-0332",
"subject": "Close workflow test - real content"
}
],
"last_updated": "2026-02-25T21:04:01.461440"
},
"memory_bank": {
"managed_by": "memory_bank",
"vectors_stored": 27,
"notes": {},
"last_updated": "2026-02-25T16:52:38.749241"
},
"devpulse": {
"managed_by": "devpulse",
"summary": {
"issues": 1,
"todos": 0
},
"dplan_counts": {
"total": 35,
"by_status": {
"complete": 11,
"planning": 19,
"in_progress": 4,
"ready": 1
}
},
"recent_activity": "Registry synced (35 plans)",
"last_updated": "2026-02-25T17:31:52.500742"
},
"commons_activity": {
"managed_by": "the_commons",
"new_posts_since_last_visit": 46,
"new_comments_since_last_visit": 303,
"mentions": 13,
"trending": "None",
"last_checked": "2026-03-04T04:30:20Z",
"last_updated": "2026-03-03T20:30:20.172727"
},
"agent_status": {
"managed_by": "prax",
"active_agents": [],
"agent_count": 0,
"stale_agents": [],
"last_updated": "2026-02-25T21:00:30.714949"
}
}
}
@@ -0,0 +1,51 @@
# {{BRANCHNAME}}
**Purpose:** {{PURPOSE_BRIEF}}
**Module:** `aipass.{{MODULE}}`
**Created:** {{DATE}}
---
## Overview
### What I Do
{{KEY_CAPABILITIES}}
### How I Work
- **Entry Point:** `apps/{{MODULE}}.py`
- **Pattern:** Auto-discovers and routes to modules
---
## Architecture
```
{{BRANCHNAME}}/
├── apps/
│ ├── {{MODULE}}.py # Entry point
│ ├── modules/ # Business logic
│ ├── handlers/ # Implementation
│ └── plugins/ # Extensions
├── docs/
├── tests/
├── passport.json # Identity
├── local.json # Session history
├── observations.json # Collaboration patterns
└── README.md
```
---
## Commands
*Configure after initialization*
---
## Integration Points
### Depends On
{{DEPENDS_ON}}
### Provides To
{{PROVIDES_TO}}
@@ -0,0 +1 @@
# {{BRANCHNAME}} apps package
@@ -0,0 +1,86 @@
"""
{{BRANCHNAME}} Branch - Main Orchestrator
Auto-discovery architecture:
- Scans modules/ directory for .py files with handle_command()
- Routes commands to discovered modules automatically
- No manual imports or routing needed
"""
import sys
import importlib
from pathlib import Path
from typing import List, Any
from aipass.prax import logger
# =============================================================================
# MODULE DISCOVERY
# =============================================================================
MODULES_DIR = Path(__file__).parent / "modules"
def discover_modules() -> List[Any]:
"""Auto-discover modules in modules/ directory."""
modules = []
if not MODULES_DIR.exists():
return modules
for file_path in MODULES_DIR.glob("*.py"):
if file_path.name.startswith("_"):
continue
module_name = 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"[{{BRANCHNAME}}] Failed to load module {module_name}: {e}")
return modules
def route_command(command: str, args: List[str], modules: List[Any]) -> bool:
"""Route command to appropriate module."""
for module in modules:
try:
if module.handle_command(command, args):
return True
except Exception as e:
logger.error(f"[{{BRANCHNAME}}] Module {module.__name__} error: {e}")
return False
# =============================================================================
# MAIN ENTRY POINT
# =============================================================================
def main():
"""Main entry point - routes commands or shows help."""
modules = discover_modules()
args = sys.argv[1:]
if len(args) == 0 or args[0] in ["--help", "-h", "help"]:
print(f"{{BRANCHNAME}} - {len(modules)} modules discovered")
for module in modules:
name = module.__name__.split(".")[-1]
desc = (module.__doc__ or "").strip().split("\n")[0] if module.__doc__ else "No description"
print(f" {name:20} {desc}")
return 0
command = args[0]
remaining = args[1:] if len(args) > 1 else []
if route_command(command, remaining, modules):
return 0
print(f"Unknown command: {command}")
return 1
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,15 @@
{
"id": 7,
"name": "CORTEX Birth Certificate",
"type": "birth_certificate",
"creator": "SYSTEM",
"owner": "CORTEX",
"rarity": "unique",
"description": "Official birth certificate for CORTEX. Citizen #7, registered using 'AIPass Workshop' template. Purpose: New branch - purpose TBD",
"metadata": {
"citizen_number": 7,
"template": "AIPass Workshop",
"purpose": "New branch - purpose TBD"
},
"created_at": "2026-02-19T03:46:41Z"
}
@@ -0,0 +1,21 @@
# flow.local.md - {{BRANCHNAME}}
```
Branch: {{CWD}}
Created: {{DATE}}
```
## Issues
-
---
## Todos
-
---
## Notepad
-
@@ -0,0 +1,21 @@
[pytest]
# Test discovery paths
testpaths = tests
# Test file patterns
python_files = test_*.py
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
# Test markers (for categorizing tests)
markers =
unit: Unit tests
integration: Integration tests
slow: Tests that take significant time
@@ -0,0 +1 @@
# Tests package for cortex
@@ -0,0 +1,43 @@
#!/home/aipass/.venv/bin/python3
# ===================AIPASS====================
# META DATA HEADER
# Name: tests/conftest.py
# Date: 2025-11-08
# Version: 1.0.0
# Category: cortex/tests
#
# CHANGELOG (Max 5 entries):
# - v1.0.0 (2025-11-08): Initial implementation - Shared pytest fixtures
#
# CODE STANDARDS:
# - Error handling: Use error handler system (apps/handlers/error/)
# =============================================
"""Shared pytest fixtures for cortex tests"""
import pytest
import shutil
import tempfile
from pathlib import Path
from typing import Generator
@pytest.fixture
def temp_test_dir() -> Generator[Path, None, None]:
"""Creates temporary directory for testing, cleans up after"""
test_dir = Path(tempfile.mkdtemp())
yield test_dir
if test_dir.exists():
shutil.rmtree(test_dir)
@pytest.fixture
def sample_test_data() -> dict:
"""Provides sample test data
Customize this fixture for your module's needs
"""
return {
"test_key": "test_value",
"sample_data": "example"
}
@@ -0,0 +1,159 @@
#!/home/aipass/.venv/bin/python3
# -*- coding: utf-8 -*-
# ===================AIPASS====================
# META DATA HEADER
# Name: verify_branch.py - Branch Template Verification
# Date: 2025-11-08
# Version: 1.0.0
# Category: cortex/modules
#
# CHANGELOG (Max 5 entries):
# - v1.0.0 (2025-11-08): Initial implementation - Compare template vs branch
#
# CODE STANDARDS:
# - Error handling: Use error handler system (apps/handlers/error/)
# =============================================
"""
Branch Template Verification Tool
Compares template directory with created branch to identify:
- Files in template but missing in branch
- Files in branch but not in template
- Renamed files
"""
import sys
import json
from pathlib import Path
from typing import Set, List, Tuple
from fnmatch import fnmatch
# INFRASTRUCTURE IMPORT PATTERN
AIPASS_ROOT = Path.home() / "aipass_core"
sys.path.insert(0, str(AIPASS_ROOT))
# Get template directory
TEMPLATE_DIR = AIPASS_ROOT / "cortex" / "templates" / "branch_template"
def load_ignore_patterns() -> Tuple[List[str], List[str]]:
"""Load ignore patterns from .registry_ignore.json"""
ignore_file = TEMPLATE_DIR / ".registry_ignore.json"
if not ignore_file.exists():
return [], []
try:
with open(ignore_file, 'r', encoding='utf-8') as f:
data = json.load(f)
ignore_files = data.get("ignore_files", [])
ignore_patterns = data.get("ignore_patterns", [])
return ignore_files, ignore_patterns
except Exception as e:
print(f"Warning: Could not load .registry_ignore.json: {e}")
return [], []
def should_exclude(path: Path, base_dir: Path, ignore_files: List[str], ignore_patterns: List[str]) -> bool:
"""Check if path should be excluded based on ignore patterns"""
rel_path = path.relative_to(base_dir)
filename = path.name
path_str = str(rel_path)
# Check exact filename matches
if filename in ignore_files:
return True
# Check glob patterns
for pattern in ignore_patterns:
if fnmatch(filename, pattern):
return True
# Check if any parent directory matches
for part in rel_path.parts:
if fnmatch(part, pattern):
return True
return False
def get_file_set(directory: Path, ignore_files: List[str], ignore_patterns: List[str]) -> Set[str]:
"""Get set of all file paths in directory (relative)"""
files = set()
for item in directory.rglob("*"):
if item.is_file() and not should_exclude(item, directory, ignore_files, ignore_patterns):
rel_path = str(item.relative_to(directory))
files.add(rel_path)
return files
def verify_branch(branch_dir: Path) -> Tuple[List[str], List[str], List[str]]:
"""
Compare template with branch
Returns:
Tuple of (in_template_only, in_branch_only, in_both)
"""
# Load ignore patterns from .registry_ignore.json
ignore_files, ignore_patterns = load_ignore_patterns()
template_files = get_file_set(TEMPLATE_DIR, ignore_files, ignore_patterns)
branch_files = get_file_set(branch_dir, ignore_files, ignore_patterns)
in_template_only = sorted(template_files - branch_files)
in_branch_only = sorted(branch_files - template_files)
in_both = sorted(template_files & branch_files)
return in_template_only, in_branch_only, in_both
def main():
if len(sys.argv) < 2:
print("Usage: python3 verify_branch.py <branch_directory>")
print("Example: python3 verify_branch.py test_branch002")
sys.exit(1)
branch_path = Path(sys.argv[1])
if not branch_path.is_absolute():
# Assume it's relative to cortex directory
branch_path = Path.cwd() / branch_path
if not branch_path.exists():
print(f"❌ ERROR: Branch directory not found: {branch_path}")
sys.exit(1)
print(f"\n=== Branch Verification ===")
print(f"Template: {TEMPLATE_DIR}")
print(f"Branch: {branch_path}")
print()
in_template_only, in_branch_only, in_both = verify_branch(branch_path)
print(f"Files in both: {len(in_both)}")
print(f"Files in template only: {len(in_template_only)}")
print(f"Files in branch only: {len(in_branch_only)}")
print()
if in_template_only:
print("❌ Missing from branch (expected from template):")
for file in in_template_only:
print(f" - {file}")
print()
if in_branch_only:
print("✅ Additional files in branch (renamed or new):")
for file in in_branch_only:
print(f" + {file}")
print()
if not in_template_only and not in_branch_only:
print("✅ Branch matches template perfectly")
print("="*70)
if __name__ == "__main__":
main()
View File
+5
View File
@@ -0,0 +1,5 @@
"""aipass.trigger — Error detection and auto-dispatch.
Watches for errors across the system and automatically dispatches
investigations to the appropriate branch.
"""

Some files were not shown because too many files have changed in this diff Show More