From 563be89470bfb4fe1d07064cce31cbe36e191f1f Mon Sep 17 00:00:00 2001 From: mmercuri Date: Sat, 25 Apr 2026 22:51:26 -0700 Subject: [PATCH] feat(instrument): port Semantic Kernel framework adapter (M2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the Microsoft Semantic Kernel adapter from the ateam reference implementation onto the layerlens.instrument base layer. M2 fan-out: slices the SK-specific adapter out of the bundled M1.C agent-tier port (PR #97) so it can land independently. Source: ateam/stratix/sdk/python/adapters/semantic_kernel/ (~929 LOC, 4 files). Template: existing langchain framework adapter package. Surface - src/layerlens/instrument/adapters/frameworks/semantic_kernel/ - __init__.py, lifecycle.py, filters.py, metadata.py - src/layerlens/instrument/adapters/frameworks/__init__.py (lazy) - tests/instrument/adapters/frameworks/test_semantic_kernel.py - samples/instrument/semantic_kernel/{main.py,README.md} - docs/adapters/frameworks-semantic_kernel.md Naming and back-compat - stratix.* import paths -> layerlens.* import paths. - STRATIXFunctionFilter / STRATIXAutoFunctionFilter / STRATIXPromptRenderFilter remain importable as deprecation aliases for ateam users; new code should use the LayerLens* names. - StratixMemoryStore class name preserved (matches the SK memory store contract users embed by name). Packaging - pyproject.toml gains semantic-kernel = ['semantic-kernel>=1.0,<2.0; python_version >= "3.10"']. Default install set is unchanged. - ruff per-file-ignores adds ARG002 for adapters/frameworks/ — the SK filter callbacks have signatures dictated by the upstream filter API. - pyright executionEnvironments relaxes a few rules under src/layerlens/instrument/adapters/frameworks because the framework instrumentation relies on runtime attribute mutation that pyright cannot follow. mypy --strict stays strict. Verification - uv run pytest tests/instrument/adapters/frameworks/test_semantic_kernel.py -x -> 12 passed - uv run mypy --strict src/layerlens/instrument/adapters/frameworks/semantic_kernel -> Success: no issues found in 4 source files - uv run pytest tests/instrument/test_lazy_imports.py tests/instrument/test_default_install.py -> 6 passed (lazy-import + default-install guards still hold) --- docs/adapters/frameworks-semantic_kernel.md | 107 ++++ pyproject.toml | 14 +- samples/instrument/semantic_kernel/README.md | 42 ++ .../instrument/semantic_kernel/__init__.py | 0 samples/instrument/semantic_kernel/main.py | 86 +++ .../adapters/frameworks/__init__.py | 17 + .../frameworks/semantic_kernel/__init__.py | 48 ++ .../frameworks/semantic_kernel/filters.py | 259 ++++++++ .../frameworks/semantic_kernel/lifecycle.py | 602 ++++++++++++++++++ .../frameworks/semantic_kernel/metadata.py | 60 ++ tests/instrument/adapters/__init__.py | 0 .../adapters/frameworks/__init__.py | 0 .../frameworks/test_semantic_kernel.py | 212 ++++++ 13 files changed, 1446 insertions(+), 1 deletion(-) create mode 100644 docs/adapters/frameworks-semantic_kernel.md create mode 100644 samples/instrument/semantic_kernel/README.md create mode 100644 samples/instrument/semantic_kernel/__init__.py create mode 100644 samples/instrument/semantic_kernel/main.py create mode 100644 src/layerlens/instrument/adapters/frameworks/__init__.py create mode 100644 src/layerlens/instrument/adapters/frameworks/semantic_kernel/__init__.py create mode 100644 src/layerlens/instrument/adapters/frameworks/semantic_kernel/filters.py create mode 100644 src/layerlens/instrument/adapters/frameworks/semantic_kernel/lifecycle.py create mode 100644 src/layerlens/instrument/adapters/frameworks/semantic_kernel/metadata.py create mode 100644 tests/instrument/adapters/__init__.py create mode 100644 tests/instrument/adapters/frameworks/__init__.py create mode 100644 tests/instrument/adapters/frameworks/test_semantic_kernel.py diff --git a/docs/adapters/frameworks-semantic_kernel.md b/docs/adapters/frameworks-semantic_kernel.md new file mode 100644 index 0000000..b29e16b --- /dev/null +++ b/docs/adapters/frameworks-semantic_kernel.md @@ -0,0 +1,107 @@ +# Semantic Kernel framework adapter + +`layerlens.instrument.adapters.frameworks.semantic_kernel.SemanticKernelAdapter` +instruments [Microsoft Semantic Kernel](https://github.com/microsoft/semantic-kernel) +using the kernel's native filter API — non-invasive, no monkey-patching. + +## Install + +```bash +pip install 'layerlens[semantic-kernel]' +``` + +Pulls `semantic-kernel>=1.0,<2.0`. Requires Python 3.10+. + +## Quick start + +```python +import asyncio +from semantic_kernel import Kernel +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion + +from layerlens.instrument.adapters.frameworks.semantic_kernel import SemanticKernelAdapter +from layerlens.instrument.transport.sink_http import HttpEventSink + +sink = HttpEventSink(adapter_name="semantic_kernel") +adapter = SemanticKernelAdapter() +adapter.add_sink(sink) +adapter.connect() + +kernel = Kernel() +kernel.add_service(OpenAIChatCompletion(ai_model_id="gpt-4o-mini")) +adapter.instrument_kernel(kernel) + +async def main() -> None: + result = await kernel.invoke_prompt("What is 2 + 2?") + print(result) + +asyncio.run(main()) + +adapter.disconnect() +sink.close() +``` + +## What's wrapped + +`adapter.instrument_kernel(kernel)` registers three Semantic Kernel filters +on the supplied kernel: + +- `function_invocation_filter` — fires before/after every `KernelFunction` + call (plugin function, prompt function, etc.). +- `prompt_rendering_filter` — fires before/after the prompt template is + rendered for prompt functions. +- `auto_function_invocation_filter` — fires when the model auto-selects a + plugin function via tool-calling. + +No methods are monkey-patched; on `disconnect()` the filter list is cleared +and the kernel returns to its original behaviour. + +## Events emitted + +| Event | Layer | When | +|---|---|---| +| `environment.config` | L4a | First plugin invocation per kernel. | +| `agent.input` | L1 | Function invocation start. | +| `agent.output` | L1 | Function invocation end (success or error). | +| `agent.code` | L2 | Per plugin function when `l2_agent_code` is true. | +| `agent.action` | L4a | Per planner step. | +| `agent.state.change` | cross-cutting | Memory store reads/writes. | +| `tool.call` | L5a | Per `auto_function_invocation` (model-selected plugin). | +| `model.invoke` | L3 | Per LLM call inside the kernel. | + +## Semantic Kernel specifics + +- **Plugin attribution**: every event includes `plugin_name`, + `function_name`, and (for prompt functions) the rendered prompt token + count when available. +- **Filter API is preferred**: filters are first-class Semantic Kernel + citizens — they survive kernel cloning and don't break the type system. + This is why this adapter uses filters instead of method-wrapping. +- **Async-first**: Semantic Kernel is async-first; all filters are async + and propagate the `next` continuation correctly. + +## Capture config + +```python +from layerlens.instrument.adapters._base import CaptureConfig + +# Recommended. +adapter = SemanticKernelAdapter(capture_config=CaptureConfig.standard()) + +# Capture rendered prompt template body. +adapter = SemanticKernelAdapter( + capture_config=CaptureConfig( + l1_agent_io=True, + l3_model_metadata=True, + l5a_tool_calls=True, + capture_content=True, + ), +) +``` + +## BYOK + +Semantic Kernel uses `OpenAIChatCompletion`, `AzureChatCompletion`, +`HuggingFacePromptExecutionSettings`, etc. for model access. The adapter +does not own those credentials. For platform-managed BYOK see +`docs/adapters/byok.md` (atlas-app M1.B). diff --git a/pyproject.toml b/pyproject.toml index ae6d1dc..46696f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,11 @@ classifiers = [ [project.optional-dependencies] cli = ["click>=8.0.0"] +# --- Instrument layer: framework adapters --- +# Adding any extra below MUST keep the default `pip install layerlens` +# install set unchanged. Verified by `tests/instrument/test_default_install.py`. +semantic-kernel = ["semantic-kernel>=1.0,<2.0; python_version >= '3.10'"] + [project.urls] Homepage = "https://github.com/LayerLens/stratix-python" Repository = "https://github.com/LayerLens/stratix-python" @@ -139,14 +144,21 @@ known-first-party = ["openai", "tests"] "tests/**.py" = ["T201", "T203", "ARG", "B007"] "examples/**.py" = ["T201", "T203"] "src/layerlens/cli/**" = ["T201", "T203"] +# Framework callbacks have signatures dictated by upstream — unused +# arguments are part of the contract, not a code smell. +"src/layerlens/instrument/adapters/frameworks/**.py" = ["ARG002"] [tool.pyright] include = ["src", "tests"] exclude = ["**/__pycache__"] reportMissingTypeStubs = false -# Less strict settings for tests and cli +# Less strict settings for tests, cli, and the dynamic-monkey-patching +# adapter code. mypy --strict stays strict for these dirs; pyright is +# relaxed here because it can't follow runtime attribute mutation that +# the framework instrumentation relies on. executionEnvironments = [ { root = "src/layerlens/cli", reportMissingImports = false, reportFunctionMemberAccess = false, reportCallIssue = false, reportArgumentType = false, reportAttributeAccessIssue = false }, + { root = "src/layerlens/instrument/adapters/frameworks", reportPossiblyUnbound = false, reportPossiblyUnboundVariable = false, reportCallIssue = false, reportAttributeAccessIssue = false, reportArgumentType = false, reportMissingImports = false, reportFunctionMemberAccess = false }, { root = "tests", reportGeneralTypeIssues = false, reportOptionalSubscript = false, reportOptionalMemberAccess = false, reportUntypedFunctionDecorator = false, reportUnknownArgumentType = false, reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnnecessaryIsInstance = false, reportUnnecessaryComparison = false, reportArgumentType = false, reportCallIssue = false }, ] diff --git a/samples/instrument/semantic_kernel/README.md b/samples/instrument/semantic_kernel/README.md new file mode 100644 index 0000000..e2e82fe --- /dev/null +++ b/samples/instrument/semantic_kernel/README.md @@ -0,0 +1,42 @@ +# Semantic Kernel sample + +Runnable end-to-end sample for the Microsoft Semantic Kernel framework +adapter. The script wires the LayerLens filters via +`SemanticKernelAdapter.instrument_kernel(kernel)` and runs a single +`invoke_prompt` call against an `OpenAIChatCompletion` service. + +## Install + +```bash +pip install 'layerlens[semantic-kernel,providers-openai]' +``` + +This pulls `semantic-kernel>=1.0,<2.0` (which itself depends on +`openai>=1.30`). Requires Python 3.10+. + +## Run + +```bash +export OPENAI_API_KEY=sk-... +export LAYERLENS_STRATIX_API_KEY=... # optional — needed only to ship spans +export LAYERLENS_STRATIX_BASE_URL=... # optional — defaults to LayerLens cloud + +python -m samples.instrument.semantic_kernel.main +``` + +The sample prints the model's response and ships an +`agent.input` / `agent.output` / `model.invoke` event triple to +atlas-app via `HttpEventSink`. If the LayerLens credentials are not set, +the sink buffers the events in memory and drops them on shutdown — the +SK call still runs. + +## What this exercises + +- `SemanticKernelAdapter` lifecycle (`connect` → `instrument_kernel` → + `disconnect`). +- All three SK filters: `function_invocation`, `prompt_rendering`, + `auto_function_invocation`. +- The HTTP transport sink batched flush path. + +For the full adapter reference see +[`docs/adapters/frameworks-semantic_kernel.md`](../../../docs/adapters/frameworks-semantic_kernel.md). diff --git a/samples/instrument/semantic_kernel/__init__.py b/samples/instrument/semantic_kernel/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/samples/instrument/semantic_kernel/main.py b/samples/instrument/semantic_kernel/main.py new file mode 100644 index 0000000..310180b --- /dev/null +++ b/samples/instrument/semantic_kernel/main.py @@ -0,0 +1,86 @@ +"""Sample: instrument a Semantic Kernel prompt invocation with LayerLens. + +Builds a ``Kernel`` with an OpenAI chat completion service, registers the +LayerLens filters via ``SemanticKernelAdapter.instrument_kernel``, and runs a +single ``invoke_prompt`` call. Filter callbacks emit ``agent.input`` / +``agent.output`` / ``model.invoke`` events that ship to atlas-app via +``HttpEventSink``. + +Required environment: + +* ``OPENAI_API_KEY`` — used by ``OpenAIChatCompletion``. +* ``LAYERLENS_STRATIX_API_KEY`` — your LayerLens API key (optional). +* ``LAYERLENS_STRATIX_BASE_URL`` — atlas-app base URL (optional). + +Run:: + + pip install 'layerlens[semantic-kernel,providers-openai]' + python -m samples.instrument.semantic_kernel.main +""" + +from __future__ import annotations + +import os +import sys +import asyncio + +from layerlens.instrument.adapters._base import CaptureConfig +from layerlens.instrument.transport.sink_http import HttpEventSink +from layerlens.instrument.adapters.frameworks.semantic_kernel import SemanticKernelAdapter + + +async def _run(kernel: object) -> str: + # Imported here to keep the top-level module importable without semantic-kernel. + from semantic_kernel.functions import KernelArguments # type: ignore[import-not-found,unused-ignore] + + result = await kernel.invoke_prompt( # type: ignore[attr-defined] + prompt="Reply with just the digit. What is 2 + 2?", + arguments=KernelArguments(), + ) + return str(result) + + +def main() -> int: + if not os.environ.get("OPENAI_API_KEY"): + print("OPENAI_API_KEY is not set; cannot run sample.", file=sys.stderr) + return 2 + + try: + from semantic_kernel import Kernel + from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion + except ImportError: + print( + "semantic-kernel not installed. Install with:\n" + " pip install 'layerlens[semantic-kernel,providers-openai]'", + file=sys.stderr, + ) + return 2 + + sink = HttpEventSink( + adapter_name="semantic_kernel", + path="/telemetry/spans", + max_batch=10, + flush_interval_s=1.0, + ) + + adapter = SemanticKernelAdapter(capture_config=CaptureConfig.standard()) + adapter.add_sink(sink) + adapter.connect() + + kernel = Kernel() + kernel.add_service(OpenAIChatCompletion(ai_model_id="gpt-4o-mini")) + adapter.instrument_kernel(kernel) + + try: + response = asyncio.run(_run(kernel)) + print(f"Response: {response}") + finally: + sink.close() + adapter.disconnect() + + print("Telemetry shipped. Check the LayerLens dashboard adapter health page.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/layerlens/instrument/adapters/frameworks/__init__.py b/src/layerlens/instrument/adapters/frameworks/__init__.py new file mode 100644 index 0000000..5928acc --- /dev/null +++ b/src/layerlens/instrument/adapters/frameworks/__init__.py @@ -0,0 +1,17 @@ +"""Framework adapters for the LayerLens Instrument layer. + +Each framework adapter wraps an agent / chain framework's lifecycle to +intercept agent runs, model invocations, tool calls, state changes, and +handoffs, emitting events through the LayerLens telemetry pipeline. + +Adapter packages exported by the registry (loaded on demand by +:class:`AdapterRegistry`): + +* ``semantic_kernel`` — Microsoft Semantic Kernel (filter API). + +Importing this package does NOT import any framework SDK. Each +``frameworks.`` package is loaded only when the user requests it +explicitly, e.g. via ``AdapterRegistry.get("semantic_kernel")``. +""" + +from __future__ import annotations diff --git a/src/layerlens/instrument/adapters/frameworks/semantic_kernel/__init__.py b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/__init__.py new file mode 100644 index 0000000..86c09c8 --- /dev/null +++ b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/__init__.py @@ -0,0 +1,48 @@ +"""LayerLens Semantic Kernel Adapter. + +Provides plugin invocation tracing, planner execution tracking, and +memory operation capture for Microsoft Semantic Kernel via the kernel's +native filter API. + +Importing this module does NOT import ``semantic-kernel`` itself — that +dependency is only required when the user calls +:meth:`SemanticKernelAdapter.instrument_kernel` against a real kernel. +""" + +from __future__ import annotations + +from layerlens.instrument.adapters.frameworks.semantic_kernel.filters import ( + LayerLensFunctionFilter, + LayerLensAutoFunctionFilter, + LayerLensPromptRenderFilter, +) +from layerlens.instrument.adapters.frameworks.semantic_kernel.metadata import ( + SKMetadataExtractor, +) +from layerlens.instrument.adapters.frameworks.semantic_kernel.lifecycle import ( + StratixMemoryStore, + SemanticKernelAdapter, +) + +# Registry lazy-loading convention. +ADAPTER_CLASS = SemanticKernelAdapter + +# Backward-compat aliases for users coming from ateam (``STRATIX*`` → +# ``LayerLens*``). The class objects are identical; only the import name +# changes. Slated for removal in the next major SDK release. +STRATIXFunctionFilter = LayerLensFunctionFilter # noqa: N816 - backward-compat alias +STRATIXAutoFunctionFilter = LayerLensAutoFunctionFilter # noqa: N816 - backward-compat alias +STRATIXPromptRenderFilter = LayerLensPromptRenderFilter # noqa: N816 - backward-compat alias + +__all__ = [ + "ADAPTER_CLASS", + "LayerLensAutoFunctionFilter", + "LayerLensFunctionFilter", + "LayerLensPromptRenderFilter", + "SKMetadataExtractor", + "STRATIXAutoFunctionFilter", + "STRATIXFunctionFilter", + "STRATIXPromptRenderFilter", + "SemanticKernelAdapter", + "StratixMemoryStore", +] diff --git a/src/layerlens/instrument/adapters/frameworks/semantic_kernel/filters.py b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/filters.py new file mode 100644 index 0000000..2e30ba8 --- /dev/null +++ b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/filters.py @@ -0,0 +1,259 @@ +""" +Semantic Kernel Filter Implementations + +Provides STRATIX-instrumented filter classes for the SK filter API: +- LayerLensFunctionFilter: Function invocation pre/post hooks +- LayerLensPromptRenderFilter: Prompt template rendering hooks +- LayerLensAutoFunctionFilter: Auto-invoked function hooks +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from layerlens.instrument.adapters.frameworks.semantic_kernel.lifecycle import SemanticKernelAdapter + +logger = logging.getLogger(__name__) + + +class LayerLensFunctionFilter: + """ + Intercepts SK function invocations via the FunctionInvocationFilter API. + + Captures plugin name, function name, arguments, result, and latency. + """ + + def __init__(self, adapter: SemanticKernelAdapter) -> None: + self._adapter = adapter + self._contexts: dict[int, dict[str, Any]] = {} + + async def __call__(self, context: Any, next: Any = None) -> None: + """SK filter callable interface: (context, next=...) -> Awaitable[None].""" + return await self.on_function_invocation(context, next) + + async def on_function_invocation( + self, + context: Any, + next_handler: Any = None, + ) -> None: + """Pre/post hook for function invocation.""" + plugin_name = self._extract_plugin_name(context) + function_name = self._extract_function_name(context) + arguments = self._extract_arguments(context) + + try: + trace_ctx = self._adapter.on_function_start( + plugin_name=plugin_name, + function_name=function_name, + arguments=arguments, + ) + except Exception: + logger.warning("Error in function start hook", exc_info=True) + trace_ctx = {} + + error = None + try: + if next_handler: + await next_handler(context) + except Exception as exc: + error = exc + raise + finally: + try: + result = self._extract_result(context) + self._adapter.on_function_end( + context=trace_ctx, + result=result, + error=error, + ) + except Exception: + logger.warning("Error in function end hook", exc_info=True) + + def on_function_invocation_sync( + self, + plugin_name: str, + function_name: str, + arguments: dict[str, Any] | None = None, + result: Any = None, + error: Exception | None = None, + ) -> None: + """Synchronous hook for testing and non-async usage.""" + try: + trace_ctx = self._adapter.on_function_start( + plugin_name=plugin_name, + function_name=function_name, + arguments=arguments, + ) + self._adapter.on_function_end( + context=trace_ctx, + result=result, + error=error, + ) + except Exception: + logger.warning("Error in sync function hook", exc_info=True) + + @staticmethod + def _extract_plugin_name(context: Any) -> str: + """Extract plugin name from SK invocation context.""" + if hasattr(context, "function"): + fn = context.function + return getattr(fn, "plugin_name", "") or getattr(fn, "skill_name", "") or "" + return getattr(context, "plugin_name", "") or "" + + @staticmethod + def _extract_function_name(context: Any) -> str: + if hasattr(context, "function"): + fn = context.function + return getattr(fn, "name", "") or "" + return getattr(context, "function_name", "") or "" + + @staticmethod + def _extract_arguments(context: Any) -> dict[str, Any] | None: + args = getattr(context, "arguments", None) + if args is None: + return None + if isinstance(args, dict): + return args + if hasattr(args, "items"): + return dict(args.items()) + return None + + @staticmethod + def _extract_result(context: Any) -> Any: + return getattr(context, "result", None) + + +class LayerLensPromptRenderFilter: + """ + Intercepts SK prompt rendering via the PromptRenderFilter API. + + Captures template text and rendered prompt string. + """ + + def __init__(self, adapter: SemanticKernelAdapter) -> None: + self._adapter = adapter + + async def __call__(self, context: Any, next: Any = None) -> None: + """SK filter callable interface.""" + return await self.on_prompt_render(context, next) + + async def on_prompt_render( + self, + context: Any, + next_handler: Any = None, + ) -> None: + """Pre/post hook for prompt rendering.""" + function_name = getattr(context, "function_name", None) or "" + template = getattr(context, "prompt_template", None) + + if next_handler: + await next_handler(context) + + try: + rendered = getattr(context, "rendered_prompt", None) + self._adapter.on_prompt_render( + template=str(template) if template else None, + rendered_prompt=str(rendered) if rendered else None, + function_name=function_name, + ) + except Exception: + logger.warning("Error in prompt render hook", exc_info=True) + + def on_prompt_render_sync( + self, + template: str | None = None, + rendered_prompt: str | None = None, + function_name: str | None = None, + ) -> None: + """Synchronous hook for testing.""" + try: + self._adapter.on_prompt_render( + template=template, + rendered_prompt=rendered_prompt, + function_name=function_name, + ) + except Exception: + logger.warning("Error in sync prompt render hook", exc_info=True) + + +class LayerLensAutoFunctionFilter: + """ + Intercepts LLM-initiated (auto-invoked) function calls via + the AutoFunctionInvocationFilter API. + + Marks all emitted events with auto_invoked=True. + """ + + def __init__(self, adapter: SemanticKernelAdapter) -> None: + self._adapter = adapter + + async def __call__(self, context: Any, next: Any = None) -> None: + """SK filter callable interface.""" + return await self.on_auto_function_invocation(context, next) + + async def on_auto_function_invocation( + self, + context: Any, + next_handler: Any = None, + ) -> None: + """Pre/post hook for auto-invoked functions.""" + plugin_name = LayerLensFunctionFilter._extract_plugin_name(context) + function_name = LayerLensFunctionFilter._extract_function_name(context) + arguments = LayerLensFunctionFilter._extract_arguments(context) + + try: + trace_ctx = self._adapter.on_function_start( + plugin_name=plugin_name, + function_name=function_name, + arguments=arguments, + auto_invoked=True, + ) + except Exception: + logger.warning("Error in auto function start hook", exc_info=True) + trace_ctx = {} + + error = None + try: + if next_handler: + await next_handler(context) + except Exception as exc: + error = exc + raise + finally: + try: + result = LayerLensFunctionFilter._extract_result(context) + self._adapter.on_function_end( + context=trace_ctx, + result=result, + error=error, + auto_invoked=True, + ) + except Exception: + logger.warning("Error in auto function end hook", exc_info=True) + + def on_auto_function_invocation_sync( + self, + plugin_name: str, + function_name: str, + arguments: dict[str, Any] | None = None, + result: Any = None, + error: Exception | None = None, + ) -> None: + """Synchronous hook for testing.""" + try: + trace_ctx = self._adapter.on_function_start( + plugin_name=plugin_name, + function_name=function_name, + arguments=arguments, + auto_invoked=True, + ) + self._adapter.on_function_end( + context=trace_ctx, + result=result, + error=error, + auto_invoked=True, + ) + except Exception: + logger.warning("Error in sync auto function hook", exc_info=True) diff --git a/src/layerlens/instrument/adapters/frameworks/semantic_kernel/lifecycle.py b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/lifecycle.py new file mode 100644 index 0000000..38eab07 --- /dev/null +++ b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/lifecycle.py @@ -0,0 +1,602 @@ +""" +STRATIX Semantic Kernel Lifecycle Hooks + +Provides the main SemanticKernelAdapter class. Instruments SK Kernel +instances via the official filter API (FunctionInvocationFilter, +PromptRenderFilter, AutoFunctionInvocationFilter). +""" + +from __future__ import annotations + +import time +import uuid +import logging +import threading +from typing import Any + +from layerlens.instrument.adapters._base.adapter import ( + AdapterInfo, + BaseAdapter, + AdapterHealth, + AdapterStatus, + ReplayableTrace, + AdapterCapability, +) +from layerlens.instrument.adapters._base.capture import CaptureConfig +from layerlens.instrument.adapters._base.pydantic_compat import PydanticCompat + +logger = logging.getLogger(__name__) + + +class SemanticKernelAdapter(BaseAdapter): + """ + Main adapter for integrating STRATIX with Microsoft Semantic Kernel. + + Instruments Kernel instances via the official SK filter API to capture + plugin invocations, planner executions, memory operations, and LLM calls. + + Usage: + adapter = SemanticKernelAdapter(stratix=stratix_instance) + adapter.connect() + kernel = adapter.instrument_kernel(kernel) + result = await kernel.invoke(my_function, arg1=val1) + """ + + FRAMEWORK = "semantic_kernel" + VERSION = "0.1.0" + # The adapter source files import nothing from ``pydantic`` directly + # (verified by grep across ``frameworks/semantic_kernel/``). The + # adapter only registers SK filter callbacks and emits dict events; + # it never touches Semantic Kernel's own Pydantic models. SK 1.0+ is + # internally Pydantic v2, but customers running older SK 0.x with + # Pydantic v1 can still use this adapter. + requires_pydantic = PydanticCompat.V1_OR_V2 + + def __init__( + self, + stratix: Any | None = None, + capture_config: CaptureConfig | None = None, + memory_service: Any | None = None, + ) -> None: + super().__init__(stratix=stratix, capture_config=capture_config) + + self._adapter_lock = threading.Lock() + self._seen_plugins: set[str] = set() + self._invocation_count: int = 0 + self._kernel_start_ns: int = 0 + self._framework_version: str | None = None + self._filters_registered: list[Any] = [] + self._memory_service = memory_service + + # --- BaseAdapter lifecycle --- + + def connect(self) -> None: + """Verify Semantic Kernel is importable and mark as connected.""" + try: + import semantic_kernel # type: ignore[import-not-found,unused-ignore] # noqa: F401 + + version = getattr(semantic_kernel, "__version__", "unknown") + logger.debug("Semantic Kernel %s detected", version) + except ImportError: + logger.debug("Semantic Kernel not installed; adapter usable in mock/test mode") + self._framework_version = self._detect_framework_version() + self._connected = True + self._status = AdapterStatus.HEALTHY + + def disconnect(self) -> None: + """Disconnect and clear state.""" + self._filters_registered.clear() + self._connected = False + self._status = AdapterStatus.DISCONNECTED + + def health_check(self) -> AdapterHealth: + return AdapterHealth( + status=self._status, + framework_name=self.FRAMEWORK, + framework_version=self._framework_version, + adapter_version=self.VERSION, + error_count=self._error_count, + circuit_open=self._circuit_open, + ) + + def get_adapter_info(self) -> AdapterInfo: + return AdapterInfo( + name="SemanticKernelAdapter", + version=self.VERSION, + framework=self.FRAMEWORK, + framework_version=self._framework_version, + capabilities=[ + AdapterCapability.TRACE_TOOLS, + AdapterCapability.TRACE_MODELS, + AdapterCapability.TRACE_STATE, + ], + description="LayerLens adapter for Microsoft Semantic Kernel", + ) + + def serialize_for_replay(self) -> ReplayableTrace: + return ReplayableTrace( + adapter_name="SemanticKernelAdapter", + framework=self.FRAMEWORK, + trace_id=str(uuid.uuid4()), + events=list(self._trace_events), + state_snapshots=[], + config={ + "capture_config": self._capture_config.model_dump(), + }, + ) + + # --- Kernel instrumentation --- + + def instrument_kernel(self, kernel: Any) -> Any: + """ + Instrument a Semantic Kernel instance with STRATIX tracing. + + Registers filter instances on the kernel for function invocations, + prompt rendering, and auto-function invocations. + + Args: + kernel: A semantic_kernel.Kernel instance + + Returns: + The modified kernel (same object, with filters attached) + """ + from layerlens.instrument.adapters.frameworks.semantic_kernel.filters import ( + LayerLensFunctionFilter, + LayerLensAutoFunctionFilter, + LayerLensPromptRenderFilter, + ) + + func_filter = LayerLensFunctionFilter(adapter=self) + prompt_filter = LayerLensPromptRenderFilter(adapter=self) + auto_filter = LayerLensAutoFunctionFilter(adapter=self) + + # Register filters via SK's filter API + try: + if hasattr(kernel, "add_filter"): + kernel.add_filter("function_invocation", func_filter) + kernel.add_filter("prompt_rendering", prompt_filter) + kernel.add_filter("auto_function_invocation", auto_filter) + self._filters_registered = [func_filter, prompt_filter, auto_filter] + else: + # Fallback: store on kernel for callback-based approach + kernel._stratix_filters = [func_filter, prompt_filter, auto_filter] + self._filters_registered = [func_filter, prompt_filter, auto_filter] + except Exception: + logger.warning("Could not register filters on kernel", exc_info=True) + + kernel._stratix_adapter = self + + # Discover registered plugins + self._discover_plugins(kernel) + + return kernel + + # --- Lifecycle hooks (called by filters) --- + + def on_function_start( + self, + plugin_name: str, + function_name: str, + arguments: dict[str, Any] | None = None, + auto_invoked: bool = False, + ) -> dict[str, Any]: + """ + Handle function invocation start. + + Returns context dict for correlation with on_function_end. + """ + with self._adapter_lock: + self._invocation_count += 1 + invocation_seq = self._invocation_count + + context = { + "start_ns": time.time_ns(), + "invocation_seq": invocation_seq, + "plugin_name": plugin_name, + "function_name": function_name, + } + + # Emit agent config on first plugin encounter + with self._adapter_lock: + if plugin_name not in self._seen_plugins: + self._seen_plugins.add(plugin_name) + self.emit_dict_event( + "environment.config", + { + "framework": "semantic_kernel", + "plugin_name": plugin_name, + "function_name": function_name, + }, + ) + + return context + + def on_function_end( + self, + context: dict[str, Any], + result: Any = None, + error: Exception | None = None, + auto_invoked: bool = False, + ) -> None: + """ + Handle function invocation end. + + Emits tool.call (L5a) for plugin functions. + """ + start_ns = context.get("start_ns", 0) + elapsed_ms = (time.time_ns() - start_ns) / 1_000_000 if start_ns else 0 + + payload: dict[str, Any] = { + "framework": "semantic_kernel", + "tool_name": f"{context.get('plugin_name', '')}.{context.get('function_name', '')}", + "plugin_name": context.get("plugin_name"), + "function_name": context.get("function_name"), + "latency_ms": elapsed_ms, + "invocation_seq": context.get("invocation_seq"), + } + + if auto_invoked: + payload["auto_invoked"] = True + + if result is not None: + payload["result_preview"] = self._truncate(self._safe_serialize(result)) + + if error: + payload["error"] = str(error) + + self.emit_dict_event("tool.call", payload) + + def on_prompt_render( + self, + template: str | None = None, + rendered_prompt: str | None = None, + function_name: str | None = None, + ) -> None: + """ + Handle prompt template rendering. + + Emits agent.code (L2) for template rendering events. + """ + payload: dict[str, Any] = { + "framework": "semantic_kernel", + "event_subtype": "prompt_render", + } + if function_name: + payload["function_name"] = function_name + if template: + payload["template_preview"] = self._truncate(template, 500) + if rendered_prompt: + payload["rendered_preview"] = self._truncate(rendered_prompt, 500) + + self.emit_dict_event("agent.code", payload) + + def on_model_invoke( + self, + provider: str | None = None, + model: str | None = None, + prompt_tokens: int | None = None, + completion_tokens: int | None = None, + latency_ms: float | None = None, + error: str | None = None, + messages: list[dict[str, str]] | None = None, + ) -> None: + """ + Handle LLM call from SK service. + + Emits model.invoke (L3) and cost.record (cross-cutting). + """ + payload: dict[str, Any] = { + "framework": "semantic_kernel", + } + if provider: + payload["provider"] = provider + if model: + payload["model"] = model + if prompt_tokens is not None: + payload["prompt_tokens"] = prompt_tokens + if completion_tokens is not None: + payload["completion_tokens"] = completion_tokens + if latency_ms is not None: + payload["latency_ms"] = latency_ms + if error: + payload["error"] = error + if self._capture_config.capture_content and messages: + payload["messages"] = messages + + self.emit_dict_event("model.invoke", payload) + + # Emit cost record + if prompt_tokens or completion_tokens: + self.emit_dict_event( + "cost.record", + { + "framework": "semantic_kernel", + "provider": provider, + "model": model, + "prompt_tokens": prompt_tokens or 0, + "completion_tokens": completion_tokens or 0, + "total_tokens": (prompt_tokens or 0) + (completion_tokens or 0), + }, + ) + + def on_planner_step( + self, + planner_type: str, + step_index: int | None = None, + plan: Any = None, + thought: str | None = None, + action: str | None = None, + observation: str | None = None, + status: str | None = None, + ) -> None: + """ + Handle planner execution step. + + Emits agent.code (L2) for plan generation and step execution. + """ + payload: dict[str, Any] = { + "framework": "semantic_kernel", + "event_subtype": "planner_step", + "planner_type": planner_type, + } + if step_index is not None: + payload["step_index"] = step_index + if plan is not None: + payload["plan_preview"] = self._truncate(str(plan), 1000) + if thought: + payload["thought"] = self._truncate(thought) + if action: + payload["action"] = action + if observation: + payload["observation"] = self._truncate(observation) + if status: + payload["status"] = status + + self.emit_dict_event("agent.code", payload) + + def on_memory_operation( + self, + operation: str, + collection: str | None = None, + key: str | None = None, + query: str | None = None, + result_count: int | None = None, + relevance_scores: list[float] | None = None, + backend_type: str | None = None, + ) -> None: + """ + Handle memory operation (save, search, get). + + Emits tool.call (L5a) for memory operations. + """ + payload: dict[str, Any] = { + "framework": "semantic_kernel", + "tool_name": f"memory.{operation}", + "operation": operation, + } + if collection: + payload["collection"] = collection + if key: + payload["key"] = key + if query: + payload["query_preview"] = self._truncate(query, 200) + if result_count is not None: + payload["result_count"] = result_count + if relevance_scores: + payload["relevance_scores"] = relevance_scores[:10] + if backend_type: + payload["backend_type"] = backend_type + + self.emit_dict_event("tool.call", payload) + + def on_kernel_invoke_start(self, input_text: Any = None) -> None: + """Handle kernel invocation start. Emits agent.input (L1).""" + with self._adapter_lock: + self._kernel_start_ns = time.time_ns() + + self.emit_dict_event( + "agent.input", + { + "framework": "semantic_kernel", + "input": self._safe_serialize(input_text), + "timestamp_ns": self._kernel_start_ns, + }, + ) + + def on_kernel_invoke_end( + self, + output: Any = None, + error: Exception | None = None, + ) -> None: + """Handle kernel invocation end. Emits agent.output (L1).""" + end_ns = time.time_ns() + duration_ns = end_ns - self._kernel_start_ns if self._kernel_start_ns else 0 + + payload: dict[str, Any] = { + "framework": "semantic_kernel", + "output": self._safe_serialize(output), + "duration_ns": duration_ns, + } + if error: + payload["error"] = str(error) + + self.emit_dict_event("agent.output", payload) + + # --- Plugin discovery --- + + def _discover_plugins(self, kernel: Any) -> None: + """Discover and register plugins from the kernel.""" + try: + plugins = getattr(kernel, "plugins", None) + if plugins is None: + return + if isinstance(plugins, dict) or hasattr(plugins, "keys"): + plugin_names = list(plugins.keys()) + else: + plugin_names = [str(p) for p in plugins] + + for name in plugin_names: + with self._adapter_lock: + if name not in self._seen_plugins: + self._seen_plugins.add(name) + self.emit_dict_event( + "environment.config", + { + "framework": "semantic_kernel", + "plugin_name": name, + "event_subtype": "plugin_registered", + }, + ) + except Exception: + logger.debug("Error discovering SK plugins", exc_info=True) + + # --- Internal helpers --- + + def _safe_serialize(self, value: Any) -> Any: + """Safely serialize a value for events.""" + try: + if value is None: + return None + if hasattr(value, "model_dump"): + return value.model_dump() + if hasattr(value, "dict"): + return value.dict() + if isinstance(value, dict): + return dict(value) + if isinstance(value, (str, int, float, bool)): + return value + return str(value) + except Exception: + return str(value) + + def _truncate(self, text: Any, max_len: int = 500) -> str: + """Truncate text to max_len.""" + text_str = str(text) if not isinstance(text, str) else text + if len(text_str) <= max_len: + return text_str + return text_str[:max_len] + "..." + + @staticmethod + def _detect_framework_version() -> str | None: + try: + import semantic_kernel # type: ignore[import-not-found,unused-ignore] + + return getattr(semantic_kernel, "__version__", None) + except ImportError: + return None + + +class StratixMemoryStore: + """Semantic Kernel memory store backed by AgentMemoryService. + + Implements the SK memory store interface (``save_information``, + ``get_nearest_matches``) by delegating to the STRATIX + ``AgentMemoryService``. This allows SK applications to use + STRATIX persistent memory without changing their code. + + Usage:: + + from stratix.memory.service import AgentMemoryService # type: ignore[import-not-found,import-untyped,unused-ignore] + + memory_svc = AgentMemoryService(crud_store) + store = StratixMemoryStore(memory_svc, agent_id="my-agent", org_id="org-1") + + # Inside SK: + await store.save_information( + collection="facts", + text="Paris is the capital of France", + id="fact-1", + ) + matches = await store.get_nearest_matches( + collection="facts", + query="capital of France", + limit=3, + ) + """ + + def __init__( + self, + memory_service: Any, + agent_id: str = "semantic_kernel", + org_id: str = "", + ) -> None: + """Initialise the memory store. + + Args: + memory_service: An ``AgentMemoryService`` instance. + agent_id: Agent identifier used for all memory entries. + org_id: Organisation identifier used for all memory entries. + """ + self._memory_service = memory_service + self._agent_id = agent_id + self._org_id = org_id + + async def save_information( + self, + collection: str, + text: str, + id: str, # noqa: A002 — matches SK interface + description: str | None = None, + additional_metadata: str | None = None, + ) -> None: + """Save a piece of information into the memory store. + + Delegates to ``AgentMemoryService.store()`` with + ``memory_type="semantic"`` and the collection as namespace. + + Args: + collection: SK memory collection name (mapped to namespace). + text: Text content to store. + id: Unique identifier for this memory. + description: Optional description (stored in metadata). + additional_metadata: Optional extra metadata string. + """ + from layerlens.instrument._vendored.memory_models import MemoryEntry + + metadata: dict[str, Any] = {"source": "semantic_kernel_memory_store"} + if description: + metadata["description"] = description + if additional_metadata: + metadata["additional"] = additional_metadata + + entry = MemoryEntry( + id=id, + org_id=self._org_id, + agent_id=self._agent_id, + memory_type="semantic", + namespace=collection, + key=id, + content=text, + importance=0.5, + metadata=metadata, + ) + self._memory_service.store(entry) + + async def get_nearest_matches( + self, + collection: str, + query: str, + limit: int = 5, + min_relevance_score: float = 0.0, + ) -> list[tuple[Any, float]]: + """Retrieve the nearest matches for a query. + + Delegates to ``AgentMemoryService.search()`` and returns results + in the SK-expected format of ``(MemoryEntry, relevance_score)`` + tuples. + + Args: + collection: SK memory collection name (used as search context). + query: Text query to match against memory content. + limit: Maximum number of results to return. + min_relevance_score: Minimum relevance threshold (reserved for + future vector search support; currently unused). + + Returns: + List of ``(MemoryEntry, score)`` tuples ordered by importance. + """ + results = self._memory_service.search(self._agent_id, query, limit=limit) + # Filter to the requested collection/namespace + filtered = [r for r in results if r.namespace == collection] + # Return as (entry, relevance) tuples — importance serves as proxy score + return [(entry, entry.importance) for entry in filtered] diff --git a/src/layerlens/instrument/adapters/frameworks/semantic_kernel/metadata.py b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/metadata.py new file mode 100644 index 0000000..ee6275e --- /dev/null +++ b/src/layerlens/instrument/adapters/frameworks/semantic_kernel/metadata.py @@ -0,0 +1,60 @@ +""" +Semantic Kernel Metadata Extraction + +Extracts plugin and kernel configuration metadata for environment.config events. +""" + +from __future__ import annotations + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + + +class SKMetadataExtractor: + """Extract metadata from Semantic Kernel components.""" + + def extract_plugin_metadata(self, plugin: Any) -> dict[str, Any]: + """Extract metadata from a registered plugin.""" + metadata: dict[str, Any] = {} + try: + metadata["plugin_name"] = getattr(plugin, "name", str(plugin)) + metadata["description"] = getattr(plugin, "description", None) + + # Extract function names + functions = getattr(plugin, "functions", None) + if functions: # noqa: SIM102 + if isinstance(functions, dict) or hasattr(functions, "keys"): + metadata["function_names"] = list(functions.keys()) + except Exception: + logger.debug("Error extracting plugin metadata", exc_info=True) + return metadata + + def extract_kernel_metadata(self, kernel: Any) -> dict[str, Any]: + """Extract metadata from a Kernel instance.""" + metadata: dict[str, Any] = {} + try: + # Extract registered plugins + plugins = getattr(kernel, "plugins", None) + if plugins: + if isinstance(plugins, dict): + metadata["plugin_count"] = len(plugins) + metadata["plugin_names"] = list(plugins.keys()) + elif hasattr(plugins, "__len__"): + metadata["plugin_count"] = len(plugins) + + # Extract registered services + services = getattr(kernel, "services", None) + if services and isinstance(services, dict): + metadata["service_count"] = len(services) + metadata["service_types"] = [type(s).__name__ for s in services.values()] + + # Extract memory backend + memory = getattr(kernel, "memory", None) + if memory: + metadata["memory_backend"] = type(memory).__name__ + + except Exception: + logger.debug("Error extracting kernel metadata", exc_info=True) + return metadata diff --git a/tests/instrument/adapters/__init__.py b/tests/instrument/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/instrument/adapters/frameworks/__init__.py b/tests/instrument/adapters/frameworks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/instrument/adapters/frameworks/test_semantic_kernel.py b/tests/instrument/adapters/frameworks/test_semantic_kernel.py new file mode 100644 index 0000000..2539048 --- /dev/null +++ b/tests/instrument/adapters/frameworks/test_semantic_kernel.py @@ -0,0 +1,212 @@ +"""Unit tests for the Microsoft Semantic Kernel adapter. + +Mocked at the SDK shape level — no real ``semantic_kernel`` runtime needed. +The adapter wires filters via ``kernel.add_filter(...)`` and exposes a +suite of lifecycle hooks (``on_function_start``, ``on_model_invoke``, +``on_planner_step``, etc.) that are called by those filters. Tests +exercise the lifecycle hooks directly + verify filter wiring. +""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from layerlens.instrument.adapters._base import AdapterStatus, CaptureConfig +from layerlens.instrument.adapters.frameworks.semantic_kernel import ( + ADAPTER_CLASS, + SemanticKernelAdapter, +) + + +class _RecordingStratix: + def __init__(self) -> None: + self.events: List[Dict[str, Any]] = [] + + def emit(self, *args: Any, **kwargs: Any) -> None: + if len(args) == 2 and isinstance(args[0], str): + self.events.append({"event_type": args[0], "payload": args[1]}) + + +class _FakeKernel: + def __init__(self, plugins: Any = None) -> None: + self.plugins = plugins or {} + self._added_filters: List[Dict[str, Any]] = [] + + def add_filter(self, filter_type: str, filter_obj: Any) -> None: + self._added_filters.append({"type": filter_type, "filter": filter_obj}) + + +def test_adapter_class_export() -> None: + assert ADAPTER_CLASS is SemanticKernelAdapter + + +def test_lifecycle() -> None: + a = SemanticKernelAdapter() + a.connect() + assert a.status == AdapterStatus.HEALTHY + a.disconnect() + assert a.status == AdapterStatus.DISCONNECTED + + +def test_adapter_info_and_health() -> None: + a = SemanticKernelAdapter() + a.connect() + info = a.get_adapter_info() + assert info.framework == "semantic_kernel" + assert info.name == "SemanticKernelAdapter" + health = a.health_check() + assert health.framework_name == "semantic_kernel" + + +def test_instrument_kernel_registers_filters_and_discovers_plugins() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + + kernel = _FakeKernel(plugins={"math": object(), "search": object()}) + adapter.instrument_kernel(kernel) + + filter_types = {f["type"] for f in kernel._added_filters} + assert filter_types == {"function_invocation", "prompt_rendering", "auto_function_invocation"} + + # Plugin discovery emits environment.config events. + configs = [e for e in stratix.events if e["event_type"] == "environment.config"] + plugin_names = {c["payload"].get("plugin_name") for c in configs} + assert "math" in plugin_names + assert "search" in plugin_names + + +def test_on_function_start_end_emits_tool_call() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + + ctx = adapter.on_function_start(plugin_name="math", function_name="add", arguments={"a": 1, "b": 2}) + adapter.on_function_end(context=ctx, result=3) + + evt = next(e for e in stratix.events if e["event_type"] == "tool.call") + assert evt["payload"]["tool_name"] == "math.add" + assert evt["payload"]["plugin_name"] == "math" + assert evt["payload"]["function_name"] == "add" + assert evt["payload"]["latency_ms"] >= 0 + + +def test_on_model_invoke_emits_invoke_and_cost() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + + adapter.on_model_invoke( + provider="azure_openai", + model="gpt-5", + prompt_tokens=10, + completion_tokens=5, + latency_ms=20.0, + ) + + invoke = next(e for e in stratix.events if e["event_type"] == "model.invoke") + assert invoke["payload"]["model"] == "gpt-5" + assert invoke["payload"]["latency_ms"] == 20.0 + + cost = next(e for e in stratix.events if e["event_type"] == "cost.record") + assert cost["payload"]["total_tokens"] == 15 + + +def test_on_prompt_render_emits_agent_code() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + + adapter.on_prompt_render( + template="Hello {{name}}", + rendered_prompt="Hello world", + function_name="greet", + ) + + evt = next(e for e in stratix.events if e["event_type"] == "agent.code") + assert evt["payload"]["event_subtype"] == "prompt_render" + assert evt["payload"]["function_name"] == "greet" + + +def test_on_planner_step_emits_agent_code() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + + adapter.on_planner_step( + planner_type="HandlebarsPlanner", + step_index=1, + thought="I need to search", + action="search", + observation="found results", + status="completed", + ) + + evt = next(e for e in stratix.events if e["event_type"] == "agent.code") + assert evt["payload"]["event_subtype"] == "planner_step" + assert evt["payload"]["planner_type"] == "HandlebarsPlanner" + assert evt["payload"]["step_index"] == 1 + + +def test_on_memory_operation_emits_tool_call() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + + adapter.on_memory_operation( + operation="search", + collection="facts", + query="capital of France", + result_count=3, + relevance_scores=[0.9, 0.8, 0.7], + backend_type="qdrant", + ) + + evt = next(e for e in stratix.events if e["event_type"] == "tool.call") + assert evt["payload"]["tool_name"] == "memory.search" + assert evt["payload"]["result_count"] == 3 + assert evt["payload"]["backend_type"] == "qdrant" + + +def test_on_kernel_invoke_start_end_emits_input_output() -> None: + stratix = _RecordingStratix() + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=CaptureConfig.full()) + adapter.connect() + adapter.on_kernel_invoke_start(input_text="hello") + adapter.on_kernel_invoke_end(output="world") + + types = [e["event_type"] for e in stratix.events] + assert "agent.input" in types + assert "agent.output" in types + + out = next(e for e in stratix.events if e["event_type"] == "agent.output") + assert out["payload"]["output"] == "world" + assert out["payload"]["duration_ns"] >= 0 + + +def test_capture_config_gates_l5a_tool_calls() -> None: + """When l5a_tool_calls is disabled, tool.call does NOT fire (model.invoke still does).""" + stratix = _RecordingStratix() + cfg = CaptureConfig(l5a_tool_calls=False) + adapter = SemanticKernelAdapter(stratix=stratix, capture_config=cfg) + adapter.connect() + + ctx = adapter.on_function_start(plugin_name="math", function_name="add") + adapter.on_function_end(context=ctx, result=3) + adapter.on_model_invoke(model="gpt-5", prompt_tokens=10, completion_tokens=5) + + types = [e["event_type"] for e in stratix.events] + assert "tool.call" not in types + assert "model.invoke" in types + + +def test_serialize_for_replay() -> None: + adapter = SemanticKernelAdapter( + stratix=_RecordingStratix(), + capture_config=CaptureConfig.full(), + ) + adapter.connect() + rt = adapter.serialize_for_replay() + assert rt.framework == "semantic_kernel" + assert rt.adapter_name == "SemanticKernelAdapter" + assert "capture_config" in rt.config