Merge pull request #421 from AIOSAI/system/devpulse-chore-gitignore-heraldmd-and-remove-from-readme
feat(system): chore: gitignore HERALD.md and remove from README
This commit is contained in:
@@ -1,6 +1,9 @@
|
||||
# Virtual environment
|
||||
.venv/
|
||||
|
||||
# Herald (session history — parked)
|
||||
HERALD.md
|
||||
|
||||
# Python
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
[](HERALD.md)
|
||||
[](#project-status)
|
||||
[](pyproject.toml)
|
||||
[](LICENSE)
|
||||
[](https://pypi.org/project/aipass/)
|
||||
@@ -232,8 +232,6 @@ setup.sh auto-detects which CLIs are installed and configures hooks for each.
|
||||
|
||||
Each agent documents its own operational status in its branch README — what works, what doesn't, and why.
|
||||
|
||||
For detailed session history, see [HERALD.md](HERALD.md).
|
||||
|
||||
---
|
||||
|
||||
## Requirements
|
||||
|
||||
@@ -46,6 +46,25 @@ def _find_repo_root() -> Path:
|
||||
_REPO_ROOT = _find_repo_root()
|
||||
|
||||
|
||||
def _find_caller_registries() -> List[Path]:
|
||||
"""Find project registries reachable from CWD (for external projects)."""
|
||||
import os
|
||||
|
||||
caller_cwd = (
|
||||
Path(os.environ.get("AIPASS_CALLER_CWD", "")).resolve() if os.environ.get("AIPASS_CALLER_CWD") else Path.cwd()
|
||||
)
|
||||
aipass_registry = (_REPO_ROOT / "AIPASS_REGISTRY.json").resolve()
|
||||
|
||||
registries = []
|
||||
for parent in [caller_cwd] + list(caller_cwd.parents):
|
||||
for reg in parent.glob("*_REGISTRY.json"):
|
||||
if reg.resolve() != aipass_registry:
|
||||
registries.append(reg)
|
||||
if registries:
|
||||
break
|
||||
return registries
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DATA STRUCTURES
|
||||
# =============================================================================
|
||||
@@ -74,17 +93,8 @@ class RolloverTrigger:
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def _read_registry() -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Read AIPASS_REGISTRY.json from repo root.
|
||||
|
||||
Registry paths are relative — resolved against repo root.
|
||||
|
||||
Returns:
|
||||
List of branch dictionaries with absolute paths
|
||||
"""
|
||||
registry_path = _REPO_ROOT / "AIPASS_REGISTRY.json"
|
||||
|
||||
def _read_single_registry(registry_path: Path, root: Path) -> List[Dict[str, Any]]:
|
||||
"""Read branches from a single registry file, resolving paths against root."""
|
||||
if not registry_path.exists():
|
||||
return []
|
||||
|
||||
@@ -93,20 +103,40 @@ def _read_registry() -> List[Dict[str, Any]]:
|
||||
data = json.load(f)
|
||||
branches = data.get("branches", [])
|
||||
|
||||
# Resolve relative paths against repo root
|
||||
for branch in branches:
|
||||
raw_path = branch.get("path", "")
|
||||
resolved = Path(raw_path)
|
||||
if not resolved.is_absolute():
|
||||
resolved = _REPO_ROOT / raw_path
|
||||
resolved = root / raw_path
|
||||
branch["path"] = str(resolved)
|
||||
|
||||
return branches
|
||||
except Exception as e:
|
||||
logger.warning(f"[detector] Failed to read registry: {e}")
|
||||
logger.warning(f"[detector] Failed to read registry {registry_path}: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def _read_registry() -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Read all project registries (AIPass + external projects from caller CWD).
|
||||
|
||||
Registry paths are relative — resolved against their respective project root.
|
||||
|
||||
Returns:
|
||||
List of branch dictionaries with absolute paths
|
||||
"""
|
||||
branches = _read_single_registry(_REPO_ROOT / "AIPASS_REGISTRY.json", _REPO_ROOT)
|
||||
|
||||
seen_paths = {b.get("path") for b in branches}
|
||||
for reg_path in _find_caller_registries():
|
||||
for branch in _read_single_registry(reg_path, reg_path.parent):
|
||||
if branch.get("path") not in seen_paths:
|
||||
branches.append(branch)
|
||||
seen_paths.add(branch.get("path"))
|
||||
|
||||
return branches
|
||||
|
||||
|
||||
def _get_memory_file_path(branch: Dict, memory_type: str) -> Path | None:
|
||||
"""
|
||||
Get path to memory file for branch.
|
||||
|
||||
@@ -429,20 +429,10 @@ def _find_repo_root() -> Path:
|
||||
return Path.cwd()
|
||||
|
||||
|
||||
def _get_branch_paths() -> list[Path]:
|
||||
"""
|
||||
Get all branch paths from AIPASS_REGISTRY.json (silent - no logging).
|
||||
|
||||
Registry paths are relative — resolved against repo root.
|
||||
|
||||
Returns:
|
||||
List of Path objects for each branch
|
||||
"""
|
||||
def _paths_from_registry(registry_path: Path, root: Path) -> list[Path]:
|
||||
"""Read branch paths from a single registry file."""
|
||||
import json
|
||||
|
||||
repo_root = _find_repo_root()
|
||||
registry_path = repo_root / "AIPASS_REGISTRY.json"
|
||||
|
||||
if not registry_path.exists():
|
||||
return []
|
||||
|
||||
@@ -456,16 +446,49 @@ def _get_branch_paths() -> list[Path]:
|
||||
raw_path = branch.get("path", "")
|
||||
branch_path = Path(raw_path)
|
||||
if not branch_path.is_absolute():
|
||||
branch_path = repo_root / raw_path
|
||||
branch_path = root / raw_path
|
||||
if branch_path.exists():
|
||||
paths.append(branch_path)
|
||||
|
||||
return paths
|
||||
except Exception as e:
|
||||
logger.warning(f"[memory_watcher] Failed to read branch paths from registry: {e}")
|
||||
logger.warning(f"[memory_watcher] Failed to read registry {registry_path}: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def _get_branch_paths() -> list[Path]:
|
||||
"""
|
||||
Get all branch paths from AIPass registry + external project registries.
|
||||
|
||||
Returns:
|
||||
List of Path objects for each branch
|
||||
"""
|
||||
import os
|
||||
|
||||
repo_root = _find_repo_root()
|
||||
paths = _paths_from_registry(repo_root / "AIPASS_REGISTRY.json", repo_root)
|
||||
seen = {p.resolve() for p in paths}
|
||||
|
||||
caller_cwd = (
|
||||
Path(os.environ.get("AIPASS_CALLER_CWD", "")).resolve() if os.environ.get("AIPASS_CALLER_CWD") else Path.cwd()
|
||||
)
|
||||
aipass_registry = (repo_root / "AIPASS_REGISTRY.json").resolve()
|
||||
|
||||
found_external = False
|
||||
for parent in [caller_cwd] + list(caller_cwd.parents):
|
||||
for reg in parent.glob("*_REGISTRY.json"):
|
||||
if reg.resolve() != aipass_registry:
|
||||
found_external = True
|
||||
for p in _paths_from_registry(reg, reg.parent):
|
||||
if p.resolve() not in seen:
|
||||
paths.append(p)
|
||||
seen.add(p.resolve())
|
||||
if found_external:
|
||||
break
|
||||
|
||||
return paths
|
||||
|
||||
|
||||
def _is_memory_file(file_path: Path) -> bool:
|
||||
"""
|
||||
Check if file is a memory file in .trinity/ (local.json or observations.json)
|
||||
|
||||
@@ -316,6 +316,7 @@ class TestReadRegistry:
|
||||
|
||||
# Point _REPO_ROOT at our tmp_path
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
|
||||
monkeypatch.setattr(detector, "_find_caller_registries", lambda: [])
|
||||
|
||||
result = detector._read_registry()
|
||||
|
||||
@@ -329,6 +330,7 @@ class TestReadRegistry:
|
||||
|
||||
# Point _REPO_ROOT at a directory with no registry file
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
|
||||
monkeypatch.setattr(detector, "_find_caller_registries", lambda: [])
|
||||
|
||||
result = detector._read_registry()
|
||||
|
||||
@@ -341,6 +343,7 @@ class TestReadRegistry:
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
|
||||
monkeypatch.setattr(detector, "_find_caller_registries", lambda: [])
|
||||
|
||||
result = detector._read_registry()
|
||||
|
||||
@@ -358,6 +361,7 @@ class TestReadRegistry:
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
|
||||
monkeypatch.setattr(detector, "_find_caller_registries", lambda: [])
|
||||
|
||||
result = detector._read_registry()
|
||||
|
||||
@@ -373,6 +377,7 @@ class TestReadRegistry:
|
||||
from aipass.memory.apps.handlers.monitor import detector
|
||||
|
||||
monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path)
|
||||
monkeypatch.setattr(detector, "_find_caller_registries", lambda: [])
|
||||
|
||||
result = detector._read_registry()
|
||||
|
||||
|
||||
@@ -34,6 +34,7 @@ from aipass.spawn.apps.handlers.registry import (
|
||||
find_registry,
|
||||
add_to_registry,
|
||||
get_next_citizen_number,
|
||||
ensure_project_has_owner,
|
||||
)
|
||||
from aipass.spawn.apps.handlers.class_registry import get_template_dir
|
||||
from aipass.spawn.apps.handlers.json import json_handler
|
||||
@@ -98,6 +99,14 @@ def grant_passport(
|
||||
# Rename any placeholder paths
|
||||
_ = rename_placeholder_paths(target, folder_name)
|
||||
|
||||
# Set owner field — first agent in the project is the owner
|
||||
passport_path = target / ".trinity" / "passport.json"
|
||||
if passport_path.exists():
|
||||
passport_data = json_handler.read_json(passport_path)
|
||||
if passport_data:
|
||||
passport_data.setdefault("citizenship", {})["owner"] = citizen_number == 1
|
||||
json_handler.write_json(passport_path, passport_data)
|
||||
|
||||
# Regenerate template registry
|
||||
regenerate_template_registry(target)
|
||||
|
||||
@@ -116,6 +125,9 @@ def grant_passport(
|
||||
purpose or "Birthright citizen - purpose TBD",
|
||||
)
|
||||
|
||||
# Ensure at least one agent in the project is the owner
|
||||
ensure_project_has_owner(reg_path)
|
||||
|
||||
# Validate
|
||||
issues = validate_no_placeholders(target)
|
||||
|
||||
|
||||
@@ -265,3 +265,34 @@ def fix_passport_registry_id(branch_dir: Path, registry_path: Path) -> bool:
|
||||
except Exception as e:
|
||||
logger.warning("[registry] Failed to fix registry_id for %s: %s", branch_dir.name, e)
|
||||
return False
|
||||
|
||||
|
||||
def ensure_project_has_owner(registry_path):
|
||||
"""If no agent in the project has owner:true, assign it to the earliest-created agent."""
|
||||
registry_path = Path(registry_path)
|
||||
reg_data = load_registry(registry_path)
|
||||
branches = _branches_as_list(reg_data.get("branches", []))
|
||||
if not branches:
|
||||
return False
|
||||
|
||||
registry_root = registry_path.parent
|
||||
for branch in branches:
|
||||
branch_path = registry_root / branch.get("path", "")
|
||||
passport_path = branch_path / ".trinity" / "passport.json"
|
||||
if passport_path.exists():
|
||||
passport = json_handler.read_json(passport_path)
|
||||
if passport and passport.get("citizenship", {}).get("owner") is True:
|
||||
return False
|
||||
|
||||
by_created = sorted(branches, key=lambda b: b.get("created", "9999-99-99"))
|
||||
for branch in by_created:
|
||||
branch_path = registry_root / branch.get("path", "")
|
||||
passport_path = branch_path / ".trinity" / "passport.json"
|
||||
if passport_path.exists():
|
||||
passport = json_handler.read_json(passport_path)
|
||||
if passport:
|
||||
passport.setdefault("citizenship", {})["owner"] = True
|
||||
json_handler.write_json(passport_path, passport)
|
||||
logger.info("[registry] Retroactively set owner=true on %s", branch.get("name", "?"))
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -46,6 +46,7 @@ from aipass.spawn.apps.handlers.registry import (
|
||||
add_to_registry,
|
||||
get_next_citizen_number,
|
||||
fix_passport_registry_id,
|
||||
ensure_project_has_owner,
|
||||
)
|
||||
from aipass.spawn.apps.handlers.class_registry import (
|
||||
get_template_dir as _get_template_dir,
|
||||
@@ -217,6 +218,14 @@ def _spawn_agent(
|
||||
# Step 2: Rename any {{BRANCH}} dirs/files that weren't caught by path replacement
|
||||
renamed = rename_placeholder_paths(target, folder_name)
|
||||
|
||||
# Step 2b: Set owner field — first agent in the project is the owner
|
||||
passport_path = target / ".trinity" / "passport.json"
|
||||
if passport_path.exists():
|
||||
passport_data = json_handler.read_json(passport_path)
|
||||
if passport_data:
|
||||
passport_data.setdefault("citizenship", {})["owner"] = citizen_number == 1
|
||||
json_handler.write_json(passport_path, passport_data)
|
||||
|
||||
# Step 3: Regenerate .template_registry.json with fresh hashes
|
||||
regenerate_template_registry(target)
|
||||
|
||||
@@ -242,7 +251,10 @@ def _spawn_agent(
|
||||
purpose or "New agent - purpose TBD",
|
||||
)
|
||||
|
||||
# Step 5: Validate no unreplaced placeholders
|
||||
# Step 5: Ensure at least one agent in the project is the owner
|
||||
ensure_project_has_owner(reg_path)
|
||||
|
||||
# Step 6: Validate no unreplaced placeholders
|
||||
issues = validate_no_placeholders(target)
|
||||
|
||||
json_handler.log_operation("branch_created", data={"branch": branch_upper})
|
||||
|
||||
@@ -137,6 +137,12 @@
|
||||
"content_hash": "2e4f4a0c1b47",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f022": {
|
||||
"path": "apps/handlers/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "dbfc0e044461",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f044": {
|
||||
"path": "apps/integrations/README.md",
|
||||
"name": "README.md",
|
||||
@@ -149,7 +155,7 @@
|
||||
"content_hash": "a4cf0a8e3b4f",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f045": {
|
||||
"f015": {
|
||||
"path": "apps/modules/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "e3b0c44298fc",
|
||||
@@ -161,6 +167,12 @@
|
||||
"content_hash": "d1e4e2b98c38",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f027": {
|
||||
"path": "apps/{{BRANCH}}.py",
|
||||
"name": "{{BRANCH}}.py",
|
||||
"content_hash": "39db2c4f8160",
|
||||
"has_branch_placeholder": true
|
||||
},
|
||||
"f028": {
|
||||
"path": "artifacts/README.md",
|
||||
"name": "README.md",
|
||||
@@ -227,6 +239,12 @@
|
||||
"content_hash": "881f06bb6574",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f039": {
|
||||
"path": "tests/conftest.py",
|
||||
"name": "conftest.py",
|
||||
"content_hash": "97f220799d19",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f040": {
|
||||
"path": "tools/README.md",
|
||||
"name": "README.md",
|
||||
@@ -245,25 +263,7 @@
|
||||
"content_hash": "28e9ae373563",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f022": {
|
||||
"path": "apps/handlers/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "dbfc0e044461",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f027": {
|
||||
"path": "apps/{{BRANCH}}.py",
|
||||
"name": "{{BRANCH}}.py",
|
||||
"content_hash": "39db2c4f8160",
|
||||
"has_branch_placeholder": true
|
||||
},
|
||||
"f039": {
|
||||
"path": "tests/conftest.py",
|
||||
"name": "conftest.py",
|
||||
"content_hash": "97f220799d19",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f015": {
|
||||
"f026": {
|
||||
"path": "apps/plugins/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "e3b0c44298fc",
|
||||
|
||||
@@ -13,6 +13,8 @@ class-aware update, and backward compatibility.
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest.mock
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
@@ -444,3 +446,163 @@ class TestMultiAgentCoexistence:
|
||||
passport = json.loads((tmp_path / name / ".trinity" / "passport.json").read_text())
|
||||
assert passport["branch_info"]["branch_name"] == name.upper()
|
||||
assert passport["identity"]["citizen_class"] == "builder"
|
||||
|
||||
|
||||
class TestPassportOwnerField:
|
||||
"""Tests verifying the owner field in passport.json."""
|
||||
|
||||
def test_first_agent_is_owner(self, tmp_path):
|
||||
"""First agent created in a project gets owner: true."""
|
||||
from aipass.spawn.apps.modules.core import _spawn_agent
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
_spawn_agent(str(tmp_path / "first"), registry_path=str(reg))
|
||||
|
||||
passport = json.loads((tmp_path / "first" / ".trinity" / "passport.json").read_text())
|
||||
assert passport["citizenship"]["owner"] is True
|
||||
|
||||
def test_second_agent_not_owner(self, tmp_path):
|
||||
"""Second agent created in a project gets owner: false."""
|
||||
from aipass.spawn.apps.modules.core import _spawn_agent
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
_spawn_agent(str(tmp_path / "first"), registry_path=str(reg))
|
||||
_spawn_agent(str(tmp_path / "second"), registry_path=str(reg))
|
||||
|
||||
p1 = json.loads((tmp_path / "first" / ".trinity" / "passport.json").read_text())
|
||||
p2 = json.loads((tmp_path / "second" / ".trinity" / "passport.json").read_text())
|
||||
assert p1["citizenship"]["owner"] is True
|
||||
assert p2["citizenship"]["owner"] is False
|
||||
|
||||
def test_third_agent_not_owner(self, tmp_path):
|
||||
"""Third agent also gets owner: false."""
|
||||
from aipass.spawn.apps.modules.core import _spawn_agent
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
for name in ["alpha", "beta", "gamma"]:
|
||||
_spawn_agent(str(tmp_path / name), registry_path=str(reg))
|
||||
|
||||
passports = {}
|
||||
for name in ["alpha", "beta", "gamma"]:
|
||||
passports[name] = json.loads((tmp_path / name / ".trinity" / "passport.json").read_text())
|
||||
|
||||
assert passports["alpha"]["citizenship"]["owner"] is True
|
||||
assert passports["beta"]["citizenship"]["owner"] is False
|
||||
assert passports["gamma"]["citizenship"]["owner"] is False
|
||||
|
||||
def test_birthright_first_agent_is_owner(self, tmp_path):
|
||||
"""First birthright agent gets owner: true."""
|
||||
from aipass.spawn.apps.handlers.passport_ops import grant_passport
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
target = tmp_path / "citizen"
|
||||
target.mkdir()
|
||||
with unittest.mock.patch("aipass.spawn.apps.handlers.passport_ops.find_registry", return_value=reg):
|
||||
grant_passport(str(target))
|
||||
|
||||
passport = json.loads((target / ".trinity" / "passport.json").read_text())
|
||||
assert passport["citizenship"]["owner"] is True
|
||||
|
||||
def test_birthright_second_agent_not_owner(self, tmp_path):
|
||||
"""Second birthright agent gets owner: false."""
|
||||
from aipass.spawn.apps.handlers.passport_ops import grant_passport
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
first = tmp_path / "first"
|
||||
first.mkdir()
|
||||
second = tmp_path / "second"
|
||||
second.mkdir()
|
||||
|
||||
with unittest.mock.patch("aipass.spawn.apps.handlers.passport_ops.find_registry", return_value=reg):
|
||||
grant_passport(str(first))
|
||||
grant_passport(str(second))
|
||||
|
||||
p1 = json.loads((first / ".trinity" / "passport.json").read_text())
|
||||
p2 = json.loads((second / ".trinity" / "passport.json").read_text())
|
||||
assert p1["citizenship"]["owner"] is True
|
||||
assert p2["citizenship"]["owner"] is False
|
||||
|
||||
|
||||
class TestRetroactiveOwner:
|
||||
"""Tests for retroactive owner assignment on legacy projects."""
|
||||
|
||||
def test_retroactive_owner_on_legacy_agents(self, tmp_path):
|
||||
"""Creating a new agent in a project where no agent has owner sets the first one."""
|
||||
from aipass.spawn.apps.modules.core import _spawn_agent
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
# Create two agents, then remove their owner field to simulate legacy
|
||||
_spawn_agent(str(tmp_path / "old_a"), registry_path=str(reg))
|
||||
_spawn_agent(str(tmp_path / "old_b"), registry_path=str(reg))
|
||||
|
||||
for name in ["old_a", "old_b"]:
|
||||
pp = tmp_path / name / ".trinity" / "passport.json"
|
||||
data = json.loads(pp.read_text())
|
||||
del data["citizenship"]["owner"]
|
||||
pp.write_text(json.dumps(data, indent=2))
|
||||
|
||||
# Create a third agent — should trigger retroactive fix
|
||||
_spawn_agent(str(tmp_path / "new_c"), registry_path=str(reg))
|
||||
|
||||
pa = json.loads((tmp_path / "old_a" / ".trinity" / "passport.json").read_text())
|
||||
pb = json.loads((tmp_path / "old_b" / ".trinity" / "passport.json").read_text())
|
||||
pc = json.loads((tmp_path / "new_c" / ".trinity" / "passport.json").read_text())
|
||||
assert pa["citizenship"]["owner"] is True
|
||||
assert pb["citizenship"].get("owner") is not True
|
||||
assert pc["citizenship"]["owner"] is False
|
||||
|
||||
def test_no_retroactive_if_owner_exists(self, tmp_path):
|
||||
"""If an existing agent already has owner:true, no retroactive change."""
|
||||
from aipass.spawn.apps.modules.core import _spawn_agent
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
_spawn_agent(str(tmp_path / "first"), registry_path=str(reg))
|
||||
_spawn_agent(str(tmp_path / "second"), registry_path=str(reg))
|
||||
_spawn_agent(str(tmp_path / "third"), registry_path=str(reg))
|
||||
|
||||
p1 = json.loads((tmp_path / "first" / ".trinity" / "passport.json").read_text())
|
||||
p2 = json.loads((tmp_path / "second" / ".trinity" / "passport.json").read_text())
|
||||
p3 = json.loads((tmp_path / "third" / ".trinity" / "passport.json").read_text())
|
||||
assert p1["citizenship"]["owner"] is True
|
||||
assert p2["citizenship"]["owner"] is False
|
||||
assert p3["citizenship"]["owner"] is False
|
||||
|
||||
def test_ensure_project_has_owner_direct(self, tmp_path):
|
||||
"""Direct call to ensure_project_has_owner fixes a legacy project."""
|
||||
from aipass.spawn.apps.handlers.registry import ensure_project_has_owner
|
||||
from aipass.spawn.apps.modules.core import _spawn_agent
|
||||
|
||||
reg = tmp_path / "TEST_REGISTRY.json"
|
||||
reg.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
|
||||
|
||||
_spawn_agent(str(tmp_path / "agent_x"), registry_path=str(reg))
|
||||
_spawn_agent(str(tmp_path / "agent_y"), registry_path=str(reg))
|
||||
|
||||
# Strip owner from both
|
||||
for name in ["agent_x", "agent_y"]:
|
||||
pp = tmp_path / name / ".trinity" / "passport.json"
|
||||
data = json.loads(pp.read_text())
|
||||
data["citizenship"].pop("owner", None)
|
||||
pp.write_text(json.dumps(data, indent=2))
|
||||
|
||||
result = ensure_project_has_owner(reg)
|
||||
assert result is True
|
||||
|
||||
px = json.loads((tmp_path / "agent_x" / ".trinity" / "passport.json").read_text())
|
||||
py = json.loads((tmp_path / "agent_y" / ".trinity" / "passport.json").read_text())
|
||||
assert px["citizenship"]["owner"] is True
|
||||
assert py["citizenship"].get("owner") is not True
|
||||
|
||||
Reference in New Issue
Block a user