diff --git a/.github/workflows/plugin-api-v2.yml b/.github/workflows/plugin-api-v2.yml deleted file mode 100644 index 7c1876b..0000000 --- a/.github/workflows/plugin-api-v2.yml +++ /dev/null @@ -1,28 +0,0 @@ -name: plugin-api-v2 - -on: - pull_request: - push: - branches: - - main - -permissions: - contents: read - -jobs: - contract: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/checkout@v4 - with: - repository: akashic-plugins/plugin-contracts - ref: 24543445c7b99ca63fcd90b5828f754a148b184c - path: .plugin-contracts - - uses: actions/setup-python@v5 - with: - python-version: "3.13" - - name: Check Plugin API v2 - env: - PYTHONPATH: .plugin-contracts - run: python -m akashic_plugin_contracts check plugin.py diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml new file mode 100644 index 0000000..a335409 --- /dev/null +++ b/.github/workflows/plugin-api-v3.yml @@ -0,0 +1,51 @@ +name: plugin-api-v3 + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +jobs: + contract: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: akashic-plugins/plugin-contracts + ref: 4dd69dd621e029e51e99aa428443fa3a4ec1f6cf + path: .plugin-contracts + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Check Plugin API v3 + env: + PYTHONPATH: .plugin-contracts + run: python -m akashic_plugin_contracts check plugin.py + + composition-parity: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: kachofugetsu09/akashic-agent + ref: b60a7ed1bcbed1f772e041336d5c858b4b9fac90 + path: .akashic-core + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: .akashic-core/requirements.txt + - name: Install pinned Core dependencies + run: python -m pip install -r .akashic-core/requirements.txt pytest pytest-asyncio + - name: Verify Tool Loop Guard v3 behavior + env: + AKASHIC_AGENT_ROOT: .akashic-core + PYTHONPATH: .akashic-core + run: python -m pytest -q tests/ diff --git a/akashic.plugin.toml b/akashic.plugin.toml new file mode 100644 index 0000000..e15629f --- /dev/null +++ b/akashic.plugin.toml @@ -0,0 +1,5 @@ +schema_version = 1 +name = "tool_loop_guard" +version = "2.0.0" +api_version = 3 +entrypoint = "plugin.py" diff --git a/plugin.py b/plugin.py index 563f7ab..f588a6f 100644 --- a/plugin.py +++ b/plugin.py @@ -1,17 +1,24 @@ from __future__ import annotations import json +from collections.abc import Mapping from dataclasses import dataclass -from typing import Any, cast +from typing import cast -from agent.lifecycle.types import PreToolCtx -from agent.plugins import Plugin, on_tool_pre -from agent.tool_hooks import HookOutcome +from agent.plugin_composition import Bail, Context +from agent.tools.events import TOOL_EXECUTION_AUTHORIZE, ToolInput _DEFAULT_REPEAT_LIMIT = 3 _DENY_PREFIX = "tool_loop_guard:" _EXCLUDED_TOOLS = frozenset({"task_output", "task_stop"}) +api_version = 3 +name = "tool_loop_guard" +version = "2.0.0" +desc = "检测连续重复的工具调用并提前截断" +author = "Akashic" +inject: tuple[()] = () + @dataclass class _LoopState: @@ -19,35 +26,18 @@ class _LoopState: repeat_count: int = 0 -class ToolLoopGuard(Plugin): - api_version = 2 - name = "tool_loop_guard" - version = "1.0.0" - desc = "检测连续重复的工具调用并提前截断" +class ToolLoopGuard: + """Own per-session repeat state and return typed authorization decisions.""" - def __init__(self) -> None: + def __init__(self, repeat_limit: int) -> None: self._states: dict[str, _LoopState] = {} - self._repeat_limit = _DEFAULT_REPEAT_LIMIT - - async def prepare(self) -> None: - config = getattr(self, "context", None) - plugin_config = getattr(config, "config", None) - raw_limit = ( - plugin_config.get("repeat_limit", _DEFAULT_REPEAT_LIMIT) - if plugin_config - else _DEFAULT_REPEAT_LIMIT - ) - try: - self._repeat_limit = max(2, int(raw_limit)) - except (TypeError, ValueError): - self._repeat_limit = _DEFAULT_REPEAT_LIMIT - - @on_tool_pre() - async def detect_repeated_tool_call(self, event: PreToolCtx) -> HookOutcome | None: - signature, active_index = self._event_signature(event) - if not signature or event.tool_batch_index != active_index: + self._repeat_limit = repeat_limit + + def authorize(self, tool_input: ToolInput) -> Bail[str] | None: + signature, active_index = self._event_signature(tool_input) + if not signature or tool_input.tool_batch_index != active_index: return None - state_key = self._state_key(event) + state_key = self._state_key(tool_input) state = self._states.setdefault(state_key, _LoopState()) if signature == state.signature: state.repeat_count += 1 @@ -56,41 +46,74 @@ async def detect_repeated_tool_call(self, event: PreToolCtx) -> HookOutcome | No state.repeat_count = 1 if state.repeat_count < self._repeat_limit: return None - return HookOutcome( - decision="deny", - reason=( - f"{_DENY_PREFIX}连续重复调用工具 " - f"{state.repeat_count} 次，已截断并进入收尾。" - ), + return Bail( + f"{_DENY_PREFIX}连续重复调用工具 " + f"{state.repeat_count} 次，已截断并进入收尾。" ) - def _state_key(self, event: PreToolCtx) -> str: - if event.session_key: - return f"{event.source}:{event.session_key}" - return f"{event.source}:{event.channel}:{event.chat_id}" + @staticmethod + def _state_key(tool_input: ToolInput) -> str: + if tool_input.session_key: + return f"{tool_input.source}:{tool_input.session_key}" + return f"{tool_input.source}:{tool_input.channel}:{tool_input.chat_id}" - def _signature(self, tool_name: str, arguments: dict[str, Any]) -> str: - args = json.dumps(arguments, ensure_ascii=False, sort_keys=True) - return f"{tool_name}:{args}" + @staticmethod + def _signature(tool_name: str, arguments: Mapping[str, object]) -> str: + encoded = json.dumps( + _thaw(arguments), + ensure_ascii=False, + sort_keys=True, + ) + return f"{tool_name}:{encoded}" - def _event_signature(self, event: PreToolCtx) -> tuple[str, int]: - if not event.tool_batch: - if event.tool_name in _EXCLUDED_TOOLS: + def _event_signature(self, tool_input: ToolInput) -> tuple[str, int]: + if not tool_input.tool_batch: + if tool_input.tool_name in _EXCLUDED_TOOLS: return "", 0 - return self._signature(event.tool_name, event.arguments), 0 + return self._signature(tool_input.tool_name, tool_input.arguments), 0 parts: list[str] = [] active_index = -1 - for index, tool_call in enumerate(event.tool_batch): + for index, tool_call in enumerate(tool_input.tool_batch): tool_name = str(tool_call.get("name", "")) if tool_name in _EXCLUDED_TOOLS: continue - arguments = tool_call.get("arguments") - if not isinstance(arguments, dict): - arguments = {} + raw_arguments = tool_call.get("arguments") + arguments: Mapping[str, object] = {} + if isinstance(raw_arguments, Mapping): + arguments = cast(Mapping[str, object], raw_arguments) if active_index < 0: active_index = index - parts.append(self._signature(tool_name, cast("dict[str, Any]", arguments))) + parts.append(self._signature(tool_name, arguments)) if active_index < 0: return "", 0 return "|".join(parts), active_index + + +async def apply(ctx: Context, config: object) -> None: + """Register one generation-scoped loop authorizer.""" + + guard = ToolLoopGuard(_repeat_limit(config)) + _ = await ctx.on(TOOL_EXECUTION_AUTHORIZE, guard.authorize) + + +def _repeat_limit(config: object) -> int: + raw_limit: object = _DEFAULT_REPEAT_LIMIT + if isinstance(config, Mapping): + raw_limit = cast(Mapping[object, object], config).get( + "repeat_limit", + _DEFAULT_REPEAT_LIMIT, + ) + try: + return max(2, int(cast(int | str, raw_limit))) + except (TypeError, ValueError): + return _DEFAULT_REPEAT_LIMIT + + +def _thaw(value: object) -> object: + if isinstance(value, Mapping): + mapping = cast(Mapping[object, object], value) + return {str(key): _thaw(item) for key, item in mapping.items()} + if isinstance(value, tuple): + return [_thaw(item) for item in cast(tuple[object, ...], value)] + return value diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 5e8306f..e375c12 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,72 +1,394 @@ from __future__ import annotations +import ast +import shutil +import tomllib +from pathlib import Path +from typing import Any + import pytest -from agent.lifecycle.types import PreToolCtx -from plugin import ToolLoopGuard +import plugin as tool_loop_guard +from agent.plugin_composition import CompositionRoot, Context, PluginRuntime +from agent.plugins.composable import ComposablePlugin +from agent.plugins.manager import PluginManager +from agent.plugins.snapshot import ( + RuntimeSnapshotCompiler, + RuntimeSnapshotStore, + bind_runtime_snapshot, + reset_runtime_snapshot, +) +from agent.tool_hooks.executor import ToolExecutor +from agent.tool_hooks.types import ToolExecutionRequest +from agent.tools.events import TOOL_RESULT, ToolResult +from bus.event_bus import EventBus + + +def _read_static_identity(root: Path) -> tuple[dict[str, object], dict[str, object]]: + """Read manifest and literal plugin identity without importing the entrypoint.""" + + # 1. Parse only the static manifest and declared source path. + manifest = tomllib.loads( + (root / "akashic.plugin.toml").read_text(encoding="utf-8") + ) + entrypoint = root / str(manifest["entrypoint"]) + tree = ast.parse(entrypoint.read_text(encoding="utf-8"), filename=str(entrypoint)) + + # 2. Resolve the identity assignments from the entrypoint AST. + values: dict[str, object] = {} + for node in tree.body: + if isinstance(node, ast.Assign): + targets = node.targets + elif isinstance(node, ast.AnnAssign): + targets = (node.target,) + else: + continue + for target in targets: + if not isinstance(target, ast.Name): + continue + if target.id not in {"name", "version", "api_version", "_VERSION"}: + continue + if isinstance(node.value, ast.Constant): + values[target.id] = node.value.value + elif isinstance(node.value, ast.Name): + values[target.id] = values[node.value.id] + return manifest, { + "name": values["name"], + "version": values["version"], + "api_version": values["api_version"], + } + + +def test_static_manifest_matches_import_free_entrypoint_identity() -> None: + root = Path(__file__).parents[1] + manifest, identity = _read_static_identity(root) + + assert manifest == { + "schema_version": 1, + "name": "tool_loop_guard", + "version": "2.0.0", + "api_version": 3, + "entrypoint": "plugin.py", + } + assert identity == { + "name": manifest["name"], + "version": manifest["version"], + "api_version": manifest["api_version"], + } + + +def _request( + *, + call_id: str, + command: str, + session_key: str = "cli:1", + tool_name: str = "shell", + tool_batch: tuple[dict[str, Any], ...] = (), + tool_batch_index: int = 0, +) -> ToolExecutionRequest: + return ToolExecutionRequest( + call_id=call_id, + tool_name=tool_name, + arguments={"command": command}, + source="passive", + session_key=session_key, + channel="cli", + chat_id="1", + tool_batch=tool_batch, + tool_batch_index=tool_batch_index, + ) + + +async def _mount_guard( + tmp_path: Path, + *, + config: object | None = None, +) -> tuple[CompositionRoot, RuntimeSnapshotStore]: + root = CompositionRoot("tool-loop-guard-test") + _ = await root.mount( + lambda ctx: tool_loop_guard.apply(ctx, config or {}), + name="tool_loop_guard", + runtime=PluginRuntime( + plugin_id="tool_loop_guard", + plugin_dir=tmp_path / "plugin", + data_dir=tmp_path / "plugin-data" / "tool_loop_guard", + workspace=tmp_path / "workspace", + config=config or {}, + ), + ) + store = RuntimeSnapshotStore() + store.install(RuntimeSnapshotCompiler().compile({}, composition_root=root)) + return root, store + + +def test_v3_namespace_is_loadable() -> None: + loaded = ComposablePlugin.from_module(tool_loop_guard) + + assert loaded.name == "tool_loop_guard" + assert loaded.version == "2.0.0" + assert loaded.inject == () + + +def test_repeat_limit_preserves_v2_fallbacks() -> None: + assert tool_loop_guard._repeat_limit({}) == 3 + assert tool_loop_guard._repeat_limit({"repeat_limit": 1}) == 2 + assert tool_loop_guard._repeat_limit({"repeat_limit": "5"}) == 5 + assert tool_loop_guard._repeat_limit({"repeat_limit": object()}) == 3 + + +@pytest.mark.asyncio +async def test_denies_on_third_repeat_without_third_invocation(tmp_path: Path) -> None: + root, store = await _mount_guard(tmp_path) + invoked: list[str] = [] + + async def invoke(_: str, arguments: dict[str, Any]) -> str: + invoked.append(str(arguments["command"])) + return "ok" + + lease = store.lease() + token = bind_runtime_snapshot(lease) + try: + results = [ + await ToolExecutor().execute( + _request(call_id=f"call-{index}", command="echo hi"), + invoke, + ) + for index in range(1, 4) + ] + finally: + reset_runtime_snapshot(token) + await lease.release() + await store.close() + + assert [item.status for item in results] == ["success", "success", "denied"] + assert "连续重复调用工具 3 次" in str(results[-1].output) + assert invoked == ["echo hi", "echo hi"] + await root.dispose() @pytest.mark.asyncio -async def test_tool_loop_guard_denies_on_third_repeat() -> None: - plugin = ToolLoopGuard() - first = await plugin.detect_repeated_tool_call( - PreToolCtx( - session_key="cli:1", - channel="cli", - chat_id="1", - tool_name="shell", - arguments={"command": "echo hi"}, - source="passive", +async def test_changed_arguments_and_sessions_have_independent_state( + tmp_path: Path, +) -> None: + root, store = await _mount_guard(tmp_path, config={"repeat_limit": 2}) + + async def invoke(_: str, __: dict[str, Any]) -> str: + return "ok" + + lease = store.lease() + token = bind_runtime_snapshot(lease) + try: + first = await ToolExecutor().execute( + _request(call_id="one", command="echo 1"), + invoke, ) - ) - second = await plugin.detect_repeated_tool_call( - PreToolCtx( - session_key="cli:1", - channel="cli", - chat_id="1", - tool_name="shell", - arguments={"command": "echo hi"}, - source="passive", + changed = await ToolExecutor().execute( + _request(call_id="two", command="echo 2"), + invoke, ) - ) - third = await plugin.detect_repeated_tool_call( - PreToolCtx( - session_key="cli:1", - channel="cli", - chat_id="1", - tool_name="shell", - arguments={"command": "echo hi"}, - source="passive", + other_session = await ToolExecutor().execute( + _request( + call_id="three", + command="echo 2", + session_key="cli:2", + ), + invoke, ) - ) - assert first is None - assert second is None - assert third is not None - assert third.decision == "deny" + denied = await ToolExecutor().execute( + _request(call_id="four", command="echo 2"), + invoke, + ) + finally: + reset_runtime_snapshot(token) + await lease.release() + await store.close() + + assert [first.status, changed.status, other_session.status, denied.status] == [ + "success", + "success", + "success", + "denied", + ] + await root.dispose() @pytest.mark.asyncio -async def test_tool_loop_guard_ignores_changed_arguments() -> None: - plugin = ToolLoopGuard() - first = await plugin.detect_repeated_tool_call( - PreToolCtx( - session_key="cli:1", - channel="cli", - chat_id="1", - tool_name="shell", - arguments={"command": "echo 1"}, - source="passive", +async def test_excluded_batch_call_does_not_own_repeat_count(tmp_path: Path) -> None: + root, store = await _mount_guard(tmp_path, config={"repeat_limit": 2}) + batch = ( + {"name": "task_output", "arguments": {"execution_id": 1}}, + { + "name": "shell", + "arguments": { + "command": "echo hi", + "nested": {"items": [1, {"name": "same"}]}, + }, + }, + ) + changed_batch = ( + batch[0], + { + "name": "shell", + "arguments": { + "command": "echo hi", + "nested": {"items": [1, {"name": "changed"}]}, + }, + }, + ) + invoked: list[int] = [] + + async def invoke(_: str, __: dict[str, Any]) -> str: + invoked.append(1) + return "ok" + + lease = store.lease() + token = bind_runtime_snapshot(lease) + try: + excluded = await ToolExecutor().execute( + _request( + call_id="batch-excluded", + command="", + tool_name="task_output", + tool_batch=batch, + tool_batch_index=0, + ), + invoke, + ) + first = await ToolExecutor().execute( + _request( + call_id="batch-first", + command="echo hi", + tool_batch=batch, + tool_batch_index=1, + ), + invoke, + ) + denied = await ToolExecutor().execute( + _request( + call_id="batch-denied", + command="echo hi", + tool_batch=batch, + tool_batch_index=1, + ), + invoke, ) + changed = await ToolExecutor().execute( + _request( + call_id="batch-changed", + command="echo hi", + tool_batch=changed_batch, + tool_batch_index=1, + ), + invoke, + ) + finally: + reset_runtime_snapshot(token) + await lease.release() + await store.close() + + assert [excluded.status, first.status, denied.status, changed.status] == [ + "success", + "success", + "denied", + "success", + ] + assert len(invoked) == 3 + await root.dispose() + + +@pytest.mark.asyncio +async def test_preflight_counts_without_publishing_result(tmp_path: Path) -> None: + root = CompositionRoot("tool-loop-guard-preflight") + _ = await root.mount( + lambda ctx: tool_loop_guard.apply(ctx, {"repeat_limit": 2}), + name="tool_loop_guard", + runtime=PluginRuntime( + plugin_id="tool_loop_guard", + plugin_dir=tmp_path / "plugin", + data_dir=tmp_path / "plugin-data" / "tool_loop_guard", + workspace=tmp_path / "workspace", + config={"repeat_limit": 2}, + ), ) - second = await plugin.detect_repeated_tool_call( - PreToolCtx( - session_key="cli:1", - channel="cli", - chat_id="1", - tool_name="shell", - arguments={"command": "echo 2"}, - source="passive", + observed: list[ToolResult] = [] + + async def observe(ctx: Context) -> None: + _ = await ctx.on(TOOL_RESULT, observed.append) + + _ = await root.mount(observe, name="observer") + store = RuntimeSnapshotStore() + store.install(RuntimeSnapshotCompiler().compile({}, composition_root=root)) + lease = store.lease() + token = bind_runtime_snapshot(lease) + try: + first = await ToolExecutor().preflight( + _request(call_id="preflight-one", command="echo hi") ) + denied = await ToolExecutor().preflight( + _request(call_id="preflight-two", command="echo hi") + ) + finally: + reset_runtime_snapshot(token) + await lease.release() + await store.close() + + assert [first.status, denied.status] == ["success", "denied"] + assert observed == [] + await root.dispose() + + +@pytest.mark.asyncio +async def test_manager_snapshot_owns_loop_state_and_cleanup(tmp_path: Path) -> None: + plugin_home = tmp_path / "plugins" + plugin_home.mkdir() + _ = shutil.copytree( + Path(__file__).parents[1], + plugin_home / "tool_loop_guard", + ignore=shutil.ignore_patterns( + ".git", + ".akashic-core", + ".pytest_cache", + "__pycache__", + ), + ) + manager = PluginManager( + plugin_dirs=[plugin_home], + event_bus=EventBus(), + tool_registry=None, + workspace=tmp_path / "workspace", + installed_cache_root=tmp_path / "plugin-home" / "cache", + ) + await manager.load_all() + generation = manager.generation("tool_loop_guard") + snapshot = manager.current_snapshot + assert generation is not None and snapshot is not None + assert isinstance(generation.instance, ComposablePlugin) + root = snapshot.composition_root + assert root is not None + assert snapshot.composition_topology is not None + assert snapshot.composition_topology.listeners == ( + "serial:tool.execution.authorize" + "[bail=akashic.tool-deny-reason.v1]:tool_loop_guard", ) - assert first is None - assert second is None + + async def invoke(_: str, __: dict[str, Any]) -> str: + return "ok" + + lease = manager._snapshot_store.lease() + token = bind_runtime_snapshot(lease) + try: + results = [ + await ToolExecutor().execute( + _request(call_id=f"manager-{index}", command="echo hi"), + invoke, + ) + for index in range(3) + ] + finally: + reset_runtime_snapshot(token) + await lease.release() + + assert [item.status for item in results] == ["success", "success", "denied"] + await manager.terminate_all() + assert root.topology_view().listeners == () + assert root.receipt().effects == ()