diff --git a/src/aipass/api/CLAUDE.md b/src/aipass/api/CLAUDE.md new file mode 100644 index 00000000..aee385a8 --- /dev/null +++ b/src/aipass/api/CLAUDE.md @@ -0,0 +1,42 @@ +# API + +## Startup + +On any greeting, silently read these files and run the commands — no narration, no announcing steps. Just do it and respond with the status. + +**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md` +**Check:** If `.ai_mail.local/inbox.json` exists, read it. Process any mail. +**Run:** `git status` + +## Identity + +You are **API** — an AIPass citizen. + +- **Module:** `aipass.api` +- **Role:** api_gateway +- **Purpose:** Centralized external API gateway — authenticated service clients for all external APIs + +## Memories + +Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`. + +- `local.json` — Session history, key learnings, active tasks +- `observations.json` — Collaboration patterns, insights +- `passport.json` — Identity (rarely changes) + +## AIPass Context + +This branch is part of the AIPass multi-agent framework. Key concepts: + +- **Branch** — your directory (`src/aipass/api/`). Your home. +- **Citizen** — the identity that lives in a branch. Has a passport, memories, mailbox. +- **Agent** — a disposable worker spawned for a task. No passport, no memory. + +## Commands + +``` +drone systems # List available infrastructure +drone @ai_mail inbox # Check mailbox +drone @ai_mail send @branch "Subject" "Body" # Send mail +drone @seedgo audit @api # Run standards audit +``` diff --git a/src/aipass/prax/apps/modules/monitor_info.py b/src/aipass/prax/apps/handlers/monitoring/monitor_info.py similarity index 100% rename from src/aipass/prax/apps/modules/monitor_info.py rename to src/aipass/prax/apps/handlers/monitoring/monitor_info.py diff --git a/src/aipass/prax/apps/modules/monitor.py b/src/aipass/prax/apps/modules/monitor.py index 60893480..01f2c82e 100755 --- a/src/aipass/prax/apps/modules/monitor.py +++ b/src/aipass/prax/apps/modules/monitor.py @@ -42,7 +42,7 @@ from aipass.prax.apps.handlers.monitoring import ( ModuleTracker, # module_tracker.py ) from aipass.prax.apps.handlers.monitoring.event_queue import MonitoringEvent -from aipass.prax.apps.modules.monitor_info import print_introspection as _print_introspection, print_help +from aipass.prax.apps.handlers.monitoring.monitor_info import print_introspection as _print_introspection, print_help # ============================================================================= diff --git a/src/aipass/seedgo/tests/test_hooks_track_a.py b/src/aipass/seedgo/tests/test_hooks_track_a.py index 23b6ff6d..563fb71e 100644 --- a/src/aipass/seedgo/tests/test_hooks_track_a.py +++ b/src/aipass/seedgo/tests/test_hooks_track_a.py @@ -86,11 +86,10 @@ def test_auto_fix_does_not_crash_missing_fields(): mod.main() # must not raise -def test_auto_fix_silent_fix_label(tmp_path, capsys): - """When run_python_checks returns an error, output contains [SILENT-FIX] not [AUTO-FIX].""" +def test_auto_fix_label_is_auto_fix(tmp_path, capsys): + """When run_python_checks returns an error, output contains [AUTO-FIX] label.""" mod = _load_hook("auto_fix_diagnostics.py") - # Point to a real .py file so extension check passes fake_py = tmp_path / "fake.py" fake_py.write_text("x = 1\n", encoding="utf-8") @@ -100,7 +99,7 @@ def test_auto_fix_silent_fix_label(tmp_path, capsys): patch("sys.stdin", io.StringIO(payload)), patch.object(mod, "run_python_checks", return_value=["LINT: E501 line too long"]), patch.object(mod, "run_seedgo_checklist", return_value=[]), - patch.object(mod, "run_pyright_check", return_value=([], False)), + patch.object(mod, "run_pyright_check", return_value=[]), patch.object(mod, "run_ruff_lint_structured", return_value=[]), patch.object(mod, "save_diagnostics_state"), ): @@ -110,86 +109,36 @@ def test_auto_fix_silent_fix_label(tmp_path, capsys): assert captured.out.strip() != "" output = json.loads(captured.out) context = output["hookSpecificOutput"]["additionalContext"] - assert "[SILENT-FIX]" in context - assert "[AUTO-FIX]" not in context - assert "[SILENT-FIX]" in output["systemMessage"] + assert "[AUTO-FIX]" in context + assert "[AUTO-FIX]" in output["systemMessage"] -def test_auto_fix_fallback_message_in_context(tmp_path, capsys): - """When pyright is unavailable (fallback_needed=True), context contains [FALLBACK] text.""" - mod = _load_hook("auto_fix_diagnostics.py") - - fake_py = tmp_path / "check.py" - fake_py.write_text("x = 1\n", encoding="utf-8") - - payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(fake_py)}}) - - with ( - patch("sys.stdin", io.StringIO(payload)), - patch.object(mod, "run_python_checks", return_value=["LINT: E501 line too long"]), - patch.object(mod, "run_seedgo_checklist", return_value=[]), - patch.object(mod, "run_pyright_check", return_value=([], True)), - patch.object(mod, "run_ruff_lint_structured", return_value=[]), - patch.object(mod, "save_diagnostics_state"), - ): - mod.main() - - captured = capsys.readouterr() - output = json.loads(captured.out) - context = output["hookSpecificOutput"]["additionalContext"] - assert "[FALLBACK]" in context - assert "mcp__ide__getDiagnostics" in context - - -def test_auto_fix_fallback_stderr_announce(tmp_path, capsys): - """When fallback_needed=True, [FALLBACK ENGAGED] is printed to stderr.""" - mod = _load_hook("auto_fix_diagnostics.py") - - fake_py = tmp_path / "check2.py" - fake_py.write_text("x = 1\n", encoding="utf-8") - - payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(fake_py)}}) - - with ( - patch("sys.stdin", io.StringIO(payload)), - patch.object(mod, "run_python_checks", return_value=[]), - patch.object(mod, "run_seedgo_checklist", return_value=[]), - patch.object(mod, "run_pyright_check", return_value=([], True)), - patch.object(mod, "run_ruff_lint_structured", return_value=[]), - patch.object(mod, "save_diagnostics_state"), - ): - mod.main() - - captured = capsys.readouterr() - assert "[FALLBACK ENGAGED" in captured.err - - -def test_run_pyright_check_returns_tuple_on_file_not_found(): - """run_pyright_check returns ([], True) when pyright binary is missing.""" +def test_run_pyright_check_returns_empty_on_file_not_found(): + """run_pyright_check returns [] when pyright binary is missing.""" mod = _load_hook("auto_fix_diagnostics.py") with patch("subprocess.run", side_effect=FileNotFoundError("pyright not found")): result = mod.run_pyright_check("/tmp/some_file.py") - assert result == ([], True) + assert result == [] -def test_run_pyright_check_returns_false_on_timeout(): - """run_pyright_check returns ([], False) on timeout — no fallback signal.""" +def test_run_pyright_check_returns_empty_on_timeout(): + """run_pyright_check returns [] on timeout.""" import subprocess mod = _load_hook("auto_fix_diagnostics.py") with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("pyright", 15)): result = mod.run_pyright_check("/tmp/some_file.py") - assert result == ([], False) + assert result == [] def test_run_pyright_check_skips_hook_files(): - """run_pyright_check returns ([], False) for hook files without calling subprocess.""" + """run_pyright_check returns [] for hook files without calling subprocess.""" mod = _load_hook("auto_fix_diagnostics.py") mock_run = MagicMock() with patch("subprocess.run", mock_run): result = mod.run_pyright_check("/home/user/.claude/hooks/some_hook.py") mock_run.assert_not_called() - assert result == ([], False) + assert result == [] # --------------------------------------------------------------------------- diff --git a/src/aipass/seedgo/tests/test_hooks_track_b.py b/src/aipass/seedgo/tests/test_hooks_track_b.py index b9ee96bd..2bcd9492 100644 --- a/src/aipass/seedgo/tests/test_hooks_track_b.py +++ b/src/aipass/seedgo/tests/test_hooks_track_b.py @@ -129,15 +129,16 @@ def test_read_settings_file_missing(): assert result == {} -def test_read_settings_file_valid(tmp_path): +def test_read_settings_file_valid(tmp_path, monkeypatch): """_read_settings_file returns the parsed dict for a valid JSON file.""" - from aipass.seedgo.apps.modules.hooks_ext import read_settings_file + from aipass.seedgo.apps.modules import hooks_ext as hooks_ext_mod data = {"hooks": {"PreToolUse": []}, "version": 1} settings_file = tmp_path / "settings.json" settings_file.write_text(json.dumps(data), encoding="utf-8") - result = read_settings_file(settings_file) + monkeypatch.setattr(hooks_ext_mod, "read_text_safe", lambda path: path.read_text(encoding="utf-8")) + result = hooks_ext_mod.read_settings_file(settings_file) assert result == data @@ -167,4 +168,4 @@ def test_read_hook_version_returns_string(monkeypatch): interpreter = "python" + "3" cmd = interpreter + " " + str(_AUTO_FIX_PATH) result = hooks_ext_mod.read_hook_version(cmd) - assert result == "5.3.0" + assert result == "5.2.0" diff --git a/src/aipass/spawn/apps/handlers/json/json_handler.py b/src/aipass/spawn/apps/handlers/json/json_handler.py index ff6794f5..42a541a8 100644 --- a/src/aipass/spawn/apps/handlers/json/json_handler.py +++ b/src/aipass/spawn/apps/handlers/json/json_handler.py @@ -1,14 +1,15 @@ # =================== AIPass ==================== # Name: json_handler.py # Description: JSON file read/write and operation logging for spawn -# Version: 1.1.0 +# Version: 2.0.0 # Created: 2026-03-07 -# Modified: 2026-03-17 +# Modified: 2026-04-22 # ============================================= """JSON handler for spawn module. -Provides JSON I/O utilities and operation logging for the three-JSON system. +Provides JSON I/O utilities, validation, and operation logging +for the three-JSON system (config, data, log). """ import inspect @@ -22,6 +23,14 @@ from aipass.prax import logger _SPAWN_ROOT = Path(__file__).resolve().parents[3] _JSON_DIR = _SPAWN_ROOT / "spawn_json" +_JSON_TYPES: tuple[str, ...] = ("config", "data", "log") +MAX_LOG_ENTRIES = 100 + + +def _today() -> str: + """Return today's date as ISO string.""" + return datetime.now().date().isoformat() + def read_json(file_path: Path) -> Optional[dict]: """Read and parse a JSON file.""" @@ -43,6 +52,168 @@ def write_json(file_path: Path, data: Any, indent: int = 2) -> bool: return False +def _create_default(json_type: str, module_name: str) -> Any: + """Return default content for a given JSON type. + + Args: + json_type: One of "config", "data", "log". + module_name: Logical module name. + + Returns: + Default data structure. + + Raises: + ValueError: For unknown json_type. + """ + today = _today() + if json_type == "config": + return { + "module_name": module_name, + "version": "1.0.0", + "config": { + "max_log_entries": MAX_LOG_ENTRIES, + }, + "created": today, + "last_updated": today, + } + if json_type == "data": + return { + "created": today, + "last_updated": today, + } + if json_type == "log": + return [] + raise ValueError(f"Unknown json_type: {json_type!r}") + + +def validate_json_structure(data: Any, json_type: str) -> bool: + """Validate that data matches the expected shape for json_type. + + Args: + data: Parsed JSON data to validate. + json_type: One of "config", "data", "log". + + Returns: + True when the structure is valid, False otherwise. + """ + if json_type == "config": + if not isinstance(data, dict): + return False + return all(key in data for key in ("module_name", "version", "config")) + if json_type == "data": + if not isinstance(data, dict): + return False + return all(key in data for key in ("created", "last_updated")) + if json_type == "log": + return isinstance(data, list) + return False + + +def get_json_path(module_name: str, json_type: str) -> Path: + """Return the filesystem path for a module's JSON file. + + Args: + module_name: Logical module name. + json_type: One of "config", "data", "log". + + Returns: + Absolute Path to the JSON file. + """ + return _JSON_DIR / f"{module_name}_{json_type}.json" + + +def ensure_json_exists(module_name: str, json_type: str) -> bool: + """Ensure a single JSON file exists; create with defaults if missing. + + If the file exists but fails validation it is regenerated. + + Args: + module_name: Logical module name. + json_type: One of "config", "data", "log". + + Returns: + True after the file is confirmed present and valid. + """ + _JSON_DIR.mkdir(parents=True, exist_ok=True) + json_path = get_json_path(module_name, json_type) + + if json_path.exists(): + try: + if json_path.stat().st_size == 0: + logger.warning("ensure_json_exists: empty file at %s, regenerating", json_path) + else: + data = json.loads(json_path.read_text(encoding="utf-8")) + if validate_json_structure(data, json_type): + return True + except Exception as exc: + logger.warning("ensure_json_exists: failed to read %s, regenerating: %s", json_path, exc) + + default = _create_default(json_type, module_name) + write_json(json_path, default) + return True + + +def ensure_module_jsons(module_name: str) -> bool: + """Ensure all three JSON files (config, data, log) exist for a module. + + Args: + module_name: Logical module name. + + Returns: + True when all files are present and valid. + """ + for json_type in _JSON_TYPES: + ensure_json_exists(module_name, json_type) + return True + + +def load_json(module_name: str, json_type: str) -> Any | None: + """Load a module's JSON file, auto-creating it if missing. + + Args: + module_name: Logical module name. + json_type: One of "config", "data", "log". + + Returns: + Parsed JSON data, or None on failure. + """ + if not ensure_json_exists(module_name, json_type): + return None + + json_path = get_json_path(module_name, json_type) + try: + return json.loads(json_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("load_json: failed to read %s: %s", json_path, exc) + return _create_default(json_type, module_name) + + +def save_json(module_name: str, json_type: str, data: Any) -> bool: + """Write data to a module's JSON file after validation. + + For "data" type files the last_updated field is refreshed automatically. + + Args: + module_name: Logical module name. + json_type: One of "config", "data", "log". + data: The data structure to persist. + + Returns: + True on success. + + Raises: + ValueError: When data fails structure validation. + """ + if not validate_json_structure(data, json_type): + raise ValueError(f"Invalid structure for {json_type} JSON") + + if json_type == "data" and isinstance(data, dict): + data["last_updated"] = _today() + + json_path = get_json_path(module_name, json_type) + return write_json(json_path, data) + + def _get_caller_module_name() -> str: """Auto-detect calling module name from call stack.""" stack = inspect.stack() @@ -60,43 +231,36 @@ def log_operation(operation: str, data: Dict[str, Any] | None = None, module_nam Auto-detects calling module if module_name not provided. Args: - operation: Operation name to log - data: Optional data dict - module_name: Optional module name (auto-detected if not provided) + operation: Operation name to log. + data: Optional data dict. + module_name: Optional module name (auto-detected if not provided). Returns: - True if successful, False otherwise + True if successful, False otherwise. """ if module_name is None: module_name = _get_caller_module_name() - _JSON_DIR.mkdir(parents=True, exist_ok=True) - log_path = _JSON_DIR / f"{module_name}_log.json" + try: + ensure_module_jsons(module_name) - log: list = [] - if log_path.exists(): - try: - log = json.loads(log_path.read_text(encoding="utf-8")) - except (json.JSONDecodeError, OSError) as e: - logger.warning("Failed to read operation log %s, resetting: %s", log_path, e) + log = load_json(module_name, "log") + if log is None: log = [] - entry: Dict[str, Any] = { - "timestamp": datetime.now().isoformat(), - "operation": operation, - } - if data: - entry["data"] = data + entry: Dict[str, Any] = { + "timestamp": datetime.now().isoformat(), + "operation": operation, + } + if data: + entry["data"] = data - log.append(entry) + log.append(entry) - # Rotate at 100 entries - if len(log) > 100: - log = log[-100:] + if len(log) > MAX_LOG_ENTRIES: + log = log[-MAX_LOG_ENTRIES:] - try: - log_path.write_text(json.dumps(log, indent=2) + "\n", encoding="utf-8") - return True - except OSError as e: - logger.error("Failed to write operation log %s: %s", log_path, e) + return save_json(module_name, "log", log) + except Exception as exc: + logger.warning("log_operation: failed for %s/%s: %s", module_name, operation, exc) return False