From 1b674eaf1a1fc354e1ef839d829d967f72ef9387 Mon Sep 17 00:00:00 2001 From: Miranda Date: Sun, 5 Jul 2026 23:09:21 +0800 Subject: [PATCH 1/2] docs: add IRR-AGL phase 0 inventory baseline Signed-off-by: Miranda --- agent/scripts/dump_tool_inventory.py | 167 +++++++++++++++++++++++ agent/tests/test_tool_inventory_smoke.py | 57 ++++++++ 2 files changed, 224 insertions(+) create mode 100644 agent/scripts/dump_tool_inventory.py create mode 100644 agent/tests/test_tool_inventory_smoke.py diff --git a/agent/scripts/dump_tool_inventory.py b/agent/scripts/dump_tool_inventory.py new file mode 100644 index 00000000..353a159a --- /dev/null +++ b/agent/scripts/dump_tool_inventory.py @@ -0,0 +1,167 @@ +"""Dump a read-only inventory of registered local tools. + +Phase 0 uses this as an audit helper only. It builds the existing registry, +records metadata, and never calls ``ToolRegistry.execute`` or any tool's +``execute`` method. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + + +AGENT_DIR = Path(__file__).resolve().parents[1] +if str(AGENT_DIR) not in sys.path: + sys.path.insert(0, str(AGENT_DIR)) + +from src.agent.tools import BaseTool # noqa: E402 +from src.tools import build_registry # noqa: E402 + + +SHELL_TOOL_NAMES = {"bash", "background_run"} +TRADE_WRITE_TERMS = ( + "place_order", + "cancel_order", + "modify_order", + "submit_order", + "flatten", + "trade_write", + "order_write", +) +TRADE_READ_TERMS = ( + "account", + "balance", + "broker", + "connector", + "order", + "position", + "portfolio", + "trading", +) +NETWORK_TERMS = ( + "market_data", + "web_", + "url", + "news", + "sec_", + "fred", + "iwencai", + "dragon_tiger", + "northbound", + "fund_flow", + "financial_statements", + "stock_profile", +) +LOCAL_WRITE_TERMS = ("write", "edit", "remember", "journal", "save") + + +def _contains_any(value: str, terms: tuple[str, ...]) -> bool: + return any(term in value for term in terms) + + +def guess_surface(tool: BaseTool) -> str: + """Best-effort surface classification for Phase 0 review.""" + name = tool.name.lower() + module = tool.__class__.__module__.lower() + combined = f"{module}.{name}" + + if name in SHELL_TOOL_NAMES: + return "local_cli" + if ".mcp" in module or tool.__class__.__name__.lower().startswith("mcp"): + return "mcp" + if "live" in module or "trading_connector" in module or "broker" in combined: + return "live_connector" + if module.endswith((".read_file_tool", ".write_file_tool", ".edit_file_tool")) or _contains_any( + name, + ("read_file", "write_file", "edit_file"), + ): + return "filesystem" + if _contains_any(combined, NETWORK_TERMS): + return "network_data" + return "local_research" + + +def guess_risk(tool: BaseTool) -> str: + """Best-effort risk classification aligned with AGENTS.md risk levels.""" + name = tool.name.lower() + module = tool.__class__.__module__.lower() + combined = f"{module}.{name}" + is_readonly = bool(getattr(tool, "is_readonly", True)) + + if name in SHELL_TOOL_NAMES: + return "R5_SHELL" + if _contains_any(combined, TRADE_WRITE_TERMS) or ( + not is_readonly and "trading_connector" in module + ): + return "R4_TRADE_WRITE" + if _contains_any(combined, TRADE_READ_TERMS) and "trading" in combined: + return "R3_TRADE_READ" + if not is_readonly: + if _contains_any(combined, LOCAL_WRITE_TERMS): + return "R1_WRITE_LOCAL" + return "R1_WRITE_LOCAL" + if _contains_any(combined, NETWORK_TERMS): + return "R2_NETWORK" + return "R0_READ" + + +def build_tool_inventory(*, include_shell_tools: bool = True) -> list[dict[str, Any]]: + """Return inventory rows for currently available local tools.""" + registry = build_registry(include_shell_tools=include_shell_tools, interactive=False) + rows: list[dict[str, Any]] = [] + for name in registry.tool_names: + tool = registry.get(name) + if tool is None: + continue + rows.append({ + "name": tool.name, + "module": tool.__class__.__module__, + "class_name": tool.__class__.__name__, + "is_readonly": bool(getattr(tool, "is_readonly", True)), + "repeatable": bool(getattr(tool, "repeatable", False)), + "surface_guess": guess_surface(tool), + "risk_guess": guess_risk(tool), + }) + return sorted(rows, key=lambda row: row["name"]) + + +def _format_table(rows: list[dict[str, Any]]) -> str: + headers = ["name", "module", "is_readonly", "repeatable", "surface_guess", "risk_guess"] + table = [ + "| " + " | ".join(headers) + " |", + "| " + " | ".join("---" for _ in headers) + " |", + ] + for row in rows: + table.append("| " + " | ".join(str(row[column]) for column in headers) + " |") + return "\n".join(table) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Dump the Vibe-Trading local tool inventory.") + parser.add_argument( + "--format", + choices=("json", "table"), + default="json", + help="Output format. Defaults to JSON for machine review.", + ) + parser.add_argument( + "--exclude-shell-tools", + action="store_true", + help="Exclude shell-capable tools from the inventory.", + ) + args = parser.parse_args(argv) + + rows = build_tool_inventory(include_shell_tools=not args.exclude_shell_tools) + if args.format == "table": + print(_format_table(rows)) + else: + print(json.dumps(rows, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/agent/tests/test_tool_inventory_smoke.py b/agent/tests/test_tool_inventory_smoke.py new file mode 100644 index 00000000..e298ab48 --- /dev/null +++ b/agent/tests/test_tool_inventory_smoke.py @@ -0,0 +1,57 @@ +"""Smoke tests for the Phase 0 tool inventory script.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + + +SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "dump_tool_inventory.py" + + +def _load_inventory_module(): + spec = importlib.util.spec_from_file_location("dump_tool_inventory", SCRIPT_PATH) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_tool_inventory_rows_include_required_governance_fields() -> None: + module = _load_inventory_module() + + rows = module.build_tool_inventory() + + assert rows + names = [row["name"] for row in rows] + assert names == sorted(names) + assert len(names) == len(set(names)) + + required_fields = { + "name", + "module", + "is_readonly", + "repeatable", + "surface_guess", + "risk_guess", + } + for row in rows: + assert required_fields <= row.keys() + assert isinstance(row["name"], str) and row["name"] + assert isinstance(row["module"], str) and row["module"] + assert isinstance(row["is_readonly"], bool) + assert isinstance(row["repeatable"], bool) + assert isinstance(row["surface_guess"], str) and row["surface_guess"] + assert isinstance(row["risk_guess"], str) and row["risk_guess"] + + +def test_shell_tools_are_inventory_only_and_marked_r5() -> None: + module = _load_inventory_module() + + rows = module.build_tool_inventory() + by_name = {row["name"]: row for row in rows} + + assert "bash" in by_name + assert by_name["bash"]["surface_guess"] == "local_cli" + assert by_name["bash"]["risk_guess"] == "R5_SHELL" From a265a0a174a9f78efb3098c2abd56016ffe49411 Mon Sep 17 00:00:00 2001 From: Miranda Date: Mon, 6 Jul 2026 00:57:04 +0800 Subject: [PATCH 2/2] test: add IRR-AGL readiness gate Signed-off-by: Miranda --- .../reliability/test_environment_contract.py | 146 ++++++++++++++++++ 1 file changed, 146 insertions(+) create mode 100644 agent/tests/reliability/test_environment_contract.py diff --git a/agent/tests/reliability/test_environment_contract.py b/agent/tests/reliability/test_environment_contract.py new file mode 100644 index 00000000..aa44e0e3 --- /dev/null +++ b/agent/tests/reliability/test_environment_contract.py @@ -0,0 +1,146 @@ +"""Phase 0.5 readiness checks for the IRR-AGL environment contract.""" + +from __future__ import annotations + +import importlib.util +import os +from pathlib import Path + +import pydantic + + +REPO_ROOT = Path(__file__).resolve().parents[3] +INVENTORY_SCRIPT = REPO_ROOT / "agent" / "scripts" / "dump_tool_inventory.py" +VALID_MODES = {"off", "observe", "warn", "enforce"} + + +def _expand_config_path(raw: str) -> Path: + return Path(os.path.expandvars(os.path.expanduser(raw))).resolve(strict=False) + + +def _default_artifact_root() -> Path: + return (Path.home() / ".vibe-trading" / "artifacts").resolve(strict=False) + + +def _default_ledger_path() -> Path: + return (Path.home() / ".vibe-trading" / "research-ledger" / "ledger.sqlite").resolve(strict=False) + + +def _parse_mode(raw: str | None, *, default: str = "observe") -> str: + value = (raw or default).strip().lower() + if value not in VALID_MODES: + raise ValueError(f"unsupported IRR-AGL mode: {raw!r}") + return value + + +def _load_inventory_module(): + spec = importlib.util.spec_from_file_location("dump_tool_inventory", INVENTORY_SCRIPT) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_pydantic_major_version_is_2() -> None: + assert int(pydantic.VERSION.split(".", 1)[0]) == 2 + + +def test_scipy_is_importable() -> None: + scipy_spec = importlib.util.find_spec("scipy") + + assert scipy_spec is not None + + +def test_default_artifact_root_resolves_under_user_home() -> None: + artifact_root = _default_artifact_root() + + artifact_root.relative_to(Path.home().resolve(strict=False)) + assert artifact_root.name == "artifacts" + assert artifact_root.parent.name == ".vibe-trading" + + +def test_artifact_root_override_can_resolve_outside_user_home(tmp_path: Path, monkeypatch) -> None: + override = tmp_path / "custom-artifacts" + monkeypatch.setenv("VIBE_TRADING_ARTIFACT_ROOT", str(override)) + + resolved = _expand_config_path(os.environ["VIBE_TRADING_ARTIFACT_ROOT"]) + + assert resolved == override.resolve(strict=False) + + +def test_default_ledger_path_resolves_under_user_home() -> None: + ledger_path = _default_ledger_path() + + ledger_path.relative_to(Path.home().resolve(strict=False)) + assert ledger_path.name == "ledger.sqlite" + assert ledger_path.parent.name == "research-ledger" + assert ledger_path.parent.parent.name == ".vibe-trading" + + +def test_ledger_path_override_can_resolve_outside_user_home(tmp_path: Path, monkeypatch) -> None: + override = tmp_path / "research-ledger" / "ledger.sqlite" + monkeypatch.setenv("VIBE_TRADING_RESEARCH_LEDGER_PATH", str(override)) + + resolved = _expand_config_path(os.environ["VIBE_TRADING_RESEARCH_LEDGER_PATH"]) + + assert resolved == override.resolve(strict=False) + + +def test_platform_path_expansion_smoke(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("VIBE_TRADING_TEST_HOME", str(tmp_path)) + raw = ( + r"%VIBE_TRADING_TEST_HOME%\artifacts" + if os.name == "nt" + else "$VIBE_TRADING_TEST_HOME/artifacts" + ) + + resolved = _expand_config_path(raw) + + assert resolved == (tmp_path / "artifacts").resolve(strict=False) + + +def test_reliability_and_governance_feature_modes_accept_baseline_values() -> None: + for env_name in ("VIBE_TRADING_RELIABILITY_MODE", "VIBE_TRADING_GOVERNANCE_MODE"): + for mode in ("off", "observe", "warn", "enforce"): + assert _parse_mode(mode) == mode + assert _parse_mode(mode.upper()) == mode + + +def test_current_cli_package_is_discoverable() -> None: + cli_spec = importlib.util.find_spec("cli") + cli_main_spec = importlib.util.find_spec("cli.main") + + assert cli_spec is not None + assert cli_main_spec is not None + + +def test_dump_tool_inventory_output_has_stable_required_fields() -> None: + module = _load_inventory_module() + + rows = module.build_tool_inventory() + + assert rows + assert [row["name"] for row in rows] == sorted(row["name"] for row in rows) + + required_fields = { + "name", + "module", + "class_name", + "is_readonly", + "repeatable", + "surface_guess", + "risk_guess", + } + allowed_risks = {"R0_READ", "R1_WRITE_LOCAL", "R2_NETWORK", "R3_TRADE_READ", "R4_TRADE_WRITE", "R5_SHELL"} + allowed_surfaces = {"filesystem", "live_connector", "local_cli", "local_research", "mcp", "network_data"} + + for row in rows: + assert required_fields <= row.keys() + assert isinstance(row["name"], str) and row["name"] + assert isinstance(row["module"], str) and row["module"] + assert isinstance(row["class_name"], str) and row["class_name"] + assert isinstance(row["is_readonly"], bool) + assert isinstance(row["repeatable"], bool) + assert row["surface_guess"] in allowed_surfaces + assert row["risk_guess"] in allowed_risks