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

feat(system): fix: cross-project support — watchdog resolves external branches (BUG-3), seedgo discovers caller project registries (BUG-2), audit @branch shorthand defaults to aipass pack

Browse Source 
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-23 01:04:03 -07:00
co-authored by @devpulse
parent 95fd5850be
commit 12da068f9a
3 changed files with 127 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/devpulse/apps/handlers/watchdog/agent.py
+57 -14
View File
@@ -51,28 +51,71 @@ def _find_repo_root(start: Path | None = None) -> Path | None:
return None
def _resolve_branch_path(agent_id: str) -> Path | None:
"""Resolve an `@branch` token (or bare name) to its absolute branch path."""
repo_root = _find_repo_root()
if repo_root is None:
logger.warning("[watchdog.agent] AIPASS_REGISTRY.json not found")
return None
registry_file = repo_root / "AIPASS_REGISTRY.json"
def _search_registry(registry_file: Path, target_email: str) -> Path | None:
"""Search a single registry file for a branch matching target_email."""
try:
registry = json.loads(registry_file.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
logger.warning("[watchdog.agent] failed to read registry: %s", exc)
logger.warning("[watchdog.agent] failed to read registry %s: %s", registry_file, exc)
return None
target = f"@{agent_id.lstrip('@').lower()}"
for branch in registry.get("branches", []):
if branch.get("email", "").lower() == target:
registry_dir = registry_file.parent
raw_branches = registry.get("branches", [])
if isinstance(raw_branches, dict):
raw_branches = list(raw_branches.values())
for branch in raw_branches:
if branch.get("email", "").lower() == target_email:
raw_path = branch.get("path", "")
path = Path(raw_path)
if not path.is_absolute():
path = repo_root / path
path = registry_dir / path
return path if path.exists() else None
name = branch.get("name", "").lower()
if f"@{name}" == target_email:
raw_path = branch.get("path", "")
path = Path(raw_path)
if not path.is_absolute():
path = registry_dir / path
return path if path.exists() else None
return None
def _resolve_branch_path(agent_id: str) -> Path | None:
"""Resolve an `@branch` token (or bare name) to its absolute branch path.
Checks the AIPass registry first, then scans external project registries
found via AIPASS_CALLER_CWD or known project paths.
"""
target = f"@{agent_id.lstrip('@').lower()}"
repo_root = _find_repo_root()
if repo_root is not None:
registry_file = repo_root / "AIPASS_REGISTRY.json"
result = _search_registry(registry_file, target)
if result is not None:
return result
caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
search_roots = []
if caller_cwd:
search_roots.append(Path(caller_cwd))
projects_dir = Path.home() / "Projects"
if projects_dir.is_dir():
search_roots.extend(sorted(projects_dir.iterdir()))
seen = set()
if repo_root:
seen.add((repo_root / "AIPASS_REGISTRY.json").resolve())
for root in search_roots:
if not root.is_dir():
continue
for reg in root.glob("*_REGISTRY.json"):
resolved = reg.resolve()
if resolved in seen:
continue
seen.add(resolved)
result = _search_registry(reg, target)
if result is not None:
return result
return None
src/aipass/seedgo/apps/handlers/audit/discovery.py
+64 -48
View File
@@ -75,9 +75,58 @@ def _find_registry() -> Path:
return Path.cwd() / "AIPASS_REGISTRY.json"
def _find_caller_registries() -> List[Path]:
"""Find registries from the caller's project via AIPASS_CALLER_CWD."""
import os
caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
if not caller_cwd:
return []
caller_path = Path(caller_cwd)
for parent in [caller_path] + list(caller_path.parents):
matches = sorted(parent.glob("*_REGISTRY.json"))
if matches:
return matches
return []
def _branches_from_registry(registry_path: Path) -> List[Dict[str, str]]:
"""Extract branch dicts from a single registry file."""
branches = []
if not registry_path.exists():
return branches
try:
with open(registry_path, "r", encoding="utf-8") as f:
registry_data = json.load(f)
registry_dir = registry_path.parent
raw_branches = registry_data.get("branches", [])
if isinstance(raw_branches, dict):
raw_branches = list(raw_branches.values())
for branch in raw_branches:
branch_name = branch.get("name", "")
raw_path = branch.get("path", "")
branch_path = Path(raw_path)
if not branch_path.is_absolute():
branch_path = (registry_dir / branch_path).resolve()
if not branch_path.exists():
continue
entry_file = None
standard_entry = branch_path / "apps" / f"{branch_name.lower()}.py"
branch_entry = branch_path / "apps" / "branch.py"
if standard_entry.exists():
entry_file = standard_entry
elif branch_entry.exists():
entry_file = branch_entry
if entry_file:
branches.append({"name": branch_name, "path": str(branch_path), "entry_file": str(entry_file)})
except (json.JSONDecodeError, IOError):
logger.info("Cannot read registry %s", registry_path)
return branches
def discover_branches(include_private: bool = False) -> List[Dict[str, str]]:
"""
Discover all AIPass branches from AIPASS_REGISTRY.json
Discover all branches from AIPASS_REGISTRY.json and caller's project registry.
Args:
include_private: If False (default), excludes branches listed in
@@ -86,56 +135,23 @@ def discover_branches(include_private: bool = False) -> List[Dict[str, str]]:
Returns:
List of dicts with 'name', 'path', 'entry_file' keys
"""
branches = []
registry_path = _find_registry()
primary_path = _find_registry()
branches = _branches_from_registry(primary_path)
if not registry_path.exists():
return branches
seen_names = {b["name"].upper() for b in branches}
for caller_reg in _find_caller_registries():
if caller_reg.resolve() == primary_path.resolve():
continue
for b in _branches_from_registry(caller_reg):
if b["name"].upper() not in seen_names:
branches.append(b)
seen_names.add(b["name"].upper())
try:
with open(registry_path, "r", encoding="utf-8") as f:
registry_data = json.load(f)
if not include_private:
branches = [b for b in branches if not _is_branch_private(b["name"])]
registry_dir = registry_path.parent
raw_branches = registry_data.get("branches", [])
# Handle both list format and dict format (keyed by name)
if isinstance(raw_branches, dict):
raw_branches = list(raw_branches.values())
for branch in raw_branches:
branch_name = branch.get("name", "")
raw_path = branch.get("path", "")
branch_path = Path(raw_path)
# Resolve relative paths against registry location
if not branch_path.is_absolute():
branch_path = (registry_dir / branch_path).resolve()
if not branch_path.exists():
continue
# Find entry point: apps/{branch_name}.py or apps/branch.py
entry_file = None
standard_entry = branch_path / "apps" / f"{branch_name.lower()}.py"
branch_entry = branch_path / "apps" / "branch.py"
if standard_entry.exists():
entry_file = standard_entry
elif branch_entry.exists():
entry_file = branch_entry
if entry_file:
branches.append({"name": branch_name, "path": str(branch_path), "entry_file": str(entry_file)})
if not include_private:
branches = [b for b in branches if not _is_branch_private(b["name"])]
json_handler.log_operation("branches_discovered", {"count": len(branches)})
return sorted(branches, key=lambda x: x["name"])
except (json.JSONDecodeError, IOError):
logger.info("Cannot read registry for branch discovery")
return branches
json_handler.log_operation("branches_discovered", {"count": len(branches)})
return sorted(branches, key=lambda x: x["name"])
def check_internal_access(branch_name: str) -> bool:
src/aipass/seedgo/apps/modules/standards_audit.py
+6 -2
View File
@@ -200,8 +200,12 @@ def handle_command(command: str, args: List[str]) -> bool:
positional.append(arg)
if len(positional) >= 1:
pack_name = positional[0]
if len(positional) >= 2:
if positional[0].startswith("@"):
pack_name = "aipass"
specific_branch = normalize_branch_arg(positional[0])
else:
pack_name = positional[0]
if len(positional) >= 2 and specific_branch is None:
branch_arg = positional[1]
if not branch_arg.startswith("@"):
error(