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

feat(aipass-new): agent = full spawn citizen at src/<pkg>/<pkg> (FPLAN-0334). new_project routes through spawn_agent() against spawn's new project_agent template (branch prompt, inbox.json, birth certificate, trinity, house entry point, agent README) - hand-rolled scaffold retired. citizen_class=manager, seat path relative, registry walk stops at project root. Birth commit excludes .venv symlink + registry lock. Boundary verified 4/4 refusals incl ai_mail cross-project check

Browse Source
This commit is contained in:
AIOSAI
2026-07-17 22:56:19 -07:00
parent f6d31285ea
commit 193336967b
18 changed files with 572 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+20
View File
@@ -42,6 +42,26 @@ and README to the house pattern: 17/17 branches at 100%.
- **Root `.gitignore`**: `projects/*` ignored (each project is its own repo);
only the future catalog README stays trackable.
### Added (second pass — the agent becomes a real citizen)
- **`aipass new` agents are now spawn-issued full citizens** (FPLAN-0334): the
hand-rolled scaffold in the new_project handler is retired for a
`spawn_agent()` call against @spawn's new `project_agent` template — branch
prompt, structured mailbox, birth certificate, trinity trio, dashboard,
house-pattern entry point, and a branch-style README. One authority issues
citizens; project agents inherit template evolution for free.
- **Agent home = `src/<project>/<agent>/`**, mirroring the host's
`src/aipass/<branch>` layout (door-test ruling: the project root is never an
agent home). Seat paths are relative like host seats; the registry walk stops
at the first project registry. The first agent is the project's **manager**
(`citizen_class: manager` — its devpulse), named after the project.
- **Birth-commit hygiene**: the `.venv` symlink (absolute host path) and the
registry lock file are no longer tracked in new projects' birth commits.
- **Boundary verified live in all four directions**: host↔project email and
dispatch all refused — project→host lands on the ai_mail cross-project check
with its feedback-channel pointer, closing the leak found in the S319
prototype probes.
### Changed
- **aipass front door rebuilt**: `--help` now follows the house pattern with a
src/aipass/aipass/apps/handlers/new_project/__init__.py
+32 -273
View File
@@ -30,6 +30,7 @@ from datetime import date
from pathlib import Path
from aipass.prax import logger
from aipass.spawn import spawn_agent
TEMPLATES = ("empty", "python")
@@ -117,7 +118,7 @@ def _write_template(target: Path, name: str, template: str) -> list[str]:
created.append("README.md")
(target / ".gitignore").write_text(
"__pycache__/\n*.pyc\n.venv/\n.trinity/\n.ai_mail.local/\n*.local.json\n*.local/\nlogs/\n",
"__pycache__/\n*.pyc\n.venv\n.trinity/\n.ai_mail.local/\n*.local.json\n*.local/\nlogs/\n.*_REGISTRY.lock\n",
encoding="utf-8",
)
created.append(".gitignore")
@@ -246,281 +247,37 @@ def _git_init(target: Path, name: str, template: str) -> None:
)
# ── agent (passport + registry seat) ────────────────────────────────────
# ── agent (via @spawn) ──────────────────────────────────────────────────
def _entry_point_content(name: str) -> str:
"""Generate the entry point script for the newborn project agent."""
reg = _registry_name(name)
return (
f'"""\n'
f"{reg} — project agent\n"
f"\n"
f"Auto-discovery architecture:\n"
f"- Scans modules/ directory for .py files with handle_command()\n"
f"- Routes commands to discovered modules automatically\n"
f'"""\n'
f"\n"
f"import importlib\n"
f"import os\n"
f"import sys\n"
f"from pathlib import Path\n"
f"from typing import Any, List\n"
f"\n"
f"PROJECT_ROOT = str(Path(__file__).resolve().parent.parent)\n"
f"if PROJECT_ROOT not in sys.path:\n"
f" sys.path.insert(0, PROJECT_ROOT)\n"
f"\n"
f'os.environ.setdefault("AIPASS_BRANCH_NAME", "{name}")\n'
f"\n"
f"from aipass.cli.apps.modules import console # noqa: E402\n"
f"from aipass.prax import logger # noqa: E402\n"
f"\n"
f'MODULES_DIR = Path(__file__).parent / "modules"\n'
f"\n"
f"\n"
f"def discover_modules() -> List[Any]:\n"
f' """Auto-discover modules in modules/ directory."""\n'
f" modules = []\n"
f" if not MODULES_DIR.exists():\n"
f" return modules\n"
f' for file_path in MODULES_DIR.glob("*.py"):\n'
f' if file_path.name.startswith("_"):\n'
f" continue\n"
f" for prefix in (\n"
f' f"aipass.{name}.apps.modules.{{file_path.stem}}",\n'
f' f"apps.modules.{{file_path.stem}}",\n'
f" ):\n"
f" try:\n"
f" mod = importlib.import_module(prefix)\n"
f' if hasattr(mod, "handle_command"):\n'
f" modules.append(mod)\n"
f" break\n"
f" except ImportError:\n"
f" continue\n"
f" return modules\n"
f"\n"
f"\n"
f"def print_introspection() -> None:\n"
f' """Bare invocation — title, purpose, --help pointer."""\n'
f" console.print()\n"
f' console.print("[bold cyan]{reg} — Project Agent[/bold cyan]")\n'
f' console.print("[dim]Resident agent of the {name} project.[/dim]")\n'
f" console.print()\n"
f" modules = discover_modules()\n"
f" if modules:\n"
f' console.print("[yellow]Modules:[/yellow]")\n'
f" for m in modules:\n"
f' cmd = getattr(m, "COMMAND", m.__name__.split(".")[-1])\n'
f' desc = (m.__doc__ or "").strip().split("\\n")[0]\n'
f' console.print(f" [green]{{cmd:16}}[/green] [dim]{{desc}}[/dim]")\n'
f" console.print()\n"
f" console.print(\"[dim]Run 'drone @{name} --help' for usage[/dim]\")\n"
f" console.print()\n"
f"\n"
f"\n"
f"def print_help() -> None:\n"
f' """Full help — usage, commands, examples."""\n'
f" console.print()\n"
f' console.print("[bold cyan]{reg} — Project Agent[/bold cyan]")\n'
f" console.print()\n"
f' console.print("[yellow]Usage:[/yellow]")\n'
f' console.print(" [green]drone @{name}[/green] [dim]<command>[/dim]")\n'
f" console.print()\n"
f' console.print("[yellow]Commands:[/yellow]")\n'
f' console.print(" [green]hello[/green] [dim]Confirm the agent is alive[/dim]")\n'
f" console.print()\n"
f' console.print("[yellow]Examples:[/yellow]")\n'
f' console.print(" [green]drone @{name} hello[/green]")\n'
f" console.print()\n"
f"\n"
f"\n"
f"def route_command(command: str, args: List[str], modules: List[Any]) -> bool:\n"
f' """Route command to appropriate module."""\n'
f" for module in modules:\n"
f" try:\n"
f" if module.handle_command(command, args):\n"
f" return True\n"
f" except Exception as e:\n"
f' logger.error("[{reg}] Module error: %s", e)\n'
f" return False\n"
f"\n"
f"\n"
f"def main():\n"
f' """Main entry point."""\n'
f" modules = discover_modules()\n"
f" args = sys.argv[1:]\n"
f"\n"
f" if not args:\n"
f" print_introspection()\n"
f" return 0\n"
f' if args[0] in ("--help", "-h", "help"):\n'
f" print_help()\n"
f" return 0\n"
f' if args[0] == "hello":\n'
f' console.print(f"[cyan]{reg}[/cyan] here. '
f'Project agent, alive and ready.")\n'
f" return 0\n"
f"\n"
f" command = args[0]\n"
f" remaining = args[1:] if len(args) > 1 else []\n"
f" if route_command(command, remaining, modules):\n"
f" return 0\n"
f' console.print(f"Unknown command: {{command}}")\n'
f" return 1\n"
f"\n"
f"\n"
f'if __name__ == "__main__":\n'
f" sys.exit(main())\n"
)
def _agent_home(project_root: Path, name: str) -> Path:
"""Compute the agent home directory: src/<pkg>/<pkg>/."""
pkg = name.replace("-", "_").lower()
return project_root / "src" / pkg / pkg
def _local_json_content(name: str) -> dict:
"""Generate initial local.json for the newborn agent."""
reg = _registry_name(name)
today = date.today().isoformat()
return {
"document_metadata": {
"document_type": "session_history",
"document_name": f"{reg}.LOCAL",
"version": "1.0.0",
"schema_version": "3.0.0",
"created": today,
"last_updated": today,
"managed_by": reg,
"tags": ["session_tracking", "work_log", reg],
},
"todos": [],
"key_learnings": [],
"sessions": [],
}
def _spawn_project_agent(project_root: Path, name: str) -> dict:
"""Create the project agent via spawn_agent().
def _observations_json_content(name: str) -> dict:
"""Generate initial observations.json for the newborn agent."""
reg = _registry_name(name)
today = date.today().isoformat()
return {
"document_metadata": {
"document_type": "observations",
"document_name": f"{reg}.OBSERVATIONS",
"version": "1.0.0",
"created": today,
"last_updated": today,
"managed_by": reg,
},
"observations": [],
}
def _write_agent(target: Path, name: str, registry_id: str, registry_file: str) -> list[str]:
"""Create the full framework agent: entry point, structure, identity, registry seat.
Models the credential linkage from the proven proto:
``registry.metadata.id == passport.citizenship.registry_id``.
Agent lives at src/<pkg>/<pkg>/ inside the project. Spawn discovers
the project-local registry (minted earlier by _write_registry) by
walking up from the agent home to the project root.
"""
today = date.today().isoformat()
reg = _registry_name(name)
created: list[str] = []
# ── apps/ entry point + skeleton ─────────────────────────────────
apps_dir = target / "apps"
apps_dir.mkdir(exist_ok=True)
(apps_dir / "__init__.py").write_text("")
created.append("apps/__init__.py")
entry_file = apps_dir / f"{name}.py"
entry_file.write_text(_entry_point_content(name), encoding="utf-8")
created.append(f"apps/{name}.py")
for sub in ("modules", "handlers"):
d = apps_dir / sub
d.mkdir(exist_ok=True)
(d / "__init__.py").write_text("")
created.append(f"apps/{sub}/__init__.py")
# ── .trinity/ identity ───────────────────────────────────────────
trinity = target / ".trinity"
trinity.mkdir(exist_ok=True)
passport = {
"document_metadata": {
"document_type": "branch_identity",
"document_name": f"{reg}.PASSPORT",
"version": "1.0.0",
"schema_version": "1.0.0",
"created": today,
"last_updated": today,
"managed_by": reg,
"tags": ["identity", "passport", "branch_profile"],
},
"branch_info": {
"branch_name": reg,
"alias": "",
"path": ".",
"module": name,
"created": today,
"git_branch": "main",
},
"identity": {
"citizen_class": "worker",
"role": "project_agent",
"purpose": f"Resident agent of the {name} project.",
"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_id": registry_id,
"communications": True,
"memory": True,
},
}
for fname, data in (
("passport.json", passport),
("local.json", _local_json_content(name)),
("observations.json", _observations_json_content(name)),
):
(trinity / fname).write_text(
json.dumps(data, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
created.append(f".trinity/{fname}")
# ── mailbox + logs ───────────────────────────────────────────────
(target / ".ai_mail.local").mkdir(exist_ok=True)
(target / "logs").mkdir(exist_ok=True)
# ── registry seat ────────────────────────────────────────────────
reg_path = target / registry_file
reg_data = json.loads(reg_path.read_text(encoding="utf-8"))
reg_data["metadata"]["total_branches"] = 1
reg_data["branches"] = [
{
"name": reg,
"registry_id": registry_id,
"path": ".",
"profile": "AIPass Project",
"description": f"Resident agent of the {name} project.",
"email": f"@{name}",
"status": "active",
"created": today,
"last_active": today,
"owner": True,
},
]
reg_path.write_text(
json.dumps(reg_data, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
home = _agent_home(project_root, name)
result = spawn_agent(
target_path=str(home),
role="project_agent",
purpose=f"Resident agent of the {name} project.",
citizen_class="project_agent",
)
created.append(registry_file)
logger.info("[aipass new] full agent created: %s (%d files)", reg, len(created))
return created
if not result.get("success"):
raise RuntimeError(f"spawn_agent failed: {result.get('error', 'unknown')}")
logger.info(
"[aipass new] agent spawned via @spawn: %s (%d files)",
result["branch_name"],
result["files_copied"],
)
return result
# ── public API ──────────────────────────────────────────────────────────
@@ -565,9 +322,9 @@ def create_project(
template_files = _write_template(target, name, template)
scaffold_files = _scaffold_aipass(target, name)
agent_files: list[str] = []
spawn_result = None
if not no_agent:
agent_files = _write_agent(target, name, registry_id, registry_file)
spawn_result = _spawn_project_agent(target, name)
_git_init(target, name, template)
logger.info("[aipass new] git repo initialized with birth commit")
@@ -579,8 +336,10 @@ def create_project(
"host": str(host),
"registry_id": registry_id,
"registry_file": registry_file,
"files": template_files + scaffold_files + agent_files,
"agent_created": not no_agent,
"files": template_files + scaffold_files,
"agent_created": spawn_result is not None,
"agent_home": str(_agent_home(target, name)) if spawn_result else None,
"spawn_result": spawn_result,
}
except Exception:
if target.exists():
src/aipass/aipass/apps/modules/new_project.py
+5 -2
View File
@@ -193,8 +193,11 @@ def handle_command(command: str, args: list[str]) -> bool:
console.print(" [dim]Agent:[/dim] skipped (--no-agent)")
console.print()
console.print("[yellow]Next steps:[/yellow]")
console.print(f" [cyan]cd {result['target']}[/cyan]")
console.print(" [cyan]claude[/cyan] [dim]# meet your project agent[/dim]")
if result["agent_created"]:
console.print(f" [cyan]cd {result['agent_home']}[/cyan]")
console.print(" [cyan]claude[/cyan] [dim]# meet your project agent[/dim]")
else:
console.print(f" [cyan]cd {result['target']}[/cyan]")
console.print()
json_handler.log_operation(
src/aipass/aipass/tests/test_new_project.py
+126 -84
View File
@@ -19,9 +19,10 @@ from unittest.mock import patch
import pytest # pyright: ignore[reportMissingImports]
from aipass.aipass.apps.handlers.new_project import (
_agent_home,
_registry_name,
_spawn_project_agent,
_validate_name,
_write_agent,
_write_registry,
_write_template,
create_project,
@@ -126,6 +127,10 @@ def test_write_template_empty(tmp_path):
assert (tmp_path / ".gitignore").exists()
assert not (tmp_path / "pyproject.toml").exists()
assert not (tmp_path / "src").exists()
gitignore = (tmp_path / ".gitignore").read_text()
assert ".venv\n" in gitignore
assert ".venv/\n" not in gitignore
assert "*_REGISTRY.lock" in gitignore
# ---------------------------------------------------------------------------
@@ -305,83 +310,74 @@ def test_create_project_registry_before_scaffold(host_env, monkeypatch):
# ---------------------------------------------------------------------------
# _write_agent
# _spawn_project_agent (delegates to spawn_agent)
# ---------------------------------------------------------------------------
def test_write_agent_creates_passport(tmp_path):
"""Passport is written with correct credential linkage."""
rid, fname = _write_registry(tmp_path, "agentapp")
files = _write_agent(tmp_path, "agentapp", rid, fname)
assert ".trinity/passport.json" in files
passport = json.loads((tmp_path / ".trinity" / "passport.json").read_text())
assert passport["citizenship"]["registry_id"] == rid
assert passport["branch_info"]["branch_name"] == "AGENTAPP"
assert passport["identity"]["role"] == "project_agent"
_SPAWN_SUCCESS = {
"success": True,
"branch_name": "DEMO",
"path": "/tmp/demo",
"files_copied": 12,
"registry_updated": True,
"validation_issues": [],
}
def test_write_agent_seats_in_registry(tmp_path):
"""Registry total_branches updated and branch entry added."""
rid, fname = _write_registry(tmp_path, "seated")
_write_agent(tmp_path, "seated", rid, fname)
reg = json.loads((tmp_path / fname).read_text())
assert reg["metadata"]["total_branches"] == 1
assert len(reg["branches"]) == 1
assert reg["branches"][0]["registry_id"] == rid
assert reg["branches"][0]["name"] == "SEATED"
def test_agent_home_simple():
"""Agent home is src/<pkg>/<pkg>/."""
from pathlib import Path
home = _agent_home(Path("/proj"), "demo")
assert home == Path("/proj/src/demo/demo")
def test_write_agent_credential_linkage(tmp_path):
"""registry.metadata.id == passport.citizenship.registry_id."""
rid, fname = _write_registry(tmp_path, "linked")
_write_agent(tmp_path, "linked", rid, fname)
reg = json.loads((tmp_path / fname).read_text())
passport = json.loads((tmp_path / ".trinity" / "passport.json").read_text())
assert reg["metadata"]["id"] == passport["citizenship"]["registry_id"]
def test_agent_home_hyphenated():
"""Hyphens normalized to underscores, matching python template."""
from pathlib import Path
home = _agent_home(Path("/proj"), "my-app")
assert home == Path("/proj/src/my_app/my_app")
def test_write_agent_creates_entry_point(tmp_path):
"""Entry point apps/<name>.py is created and contains hello handler."""
rid, fname = _write_registry(tmp_path, "myagent")
files = _write_agent(tmp_path, "myagent", rid, fname)
assert "apps/myagent.py" in files
content = (tmp_path / "apps" / "myagent.py").read_text()
assert "def main()" in content
assert "def print_introspection()" in content
assert "def print_help()" in content
assert '"hello"' in content
def test_spawn_project_agent_calls_spawn(tmp_path):
"""Calls spawn_agent with correct citizen_class, purpose, and agent_home path."""
with patch(
"aipass.aipass.apps.handlers.new_project.spawn_agent",
return_value=_SPAWN_SUCCESS,
) as mock_spawn:
result = _spawn_project_agent(tmp_path, "demo")
expected_home = str(tmp_path / "src" / "demo" / "demo")
mock_spawn.assert_called_once_with(
target_path=expected_home,
role="project_agent",
purpose="Resident agent of the demo project.",
citizen_class="project_agent",
)
assert result["success"] is True
assert result["branch_name"] == "DEMO"
def test_write_agent_creates_apps_skeleton(tmp_path):
"""Apps skeleton: __init__.py, modules/, handlers/."""
rid, fname = _write_registry(tmp_path, "skel")
files = _write_agent(tmp_path, "skel", rid, fname)
assert "apps/__init__.py" in files
assert "apps/modules/__init__.py" in files
assert "apps/handlers/__init__.py" in files
assert (tmp_path / "apps" / "modules" / "__init__.py").exists()
assert (tmp_path / "apps" / "handlers" / "__init__.py").exists()
def test_spawn_project_agent_raises_on_failure(tmp_path):
"""Raises RuntimeError when spawn_agent returns success=False."""
with (
patch(
"aipass.aipass.apps.handlers.new_project.spawn_agent",
return_value={"success": False, "error": "template missing"},
),
pytest.raises(RuntimeError, match="spawn_agent failed.*template missing"),
):
_spawn_project_agent(tmp_path, "broken")
def test_write_agent_creates_trinity_full_set(tmp_path):
"""Full .trinity/ set: passport, local.json, observations.json."""
rid, fname = _write_registry(tmp_path, "fullset")
files = _write_agent(tmp_path, "fullset", rid, fname)
assert ".trinity/passport.json" in files
assert ".trinity/local.json" in files
assert ".trinity/observations.json" in files
local = json.loads((tmp_path / ".trinity" / "local.json").read_text())
assert local["document_metadata"]["managed_by"] == "FULLSET"
obs = json.loads((tmp_path / ".trinity" / "observations.json").read_text())
assert obs["document_metadata"]["managed_by"] == "FULLSET"
def test_write_agent_creates_mailbox_and_logs(tmp_path):
"""Mailbox and logs directories created."""
rid, fname = _write_registry(tmp_path, "dirs")
_write_agent(tmp_path, "dirs", rid, fname)
assert (tmp_path / ".ai_mail.local").is_dir()
assert (tmp_path / "logs").is_dir()
def test_spawn_project_agent_returns_spawn_result(tmp_path):
"""Returns the full result dict from spawn_agent."""
with patch(
"aipass.aipass.apps.handlers.new_project.spawn_agent",
return_value={**_SPAWN_SUCCESS, "citizen_number": 1},
):
result = _spawn_project_agent(tmp_path, "demo")
assert result["files_copied"] == 12
assert result["citizen_number"] == 1
# ---------------------------------------------------------------------------
@@ -390,7 +386,42 @@ def test_write_agent_creates_mailbox_and_logs(tmp_path):
def test_create_project_with_agent(host_env, monkeypatch):
"""WITH-agent path: full framework agent created."""
"""WITH-agent path: spawn_agent called, result propagated."""
monkeypatch.chdir(host_env)
spawn_ok = {
"success": True,
"branch_name": "WITHAGENT",
"path": str(host_env / "projects" / "withagent"),
"files_copied": 15,
"registry_updated": True,
"validation_issues": [],
}
with (
patch("subprocess.run", side_effect=_mock_git_run),
patch(
"aipass.aipass.apps.handlers.init.bootstrap._detect_aipass_home",
return_value=None,
),
patch("aipass.aipass.apps.handlers.init.bootstrap._enroll_project"),
patch(
"aipass.aipass.apps.handlers.new_project.spawn_agent",
return_value=spawn_ok,
) as mock_spawn,
):
result = create_project("withagent", template="empty", no_agent=False)
assert result["agent_created"] is True
assert result["spawn_result"] == spawn_ok
expected_home = str(host_env / "projects" / "withagent" / "src" / "withagent" / "withagent")
assert result["agent_home"] == expected_home
mock_spawn.assert_called_once()
call_kwargs = mock_spawn.call_args[1]
assert call_kwargs["target_path"] == expected_home
assert call_kwargs["citizen_class"] == "project_agent"
def test_create_project_spawn_failure_cleans_up(host_env, monkeypatch):
"""spawn_agent failure triggers cleanup — no partial project left."""
monkeypatch.chdir(host_env)
with (
patch("subprocess.run", side_effect=_mock_git_run),
@@ -399,24 +430,34 @@ def test_create_project_with_agent(host_env, monkeypatch):
return_value=None,
),
patch("aipass.aipass.apps.handlers.init.bootstrap._enroll_project"),
patch(
"aipass.aipass.apps.handlers.new_project.spawn_agent",
return_value={"success": False, "error": "template missing"},
),
pytest.raises(RuntimeError, match="spawn_agent failed"),
):
result = create_project("withagent", template="empty", no_agent=False)
create_project("failspawn", template="empty", no_agent=False)
assert result["agent_created"] is True
target = Path(result["target"])
assert (target / ".trinity" / "passport.json").exists()
assert (target / ".trinity" / "local.json").exists()
assert (target / ".trinity" / "observations.json").exists()
assert (target / "apps" / "withagent.py").exists()
assert (target / "apps" / "modules" / "__init__.py").exists()
assert (target / "apps" / "handlers" / "__init__.py").exists()
assert (target / ".ai_mail.local").is_dir()
assert (target / "logs").is_dir()
passport = json.loads((target / ".trinity" / "passport.json").read_text())
assert passport["citizenship"]["registry_id"] == result["registry_id"]
reg = json.loads((target / result["registry_file"]).read_text())
assert reg["metadata"]["total_branches"] == 1
assert reg["branches"][0]["registry_id"] == result["registry_id"]
assert not (host_env / "projects" / "failspawn").exists()
def test_create_project_no_agent_next_steps(host_env, monkeypatch):
"""no_agent output omits 'meet your project agent' line."""
from aipass.aipass.apps.modules.new_project import handle_command
monkeypatch.chdir(host_env)
with (
patch("subprocess.run", side_effect=_mock_git_run),
patch(
"aipass.aipass.apps.handlers.init.bootstrap._detect_aipass_home",
return_value=None,
),
patch("aipass.aipass.apps.handlers.init.bootstrap._enroll_project"),
patch("aipass.aipass.apps.modules.new_project.console") as mock_con,
):
handle_command("new", ["cosmtest", "--template", "empty", "--no-agent"])
printed = " ".join(str(a) for call in mock_con.print.call_args_list for a in call[0])
assert "meet your project agent" not in printed
def test_create_project_no_agent_flag(host_env, monkeypatch):
@@ -433,8 +474,9 @@ def test_create_project_no_agent_flag(host_env, monkeypatch):
result = create_project("noagent", template="empty", no_agent=True)
assert result["agent_created"] is False
assert result["agent_home"] is None
target = Path(result["target"])
assert not (target / ".trinity").exists()
assert not (target / "src" / "noagent" / "noagent").exists()
reg = json.loads((target / result["registry_file"]).read_text())
assert reg["metadata"]["total_branches"] == 0
src/aipass/spawn/apps/handlers/class_registry.py
+5
View File
@@ -27,6 +27,11 @@ CITIZEN_CLASSES = {
"description": "Full 3-layer branch with apps/, modules/, handlers/",
"default": True,
},
"project_agent": {
"template_dir": "project_agent",
"description": "Project-root resident agent (manager class, collision-safe)",
"default": False,
},
}
# The default class when none is specified
src/aipass/spawn/apps/handlers/placeholders.py
+6 -5
View File
@@ -41,11 +41,12 @@ def build_replacements_dict(target_dir, branch_name, **overrides):
lower = branch_name.lower().replace("-", "_")
now = datetime.now()
registry_id = ""
registry_path = find_registry(start_path=Path(target_dir).parent)
if registry_path.exists():
data = json.loads(registry_path.read_text(encoding="utf-8"))
registry_id = data.get("metadata", {}).get("id", "")
registry_id = overrides.get("registry_id", "")
if not registry_id:
registry_path = find_registry(start_path=Path(target_dir).parent)
if registry_path.exists():
data = json.loads(registry_path.read_text(encoding="utf-8"))
registry_id = data.get("metadata", {}).get("id", "")
replacements = {
"BRANCHNAME": upper,
src/aipass/spawn/apps/modules/core.py
+8
View File
@@ -260,6 +260,13 @@ def _spawn_agent(
reg_path = _find_project_registry(target)
citizen_number = get_next_citizen_number(reg_path)
# Read registry_id from the resolved registry for credential linkage
resolved_registry_id = ""
if reg_path.exists():
reg_data = json_handler.read_json(reg_path)
if reg_data:
resolved_registry_id = reg_data.get("metadata", {}).get("id", "")
# Build placeholder replacements
meta_tabs = _load_meta_tabs()
replacements = build_replacements_dict(
@@ -272,6 +279,7 @@ def _spawn_agent(
citizen_number=citizen_number,
citizen_class=citizen_class,
meta_tabs=meta_tabs,
registry_id=resolved_registry_id,
)
# Step 1: Copy template with placeholder replacement in content
src/aipass/spawn/templates/project_agent/.aipass/aipass_local_prompt.md
+35
View File
@@ -0,0 +1,35 @@
# {{BRANCHNAME}} — Branch Prompt
*Injected every turn. Breadcrumbs only — details in README, --help, .trinity/ memories.*
## Identity
You are {{BRANCHNAME}} — resident project agent and manager of this project.
## What I Do
- Manage this project as its resident agent (citizen_class: manager)
- Route commands to discovered modules
- Coordinate project work and maintain project context
## Key Commands
```
drone @{{BRANCH}} hello # Confirm agent is alive
drone @{{BRANCH}} --help # Full help text
drone @{{BRANCH}} # Show connected modules
```
## Architecture
```
apps/
├── {{BRANCH}}.py # Entry point
├── modules/ # Business logic (auto-discovered)
└── handlers/ # Implementation details
```
## Integration
- **Depends on:** @prax for logging, @cli for console output
- **Serves:** Project users — routes commands, manages context
src/aipass/spawn/templates/project_agent/.trinity/local.json
+35
View File
@@ -0,0 +1,35 @@
{
"document_metadata": {
"document_type": "session_history",
"document_name": "{{BRANCHNAME}}.LOCAL",
"version": "2.0.0",
"schema_version": "3.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": [
"session_tracking",
"work_log",
"{{BRANCHNAME}}"
],
"_usage": "Automated file — add entries within your sections, newest on top. Rollover auto-archives sessions/key_learnings (+ observations.json) to @memory; todos[] are OPERATIONAL and NEVER rolled — prune done ones by hand at /prep. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"last_health_check": "{{DATE}}"
}
},
"todos_meta": "{{TODOS_META}}",
"todos": [],
"key_learnings_meta": "{{KEY_LEARNINGS_META}}",
"key_learnings": [],
"sessions_meta": "{{SESSIONS_META}}",
"sessions": [
{
"number": 1,
"date": "{{DATE}}",
"summary": "Branch initialized - {{BRANCHNAME}} created by aipass new.",
"status": "completed",
"tags": []
}
]
}
src/aipass/spawn/templates/project_agent/.trinity/observations.json
+34
View File
@@ -0,0 +1,34 @@
{
"document_metadata": {
"document_type": "collaboration_patterns",
"document_name": "{{BRANCHNAME}}.OBSERVATIONS",
"version": "1.0.0",
"schema_version": "3.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": [
"collaboration",
"patterns",
"{{BRANCHNAME}}"
],
"_usage": "Automated file — add entries within your sections, newest on top. Rollover auto-archives sessions/key_learnings (+ observations.json) to @memory; todos[] are OPERATIONAL and NEVER rolled — prune done ones by hand at /prep. Limits live in @memory's memory.config.json.",
"status": {
"health": "healthy",
"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_meta": "{{OBSERVATIONS_META}}",
"observations": [
{
"number": 1,
"date": "{{DATE}}",
"note": "Branch initialized. Ready to begin capturing collaboration patterns.",
"tags": []
}
]
}
src/aipass/spawn/templates/project_agent/.trinity/passport.json
+38
View File
@@ -0,0 +1,38 @@
{
"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}}",
"git_branch": "main"
},
"identity": {
"citizen_class": "manager",
"role": "{{ROLE}}",
"purpose": "{{PURPOSE_BRIEF}}",
"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_id": "{{REGISTRY_ID}}",
"communications": true,
"memory": true
}
}
src/aipass/spawn/templates/project_agent/README.md
+55
View File
@@ -0,0 +1,55 @@
# {{BRANCHNAME}}
**Resident project agent and manager.**
**Module:** `{{MODULE}}` | **Class:** manager | **Created:** {{DATE}}
---
## What I Do
- Manage this project as its resident agent (citizen_class: manager)
- Route commands to discovered modules
- Coordinate project work and maintain project context
---
## Commands
All commands run through `drone @{{BRANCH}} <command>`.
```bash
drone @{{BRANCH}} # Show connected modules
drone @{{BRANCH}} hello # Confirm the agent is alive
drone @{{BRANCH}} --help # Full help text
```
---
## Architecture
```
apps/
├── {{BRANCH}}.py # Entry point — CLI routing, introspection, help
├── modules/ # Business logic (auto-discovered)
└── handlers/ # Implementation details
```
### Three-Layer Design
1. **Entry point** (`{{BRANCH}}.py`) — Routes CLI commands, never imports handlers directly
2. **Modules** (`modules/`) — Business logic coordinators, parse arguments, delegate to handlers
3. **Handlers** (`handlers/`) — Implementation details, pure functions where possible
---
## Integration
### Depends On
- **aipass.prax** — Logging via `system_logger`
- **aipass.cli** — Console output (header, error, warning)
---
*Created by `aipass new` via spawn project_agent template.*
src/aipass/spawn/templates/project_agent/apps/__init__.py
View File
src/aipass/spawn/templates/project_agent/apps/handlers/README.md
+5
View File
@@ -0,0 +1,5 @@
# Handlers
Implementation details for `{{BRANCHNAME}}`. Pure functions where possible.
Handlers do the actual work — file I/O, data transforms, external calls. They never import from modules/ (no circular deps).
src/aipass/spawn/templates/project_agent/apps/handlers/__init__.py
View File
src/aipass/spawn/templates/project_agent/apps/modules/README.md
+5
View File
@@ -0,0 +1,5 @@
# Modules
Business logic for `{{BRANCHNAME}}`. One module per command.
Modules orchestrate work by calling handlers. They are the public API of the branch — drone routes commands here.
src/aipass/spawn/templates/project_agent/apps/modules/__init__.py
View File
src/aipass/spawn/templates/project_agent/apps/{{BRANCH}}.py
+163
View File
@@ -0,0 +1,163 @@
"""
{{BRANCHNAME}} — Project Agent
Auto-discovery architecture:
- Scans modules/ directory for .py files with handle_command()
- Routes commands to discovered modules automatically
"""
import importlib
import os
import sys
from pathlib import Path
from typing import Any, List
PROJECT_ROOT = str(Path(__file__).resolve().parent.parent)
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
os.environ.setdefault("AIPASS_BRANCH_NAME", "{{BRANCH}}")
from aipass.cli.apps.modules import console # noqa: E402
from aipass.prax import logger # noqa: E402
MODULES_DIR = Path(__file__).parent / "modules"
def _module_import_path(stem: str) -> str:
for prefix in (
f"aipass.{{BRANCH}}.apps.modules.{stem}",
f"apps.modules.{stem}",
):
try:
importlib.import_module(prefix)
return prefix
except ImportError:
continue
return f"apps.modules.{stem}"
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 = _module_import_path(file_path.stem)
try:
module = importlib.import_module(module_name)
if hasattr(module, "handle_command"):
modules.append(module)
except Exception as e:
logger.error("[{{BRANCHNAME}}] Failed to load module %s: %s", 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("[{{BRANCHNAME}}] Module %s error: %s", module.__name__, e)
return False
def print_introspection() -> None:
"""Bare invocation — title, purpose, module list, help pointer."""
console.print()
console.print("[bold cyan]{{BRANCHNAME}} — Project Agent[/bold cyan]")
console.print("[dim]{{PURPOSE_BRIEF}}[/dim]")
console.print()
modules = discover_modules()
if modules:
console.print("[yellow]Modules:[/yellow]")
for m in modules:
cmd = getattr(m, "COMMAND", m.__name__.split(".")[-1])
desc = (m.__doc__ or "").strip().split("\n")[0]
console.print(f" [green]{cmd:16}[/green] [dim]{desc}[/dim]")
console.print()
console.print("[dim]Run 'drone @{{BRANCH}} --help' for usage information[/dim]")
console.print()
def print_help() -> None:
"""Full help — usage, commands, examples."""
console.print()
console.print("[bold cyan]{{BRANCHNAME}} — Project Agent[/bold cyan]")
console.print()
console.print("[dim]{{PURPOSE_BRIEF}}[/dim]")
console.print()
console.print("[yellow]Usage:[/yellow]")
console.print(" [green]drone @{{BRANCH}}[/green] [dim]<command>[/dim]")
console.print()
console.print("[yellow]Commands:[/yellow]")
console.print(" [green]hello[/green] [dim]Confirm the agent is alive[/dim]")
modules = discover_modules()
for m in modules:
cmd = getattr(m, "COMMAND", m.__name__.split(".")[-1])
desc = (m.__doc__ or "").strip().split("\n")[0]
console.print(f" [green]{cmd:10}[/green] [dim]{desc}[/dim]")
console.print()
console.print("[yellow]Examples:[/yellow]")
console.print()
console.print(" [dim]# Check the agent is alive[/dim]")
console.print(" [green]drone @{{BRANCH}} hello[/green]")
console.print()
console.print(" [dim]# Show connected modules[/dim]")
console.print(" [green]drone @{{BRANCH}}[/green]")
console.print()
def main():
"""Main entry point."""
args = sys.argv[1:]
if not args:
print_introspection()
return 0
if args[0] in ("--help", "-h", "help"):
print_help()
return 0
if args[0] == "hello":
console.print("[cyan]{{BRANCHNAME}}[/cyan] here. Project agent, alive and ready.")
return 0
command = args[0]
remaining = args[1:] if len(args) > 1 else []
modules = discover_modules()
if remaining and remaining[0] in ("--help", "-h"):
remaining = ["--help"]
if route_command(command, remaining, modules):
return 0
console.print(f"Unknown command: {command}")
console.print("[dim]Run 'drone @{{BRANCH}} --help' for usage information[/dim]")
return 1
if __name__ == "__main__":
if sys.platform == "win32":
os.environ.setdefault("PYTHONUTF8", "1")
for _stream in (sys.stdout, sys.stderr):
_reconfigure = getattr(_stream, "reconfigure", None)
if _reconfigure is not None:
_reconfigure(encoding="utf-8", errors="replace")
try:
sys.exit(main())
except KeyboardInterrupt:
logger.warning("{{BRANCHNAME}} interrupted by user (KeyboardInterrupt)")
console.print("\n\nOperation cancelled by user")
sys.exit(0)
except Exception as e:
logger.error("{{BRANCHNAME}} entry point error: %s", e, exc_info=True)
console.print(f"\nError: {e}")
sys.exit(1)