diff --git a/CONTRACTS.md b/CONTRACTS.md index e853642..6826827 100644 --- a/CONTRACTS.md +++ b/CONTRACTS.md @@ -279,6 +279,18 @@ is added later (adding one is a breaking change). `module:on_session_ready_failed` event with payload `{"module_id": str, "error": str}`, in addition to the WARNING log. Log-only failures are invisible to observability hooks. +**Same mechanism for provider/tool/hook load failures:** A provider, tool, or hook that +raises during `mount()` is caught, logged as a WARNING with `exc_info=True`, and does +**not** abort the session — the remaining modules still load and the session still starts. +This is intentional non-interference (a single broken optional module must not take down +an otherwise-working session) but it is silent by default: nothing short of grepping logs +tells you a configured module never mounted. The kernel emits a `module:load_failed` event +with payload `{"module_type": "provider"|"tool"|"hook", "module_id": str, "error": str}` so +a hook module can observe the gap and decide policy (abort, notify the user, inject a +system message telling the model the tool is unavailable, etc.). Required modules +(orchestrator, context manager) are unaffected by this — their failure still raises +`RuntimeError` and aborts initialization, as documented above. + #### When to use `on_session_ready()` - **Discovering contributions from other modules** — read the fully-composed coordinator diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs index 489d741..c33e0fe 100644 --- a/bindings/python/src/lib.rs +++ b/bindings/python/src/lib.rs @@ -253,6 +253,10 @@ fn _engine(m: &Bound<'_, PyModule>) -> PyResult<()> { "MODULE_ON_SESSION_READY_FAILED", amplifier_core::events::MODULE_ON_SESSION_READY_FAILED, )?; + m.add( + "MODULE_LOAD_FAILED", + amplifier_core::events::MODULE_LOAD_FAILED, + )?; // Aggregate list of all events m.add("ALL_EVENTS", amplifier_core::events::ALL_EVENTS.to_vec())?; diff --git a/bindings/python/tests/test_event_constants.py b/bindings/python/tests/test_event_constants.py index 2d62cdc..7acf1e2 100644 --- a/bindings/python/tests/test_event_constants.py +++ b/bindings/python/tests/test_event_constants.py @@ -93,7 +93,7 @@ def test_all_events_is_list(self): def test_all_events_count(self): from amplifier_core.events import ALL_EVENTS - assert len(ALL_EVENTS) == 42, f"Expected 42 events, got {len(ALL_EVENTS)}" + assert len(ALL_EVENTS) == 43, f"Expected 43 events, got {len(ALL_EVENTS)}" def test_all_events_contains_all_constants(self): import amplifier_core.events as events diff --git a/bindings/python/tests/test_python_stubs.py b/bindings/python/tests/test_python_stubs.py index 0b0b582..67f005c 100644 --- a/bindings/python/tests/test_python_stubs.py +++ b/bindings/python/tests/test_python_stubs.py @@ -23,7 +23,7 @@ def test_events_reexport_provider_throttle(): def test_events_reexport_all_events(): from amplifier_core.events import ALL_EVENTS - assert len(ALL_EVENTS) == 42 + assert len(ALL_EVENTS) == 43 def test_capabilities_reexport_tools(): diff --git a/bindings/python/tests/test_schema_sync.py b/bindings/python/tests/test_schema_sync.py index 2388287..0702e57 100644 --- a/bindings/python/tests/test_schema_sync.py +++ b/bindings/python/tests/test_schema_sync.py @@ -126,7 +126,7 @@ def test_event_constants_match(): assert TOOL_ERROR == "tool:error" assert CANCEL_REQUESTED == "cancel:requested" assert CANCEL_COMPLETED == "cancel:completed" - assert len(ALL_EVENTS) == 42 + assert len(ALL_EVENTS) == 43 def test_hook_result_json_roundtrip(): diff --git a/crates/amplifier-core/src/events.rs b/crates/amplifier-core/src/events.rs index 6cb2b9b..d81e28b 100644 --- a/crates/amplifier-core/src/events.rs +++ b/crates/amplifier-core/src/events.rs @@ -156,6 +156,12 @@ pub const CANCEL_COMPLETED: &str = "cancel:completed"; /// Payload: {module_id: str, error: str} pub const MODULE_ON_SESSION_READY_FAILED: &str = "module:on_session_ready_failed"; +/// Emitted when a provider, tool, or hook module raises during load/mount and +/// the session continues without it (the failure is caught and logged, not +/// re-raised, so the module is silently absent unless something observes this +/// event). Payload: {module_type: "provider"|"tool"|"hook", module_id: str, error: str} +pub const MODULE_LOAD_FAILED: &str = "module:load_failed"; + // --- Aggregate --- /// All canonical event names, for iteration and validation. @@ -205,6 +211,7 @@ pub const ALL_EVENTS: &[&str] = &[ CANCEL_REQUESTED, CANCEL_COMPLETED, MODULE_ON_SESSION_READY_FAILED, + MODULE_LOAD_FAILED, ]; #[cfg(test)] @@ -327,6 +334,19 @@ mod tests { ); } + #[test] + fn test_module_on_session_ready_failed_event_value() { + assert_eq!( + MODULE_ON_SESSION_READY_FAILED, + "module:on_session_ready_failed" + ); + } + + #[test] + fn test_module_load_failed_event_value() { + assert_eq!(MODULE_LOAD_FAILED, "module:load_failed"); + } + #[test] fn test_all_events_contains_new_constants() { assert!( @@ -347,7 +367,7 @@ mod tests { #[test] fn all_events_count() { - assert_eq!(ALL_EVENTS.len(), 42, "expected 42 canonical events"); + assert_eq!(ALL_EVENTS.len(), 43, "expected 43 canonical events"); } #[test] @@ -391,6 +411,8 @@ mod tests { APPROVAL_DENIED, CANCEL_REQUESTED, CANCEL_COMPLETED, + MODULE_ON_SESSION_READY_FAILED, + MODULE_LOAD_FAILED, ]; for event in expected { assert!(ALL_EVENTS.contains(event), "ALL_EVENTS missing: {event}"); diff --git a/python/amplifier_core/_session_init.py b/python/amplifier_core/_session_init.py index 1aab6f6..03ede4b 100644 --- a/python/amplifier_core/_session_init.py +++ b/python/amplifier_core/_session_init.py @@ -19,6 +19,33 @@ def _safe_exception_str(e: BaseException) -> str: return repr(e) +async def _emit_module_load_failed( + coordinator: Any, module_type: str, module_id: str, error: BaseException +) -> None: + """Emit the module:load_failed observability event for a provider, tool, + or hook that raised during load/mount. + + This is a mechanism only: the kernel makes the failure observable via the + canonical event stream. It does not decide whether the session should + abort -- that policy choice belongs to a hook module subscribed to this + event. Mirrors the on_session_ready failure pattern below: event emission + failure must never suppress the original WARNING log. + """ + from .events import MODULE_LOAD_FAILED + + try: + await coordinator.hooks.emit( + MODULE_LOAD_FAILED, + { + "module_type": module_type, + "module_id": module_id, + "error": _safe_exception_str(error), + }, + ) + except Exception: + pass # Event emission failure must not suppress the original warning + + async def initialize_session( config: dict[str, Any], coordinator: Any, @@ -190,6 +217,7 @@ async def initialize_session( f"Failed to load provider '{module_id}': {_safe_exception_str(e)}", exc_info=True, ) + await _emit_module_load_failed(coordinator, "provider", module_id, e) # Load tools for tool_config in config.get("tools", []): @@ -215,6 +243,7 @@ async def initialize_session( f"Failed to load tool '{module_id}': {_safe_exception_str(e)}", exc_info=True, ) + await _emit_module_load_failed(coordinator, "tool", module_id, e) # Load hooks for hook_config in config.get("hooks", []): @@ -240,6 +269,7 @@ async def initialize_session( f"Failed to load hook '{module_id}': {_safe_exception_str(e)}", exc_info=True, ) + await _emit_module_load_failed(coordinator, "hook", module_id, e) # Phase 6 — on_session_ready callbacks # Called after ALL modules have been mounted. Each callback receives the @@ -260,6 +290,7 @@ async def initialize_session( exc_info=True, ) from .events import MODULE_ON_SESSION_READY_FAILED + try: await coordinator.hooks.emit( MODULE_ON_SESSION_READY_FAILED, diff --git a/python/amplifier_core/events.py b/python/amplifier_core/events.py index 54cff91..fa0d8a8 100644 --- a/python/amplifier_core/events.py +++ b/python/amplifier_core/events.py @@ -62,6 +62,7 @@ CANCEL_COMPLETED, # Module lifecycle events MODULE_ON_SESSION_READY_FAILED, + MODULE_LOAD_FAILED, ALL_EVENTS, ) @@ -109,4 +110,5 @@ "CANCEL_COMPLETED", "ALL_EVENTS", "MODULE_ON_SESSION_READY_FAILED", + "MODULE_LOAD_FAILED", ] diff --git a/tests/test_session_init_module_load_failed.py b/tests/test_session_init_module_load_failed.py new file mode 100644 index 0000000..ce7c12c --- /dev/null +++ b/tests/test_session_init_module_load_failed.py @@ -0,0 +1,223 @@ +"""Tests verifying the module:load_failed observability event. + +Provider, tool, and hook modules that raise during load/mount are caught and +logged as a WARNING (non-fatal -- the session still starts with the other +modules loaded). Before this fix, that WARNING was the *only* trace of the +failure: nothing was observable through the kernel's event surface, so a +hook module had no way to detect that a configured tool/provider/hook never +mounted. These tests pin down the mechanism fix: a `module:load_failed` +event, carrying which module type failed, its module_id, and the error. + +Mirrors the mocked style of test_session_init_on_session_ready.py. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest +from amplifier_core._session_init import initialize_session +from amplifier_core.events import MODULE_LOAD_FAILED + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + + +def _make_mocks(failing_module_ids, calls=None): + """Build a mock loader + coordinator whose loader.load() raises for any + module_id in ``failing_module_ids``, and otherwise succeeds -- including + for the required orchestrator/context modules ("loop-basic", + "context-simple"), which must mount successfully so the provider/tool/ + hook loops under test are actually reached. + + Args: + failing_module_ids: module_id or iterable of module_ids whose + load() call should raise. + calls: optional list to record every module_id load() was called + with, in order (used to assert non-interference). + + Returns: + (mock_loader, mock_coordinator) tuple. + """ + if isinstance(failing_module_ids, str): + failing_module_ids = {failing_module_ids} + else: + failing_module_ids = set(failing_module_ids) + + async def load_side_effect(module_id, config, source_hint=None, coordinator=None): + if calls is not None: + calls.append(module_id) + if module_id in failing_module_ids: + raise RuntimeError(f"{module_id} mount() deliberately raised") + + async def mount_fn(coordinator): + return None + + return mount_fn + + mock_loader = MagicMock() + mock_loader.load = AsyncMock(side_effect=load_side_effect) + mock_loader.get_on_session_ready_queue = MagicMock(return_value=[]) + mock_loader._on_session_ready_queue = [] + + mock_coordinator = MagicMock() + mock_coordinator.loader = mock_loader + mock_coordinator.register_cleanup = MagicMock() + mock_coordinator.get = MagicMock(return_value={}) + mock_coordinator.hooks = MagicMock() + mock_coordinator.hooks.emit = AsyncMock() + + return mock_loader, mock_coordinator + + +def _tracking_emit(emitted_events): + async def _emit(event, payload): + emitted_events.append((event, payload)) + + return _emit + + +_BASE_CONFIG = { + "session": {"orchestrator": "loop-basic", "context": "context-simple"}, +} + + +# --------------------------------------------------------------------------- +# Tool load failure -- the exact defect described in the issue +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_tool_load_failure_emits_module_load_failed(): + """A tool whose mount() raises is caught (session continues) AND emits + module:load_failed. Before the fix, the exception was logged at WARNING + and nothing else observed it -- this test fails against the old code + because zero events were ever emitted for a tool load failure.""" + + _, mock_coordinator = _make_mocks(failing_module_ids="tool-computer-use") + emitted_events = [] + mock_coordinator.hooks.emit = AsyncMock(side_effect=_tracking_emit(emitted_events)) + + config = { + **_BASE_CONFIG, + "providers": [], + "tools": [{"module": "tool-computer-use"}], + "hooks": [], + } + + # Must not raise -- tool failures are non-fatal by design (unlike + # orchestrator/context). The defect is silence, not the non-fatal-ness. + await initialize_session( + config, mock_coordinator, session_id="test-session", parent_id=None + ) + + failures = [(e, p) for e, p in emitted_events if e == MODULE_LOAD_FAILED] + assert len(failures) == 1, ( + f"Expected exactly one module:load_failed event, got: {emitted_events}" + ) + payload = failures[0][1] + assert payload["module_type"] == "tool" + assert payload["module_id"] == "tool-computer-use" + assert "tool-computer-use mount() deliberately raised" in payload["error"] + + +@pytest.mark.asyncio +async def test_tool_load_failure_does_not_abort_remaining_tools(): + """One tool's mount() raising must not prevent the next tool from loading + -- non-interference must be preserved by this fix, not just the event.""" + + calls: list[str] = [] + _, mock_coordinator = _make_mocks(failing_module_ids="tool-broken", calls=calls) + + config = { + **_BASE_CONFIG, + "providers": [], + "tools": [{"module": "tool-broken"}, {"module": "tool-ok"}], + "hooks": [], + } + + await initialize_session( + config, mock_coordinator, session_id="test-session", parent_id=None + ) + + assert calls == ["loop-basic", "context-simple", "tool-broken", "tool-ok"], ( + "tool-ok must still be attempted after tool-broken raised" + ) + + +# --------------------------------------------------------------------------- +# Provider and hook load failure -- identical shape to tools; same fix applies +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_provider_load_failure_emits_module_load_failed(): + _, mock_coordinator = _make_mocks(failing_module_ids="provider-anthropic") + emitted_events = [] + mock_coordinator.hooks.emit = AsyncMock(side_effect=_tracking_emit(emitted_events)) + + config = { + **_BASE_CONFIG, + "providers": [{"module": "provider-anthropic"}], + "tools": [], + "hooks": [], + } + + await initialize_session( + config, mock_coordinator, session_id="test-session", parent_id=None + ) + + failures = [(e, p) for e, p in emitted_events if e == MODULE_LOAD_FAILED] + assert len(failures) == 1 + assert failures[0][1]["module_type"] == "provider" + assert failures[0][1]["module_id"] == "provider-anthropic" + + +@pytest.mark.asyncio +async def test_hook_load_failure_emits_module_load_failed(): + _, mock_coordinator = _make_mocks(failing_module_ids="hook-logging") + emitted_events = [] + mock_coordinator.hooks.emit = AsyncMock(side_effect=_tracking_emit(emitted_events)) + + config = { + **_BASE_CONFIG, + "providers": [], + "tools": [], + "hooks": [{"module": "hook-logging"}], + } + + await initialize_session( + config, mock_coordinator, session_id="test-session", parent_id=None + ) + + failures = [(e, p) for e, p in emitted_events if e == MODULE_LOAD_FAILED] + assert len(failures) == 1 + assert failures[0][1]["module_type"] == "hook" + assert failures[0][1]["module_id"] == "hook-logging" + + +# --------------------------------------------------------------------------- +# Emission failure must not suppress the original warning / abort the session +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_event_emission_failure_does_not_propagate(): + """If coordinator.hooks.emit() itself raises, initialize_session must + still complete -- mirrors the on_session_ready precedent at + _session_init.py's Phase 6 (event emission failure must never suppress + the original warning or abort session init).""" + + _, mock_coordinator = _make_mocks(failing_module_ids="tool-computer-use") + mock_coordinator.hooks.emit = AsyncMock(side_effect=RuntimeError("emit is broken")) + + config = { + **_BASE_CONFIG, + "providers": [], + "tools": [{"module": "tool-computer-use"}], + "hooks": [], + } + + # Must not raise even though hooks.emit() itself raises. + await initialize_session( + config, mock_coordinator, session_id="test-session", parent_id=None + ) diff --git a/tests/test_session_init_module_load_failed_integration.py b/tests/test_session_init_module_load_failed_integration.py new file mode 100644 index 0000000..cc06dc1 --- /dev/null +++ b/tests/test_session_init_module_load_failed_integration.py @@ -0,0 +1,179 @@ +""" +Integration test: real session init loading pipeline, real ModuleLoader, +a tool module whose mount() raises. + +This is the reproduction shape for the silent-failure defect: a tool module +that deliberately raises during mount() (exactly as a real tool module would +when it detects it cannot serve the current platform and fails loud by +design) must be observable through the kernel's event surface, not just a +WARNING log line nobody watches. Exercises the real ModuleLoader.load() -> +source resolution -> filesystem discovery -> mount path WITHOUT mocking the +loader, mirroring test_session_init_integration.py's real-loader pattern. +""" + +import importlib +import os +import shutil +import sys +import tempfile + +import pytest +from amplifier_core._session_init import initialize_session +from amplifier_core.events import MODULE_LOAD_FAILED +from amplifier_core.loader import ModuleLoader +from amplifier_core.testing import MockCoordinator + +# --------------------------------------------------------------------------- +# Fixture helpers +# --------------------------------------------------------------------------- + +ORCH_MODULE_NAME = "amplifier_module_lf_test_orch" +CTX_MODULE_NAME = "amplifier_module_lf_test_ctx" +FAILING_TOOL_MODULE_NAME = "amplifier_module_lf_test_failing_tool" + +ORCH_INIT_PY = """\ +__amplifier_module_type__ = "orchestrator" + + +async def mount(coordinator, config=None): + class FakeOrch: + async def execute(self, prompt, context, providers, tools, hooks, **kwargs): + return f"echo: {prompt}" + + await coordinator.mount("orchestrator", FakeOrch()) + return None +""" + +CTX_INIT_PY = """\ +__amplifier_module_type__ = "context" + + +async def mount(coordinator, config=None): + class FakeCtx: + async def add_message(self, msg): + pass + + async def get_messages(self): + return [] + + async def get_messages_for_request(self, request=None): + return [] + + async def set_messages(self, msgs): + pass + + async def clear(self): + pass + + await coordinator.mount("context", FakeCtx()) + return None +""" + +# A tool module whose mount() deliberately raises -- exactly the shape of a +# real-world tool that detects it cannot serve the current platform and +# fails loud by design (per the incident this fix responds to). +FAILING_TOOL_INIT_PY = """\ +__amplifier_module_type__ = "tool" + + +class UnsupportedPlatformError(RuntimeError): + pass + + +async def mount(coordinator, config=None): + raise UnsupportedPlatformError( + "cannot mount: this tool does not support the current platform" + ) +""" + + +@pytest.fixture +def fixture_dir(): + """Create a temp directory with orchestrator, context, and a tool module + whose mount() raises.""" + tmp = tempfile.mkdtemp(prefix="amp_integ_load_failed_test_") + + for pkg_name, init_py in ( + (ORCH_MODULE_NAME, ORCH_INIT_PY), + (CTX_MODULE_NAME, CTX_INIT_PY), + (FAILING_TOOL_MODULE_NAME, FAILING_TOOL_INIT_PY), + ): + pkg = os.path.join(tmp, pkg_name) + os.makedirs(pkg) + with open(os.path.join(pkg, "__init__.py"), "w") as fh: + fh.write(init_py) + + sys.path.insert(0, tmp) + importlib.invalidate_caches() + + yield tmp + + try: + sys.path.remove(tmp) + except ValueError: + pass + for name in [ORCH_MODULE_NAME, CTX_MODULE_NAME, FAILING_TOOL_MODULE_NAME]: + sys.modules.pop(name, None) + + shutil.rmtree(tmp, ignore_errors=True) + + +# --------------------------------------------------------------------------- +# The reproduction test +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_real_loader_tool_mount_failure_is_surfaced_not_swallowed(fixture_dir): + """A tool module's mount() raising, loaded through a REAL session (real + ModuleLoader, real filesystem discovery, real coordinator), must surface + as a module:load_failed event -- not just a log line. + + Without the fix in _session_init.py's tool-loading loop, this assertion + fails: zero events are emitted for a tool load failure (the exception is + caught and only logger.warning() sees it), which is the exact silent + failure this PR addresses. + """ + config = { + "session": { + "orchestrator": "lf-test-orch", + "context": "lf-test-ctx", + }, + "providers": [], + "tools": [{"module": "lf-test-failing-tool"}], + "hooks": [], + } + + coordinator = MockCoordinator() + loader = ModuleLoader(coordinator=coordinator) + coordinator.loader = loader + + captured: list[dict] = [] + + def capture_load_failed(event, data): + captured.append(dict(data)) + + coordinator.hooks.register( + MODULE_LOAD_FAILED, capture_load_failed, 0, name="test-capture" + ) + + # Must not raise: a broken optional tool must not abort session init. + await initialize_session( + config, coordinator, session_id="integ-load-failed", parent_id=None + ) + + # The orchestrator and context must still have mounted successfully. + assert coordinator.get("orchestrator") is not None + assert coordinator.get("context") is not None + + # The tool must NOT be mounted (its mount() raised). + tools = coordinator.get("tools") or {} + assert "lf-test-failing-tool" not in tools + + # The failure must be observable via the event surface, not just logs. + assert len(captured) == 1, ( + f"Expected exactly one module:load_failed event, got: {captured}" + ) + assert captured[0]["module_type"] == "tool" + assert captured[0]["module_id"] == "lf-test-failing-tool" + assert "cannot mount" in captured[0]["error"]