Index

agent_cover.instrumentation

Instrumentation Core.

This package orchestrates the interception of agent activities. It employs a Hybrid Instrumentation Strategy supported by a shared infrastructure layer.

🧠 Instrumentation Strategies

AgentCover distinguishes between Structured Objects (like Classes/Methods) and Unstructured Data (like Strings).

1. Direct Patching (Runtime Wrapping)

We wrap defined Python objects to intercept their lifecycle. Crucially, these patched objects interact to establish and respect the Execution Scope.

A. The Context System (Avoiding False Positives)

Some components (Tools, Prompts) are ambiguous: they can be called by the Agent or by your test setup code (e.g., asserting a prompt matches an expected string). To prevent false positives, we distinguish between Providers and Consumers:

• Scope Providers: Agents wrap the main entry point (e.g., invoke). They activate the AgentContextManager, signaling "Agent Logic is running".

• Scope Consumers: Tools and Prompts wrap execution methods. They check the AgentContextManager. If the scope is inactive, they ignore the call.

B. Context-Agnostic Inspectors

Certain components inherently belong to the agent's runtime or represent system boundaries. They do not check the AgentContextManager because their execution implies a relevant event:

• LLM Providers: Wraps API calls (e.g., openai.create).

• Why Agnostic: An LLM call is an expensive I/O operation that defines the agent's behavior. Whether it happens inside a framework or in a custom script, it is always a "Decision Point" worth analyzing. It represents the result of the agent's thinking, so we capture it unconditionally.

• PromptFlow Runtime: Wraps PromptFlow internals (render_jinja_template, @tool).

• Why Agnostic: PromptFlow is an orchestration framework. If its internal rendering functions are called, the agent execution is essentially active by definition.

• Callbacks (detecting Raw Strings): Injects listeners into framework event loops (e.g., LangChain).

• Why Agnostic: The Agent's CallbackManager is architecturally bound to the agent's lifecycle: it only executes when the agent runs, guaranteeing an implicit valid scope.

Configuration (targets.py): To decouple patching logic from library definitions, each instrumentor relies on a specific targets.py registry (e.g., agents.targets).

2. Callback Injection (Event Listening)

Used for: Raw Strings.

Why: You cannot "patch" a python string definition (e.g., PROMPT = "..."). To track if a raw string was used, we must wait until it passes through a choke point. In frameworks like LangChain, the Callback System is that choke point.

• We inject a CoverageCallbackHandler that listens for on_llm_start.
• We scan the compiled prompt text sent to the LLM.
• We match this text against the regex patterns discovered by scan_raw_string_prompts.

3. Static Analysis (Source Scanning)

Used for: Discovery, Raw Strings Scanner, Structures Scanner, PromptFlow Scanner.

Why: Before code runs, we need a "map" of what exists. We scan source files and loaded modules to find:

• Files & Modules: The discovery module walks the file system to import user code.
• Heuristics: The raw_strings scanner reads source code to find global variables matching specific patterns.
• Data Structures: The structures scanner inspects Pydantic models to auto-generate decision rules.
• Definitions: The promptflow scanner parses YAML DAGs to identify Jinja templates.

🛠️ Core Infrastructure

The strategies above rely on a set of shared components defined in this package:

• BaseInstrumentor (base.py): The abstract base class for all instrumentors. It handles the lifecycle of applying/reverting patches (instrument()/uninstrument()), ensures idempotency (preventing double-patching), and manages Version Checkers to skip unsupported library versions.

• OutputAnalyzer (analyzer.py): The verification engine. It receives raw text payloads (from LLM providers or callbacks) and performs "Fuzzy Parsing" to check if they satisfy the business rules defined in agent-cover.yaml.

• PatchManager (base.py): A utility that performs the actual setattr / getattr operations safely, ensuring that original methods can always be restored during cleanup.

• WrapperStrategies (strategies.py): A collection of specialized wrappers (Sync, Async, Generator) that handle the complex logic of maintaining the ContextVar state across different execution models.

• TargetConfig (definitions.py): Data classes used to strictly type the configuration of targets, ensuring validation of module paths and version strings.

Usage

While mostly used internally by the pytest plugin, you can manually trigger instrumentation:

from agent_cover.instrumentation import instrument_all
from agent_cover.registry import get_registry

# 1. Apply patches
instrument_all()

# 2. Run your agent code
my_agent.invoke("Hello world")

# 3. Check registry
reg = get_registry()
print(reg.executions)

Classes

AgentInstrumentor

Bases: BaseInstrumentor

Instruments agent classes with context management strategies.

This class handles the lifecycle of patching agent runners. It identifies target modules, loads them if necessary, resolves the specific classes, and wraps execution methods with the appropriate context managers.

Attributes:

Name	Type	Description
targets_provider		A callable returning a list of targets to instrument.
strategies		A dictionary mapping strategy keys to WrapperStrategy instances.
registry		The agent registry instance.
context_manager		The context manager for handling agent contexts.
patch_manager		The manager responsible for applying and reverting patches.
module_iterator		A callable returning the current snapshot of sys.modules.
importer_func		A callable used to import modules by name.
is_instrumented		A boolean indicating if instrumentation has been applied.

Methods:

Name	Description
instrument	Performs the instrumentation process on all defined targets.

Source code in src/agent_cover/instrumentation/agents/patcher.py

class AgentInstrumentor(BaseInstrumentor):
    """Instruments agent classes with context management strategies.

    This class handles the lifecycle of patching agent runners. It identifies
    target modules, loads them if necessary, resolves the specific classes,
    and wraps execution methods with the appropriate context managers.

    Attributes:
        targets_provider: A callable returning a list of targets to instrument.
        strategies: A dictionary mapping strategy keys to WrapperStrategy instances.
        registry: The agent registry instance.
        context_manager: The context manager for handling agent contexts.
        patch_manager: The manager responsible for applying and reverting patches.
        module_iterator: A callable returning the current snapshot of sys.modules.
        importer_func: A callable used to import modules by name.
        is_instrumented: A boolean indicating if instrumentation has been applied.

    Methods:
        instrument: Performs the instrumentation process on all defined targets.
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        context_manager: Optional[AgentContextManager] = None,
        patch_manager: Optional[PatchManager] = None,
        module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
        importer_func: Optional[Callable[[str], Any]] = None,
        targets_provider: Optional[Callable[[], TargetList]] = None,
        strategies: Optional[Dict[str, WrapperStrategy]] = None,
    ):
        """Initializes the AgentInstrumentor.

        Args:
            registry: Optional registry for tracking agents.
            context_manager: Optional manager for agent context.
            patch_manager: Optional manager for handling patches.
            module_iterator: Optional callable to get current modules.
            importer_func: Optional callable to import modules.
            targets_provider: Optional callable to provide instrumentation targets.
            strategies: Optional dictionary of instrumentation strategies.
        """
        super().__init__(
            registry, context_manager, patch_manager, module_iterator, importer_func
        )
        self.targets_provider = targets_provider or _default_targets_provider
        self.strategies = strategies or DEFAULT_STRATEGIES

    def instrument(self):
        """Applies instrumentation to all discoverable agent targets.

        This method iterates through the targets provided by the targets_provider.
        It attempts to load the module if it is not present, resolves the target
        class using the base class resolver, and applies the configured wrapper
        strategies to the specified methods.
        """
        if self.is_instrumented:
            return

        raw_targets = self.targets_provider()
        targets = self._normalize_targets(raw_targets)

        modules_snapshot = self.module_iterator()

        for target in targets:
            if not self._should_instrument(target):
                continue

            mod_name = target.module
            cls_name = target.class_name
            method_map = target.methods

            try:
                # Attempt lazy import
                if mod_name not in modules_snapshot:
                    try:
                        self.importer(mod_name)
                        modules_snapshot = self.module_iterator()
                    except ModuleNotFoundError:
                        Log.log_skip_missing_module(logger, mod_name)
                        continue
                    except Exception as e:
                        logger.warning(e, exc_info=True)
                        continue

                if mod_name in modules_snapshot:
                    mod = modules_snapshot[mod_name]

                    # [CHANGE] Use the robust resolver from the base class
                    cls = self._resolve_target_class(mod, cls_name)

                    if cls and "_ac_agent_patched" not in cls.__dict__:
                        self._patch_agent_runner(cls, method_map)
                        cls._ac_agent_patched = True

            except Exception as e:
                logger.warning(e, exc_info=True)

        self.is_instrumented = True

    def _patch_agent_runner(self, cls: Any, method_map: Dict[str, str]):
        """Applies wrappers to the specified methods of an agent class.

        Args:
            cls: The class object to patch.
            method_map: A dictionary mapping method names to strategy keys.
        """
        for method_name, strategy_key in method_map.items():
            if not hasattr(cls, method_name):
                continue

            strategy = self.strategies.get(strategy_key)
            if not strategy:
                continue

            original = getattr(cls, method_name)
            wrapper = strategy.wrap(original, self.context_manager)

            self._safe_patch(cls, method_name, wrapper)

Functions

__init__(registry=None, context_manager=None, patch_manager=None, module_iterator=None, importer_func=None, targets_provider=None, strategies=None)

Initializes the AgentInstrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	Optional registry for tracking agents.	None
context_manager	Optional[AgentContextManager]	Optional manager for agent context.	None
patch_manager	Optional[PatchManager]	Optional manager for handling patches.	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	Optional callable to get current modules.	None
importer_func	Optional[Callable[[str], Any]]	Optional callable to import modules.	None
targets_provider	Optional[Callable[[], TargetList]]	Optional callable to provide instrumentation targets.	None
strategies	Optional[Dict[str, WrapperStrategy]]	Optional dictionary of instrumentation strategies.	None

Source code in src/agent_cover/instrumentation/agents/patcher.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    context_manager: Optional[AgentContextManager] = None,
    patch_manager: Optional[PatchManager] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    importer_func: Optional[Callable[[str], Any]] = None,
    targets_provider: Optional[Callable[[], TargetList]] = None,
    strategies: Optional[Dict[str, WrapperStrategy]] = None,
):
    """Initializes the AgentInstrumentor.

    Args:
        registry: Optional registry for tracking agents.
        context_manager: Optional manager for agent context.
        patch_manager: Optional manager for handling patches.
        module_iterator: Optional callable to get current modules.
        importer_func: Optional callable to import modules.
        targets_provider: Optional callable to provide instrumentation targets.
        strategies: Optional dictionary of instrumentation strategies.
    """
    super().__init__(
        registry, context_manager, patch_manager, module_iterator, importer_func
    )
    self.targets_provider = targets_provider or _default_targets_provider
    self.strategies = strategies or DEFAULT_STRATEGIES

instrument()

Applies instrumentation to all discoverable agent targets.

This method iterates through the targets provided by the targets_provider. It attempts to load the module if it is not present, resolves the target class using the base class resolver, and applies the configured wrapper strategies to the specified methods.

Source code in src/agent_cover/instrumentation/agents/patcher.py

def instrument(self):
    """Applies instrumentation to all discoverable agent targets.

    This method iterates through the targets provided by the targets_provider.
    It attempts to load the module if it is not present, resolves the target
    class using the base class resolver, and applies the configured wrapper
    strategies to the specified methods.
    """
    if self.is_instrumented:
        return

    raw_targets = self.targets_provider()
    targets = self._normalize_targets(raw_targets)

    modules_snapshot = self.module_iterator()

    for target in targets:
        if not self._should_instrument(target):
            continue

        mod_name = target.module
        cls_name = target.class_name
        method_map = target.methods

        try:
            # Attempt lazy import
            if mod_name not in modules_snapshot:
                try:
                    self.importer(mod_name)
                    modules_snapshot = self.module_iterator()
                except ModuleNotFoundError:
                    Log.log_skip_missing_module(logger, mod_name)
                    continue
                except Exception as e:
                    logger.warning(e, exc_info=True)
                    continue

            if mod_name in modules_snapshot:
                mod = modules_snapshot[mod_name]

                # [CHANGE] Use the robust resolver from the base class
                cls = self._resolve_target_class(mod, cls_name)

                if cls and "_ac_agent_patched" not in cls.__dict__:
                    self._patch_agent_runner(cls, method_map)
                    cls._ac_agent_patched = True

        except Exception as e:
            logger.warning(e, exc_info=True)

    self.is_instrumented = True

CoverageCallbackHandler

Bases: BaseCallbackHandler

A custom callback handler that integrates with AgentCover to track and analyze LLM and tool usage.

This handler acts as a bridge between the LangChain execution lifecycle and the AgentRegistry. It performs two main tasks:

1. Raw String Detection: During on_llm_start, it scans the prompt text to see if it matches any raw string patterns registered in the system.
2. Output Analysis: During on_llm_end or on_chain_end, it captures the generated text and passes it to the OutputAnalyzer to verify if business decisions/rules are met.

Attributes:

Name	Type	Description
registry	AgentRegistry	The registry instance used to track execution coverage.
analyzer	OutputAnalyzer	The component used to analyze LLM generations against configured decisions.

Examples:

Manually using the handler without the global instrumentor:

from langchain_chat.chat_models import ChatOpenAI
from agent_cover.instrumentation.callbacks import CoverageCallbackHandler

# Create the handler
handler = CoverageCallbackHandler()

# Pass it to a LangChain model
llm = ChatOpenAI(callbacks=[handler])
llm.invoke("Hello world")

Source code in src/agent_cover/instrumentation/callbacks.py

class CoverageCallbackHandler(BaseCallbackHandler):
    """A custom callback handler that integrates with AgentCover to track and analyze LLM and tool usage.

    This handler acts as a bridge between the LangChain execution lifecycle and the
    [AgentRegistry][agent_cover.registry.AgentRegistry]. It performs two main tasks:

    1.  **Raw String Detection**: During `on_llm_start`, it scans the prompt text to see if it
        matches any raw string patterns registered in the system.
    2.  **Output Analysis**: During `on_llm_end` or `on_chain_end`, it captures the generated
        text and passes it to the [OutputAnalyzer][agent_cover.instrumentation.analyzer.OutputAnalyzer]
        to verify if business decisions/rules are met.

    Attributes:
        registry (AgentRegistry): The registry instance used to track execution coverage.
        analyzer (OutputAnalyzer): The component used to analyze LLM generations against configured decisions.

    Examples:
        Manually using the handler without the global instrumentor:

        ```python
        from langchain_chat.chat_models import ChatOpenAI
        from agent_cover.instrumentation.callbacks import CoverageCallbackHandler

        # Create the handler
        handler = CoverageCallbackHandler()

        # Pass it to a LangChain model
        llm = ChatOpenAI(callbacks=[handler])
        llm.invoke("Hello world")
        ```
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        analyzer: Optional[OutputAnalyzer] = None,
    ):
        """Initializes the CoverageCallbackHandler.

        Args:
            registry: An optional `AgentRegistry` instance. If not provided,
                the global singleton registry is used.
            analyzer: An optional `OutputAnalyzer` instance. If not provided,
                a new instance is created linked to the registry.
        """
        super().__init__()
        self.registry = (
            registry or get_registry()
        )  # Use provided registry or get the default global registry
        self.analyzer = analyzer or OutputAnalyzer(
            registry=self.registry
        )  # Use provided analyzer or create a new one

    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
        """Handles the start of an LLM call to detect raw string usage.

        It concatenates all input prompts into a single text block and checks if
        this block matches any regex patterns defined for "Raw String" prompts
        (variables tracked by [scan_raw_string_prompts][agent_cover.instrumentation.raw_strings.scanner.scan_raw_string_prompts]).

        Args:
            serialized: A dictionary containing the serialized LLM call information.
            prompts: A list of string prompts passed to the LLM.
            **kwargs: Additional keyword arguments.
        """
        full_text = " ".join(prompts)  # Concatenate all prompts into a single string
        self._check_raw_strings(full_text)  # Check for raw strings in the combined text

    async def on_llm_start_async(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
        """Asynchronously handles the start of an LLM call.

        Delegates logic to [on_llm_start][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_llm_start].
        """
        self.on_llm_start(
            serialized, prompts, **kwargs
        )  # Delegate to the synchronous method

    def on_chat_model_start(
        self, serialized: Dict[str, Any], messages: List[List[Any]], **kwargs: Any
    ) -> None:
        """Handles the start of a chat model call.

        Iterates through the list of message objects (e.g., `HumanMessage`, `SystemMessage`),
        extracts their `content`, and performs raw string detection.

        Args:
            serialized: Metadata about the chat model.
            messages: A list of lists, where each inner list contains LangChain message objects.
            **kwargs: Additional arguments.
        """
        full_text = ""  # Initialize an empty string to store the combined text
        for sublist in messages:  # Iterate through the sublists of messages
            for msg in sublist:  # Iterate through the messages in each sublist
                if hasattr(
                    msg, "content"
                ):  # Check if the message has a content attribute
                    full_text += (
                        str(msg.content) + " "
                    )  # Append the content to the full text
        self._check_raw_strings(full_text)  # Check for raw strings in the combined text

    async def on_chat_model_start_async(
        self, serialized: Dict[str, Any], messages: List[List[Any]], **kwargs: Any
    ) -> None:
        """Asynchronously handles the start of a chat model call.

        Delegates logic to [on_chat_model_start][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_chat_model_start].
        """
        self.on_chat_model_start(
            serialized, messages, **kwargs
        )  # Delegate to the synchronous method

    def _check_raw_strings(self, text: str) -> None:
        """Internal helper to match text against registered string constants.

        This method performs a regex search using patterns stored in the registry
        (metadata field `regex_pattern`). If a match is found, the specific string
        constant is marked as executed.

        Args:
            text: The combined prompt text to scan.
        """
        if not text:  # If the text is empty, return
            return
        for (
            key,
            data,
        ) in (
            self.registry.definitions.items()
        ):  # Iterate through the definitions in the registry
            if (
                data.get("class") == "StringConstant"
            ):  # Check if the definition is a string constant
                regex = data.get(
                    "regex_pattern"
                )  # Get the regex pattern for the string constant
                if regex and re.search(
                    regex, text, re.DOTALL | re.IGNORECASE
                ):  # Check if the regex matches the text
                    self.registry.register_execution(
                        key
                    )  # Register the execution of the key

    def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
    ) -> None:
        """Handles the start of a tool call.

        Note:
            Currently, this method is a placeholder. Tool execution tracking is primarily
            handled by the [ToolInstrumentor][agent_cover.instrumentation.tools.patcher.ToolInstrumentor]
            which patches the tool methods directly, rather than relying on callbacks.
        """
        pass  # Placeholder for tool start handling

    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
        """Handles the end of an LLM call to analyze output quality.

        Extracts the generated text from the `response` object and passes it to the
        `analyzer` to check for Decision Coverage (e.g., "Did the agent output 'REFUND'?").

        Args:
            response: The `LLMResult` object from LangChain containing generations.
            **kwargs: Additional arguments.
        """
        if (
            not response or not response.generations
        ):  # Check if the response or generations are empty
            return
        for gen_list in response.generations:  # Iterate through the generation lists
            for gen in gen_list:  # Iterate through the generations
                if gen.text:  # Check if the generation has text
                    self.analyzer.analyze(
                        gen.text
                    )  # Analyze the generated text using the analyzer

    async def on_llm_end_async(self, response: Any, **kwargs: Any) -> None:
        """Asynchronously handles the end of an LLM call.

        Delegates logic to [on_llm_end][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_llm_end].
        """
        self.on_llm_end(response, **kwargs)  # Delegate to the synchronous method

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        """Handles the end of a chain execution.

        Analyzes the final chain outputs. This is useful for agents that return
        structured dictionaries rather than just raw strings.

        Args:
            outputs: The final dictionary returned by the chain.
            **kwargs: Additional keyword arguments.
        """
        self.analyzer.analyze(outputs)  # Analyze the outputs using the analyzer

    async def on_chain_end_async(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        """Asynchronously handles the end of a chain execution.

        Delegates logic to [on_chain_end][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_chain_end].
        """
        self.on_chain_end(outputs, **kwargs)  # Delegate to the synchronous method

Functions

__init__(registry=None, analyzer=None)

Initializes the CoverageCallbackHandler.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	An optional AgentRegistry instance. If not provided, the global singleton registry is used.	None
analyzer	Optional[OutputAnalyzer]	An optional OutputAnalyzer instance. If not provided, a new instance is created linked to the registry.	None

Source code in src/agent_cover/instrumentation/callbacks.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    analyzer: Optional[OutputAnalyzer] = None,
):
    """Initializes the CoverageCallbackHandler.

    Args:
        registry: An optional `AgentRegistry` instance. If not provided,
            the global singleton registry is used.
        analyzer: An optional `OutputAnalyzer` instance. If not provided,
            a new instance is created linked to the registry.
    """
    super().__init__()
    self.registry = (
        registry or get_registry()
    )  # Use provided registry or get the default global registry
    self.analyzer = analyzer or OutputAnalyzer(
        registry=self.registry
    )  # Use provided analyzer or create a new one

on_chain_end(outputs, **kwargs)

Handles the end of a chain execution.

Analyzes the final chain outputs. This is useful for agents that return structured dictionaries rather than just raw strings.

Parameters:

Name	Type	Description	Default
outputs	Dict[str, Any]	The final dictionary returned by the chain.	required
**kwargs	Any	Additional keyword arguments.	{}

Source code in src/agent_cover/instrumentation/callbacks.py

def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
    """Handles the end of a chain execution.

    Analyzes the final chain outputs. This is useful for agents that return
    structured dictionaries rather than just raw strings.

    Args:
        outputs: The final dictionary returned by the chain.
        **kwargs: Additional keyword arguments.
    """
    self.analyzer.analyze(outputs)  # Analyze the outputs using the analyzer

on_chain_end_async(outputs, **kwargs) async

Asynchronously handles the end of a chain execution.

Delegates logic to on_chain_end.

Source code in src/agent_cover/instrumentation/callbacks.py

async def on_chain_end_async(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
    """Asynchronously handles the end of a chain execution.

    Delegates logic to [on_chain_end][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_chain_end].
    """
    self.on_chain_end(outputs, **kwargs)  # Delegate to the synchronous method

on_chat_model_start(serialized, messages, **kwargs)

Handles the start of a chat model call.

Iterates through the list of message objects (e.g., HumanMessage, SystemMessage), extracts their content, and performs raw string detection.

Parameters:

Name	Type	Description	Default
serialized	Dict[str, Any]	Metadata about the chat model.	required
messages	List[List[Any]]	A list of lists, where each inner list contains LangChain message objects.	required
**kwargs	Any	Additional arguments.	{}

Source code in src/agent_cover/instrumentation/callbacks.py

def on_chat_model_start(
    self, serialized: Dict[str, Any], messages: List[List[Any]], **kwargs: Any
) -> None:
    """Handles the start of a chat model call.

    Iterates through the list of message objects (e.g., `HumanMessage`, `SystemMessage`),
    extracts their `content`, and performs raw string detection.

    Args:
        serialized: Metadata about the chat model.
        messages: A list of lists, where each inner list contains LangChain message objects.
        **kwargs: Additional arguments.
    """
    full_text = ""  # Initialize an empty string to store the combined text
    for sublist in messages:  # Iterate through the sublists of messages
        for msg in sublist:  # Iterate through the messages in each sublist
            if hasattr(
                msg, "content"
            ):  # Check if the message has a content attribute
                full_text += (
                    str(msg.content) + " "
                )  # Append the content to the full text
    self._check_raw_strings(full_text)  # Check for raw strings in the combined text

on_chat_model_start_async(serialized, messages, **kwargs) async

Asynchronously handles the start of a chat model call.

Delegates logic to on_chat_model_start.

Source code in src/agent_cover/instrumentation/callbacks.py

async def on_chat_model_start_async(
    self, serialized: Dict[str, Any], messages: List[List[Any]], **kwargs: Any
) -> None:
    """Asynchronously handles the start of a chat model call.

    Delegates logic to [on_chat_model_start][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_chat_model_start].
    """
    self.on_chat_model_start(
        serialized, messages, **kwargs
    )  # Delegate to the synchronous method

on_llm_end(response, **kwargs)

Handles the end of an LLM call to analyze output quality.

Extracts the generated text from the response object and passes it to the analyzer to check for Decision Coverage (e.g., "Did the agent output 'REFUND'?").

Parameters:

Name	Type	Description	Default
response	Any	The LLMResult object from LangChain containing generations.	required
**kwargs	Any	Additional arguments.	{}

Source code in src/agent_cover/instrumentation/callbacks.py

def on_llm_end(self, response: Any, **kwargs: Any) -> None:
    """Handles the end of an LLM call to analyze output quality.

    Extracts the generated text from the `response` object and passes it to the
    `analyzer` to check for Decision Coverage (e.g., "Did the agent output 'REFUND'?").

    Args:
        response: The `LLMResult` object from LangChain containing generations.
        **kwargs: Additional arguments.
    """
    if (
        not response or not response.generations
    ):  # Check if the response or generations are empty
        return
    for gen_list in response.generations:  # Iterate through the generation lists
        for gen in gen_list:  # Iterate through the generations
            if gen.text:  # Check if the generation has text
                self.analyzer.analyze(
                    gen.text
                )  # Analyze the generated text using the analyzer

on_llm_end_async(response, **kwargs) async

Asynchronously handles the end of an LLM call.

Delegates logic to on_llm_end.

Source code in src/agent_cover/instrumentation/callbacks.py

async def on_llm_end_async(self, response: Any, **kwargs: Any) -> None:
    """Asynchronously handles the end of an LLM call.

    Delegates logic to [on_llm_end][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_llm_end].
    """
    self.on_llm_end(response, **kwargs)  # Delegate to the synchronous method

on_llm_start(serialized, prompts, **kwargs)

Handles the start of an LLM call to detect raw string usage.

It concatenates all input prompts into a single text block and checks if this block matches any regex patterns defined for "Raw String" prompts (variables tracked by scan_raw_string_prompts).

Parameters:

Name	Type	Description	Default
serialized	Dict[str, Any]	A dictionary containing the serialized LLM call information.	required
prompts	List[str]	A list of string prompts passed to the LLM.	required
**kwargs	Any	Additional keyword arguments.	{}

Source code in src/agent_cover/instrumentation/callbacks.py

def on_llm_start(
    self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> None:
    """Handles the start of an LLM call to detect raw string usage.

    It concatenates all input prompts into a single text block and checks if
    this block matches any regex patterns defined for "Raw String" prompts
    (variables tracked by [scan_raw_string_prompts][agent_cover.instrumentation.raw_strings.scanner.scan_raw_string_prompts]).

    Args:
        serialized: A dictionary containing the serialized LLM call information.
        prompts: A list of string prompts passed to the LLM.
        **kwargs: Additional keyword arguments.
    """
    full_text = " ".join(prompts)  # Concatenate all prompts into a single string
    self._check_raw_strings(full_text)  # Check for raw strings in the combined text

on_llm_start_async(serialized, prompts, **kwargs) async

Asynchronously handles the start of an LLM call.

Delegates logic to on_llm_start.

Source code in src/agent_cover/instrumentation/callbacks.py

async def on_llm_start_async(
    self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> None:
    """Asynchronously handles the start of an LLM call.

    Delegates logic to [on_llm_start][agent_cover.instrumentation.callbacks.CoverageCallbackHandler.on_llm_start].
    """
    self.on_llm_start(
        serialized, prompts, **kwargs
    )  # Delegate to the synchronous method

on_tool_start(serialized, input_str, **kwargs)

Handles the start of a tool call.

Note

Currently, this method is a placeholder. Tool execution tracking is primarily handled by the ToolInstrumentor which patches the tool methods directly, rather than relying on callbacks.

Source code in src/agent_cover/instrumentation/callbacks.py

def on_tool_start(
    self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
) -> None:
    """Handles the start of a tool call.

    Note:
        Currently, this method is a placeholder. Tool execution tracking is primarily
        handled by the [ToolInstrumentor][agent_cover.instrumentation.tools.patcher.ToolInstrumentor]
        which patches the tool methods directly, rather than relying on callbacks.
    """
    pass  # Placeholder for tool start handling

GlobalCallbackInstrumentor

Bases: BaseInstrumentor

Instrumentor that globally injects the CoverageCallbackHandler into LangChain.

This class performs monkey-patching on langchain_core.callbacks.manager.CallbackManager and AsyncCallbackManager. It wraps their __init__ methods to automatically append a CoverageCallbackHandler to every new manager instance.

Attributes:

Name	Type	Description
registry	AgentRegistry	The registry instance.
handler_factory	Callable	A factory function that returns a new handler instance.

Methods:

Name	Description
instrument	Applies the instrumentation to Langchain's callback managers.
_patch_manager_init	Internal method to patch the init of a manager class.

Source code in src/agent_cover/instrumentation/callbacks.py

class GlobalCallbackInstrumentor(BaseInstrumentor):
    """Instrumentor that globally injects the CoverageCallbackHandler into LangChain.

    This class performs monkey-patching on `langchain_core.callbacks.manager.CallbackManager`
    and `AsyncCallbackManager`. It wraps their `__init__` methods to automatically
    append a [CoverageCallbackHandler][agent_cover.instrumentation.callbacks.CoverageCallbackHandler]
    to every new manager instance.

    Attributes:
        registry (AgentRegistry): The registry instance.
        handler_factory (Callable): A factory function that returns a new handler instance.

    Methods:
        instrument: Applies the instrumentation to Langchain's callback managers.
        _patch_manager_init: Internal method to patch the __init__ of a manager class.
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        patch_manager: Optional[PatchManager] = None,
        handler_factory: Optional[Callable] = None,
    ):
        """Initializes the instrumentor.

        Args:
            registry: The registry instance.
            patch_manager: The patch manager instance to ensure safe patching.
            handler_factory: Optional factory for dependency injection (testing).
                If None, defaults to creating a real `CoverageCallbackHandler`.
        """
        super().__init__(
            registry=registry, patch_manager=patch_manager
        )  # Initialize the base class with the provided or default registry

        # If a handler factory is not provided (e.g., in a test with a mock), use the default factory
        if handler_factory:
            self.handler_factory = handler_factory  # Use the provided handler factory
        else:
            # Default factory: creates a real handler connected to the current registry
            def _default_factory() -> CoverageCallbackHandler:
                """Default factory function to create a CoverageCallbackHandler instance."""
                return CoverageCallbackHandler(registry=self.registry)

            self.handler_factory = _default_factory  # Assign the default factory

    def instrument(self):
        """Applies the patches to LangChain's CallbackManagers.

        If `langchain_core` is not installed, this method returns silently.
        """
        if self.is_instrumented:  # If already instrumented, return
            return

        try:
            # Import CallbackManager and AsyncCallbackManager from langchain_core
            from langchain_core.callbacks.manager import (
                AsyncCallbackManager,
                CallbackManager,
            )
        except ImportError:
            # If the import fails, return (Langchain not installed)
            return

        # Patch the __init__ method of CallbackManager
        self._patch_manager_init(CallbackManager)
        # Patch the __init__ method of AsyncCallbackManager
        self._patch_manager_init(AsyncCallbackManager)

        self.is_instrumented = True  # Set the is_instrumented flag to True

    def _patch_manager_init(self, manager_cls: type) -> None:
        """Internal helper to patch the `__init__` of a specific manager class.

        Args:
            manager_cls: The CallbackManager or AsyncCallbackManager class to patch.
        """
        if hasattr(
            manager_cls, "__init__"
        ):  # Check if the class has an __init__ method
            # Accessing __init__ on a type is considered unsafe by mypy
            original_init = manager_cls.__init__  # type: ignore[misc]

            def patched_init(
                instance,
                handlers: Optional[List[Any]] = None,
                tags: Optional[List[str]] = None,
                inheritable_tags: Optional[List[str]] = None,
                metadata: Optional[Dict[str, Any]] = None,
                **kwargs: Any,
            ) -> None:
                """The patched __init__ method.

                This method is a wrapper around the original __init__ method. It adds the
                CoverageCallbackHandler to the list of handlers before calling the
                original __init__ method.

                Args:
                    instance: The instance of the CallbackManager or AsyncCallbackManager.
                    handlers: An optional list of existing callback handlers.
                    tags: Optional tags for the callback manager.
                    inheritable_tags: Optional inheritable tags for the callback manager.
                    metadata: Optional metadata for the callback manager.
                    **kwargs: Additional keyword arguments.
                """
                if (
                    handlers is None
                ):  # If no handlers are provided, create an empty list
                    handlers = []

                # Use the injected factory (which can return a Mock) to create the handler
                handlers.append(
                    self.handler_factory()
                )  # Append the CoverageCallbackHandler to the list of handlers

                original_init(
                    instance,
                    handlers=handlers,
                    tags=tags,
                    inheritable_tags=inheritable_tags,
                    metadata=metadata,
                    **kwargs,
                )  # Call the original __init__ method

            self._safe_patch(
                manager_cls, "__init__", patched_init
            )  # Apply the patch using _safe_patch

Functions

__init__(registry=None, patch_manager=None, handler_factory=None)

Initializes the instrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The registry instance.	None
patch_manager	Optional[PatchManager]	The patch manager instance to ensure safe patching.	None
handler_factory	Optional[Callable]	Optional factory for dependency injection (testing). If None, defaults to creating a real CoverageCallbackHandler.	None

Source code in src/agent_cover/instrumentation/callbacks.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    patch_manager: Optional[PatchManager] = None,
    handler_factory: Optional[Callable] = None,
):
    """Initializes the instrumentor.

    Args:
        registry: The registry instance.
        patch_manager: The patch manager instance to ensure safe patching.
        handler_factory: Optional factory for dependency injection (testing).
            If None, defaults to creating a real `CoverageCallbackHandler`.
    """
    super().__init__(
        registry=registry, patch_manager=patch_manager
    )  # Initialize the base class with the provided or default registry

    # If a handler factory is not provided (e.g., in a test with a mock), use the default factory
    if handler_factory:
        self.handler_factory = handler_factory  # Use the provided handler factory
    else:
        # Default factory: creates a real handler connected to the current registry
        def _default_factory() -> CoverageCallbackHandler:
            """Default factory function to create a CoverageCallbackHandler instance."""
            return CoverageCallbackHandler(registry=self.registry)

        self.handler_factory = _default_factory  # Assign the default factory

instrument()

Applies the patches to LangChain's CallbackManagers.

If langchain_core is not installed, this method returns silently.

Source code in src/agent_cover/instrumentation/callbacks.py

def instrument(self):
    """Applies the patches to LangChain's CallbackManagers.

    If `langchain_core` is not installed, this method returns silently.
    """
    if self.is_instrumented:  # If already instrumented, return
        return

    try:
        # Import CallbackManager and AsyncCallbackManager from langchain_core
        from langchain_core.callbacks.manager import (
            AsyncCallbackManager,
            CallbackManager,
        )
    except ImportError:
        # If the import fails, return (Langchain not installed)
        return

    # Patch the __init__ method of CallbackManager
    self._patch_manager_init(CallbackManager)
    # Patch the __init__ method of AsyncCallbackManager
    self._patch_manager_init(AsyncCallbackManager)

    self.is_instrumented = True  # Set the is_instrumented flag to True

InspectionProvider

Helper class to abstract the calls to inspect that touch the filesystem.

Methods:

Name	Description
get_file	Any) -> str: Returns the file path of the given object.
get_source_lines	Any) -> Tuple[List[str], int]: Returns the source code lines and starting line number of the given object.

Source code in src/agent_cover/instrumentation/structures/scanner.py

class InspectionProvider:
    """Helper class to abstract the calls to inspect that touch the filesystem.

    Attributes:
        None

    Methods:
        get_file(obj: Any) -> str: Returns the file path of the given object.
        get_source_lines(obj: Any) -> Tuple[List[str], int]: Returns the source code lines and starting line number of the given object.
    """

    def get_file(self, obj: Any) -> str:
        """Returns the file path of the given object.

        Args:
            obj: The object to get the file path for.

        Returns:
            The file path of the object.
        """
        return inspect.getfile(obj)

    def get_source_lines(self, obj: Any) -> Tuple[List[str], int]:
        """Returns the source code lines and starting line number of the given object.

        Args:
            obj: The object to get the source code lines for.

        Returns:
            A tuple containing the source code lines as a list of strings and the starting line number.
        """
        return inspect.getsourcelines(obj)

Functions

get_file(obj)

Returns the file path of the given object.

Parameters:

Name	Type	Description	Default
obj	Any	The object to get the file path for.	required

Returns:

Type	Description
str	The file path of the object.

Source code in src/agent_cover/instrumentation/structures/scanner.py

def get_file(self, obj: Any) -> str:
    """Returns the file path of the given object.

    Args:
        obj: The object to get the file path for.

    Returns:
        The file path of the object.
    """
    return inspect.getfile(obj)

get_source_lines(obj)

Returns the source code lines and starting line number of the given object.

Parameters:

Name	Type	Description	Default
obj	Any	The object to get the source code lines for.	required

Returns:

Type	Description
Tuple[List[str], int]	A tuple containing the source code lines as a list of strings and the starting line number.

Source code in src/agent_cover/instrumentation/structures/scanner.py

def get_source_lines(self, obj: Any) -> Tuple[List[str], int]:
    """Returns the source code lines and starting line number of the given object.

    Args:
        obj: The object to get the source code lines for.

    Returns:
        A tuple containing the source code lines as a list of strings and the starting line number.
    """
    return inspect.getsourcelines(obj)

LLMProviderInstrumentor

Bases: BaseInstrumentor

Instrumentor that sits at the edge of the system (External APIs).

Unlike other instrumentors that track code coverage, this tracks Data Coverage. It intercepts the raw string response from the LLM to analyze if business logic requirements (Decisions) were met.

Attributes:

Name	Type	Description
registry	Optional[AgentRegistry]	The registry for agent components.
analyzer	OutputAnalyzer	The component responsible for analyzing extracted text.
importer_func	Callable	Function used to import modules dynamically.
module_iterator	Callable	Function that returns the current mapping of loaded modules.
targets_provider	Callable	Function that returns a list of targets to instrument.
extractors	List[PayloadExtractor]	List of strategies used to parse LLM results.
is_instrumented	bool	Flag indicating if instrumentation has already run.

Methods:

Name	Description
instrument	Performs the actual patching of target methods.
_resolve_target	helper to find the specific object/method to patch.
_create_wrapper	Creates the closure that wraps the original method.
_delegate_extraction	Iterates through extractors to parse the result.

Source code in src/agent_cover/instrumentation/llm_providers/patcher.py

class LLMProviderInstrumentor(BaseInstrumentor):
    """Instrumentor that sits at the edge of the system (External APIs).

    Unlike other instrumentors that track code coverage, this tracks **Data Coverage**.
    It intercepts the raw string response from the LLM to analyze if business logic
    requirements (Decisions) were met.

    Attributes:
        registry (Optional[AgentRegistry]): The registry for agent components.
        analyzer (OutputAnalyzer): The component responsible for analyzing extracted text.
        importer_func (Callable): Function used to import modules dynamically.
        module_iterator (Callable): Function that returns the current mapping of loaded modules.
        targets_provider (Callable): Function that returns a list of targets to instrument.
        extractors (List[PayloadExtractor]): List of strategies used to parse LLM results.
        is_instrumented (bool): Flag indicating if instrumentation has already run.

    Methods:
        instrument(): Performs the actual patching of target methods.
        _resolve_target(mod, obj_name, patch_type): helper to find the specific object/method to patch.
        _create_wrapper(original_func): Creates the closure that wraps the original method.
        _delegate_extraction(result): Iterates through extractors to parse the result.
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        analyzer: Optional[OutputAnalyzer] = None,
        patch_manager: Optional[PatchManager] = None,
        importer_func: Optional[Callable[[str], Any]] = None,
        module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
        targets_provider: Optional[Callable[[], TargetList]] = None,
        extractors: Optional[List[PayloadExtractor]] = None,
    ):
        """Initializes the LLMProviderInstrumentor.

        Args:
            registry (Optional[AgentRegistry]): The registry instance.
            analyzer (Optional[OutputAnalyzer]): Custom analyzer instance.
            patch_manager (Optional[PatchManager]): The patch manager for safe patching.
            importer_func (Optional[Callable]): Custom import function.
            module_iterator (Optional[Callable]): Custom module iterator provider (for dependency injection).
            targets_provider (Optional[Callable]): Custom provider for patch targets.
            extractors (Optional[List[PayloadExtractor]]): Custom list of payload extractors.
        """
        super().__init__(
            registry=registry,
            patch_manager=patch_manager,
            importer_func=importer_func,
            module_iterator=module_iterator,
        )
        self.analyzer = analyzer or OutputAnalyzer(registry=self.registry)
        self.targets_provider = targets_provider or _default_targets_provider

        self.extractors = extractors or [OpenAIExtractor(), StringExtractor()]

    def instrument(self):
        """Applies patches to the defined LLM provider targets.

        This method iterates through the targets provided by `targets_provider`.
        It loads the necessary modules (using a snapshot to allow for test isolation)
        and wraps the specified methods/functions to enable output analysis.
        """
        if self.is_instrumented:
            return

        raw_targets = self.targets_provider()
        targets = self._normalize_targets(raw_targets)

        # [MODIFICATION] Use module_iterator for test isolation
        modules_snapshot = self.module_iterator()

        for target in targets:
            if not self._should_instrument(target):
                continue

            mod_name = target.module
            obj_name = target.class_name
            patch_type = target.params.get("type")

            try:
                # Import: Check snapshot instead of global sys.modules
                if mod_name not in modules_snapshot:
                    try:
                        self.importer(mod_name)
                        # Refresh snapshot after import
                        modules_snapshot = self.module_iterator()
                    except ModuleNotFoundError:
                        Log.log_skip_missing_module(logger, mod_name)
                        continue

                # Patching logic
                if mod_name in modules_snapshot:
                    mod = modules_snapshot[mod_name]
                    target_obj, method_name = self._resolve_target(
                        mod, obj_name, patch_type
                    )

                    if target_obj is None:
                        Log.log_skip_missing_attr(logger, mod_name, obj_name)
                        continue

                    if target_obj and method_name and hasattr(target_obj, method_name):
                        original_func = getattr(target_obj, method_name)
                        wrapper = self._create_wrapper(original_func)
                        self._safe_patch(target_obj, method_name, wrapper)

            except Exception as e:
                logger.warning(f"Error: {e}", exc_info=True)

        self.is_instrumented = True

    def _resolve_target(
        self, mod: Any, obj_name: str, patch_type: str
    ) -> Tuple[Optional[Any], Optional[str]]:
        """Resolves the module and method names into actual objects.

        Args:
            mod (Any): The module object.
            obj_name (str): The name of the class or function to target.
            patch_type (str): The type of patch (class method or function).

        Returns:
            Tuple[Optional[Any], Optional[str]]: A tuple containing the target object
            (class or module) and the method name to patch.
        """
        target_obj = None
        method_name = None

        if patch_type == TYPE_CLASS_METHOD and "." in obj_name:
            cls_name, m_name = obj_name.split(".")
            if hasattr(mod, cls_name):
                target_obj = getattr(mod, cls_name)
                method_name = m_name
        elif patch_type == TYPE_FUNCTION:
            if hasattr(mod, obj_name):
                target_obj = mod
                method_name = obj_name
            elif "." in obj_name:
                cls_name, m_name = obj_name.split(".")
                if hasattr(mod, cls_name):
                    target_obj = getattr(mod, cls_name)
                    method_name = m_name
        return target_obj, method_name

    def _create_wrapper(self, original_func: Callable) -> Callable:
        """Creates a wrapper function around the original method.

        The wrapper executes the original function, captures the result, attempts to
        extract the payload, and sends it to the analyzer without altering the
        original return value.

        Args:
            original_func (Callable): The original function/method being patched.

        Returns:
            Callable: The wrapped function.
        """

        @functools.wraps(original_func)
        def wrapper(*args, **kwargs):
            result = original_func(*args, **kwargs)
            try:
                payload = self._delegate_extraction(result)
                if payload:
                    self.analyzer.analyze(payload)
            except Exception as e:
                logger.warning(e, exc_info=True)
            return result

        return wrapper

    def _delegate_extraction(self, result: Any) -> Optional[str]:
        """Delegates the result to the registered extractors.

        Args:
            result (Any): The return value from the instrumented function.

        Returns:
            Optional[str]: The extracted text if any extractor succeeds, otherwise None.
        """
        for extractor in self.extractors:
            try:
                val = extractor.extract(result)
                if val:
                    return val
            except Exception as e:
                logger.warning(e, exc_info=True)
                continue
        return None

Functions

__init__(registry=None, analyzer=None, patch_manager=None, importer_func=None, module_iterator=None, targets_provider=None, extractors=None)

Initializes the LLMProviderInstrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The registry instance.	None
analyzer	Optional[OutputAnalyzer]	Custom analyzer instance.	None
patch_manager	Optional[PatchManager]	The patch manager for safe patching.	None
importer_func	Optional[Callable]	Custom import function.	None
module_iterator	Optional[Callable]	Custom module iterator provider (for dependency injection).	None
targets_provider	Optional[Callable]	Custom provider for patch targets.	None
extractors	Optional[List[PayloadExtractor]]	Custom list of payload extractors.	None

Source code in src/agent_cover/instrumentation/llm_providers/patcher.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    analyzer: Optional[OutputAnalyzer] = None,
    patch_manager: Optional[PatchManager] = None,
    importer_func: Optional[Callable[[str], Any]] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    targets_provider: Optional[Callable[[], TargetList]] = None,
    extractors: Optional[List[PayloadExtractor]] = None,
):
    """Initializes the LLMProviderInstrumentor.

    Args:
        registry (Optional[AgentRegistry]): The registry instance.
        analyzer (Optional[OutputAnalyzer]): Custom analyzer instance.
        patch_manager (Optional[PatchManager]): The patch manager for safe patching.
        importer_func (Optional[Callable]): Custom import function.
        module_iterator (Optional[Callable]): Custom module iterator provider (for dependency injection).
        targets_provider (Optional[Callable]): Custom provider for patch targets.
        extractors (Optional[List[PayloadExtractor]]): Custom list of payload extractors.
    """
    super().__init__(
        registry=registry,
        patch_manager=patch_manager,
        importer_func=importer_func,
        module_iterator=module_iterator,
    )
    self.analyzer = analyzer or OutputAnalyzer(registry=self.registry)
    self.targets_provider = targets_provider or _default_targets_provider

    self.extractors = extractors or [OpenAIExtractor(), StringExtractor()]

instrument()

Applies patches to the defined LLM provider targets.

This method iterates through the targets provided by targets_provider. It loads the necessary modules (using a snapshot to allow for test isolation) and wraps the specified methods/functions to enable output analysis.

Source code in src/agent_cover/instrumentation/llm_providers/patcher.py

def instrument(self):
    """Applies patches to the defined LLM provider targets.

    This method iterates through the targets provided by `targets_provider`.
    It loads the necessary modules (using a snapshot to allow for test isolation)
    and wraps the specified methods/functions to enable output analysis.
    """
    if self.is_instrumented:
        return

    raw_targets = self.targets_provider()
    targets = self._normalize_targets(raw_targets)

    # [MODIFICATION] Use module_iterator for test isolation
    modules_snapshot = self.module_iterator()

    for target in targets:
        if not self._should_instrument(target):
            continue

        mod_name = target.module
        obj_name = target.class_name
        patch_type = target.params.get("type")

        try:
            # Import: Check snapshot instead of global sys.modules
            if mod_name not in modules_snapshot:
                try:
                    self.importer(mod_name)
                    # Refresh snapshot after import
                    modules_snapshot = self.module_iterator()
                except ModuleNotFoundError:
                    Log.log_skip_missing_module(logger, mod_name)
                    continue

            # Patching logic
            if mod_name in modules_snapshot:
                mod = modules_snapshot[mod_name]
                target_obj, method_name = self._resolve_target(
                    mod, obj_name, patch_type
                )

                if target_obj is None:
                    Log.log_skip_missing_attr(logger, mod_name, obj_name)
                    continue

                if target_obj and method_name and hasattr(target_obj, method_name):
                    original_func = getattr(target_obj, method_name)
                    wrapper = self._create_wrapper(original_func)
                    self._safe_patch(target_obj, method_name, wrapper)

        except Exception as e:
            logger.warning(f"Error: {e}", exc_info=True)

    self.is_instrumented = True

PromptFlowInstrumentor

Bases: BaseInstrumentor

Instruments Microsoft PromptFlow components.

PromptFlow defines flows using a mix of YAML configuration (flow.dag.yaml), Python tools (@tool), and Jinja2 templates. This instrumentor applies patches to track:

1. Template Rendering: Patches internal render functions to track when a Jinja template is used.
2. Tool Execution: Wraps the @tool decorator to track when a node in the flow is executed.

It works in tandem with scan_promptflow_definitions which statically registers the files found in the DAG.

Attributes:

Name	Type	Description
analyzer	OutputAnalyzer	The analyzer used to process execution results.
targets_provider	Callable	A function that returns a list of targets to instrument.
registry	AgentRegistry	The registry where execution data is stored.
patch_manager	PatchManager	Manager for applying and reverting patches.
module_iterator	Callable	Function to retrieve current sys.modules.
importer_func	Callable	Function to import modules dynamically.
is_instrumented	bool	Flag indicating if instrumentation has already run.

Methods:

Name	Description
instrument	Applies patches to the target PromptFlow modules.

Source code in src/agent_cover/instrumentation/promptflow/patcher.py

class PromptFlowInstrumentor(BaseInstrumentor):
    """Instruments Microsoft PromptFlow components.

    PromptFlow defines flows using a mix of YAML configuration (`flow.dag.yaml`),
    Python tools (`@tool`), and Jinja2 templates. This instrumentor applies patches
    to track:

    1.  **Template Rendering**: Patches internal render functions to track when a Jinja template is used.
    2.  **Tool Execution**: Wraps the `@tool` decorator to track when a node in the flow is executed.

    It works in tandem with [`scan_promptflow_definitions`][agent_cover.instrumentation.promptflow.scanner.scan_promptflow_definitions]
    which statically registers the files found in the DAG.

    Attributes:
        analyzer (OutputAnalyzer): The analyzer used to process execution results.
        targets_provider (Callable): A function that returns a list of targets to instrument.
        registry (AgentRegistry): The registry where execution data is stored.
        patch_manager (PatchManager): Manager for applying and reverting patches.
        module_iterator (Callable): Function to retrieve current sys.modules.
        importer_func (Callable): Function to import modules dynamically.
        is_instrumented (bool): Flag indicating if instrumentation has already run.

    Methods:
        instrument(): Applies patches to the target PromptFlow modules.
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        analyzer: Optional[OutputAnalyzer] = None,
        patch_manager: Optional[PatchManager] = None,
        module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
        importer_func: Optional[Callable[[str], Any]] = None,
        targets_provider: Optional[Callable[[], TargetList]] = None,
    ):
        """Initializes the PromptFlowInstrumentor.

        Args:
            registry: The agent registry instance.
            analyzer: Optional output analyzer instance.
            patch_manager: Optional patch manager instance.
            module_iterator: Optional callable to get current modules.
            importer_func: Optional callable to import modules.
            targets_provider: Optional callable to provide instrumentation targets.
        """
        super().__init__(
            registry=registry,
            patch_manager=patch_manager,
            module_iterator=module_iterator,
            importer_func=importer_func,
        )
        self.analyzer = analyzer or OutputAnalyzer(registry=self.registry)
        self.targets_provider = targets_provider or _default_targets_provider

    def instrument(self):
        """Applies runtime patches to PromptFlow's core modules.

        It targets:
        - `promptflow.tools.common.render_jinja_template`: To catch prompt usage.
        - `promptflow.core.tool`: To catch python tool usage.

        Note:
            This method is fault-tolerant; if PromptFlow is not installed, it simply
            skips instrumentation without raising errors.
        """
        if self.is_instrumented:
            return

        raw_targets = self.targets_provider()
        targets = self._normalize_targets(raw_targets)

        patched_count = 0

        # [MODIFICATION] Use module_iterator
        modules_snapshot = self.module_iterator()

        for target in targets:
            if not self._should_instrument(target):
                continue

            mod_name = target.module
            obj_name = target.class_name
            target_type = target.params.get("type")

            try:
                if mod_name not in modules_snapshot:
                    try:
                        self.importer(mod_name)
                        modules_snapshot = self.module_iterator()
                    except ModuleNotFoundError:
                        Log.log_skip_missing_module(logger, mod_name)
                        continue
                    except Exception as e:
                        logger.warning(e, exc_info=True)
                        continue

                if mod_name in modules_snapshot:
                    mod = modules_snapshot[mod_name]

                    if hasattr(mod, obj_name):
                        if target_type == "render":
                            self._patch_render_function(mod, obj_name)
                            patched_count += 1
                        elif target_type == "decorator":
                            self._patch_tool_decorator(mod, obj_name)
                            patched_count += 1

            except Exception as e:
                logger.warning(
                    f"PromptFlow Patch Error ({mod_name}): {e}", exc_info=True
                )

        if patched_count > 0:
            logger.debug(f"Instrumented {patched_count} PromptFlow internals.")

        self.is_instrumented = True

    # --- RENDER PATCHER ---
    def _patch_render_function(self, module: Any, func_name: str):
        """Patches the PromptFlow render function to track Jinja2 template usage.

        This method wraps the internal rendering logic to:
        1. Calculate content hashes and link them to static file definitions.
        2. Analyze the rendered prompt text for decision/business logic coverage.

        Args:
            module (Any): The module containing the render function.
            func_name (str): The name of the rendering function (e.g., 'render_jinja_template').
        """
        original_func = getattr(module, func_name)

        if getattr(original_func, "_ac_patched", False):
            return

        @functools.wraps(original_func)
        def wrapper(template, **kwargs):
            # Execute original rendering to get the final string sent to the LLM
            rendered_prompt = original_func(template, **kwargs)

            try:
                if isinstance(template, str):
                    # 1. Track Prompt Coverage (Jinja)
                    # We use a runtime hash to match this specific template version
                    # back to the statically scanned file IDs.
                    content_hash = f"RUNTIME:PF:{hash(template)}"
                    canonical_id = self.registry.get_canonical_id(
                        template, content_hash
                    )

                    if canonical_id.startswith("FILE:"):
                        self.registry.register_execution(canonical_id)
                        logger.debug(f"PromptFlow template hit: {canonical_id}")

                # 2. Track Decision Coverage (Business Logic)
                # We pass the fully rendered text to the analyzer to check for
                # expected values.
                if rendered_prompt:
                    self.analyzer.analyze(rendered_prompt)

            except Exception as e:
                logger.error(f"Error in PromptFlow render patch: {e}", exc_info=True)

            return rendered_prompt

        self._safe_patch(module, func_name, wrapper)

    # --- TOOL DECORATOR PATCHER ---
    def _patch_tool_decorator(self, module: Any, func_name: str):
        """Patches the @tool decorator to instrument user functions.

        Args:
            module: The module containing the decorator.
            func_name: The name of the decorator function.
        """
        original_tool = getattr(module, func_name)
        if getattr(original_tool, "_ac_patched", False):
            return

        @functools.wraps(original_tool)
        def tool_wrapper(func_or_none=None, **kwargs):
            if func_or_none is not None and callable(func_or_none):
                instrumented_func = self._instrument_user_function(func_or_none)
                return original_tool(instrumented_func)

            def partial_wrapper(user_func):
                instrumented_func = self._instrument_user_function(user_func)
                return original_tool(**kwargs)(instrumented_func)

            return partial_wrapper

        self._safe_patch(module, func_name, tool_wrapper)

    def _instrument_user_function(self, user_func: Callable) -> Callable:
        """Wraps a user-defined tool function to manage context and track execution.

        This wrapper is critical for PromptFlow as it manually activates the
        AgentContextManager. Since PromptFlow doesn't use standard AgentExecutors,
        this ensures that subsequent tool/prompt calls are recorded.

        Args:
            user_func (Callable): The actual tool function decorated by @tool.

        Returns:
            Callable: The instrumented function with context management and tracking.
        """
        try:
            # Resolve tool identification metadata
            source_file = inspect.getsourcefile(user_func)
            source_file = os.path.abspath(source_file) if source_file else "unknown"
            func_name = user_func.__name__
            canonical_id = f"{source_file}:TOOL:{func_name}"

            # Register the tool definition if not already present
            if canonical_id not in self.registry.definitions:
                self.registry.register_definition(
                    key=canonical_id,
                    kind="TOOL",
                    metadata={
                        "class": "PromptFlow::Python",
                        "tool_name": func_name,
                        "preview": f"@tool def {func_name}(...)",
                        "file_path": source_file,
                        "line_number": 0,
                    },
                )
        except Exception as e:
            logger.warning(f"Failed to register PromptFlow tool definition: {e}")

        @functools.wraps(user_func)
        def execution_wrapper(*args, **kwargs):
            # FORCED CONTEXT ACTIVATION
            # We manually set the agent context to active so that the instrumentation
            # core knows we are inside a valid agent logic flow.

            ctx = get_global_context_manager()
            token = ctx.set_active(True)

            logger.debug(
                f"Executing PromptFlow node: {user_func.__name__} (PID: {os.getpid()})"
            )

            try:
                result = user_func(*args, **kwargs)

                # Register execution of the node itself
                try:
                    # Re-calculate ID for recording the execution hit
                    s_file = inspect.getsourcefile(user_func)
                    if s_file:
                        c_id = f"{os.path.abspath(s_file)}:TOOL:{user_func.__name__}"
                        self.registry.register_execution(c_id)
                except Exception as e:
                    logger.warning(e)

                # Pass the tool result to the analyzer for data coverage
                try:
                    self.analyzer.analyze(result)
                except Exception as e:
                    logger.warning(e)

                return result
            finally:
                # Always restore the previous context state
                ctx.reset(token)

        return execution_wrapper

Functions

__init__(registry=None, analyzer=None, patch_manager=None, module_iterator=None, importer_func=None, targets_provider=None)

Initializes the PromptFlowInstrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The agent registry instance.	None
analyzer	Optional[OutputAnalyzer]	Optional output analyzer instance.	None
patch_manager	Optional[PatchManager]	Optional patch manager instance.	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	Optional callable to get current modules.	None
importer_func	Optional[Callable[[str], Any]]	Optional callable to import modules.	None
targets_provider	Optional[Callable[[], TargetList]]	Optional callable to provide instrumentation targets.	None

Source code in src/agent_cover/instrumentation/promptflow/patcher.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    analyzer: Optional[OutputAnalyzer] = None,
    patch_manager: Optional[PatchManager] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    importer_func: Optional[Callable[[str], Any]] = None,
    targets_provider: Optional[Callable[[], TargetList]] = None,
):
    """Initializes the PromptFlowInstrumentor.

    Args:
        registry: The agent registry instance.
        analyzer: Optional output analyzer instance.
        patch_manager: Optional patch manager instance.
        module_iterator: Optional callable to get current modules.
        importer_func: Optional callable to import modules.
        targets_provider: Optional callable to provide instrumentation targets.
    """
    super().__init__(
        registry=registry,
        patch_manager=patch_manager,
        module_iterator=module_iterator,
        importer_func=importer_func,
    )
    self.analyzer = analyzer or OutputAnalyzer(registry=self.registry)
    self.targets_provider = targets_provider or _default_targets_provider

instrument()

Applies runtime patches to PromptFlow's core modules.

It targets: - promptflow.tools.common.render_jinja_template: To catch prompt usage. - promptflow.core.tool: To catch python tool usage.

Note

This method is fault-tolerant; if PromptFlow is not installed, it simply skips instrumentation without raising errors.

Source code in src/agent_cover/instrumentation/promptflow/patcher.py

def instrument(self):
    """Applies runtime patches to PromptFlow's core modules.

    It targets:
    - `promptflow.tools.common.render_jinja_template`: To catch prompt usage.
    - `promptflow.core.tool`: To catch python tool usage.

    Note:
        This method is fault-tolerant; if PromptFlow is not installed, it simply
        skips instrumentation without raising errors.
    """
    if self.is_instrumented:
        return

    raw_targets = self.targets_provider()
    targets = self._normalize_targets(raw_targets)

    patched_count = 0

    # [MODIFICATION] Use module_iterator
    modules_snapshot = self.module_iterator()

    for target in targets:
        if not self._should_instrument(target):
            continue

        mod_name = target.module
        obj_name = target.class_name
        target_type = target.params.get("type")

        try:
            if mod_name not in modules_snapshot:
                try:
                    self.importer(mod_name)
                    modules_snapshot = self.module_iterator()
                except ModuleNotFoundError:
                    Log.log_skip_missing_module(logger, mod_name)
                    continue
                except Exception as e:
                    logger.warning(e, exc_info=True)
                    continue

            if mod_name in modules_snapshot:
                mod = modules_snapshot[mod_name]

                if hasattr(mod, obj_name):
                    if target_type == "render":
                        self._patch_render_function(mod, obj_name)
                        patched_count += 1
                    elif target_type == "decorator":
                        self._patch_tool_decorator(mod, obj_name)
                        patched_count += 1

        except Exception as e:
            logger.warning(
                f"PromptFlow Patch Error ({mod_name}): {e}", exc_info=True
            )

    if patched_count > 0:
        logger.debug(f"Instrumented {patched_count} PromptFlow internals.")

    self.is_instrumented = True

PromptInstrumentor

Bases: BaseInstrumentor

Instruments prompt classes to track their definition and runtime usage.

This instrumentor applies a dual-strategy approach: 1. Init Strategy: Patches __init__ to register the prompt instance in the registry immediately upon creation. This establishes the "Total Prompts" count. 2. Execution Strategy: Patches methods like format or format_messages to track when a prompt is actually used by the agent.

Attributes:

Name	Type	Description
registry	AgentRegistry	The registry to store coverage data.
init_strategy	PromptInitStrategy	Strategy for wrapping initialization.
exec_strategy	PromptExecutionStrategy	Strategy for wrapping execution methods.

Examples:

How it tracks a LangChain prompt:

# 1. __init__ triggers registration (Definition Coverage)
prompt = PromptTemplate.from_template("Hello {name}")

# 2. format() triggers execution (Runtime Coverage)
prompt.format(name="World")

Source code in src/agent_cover/instrumentation/prompts/patcher.py

class PromptInstrumentor(BaseInstrumentor):
    """Instruments prompt classes to track their definition and runtime usage.

    This instrumentor applies a dual-strategy approach:
    1.  **Init Strategy**: Patches `__init__` to register the prompt instance in the registry immediately upon creation. This establishes the "Total Prompts" count.
    2.  **Execution Strategy**: Patches methods like `format` or `format_messages` to track when a prompt is actually used by the agent.

    Attributes:
        registry (AgentRegistry): The registry to store coverage data.
        init_strategy (PromptInitStrategy): Strategy for wrapping initialization.
        exec_strategy (PromptExecutionStrategy): Strategy for wrapping execution methods.

    Examples:
        How it tracks a LangChain prompt:

        ```python
        # 1. __init__ triggers registration (Definition Coverage)
        prompt = PromptTemplate.from_template("Hello {name}")

        # 2. format() triggers execution (Runtime Coverage)
        prompt.format(name="World")
        ```
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        context_manager: Optional[AgentContextManager] = None,
        patch_manager: Optional[PatchManager] = None,
        module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
        importer_func: Optional[Callable[[str], Any]] = None,
        targets_provider: Optional[Callable[[], TargetList]] = None,
        # Strategies injection
        init_strategy: Optional[PromptInitStrategy] = None,
        exec_strategy: Optional[PromptExecutionStrategy] = None,
        stack_walker: Optional[Callable[[Any], Iterator[Any]]] = None,
    ):
        """Initializes the PromptInstrumentor.

        Args:
            registry: The AgentRegistry.
            context_manager: The AgentContextManager.
            patch_manager: The PatchManager.
            module_iterator: A callable to iterate through modules.
            importer_func: A function to import modules.
            targets_provider: A callable to provide instrumentation targets.
            init_strategy: The strategy for wrapping the __init__ method.
            exec_strategy: The strategy for wrapping execution methods.
            stack_walker:  A callable used to walk the stack and
            identify the prompt's definition location.
        """
        super().__init__(
            registry, context_manager, patch_manager, module_iterator, importer_func
        )
        self.targets_provider = targets_provider or _default_targets_provider
        self.stack_walker = stack_walker

        # Passiamo lo stack walker alla strategia
        self.init_strategy = init_strategy or PromptInitStrategy(
            self.registry, stack_walker=self.stack_walker
        )
        self.exec_strategy = exec_strategy or PromptExecutionStrategy()

    def instrument(self):
        """Instruments the prompts.

        This method retrieves the targets, iterates through the modules, and applies
        the appropriate strategies to patch the prompt classes.
        """
        if self.is_instrumented:
            return

        raw_targets = self.targets_provider()
        targets = self._normalize_targets(raw_targets)
        modules_snapshot = self.module_iterator()

        for target in targets:
            if not self._should_instrument(target):
                continue

            mod_name = target.module
            cls_name = target.class_name
            methods_to_patch = target.methods

            # Read content attribute from config
            content_attr = target.params.get("content_attribute")

            try:
                if mod_name not in modules_snapshot:
                    try:
                        self.importer(mod_name)
                        modules_snapshot = self.module_iterator()
                    except ModuleNotFoundError:
                        Log.log_skip_missing_module(logger, mod_name)
                        continue
                    except Exception as e:
                        logger.warning(e, exc_info=True)
                        continue

                if mod_name in modules_snapshot:
                    mod = modules_snapshot[mod_name]

                    cls = self._resolve_target_class(mod, cls_name)

                    if cls:
                        # Pass content_attr to the patching method
                        self._patch_prompt_class(cls, methods_to_patch, content_attr)
                        logger.debug(
                            f"Patched Prompt: {cls_name} (content_attr={content_attr})"
                        )

            except Exception as e:
                logger.warning(f"Error patching {cls_name}: {e}", exc_info=True)

        self.is_instrumented = True

    def _patch_prompt_class(
        self, cls, methods_to_patch: List[str], content_attr: Optional[str] = None
    ):
        """Patches a prompt class.

        This method applies the instrumentation strategies to the __init__ and
        specified methods of a prompt class.

        Args:
            cls: The prompt class.
            methods_to_patch: A list of methods to patch.
            content_attr: The attribute to read for content hashing.
        """
        if hasattr(cls, "__init__"):
            original_init = cls.__init__
            # Pass content_attr to the strategy wrapper
            wrapper_init = self.init_strategy.wrap(
                original_init, cls.__name__, content_attr
            )
            self._safe_patch(cls, "__init__", wrapper_init)

        for method_name in methods_to_patch:
            if hasattr(cls, method_name):
                original_method = getattr(cls, method_name)
                wrapper_method = self.exec_strategy.wrap(
                    original_method, self.context_manager
                )
                self._safe_patch(cls, method_name, wrapper_method)

Functions

__init__(registry=None, context_manager=None, patch_manager=None, module_iterator=None, importer_func=None, targets_provider=None, init_strategy=None, exec_strategy=None, stack_walker=None)

Initializes the PromptInstrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The AgentRegistry.	None
context_manager	Optional[AgentContextManager]	The AgentContextManager.	None
patch_manager	Optional[PatchManager]	The PatchManager.	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	A callable to iterate through modules.	None
importer_func	Optional[Callable[[str], Any]]	A function to import modules.	None
targets_provider	Optional[Callable[[], TargetList]]	A callable to provide instrumentation targets.	None
init_strategy	Optional[PromptInitStrategy]	The strategy for wrapping the init method.	None
exec_strategy	Optional[PromptExecutionStrategy]	The strategy for wrapping execution methods.	None
stack_walker	Optional[Callable[[Any], Iterator[Any]]]	A callable used to walk the stack and	None

Source code in src/agent_cover/instrumentation/prompts/patcher.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    context_manager: Optional[AgentContextManager] = None,
    patch_manager: Optional[PatchManager] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    importer_func: Optional[Callable[[str], Any]] = None,
    targets_provider: Optional[Callable[[], TargetList]] = None,
    # Strategies injection
    init_strategy: Optional[PromptInitStrategy] = None,
    exec_strategy: Optional[PromptExecutionStrategy] = None,
    stack_walker: Optional[Callable[[Any], Iterator[Any]]] = None,
):
    """Initializes the PromptInstrumentor.

    Args:
        registry: The AgentRegistry.
        context_manager: The AgentContextManager.
        patch_manager: The PatchManager.
        module_iterator: A callable to iterate through modules.
        importer_func: A function to import modules.
        targets_provider: A callable to provide instrumentation targets.
        init_strategy: The strategy for wrapping the __init__ method.
        exec_strategy: The strategy for wrapping execution methods.
        stack_walker:  A callable used to walk the stack and
        identify the prompt's definition location.
    """
    super().__init__(
        registry, context_manager, patch_manager, module_iterator, importer_func
    )
    self.targets_provider = targets_provider or _default_targets_provider
    self.stack_walker = stack_walker

    # Passiamo lo stack walker alla strategia
    self.init_strategy = init_strategy or PromptInitStrategy(
        self.registry, stack_walker=self.stack_walker
    )
    self.exec_strategy = exec_strategy or PromptExecutionStrategy()

instrument()

Instruments the prompts.

This method retrieves the targets, iterates through the modules, and applies the appropriate strategies to patch the prompt classes.

Source code in src/agent_cover/instrumentation/prompts/patcher.py

def instrument(self):
    """Instruments the prompts.

    This method retrieves the targets, iterates through the modules, and applies
    the appropriate strategies to patch the prompt classes.
    """
    if self.is_instrumented:
        return

    raw_targets = self.targets_provider()
    targets = self._normalize_targets(raw_targets)
    modules_snapshot = self.module_iterator()

    for target in targets:
        if not self._should_instrument(target):
            continue

        mod_name = target.module
        cls_name = target.class_name
        methods_to_patch = target.methods

        # Read content attribute from config
        content_attr = target.params.get("content_attribute")

        try:
            if mod_name not in modules_snapshot:
                try:
                    self.importer(mod_name)
                    modules_snapshot = self.module_iterator()
                except ModuleNotFoundError:
                    Log.log_skip_missing_module(logger, mod_name)
                    continue
                except Exception as e:
                    logger.warning(e, exc_info=True)
                    continue

            if mod_name in modules_snapshot:
                mod = modules_snapshot[mod_name]

                cls = self._resolve_target_class(mod, cls_name)

                if cls:
                    # Pass content_attr to the patching method
                    self._patch_prompt_class(cls, methods_to_patch, content_attr)
                    logger.debug(
                        f"Patched Prompt: {cls_name} (content_attr={content_attr})"
                    )

        except Exception as e:
            logger.warning(f"Error patching {cls_name}: {e}", exc_info=True)

    self.is_instrumented = True

ToolInstrumentor

Bases: BaseInstrumentor

Instrumentor for Tool classes and decorators.

This class handles the complexity of tracking tools across different frameworks. It distinguishes between: - Class-based Tools: (e.g., LangChain's BaseTool) where we patch _run and _arun. - Decorated Tools: (e.g., @tool) where we wrap the decorated function.

It ensures that every tool available to the agent is registered as a "coverage target".

Methods:

Name	Description
register_existing_tools	Scans the memory for tools that were instantiated before instrumentation started.
instrument	Applies patches to the classes defined in targets.py.

Source code in src/agent_cover/instrumentation/tools/patcher.py

class ToolInstrumentor(BaseInstrumentor):
    """Instrumentor for Tool classes and decorators.

    This class handles the complexity of tracking tools across different frameworks.
    It distinguishes between:
    - **Class-based Tools**: (e.g., LangChain's `BaseTool`) where we patch `_run` and `_arun`.
    - **Decorated Tools**: (e.g., `@tool`) where we wrap the decorated function.

    It ensures that every tool available to the agent is registered as a "coverage target".

    Methods:
        register_existing_tools: Scans the memory for tools that were instantiated *before* instrumentation started.
        instrument: Applies patches to the classes defined in `targets.py`.
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        context_manager=None,
        patch_manager: Optional[PatchManager] = None,
        module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
        importer_func: Optional[Callable[[str], Any]] = None,
        targets_provider: Optional[Callable[[], TargetList]] = None,
        stack_walker: Optional[Callable[[Any], Iterator[Any]]] = None,
    ):
        """Initializes the ToolInstrumentor.

        Args:
            registry: The agent registry to record data into.
            context_manager: The context manager for tracking execution flow.
            patch_manager: The manager responsible for applying safe patches.
            module_iterator: A callable that returns the current snapshot of sys.modules.
            importer_func: A callable to import modules dynamically.
            targets_provider: A callable returning specific tool classes to target.
            stack_walker: A callable to walk the stack for location resolution.
        """
        super().__init__(
            registry, context_manager, patch_manager, module_iterator, importer_func
        )
        self.targets_provider = targets_provider or _default_targets_provider
        self.stack_walker = stack_walker

    def instrument(self) -> None:
        """Performs the instrumentation process for tools.

        It iterates through the targets, resolves the classes (e.g. `langchain.tools.BaseTool`),
        and applies a wrapper that records execution to the [`AgentRegistry`][agent_cover.registry.AgentRegistry]
        before delegating to the original method.
        """
        if self.is_instrumented:
            return

        raw_targets = self.targets_provider()
        targets = self._normalize_targets(raw_targets)
        modules_snapshot = self.module_iterator()

        for target in targets:
            if not self._should_instrument(target):
                continue

            mod_name = target.module
            cls_name = target.class_name
            methods_to_patch = target.methods

            # Read the "name_attribute" configuration from the params (default: "name")
            name_attr = target.params.get("name_attribute", "name")

            try:
                if mod_name not in modules_snapshot:
                    try:
                        self.importer(mod_name)
                        modules_snapshot = self.module_iterator()
                    except ModuleNotFoundError:
                        Log.log_skip_missing_module(logger, mod_name)
                        continue
                    except Exception as e:
                        logger.warning(e, exc_info=True)
                        continue

                if mod_name in modules_snapshot:
                    mod = modules_snapshot[mod_name]

                    cls = self._resolve_target_class(mod, cls_name)

                    if cls and "_ac_tool_patched" not in cls.__dict__:
                        # Pass the configured attribute to the patch method
                        if isinstance(methods_to_patch, list):
                            self._patch_tool_class(cls, methods_to_patch, name_attr)
                            logger.debug(
                                f"Patched Tool: {cls_name} (name_attr={name_attr})"
                            )

            except Exception as e:
                logger.warning(f"Error patching tool {cls_name}: {e}", exc_info=True)

        self.register_existing_tools()
        self.is_instrumented = True

    def _patch_tool_class(
        self, cls: Any, methods_to_patch: List[str], name_attr: str = "name"
    ) -> None:
        """Patches a specific tool class.

        Wraps the `__init__` method to register the tool definition upon instantiation
        and wraps execution methods to record tool usage.

        Args:
            cls: The class to patch.
            methods_to_patch: A list of method names (e.g., 'invoke', '_run') to patch.
            name_attr: The dot-notation path to the tool name attribute (e.g. 'name' or 'metadata.name').
        """
        # We save the configuration to the class for runtime use (e.g. in _get_safe_tool_id)
        setattr(cls, "_ac_name_attr", name_attr)

        original_init = cls.__init__

        def patched_init(instance, *args, **kwargs):
            original_init(instance, *args, **kwargs)
            try:
                # Use the helper to retrieve the name based on the configuration
                tool_name = _get_nested_attr(instance, name_attr)

                if tool_name:
                    # Passing the injected stack_walker
                    raw_loc = get_definition_location(
                        registry=self.registry, stack_walker=self.stack_walker
                    )
                    file_path = raw_loc.split(":")[0]
                    line_num = 0
                    try:
                        line_num = int(raw_loc.split(":")[1])
                    except Exception as e:
                        logger.warning(e, exc_info=True)

                    canonical_id = f"{file_path}:TOOL:{tool_name}"
                    object.__setattr__(instance, "_ac_tool_id", canonical_id)

                    self.registry.register_definition(
                        key=canonical_id,
                        kind="TOOL",
                        metadata={
                            "class": cls.__name__,
                            "tool_name": tool_name,
                            "preview": f"Tool: {tool_name}",
                            "line_number": line_num,
                        },
                    )
            except Exception as e:
                logger.warning(f"Init patch error: {e}", exc_info=True)

        self._safe_patch(cls, "__init__", patched_init)

        for method_name in methods_to_patch:
            if hasattr(cls, method_name):
                self._apply_execution_patch(cls, method_name)

        cls._ac_tool_patched = True

    def _apply_execution_patch(self, cls: Any, method_name: str) -> None:
        """Applies a wrapper to an execution method (sync or async).

        Args:
            cls: The class containing the method.
            method_name: The name of the method to wrap.
        """
        original_method = getattr(cls, method_name)
        is_async = inspect.iscoroutinefunction(original_method)

        if is_async:

            @functools.wraps(original_method)
            async def wrapper(instance, *args, **kwargs):
                self._record_tool_usage(instance)
                return await original_method(instance, *args, **kwargs)
        else:

            @functools.wraps(original_method)
            def wrapper(instance, *args, **kwargs):
                self._record_tool_usage(instance)
                return original_method(instance, *args, **kwargs)

        self._safe_patch(cls, method_name, wrapper)

    def _record_tool_usage(self, tool_instance: Any) -> None:
        """Records the usage of a tool in the registry if the context is active.

        Args:
            tool_instance: The instance of the tool being used.
        """
        # This prevents false positives when tools are used in isolation (e.g. unit tests).
        if self.context_manager and not self.context_manager.is_active():
            return

        tool_id = self._get_safe_tool_id(tool_instance)
        if tool_id:
            self.registry.register_execution(tool_id)

    def _get_safe_tool_id(self, tool_instance: Any) -> Optional[str]:
        """Retrieves the canonical ID for a tool instance.

        Args:
            tool_instance: The tool instance.

        Returns:
            Optional[str]: The tool ID string, or None if it cannot be determined.
        """
        if hasattr(tool_instance, "_ac_tool_id"):
            return tool_instance._ac_tool_id

        # Retrieves the configuration injected into the class during patching.
        # If it doesn't exist (e.g., uninstrumented tool or existing_tool), fallback to "name".
        name_attr = getattr(tool_instance, "_ac_name_attr", "name")

        tool_name = _get_nested_attr(tool_instance, name_attr)

        if not tool_name:
            return None
        return f"TOOL:{tool_name}"

    def register_existing_tools(self) -> None:
        """Scans currently loaded modules for tools that may have been missed.

        This is useful for tools instantiated before the instrumentor was initialized.
        It scans variables in loaded modules to find objects that look like tools.
        """
        cwd = os.getcwd()
        modules_snapshot = self.module_iterator()

        for mod_name, mod in list(modules_snapshot.items()):
            if not hasattr(mod, "__file__") or not mod.__file__:
                continue
            mod_file = os.path.abspath(mod.__file__)
            if not mod_file.startswith(cwd) or "site-packages" in mod_file:
                continue

            for name, val in list(vars(mod).items()):
                if name.startswith("_"):
                    continue

                # Basic heuristic: must have invoke/_run
                if (
                    hasattr(val, "invoke")
                    or hasattr(val, "_run")
                    or hasattr(val, "call")
                ):
                    try:
                        if hasattr(val, "_ac_tool_id"):
                            continue

                        # Get the path to the name attribute (if the class has been patched, it will have it)
                        name_attr = getattr(val, "_ac_name_attr", "name")
                        tool_name = _get_nested_attr(val, name_attr)

                        if not tool_name:
                            continue

                        canonical_id = f"{mod_file}:TOOL:{tool_name}"

                        try:
                            object.__setattr__(val, "_ac_tool_id", canonical_id)
                        except Exception as e:
                            logger.warning(e, exc_info=True)

                        if canonical_id not in self.registry.definitions:
                            self.registry.register_definition(
                                key=canonical_id,
                                kind="TOOL",
                                metadata={
                                    "class": val.__class__.__name__,
                                    "tool_name": tool_name,
                                    "preview": f"Tool: {tool_name}",
                                    "line_number": 0,
                                },
                            )
                    except Exception as e:
                        logger.warning(e, exc_info=True)

Functions

__init__(registry=None, context_manager=None, patch_manager=None, module_iterator=None, importer_func=None, targets_provider=None, stack_walker=None)

Initializes the ToolInstrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The agent registry to record data into.	None
context_manager		The context manager for tracking execution flow.	None
patch_manager	Optional[PatchManager]	The manager responsible for applying safe patches.	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	A callable that returns the current snapshot of sys.modules.	None
importer_func	Optional[Callable[[str], Any]]	A callable to import modules dynamically.	None
targets_provider	Optional[Callable[[], TargetList]]	A callable returning specific tool classes to target.	None
stack_walker	Optional[Callable[[Any], Iterator[Any]]]	A callable to walk the stack for location resolution.	None

Source code in src/agent_cover/instrumentation/tools/patcher.py

def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    context_manager=None,
    patch_manager: Optional[PatchManager] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    importer_func: Optional[Callable[[str], Any]] = None,
    targets_provider: Optional[Callable[[], TargetList]] = None,
    stack_walker: Optional[Callable[[Any], Iterator[Any]]] = None,
):
    """Initializes the ToolInstrumentor.

    Args:
        registry: The agent registry to record data into.
        context_manager: The context manager for tracking execution flow.
        patch_manager: The manager responsible for applying safe patches.
        module_iterator: A callable that returns the current snapshot of sys.modules.
        importer_func: A callable to import modules dynamically.
        targets_provider: A callable returning specific tool classes to target.
        stack_walker: A callable to walk the stack for location resolution.
    """
    super().__init__(
        registry, context_manager, patch_manager, module_iterator, importer_func
    )
    self.targets_provider = targets_provider or _default_targets_provider
    self.stack_walker = stack_walker

instrument()

Performs the instrumentation process for tools.

It iterates through the targets, resolves the classes (e.g. langchain.tools.BaseTool), and applies a wrapper that records execution to the AgentRegistry before delegating to the original method.

Source code in src/agent_cover/instrumentation/tools/patcher.py

def instrument(self) -> None:
    """Performs the instrumentation process for tools.

    It iterates through the targets, resolves the classes (e.g. `langchain.tools.BaseTool`),
    and applies a wrapper that records execution to the [`AgentRegistry`][agent_cover.registry.AgentRegistry]
    before delegating to the original method.
    """
    if self.is_instrumented:
        return

    raw_targets = self.targets_provider()
    targets = self._normalize_targets(raw_targets)
    modules_snapshot = self.module_iterator()

    for target in targets:
        if not self._should_instrument(target):
            continue

        mod_name = target.module
        cls_name = target.class_name
        methods_to_patch = target.methods

        # Read the "name_attribute" configuration from the params (default: "name")
        name_attr = target.params.get("name_attribute", "name")

        try:
            if mod_name not in modules_snapshot:
                try:
                    self.importer(mod_name)
                    modules_snapshot = self.module_iterator()
                except ModuleNotFoundError:
                    Log.log_skip_missing_module(logger, mod_name)
                    continue
                except Exception as e:
                    logger.warning(e, exc_info=True)
                    continue

            if mod_name in modules_snapshot:
                mod = modules_snapshot[mod_name]

                cls = self._resolve_target_class(mod, cls_name)

                if cls and "_ac_tool_patched" not in cls.__dict__:
                    # Pass the configured attribute to the patch method
                    if isinstance(methods_to_patch, list):
                        self._patch_tool_class(cls, methods_to_patch, name_attr)
                        logger.debug(
                            f"Patched Tool: {cls_name} (name_attr={name_attr})"
                        )

        except Exception as e:
            logger.warning(f"Error patching tool {cls_name}: {e}", exc_info=True)

    self.register_existing_tools()
    self.is_instrumented = True

register_existing_tools()

Scans currently loaded modules for tools that may have been missed.

This is useful for tools instantiated before the instrumentor was initialized. It scans variables in loaded modules to find objects that look like tools.

Source code in src/agent_cover/instrumentation/tools/patcher.py

def register_existing_tools(self) -> None:
    """Scans currently loaded modules for tools that may have been missed.

    This is useful for tools instantiated before the instrumentor was initialized.
    It scans variables in loaded modules to find objects that look like tools.
    """
    cwd = os.getcwd()
    modules_snapshot = self.module_iterator()

    for mod_name, mod in list(modules_snapshot.items()):
        if not hasattr(mod, "__file__") or not mod.__file__:
            continue
        mod_file = os.path.abspath(mod.__file__)
        if not mod_file.startswith(cwd) or "site-packages" in mod_file:
            continue

        for name, val in list(vars(mod).items()):
            if name.startswith("_"):
                continue

            # Basic heuristic: must have invoke/_run
            if (
                hasattr(val, "invoke")
                or hasattr(val, "_run")
                or hasattr(val, "call")
            ):
                try:
                    if hasattr(val, "_ac_tool_id"):
                        continue

                    # Get the path to the name attribute (if the class has been patched, it will have it)
                    name_attr = getattr(val, "_ac_name_attr", "name")
                    tool_name = _get_nested_attr(val, name_attr)

                    if not tool_name:
                        continue

                    canonical_id = f"{mod_file}:TOOL:{tool_name}"

                    try:
                        object.__setattr__(val, "_ac_tool_id", canonical_id)
                    except Exception as e:
                        logger.warning(e, exc_info=True)

                    if canonical_id not in self.registry.definitions:
                        self.registry.register_definition(
                            key=canonical_id,
                            kind="TOOL",
                            metadata={
                                "class": val.__class__.__name__,
                                "tool_name": tool_name,
                                "preview": f"Tool: {tool_name}",
                                "line_number": 0,
                            },
                        )
                except Exception as e:
                    logger.warning(e, exc_info=True)

Functions

instrument_all(registry=None, context_manager=None, patch_manager=None, analyzer=None, config=None, importer_func=None, module_iterator=None, inspection_provider=None, targets_provider_map=None, stack_walker=None)

Applies all available instrumentation strategies.

This function orchestrates the initialization and execution of various instrumentors, including raw strings, callbacks, agents, prompts, tools, LLM providers, and PromptFlow. It uses dependency injection to allow custom implementations for testing purposes.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The agent registry to use. If None, the global registry is used.	None
context_manager	Optional[AgentContextManager]	The context manager for tracking agent state.	None
patch_manager	Optional[PatchManager]	The manager for applying safe patches.	None
analyzer	Optional[OutputAnalyzer]	The analyzer for processing outputs.	None
config	Optional[AgentCoverConfig]	The configuration object.	None
importer_func	Optional[Callable[[str], Any]]	Optional function to import modules (for DI).	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	Optional function to iterate over modules (for DI).	None
inspection_provider	Optional[InspectionProvider]	Optional provider for code inspection (for DI).	None
targets_provider_map	Optional[Dict[str, Callable]]	Optional map of target providers (for DI).	None
stack_walker	Optional[Callable[[Any], Iterator[Any]]]	Optional function to walk the stack (for DI).	None

Returns:

Type	Description
List[BaseInstrumentor]	List[BaseInstrumentor]: A list of successfully initialized and
List[BaseInstrumentor]	active instrumentor instances.

Source code in src/agent_cover/instrumentation/__init__.py

def instrument_all(
    registry: Optional[AgentRegistry] = None,
    context_manager: Optional[AgentContextManager] = None,
    patch_manager: Optional[PatchManager] = None,
    analyzer: Optional[OutputAnalyzer] = None,
    config: Optional[AgentCoverConfig] = None,
    # --- DEPENDENCY INJECTION FOR TESTS ---
    importer_func: Optional[Callable[[str], Any]] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    inspection_provider: Optional[InspectionProvider] = None,
    targets_provider_map: Optional[Dict[str, Callable]] = None,
    stack_walker: Optional[Callable[[Any], Iterator[Any]]] = None,
) -> List[BaseInstrumentor]:
    """Applies all available instrumentation strategies.

    This function orchestrates the initialization and execution of various
    instrumentors, including raw strings, callbacks, agents, prompts, tools,
    LLM providers, and PromptFlow. It uses dependency injection to allow
    custom implementations for testing purposes.

    Args:
        registry: The agent registry to use. If None, the global registry is used.
        context_manager: The context manager for tracking agent state.
        patch_manager: The manager for applying safe patches.
        analyzer: The analyzer for processing outputs.
        config: The configuration object.
        importer_func: Optional function to import modules (for DI).
        module_iterator: Optional function to iterate over modules (for DI).
        inspection_provider: Optional provider for code inspection (for DI).
        targets_provider_map: Optional map of target providers (for DI).
        stack_walker: Optional function to walk the stack (for DI).

    Returns:
        List[BaseInstrumentor]: A list of successfully initialized and
        active instrumentor instances.
    """
    if registry is None:
        registry = get_registry()
    if context_manager is None:
        context_manager = get_global_context_manager()
    if patch_manager is None:
        patch_manager = DefaultPatchManager()
    if config is None:
        config = get_config()
    if analyzer is None:
        analyzer = OutputAnalyzer(registry=registry, config=config)

    if targets_provider_map is None:
        targets_provider_map = {}

    logger.debug("Initializing instrumentation...")

    active_instrumentors: List[BaseInstrumentor] = []

    # 1. Raw Strings
    try:
        scan_raw_string_prompts(registry=registry, module_iterator=module_iterator)
    except Exception as e:
        logger.warning(f"Failed raw strings scan: {e}", exc_info=True)

    # 2. Callbacks
    try:

        def _handler_factory():
            return CoverageCallbackHandler(registry=registry, analyzer=analyzer)

        cb_inst = GlobalCallbackInstrumentor(
            registry=registry,
            patch_manager=patch_manager,
            handler_factory=_handler_factory,
        )
        cb_inst.instrument()
        active_instrumentors.append(cb_inst)
    except Exception as e:
        logger.warning(f"Failed 2. Callbacks {e}", exc_info=True)

    # 3. Agents
    try:
        agent_inst = AgentInstrumentor(
            registry=registry,
            context_manager=context_manager,
            patch_manager=patch_manager,
            importer_func=importer_func,
            module_iterator=module_iterator,
            targets_provider=targets_provider_map.get("agents"),
        )
        agent_inst.instrument()
        active_instrumentors.append(agent_inst)
    except Exception as e:
        logger.warning(f"Failed 3. Agents {e}", exc_info=True)

    # 4. Prompts
    try:
        p_inst = PromptInstrumentor(
            registry=registry,
            context_manager=context_manager,
            patch_manager=patch_manager,
            importer_func=importer_func,
            module_iterator=module_iterator,
            targets_provider=targets_provider_map.get("prompts"),
            stack_walker=stack_walker,
        )
        p_inst.instrument()
        active_instrumentors.append(p_inst)
    except Exception as e:
        logger.warning(f"Failed 4. Prompts {e}", exc_info=True)

    # 5. Tools
    try:
        t_inst = ToolInstrumentor(
            registry=registry,
            patch_manager=patch_manager,
            importer_func=importer_func,
            module_iterator=module_iterator,
            targets_provider=targets_provider_map.get("tools"),
            stack_walker=stack_walker,
        )
        t_inst.instrument()
        active_instrumentors.append(t_inst)
    except Exception as e:
        logger.warning(f"Failed 5. Tools {e}", exc_info=True)

    # 6. LLM Providers
    try:
        l_inst = LLMProviderInstrumentor(
            registry=registry,
            analyzer=analyzer,
            patch_manager=patch_manager,
            importer_func=importer_func,
            module_iterator=module_iterator,
            targets_provider=targets_provider_map.get("llm"),
        )
        l_inst.instrument()
        active_instrumentors.append(l_inst)
    except Exception as e:
        logger.warning(f"Failed 6. LLM Providers {e}", exc_info=True)

    # 7. PromptFlow
    try:
        pf_inst = PromptFlowInstrumentor(
            registry=registry,
            analyzer=analyzer,
            patch_manager=patch_manager,
            importer_func=importer_func,
            module_iterator=module_iterator,
            targets_provider=targets_provider_map.get("promptflow"),
        )
        pf_inst.instrument()
        active_instrumentors.append(pf_inst)
    except Exception as e:
        logger.warning(
            f"CRITICAL: Failed to initialize PromptFlow instrumentor: {e}",
            exc_info=True,
        )

    # 8. Structures
    try:
        scan_pydantic_models(
            registry=registry,
            config=config,
            module_iterator=module_iterator,
            inspector=inspection_provider,
        )
    except Exception as e:
        logger.warning(f"Failed 8. Structures: {e}", exc_info=True)

    return active_instrumentors

instrument_promptflow(registry=None)

Legacy helper function to instantiate and run the instrumentor.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	Optional AgentRegistry instance.	None

Returns:

Name	Type	Description
PromptFlowInstrumentor		The instrumentor instance.

Source code in src/agent_cover/instrumentation/promptflow/patcher.py

def instrument_promptflow(registry: Optional[AgentRegistry] = None):
    """Legacy helper function to instantiate and run the instrumentor.

    Args:
        registry: Optional AgentRegistry instance.

    Returns:
        PromptFlowInstrumentor: The instrumentor instance.
    """
    inst = PromptFlowInstrumentor(registry=registry)
    inst.instrument()
    return inst

register_existing_prompts(registry=None, root_path=None, module_iterator=None)

Registers existing prompts by scanning modules.

This function scans modules for prompts that are already defined.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The AgentRegistry to register the prompts with.	None
root_path	Optional[str]	The root path to search for modules.	None
module_iterator	Optional[Callable]	A callable to iterate through modules.	None

Source code in src/agent_cover/instrumentation/prompts/patcher.py

def register_existing_prompts(
    registry: Optional[AgentRegistry] = None,
    root_path: Optional[str] = None,
    module_iterator: Optional[Callable] = None,
):
    """Registers existing prompts by scanning modules.

    This function scans modules for prompts that are already defined.

    Args:
        registry: The AgentRegistry to register the prompts with.
        root_path: The root path to search for modules.
        module_iterator: A callable to iterate through modules.
    """
    if registry is None:
        registry = get_registry()
    if root_path is None:
        root_path = os.getcwd()
    if module_iterator is None:
        module_iterator = _default_module_iterator

    modules_snapshot = module_iterator()

    for mod_name, mod in modules_snapshot.items():
        if not hasattr(mod, "__file__") or not mod.__file__:
            continue

        try:
            mod_file = os.path.abspath(mod.__file__)

            if not mod_file.startswith(root_path) or "site-packages" in mod_file:
                continue

            for name, val in list(vars(mod).items()):
                if not isinstance(val, type) and (
                    hasattr(val, "template") or hasattr(val, "messages")
                ):
                    content = ""
                    # Note: Static scanning relies on common attribute names as we don't have
                    # the target config here easily without a class lookup map.
                    if hasattr(val, "template") and isinstance(val.template, str):
                        content = val.template
                    elif hasattr(val, "messages"):
                        content = str(val.messages)

                    if content:
                        raw_id = f"{mod_file}:PROMPT:{name}"
                        canonical_id = registry.get_canonical_id(content, raw_id)

                        try:
                            object.__setattr__(val, "_coverage_id", canonical_id)
                        except Exception as e:
                            logger.warning(e, exc_info=True)

                        if canonical_id not in registry.definitions:
                            registry.register_definition(
                                key=canonical_id,
                                kind="PROMPT",
                                metadata={
                                    "class": val.__class__.__name__,
                                    "preview": content[:50].replace("\n", " "),
                                    "file_path": mod_file,
                                    "variable_name": name,
                                },
                            )
                            logger.debug(
                                f"Discovered static prompt: {name} in {mod_name}"
                            )

        except Exception as e:
            logger.warning(f"Error scanning module {mod_name}: {e}", exc_info=True)

scan_pydantic_models(registry=None, config=None, root_path=None, adapter_registry=None, module_iterator=None, inspector=None)

Scans loaded modules to automatically generate Business Logic decisions.

This function inspects Pydantic models and Dataclasses to find fields with finite sets of expected values. It automatically populates the AgentCoverConfig with these rules.

Supported Types: - Enum: Expects all enum members. - Literal: Expects all literal values (e.g., Literal["yes", "no"]). - Bool: Expects True and False.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The AgentRegistry instance. Defaults to the result of get_registry().	None
config	Optional[AgentCoverConfig]	The AgentCoverConfig instance. Defaults to the result of get_config().	None
root_path	Optional[str]	The root path to scan for modules. Defaults to the current working directory.	None
adapter_registry	Optional[AdapterRegistry]	The AdapterRegistry instance. Defaults to the result of get_default_adapter_registry().	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	A callable that returns an iterator over the modules. Defaults to _default_module_iterator.	None
inspector	Optional[InspectionProvider]	An instance of InspectionProvider for abstracting inspect calls. Defaults to None.	None

Examples:

If you have this model:

class Sentiment(BaseModel):
    label: Literal["POSITIVE", "NEGATIVE"]

This scanner creates a Decision rule expecting both "POSITIVE" and "NEGATIVE" to appear in the label field during testing.

Source code in src/agent_cover/instrumentation/structures/scanner.py

def scan_pydantic_models(
    registry: Optional[AgentRegistry] = None,
    config: Optional[AgentCoverConfig] = None,
    root_path: Optional[str] = None,
    adapter_registry: Optional[AdapterRegistry] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    inspector: Optional[InspectionProvider] = None,
):
    """Scans loaded modules to automatically generate Business Logic decisions.

    This function inspects Pydantic models and Dataclasses to find fields with
    finite sets of expected values. It automatically populates the
    [`AgentCoverConfig`][agent_cover.config.AgentCoverConfig] with these rules.

    Supported Types:
    - **Enum**: Expects all enum members.
    - **Literal**: Expects all literal values (e.g., `Literal["yes", "no"]`).
    - **Bool**: Expects `True` and `False`.

    Args:
        registry: The AgentRegistry instance. Defaults to the result of get_registry().
        config: The AgentCoverConfig instance. Defaults to the result of get_config().
        root_path: The root path to scan for modules. Defaults to the current working directory.
        adapter_registry: The AdapterRegistry instance. Defaults to the result of get_default_adapter_registry().
        module_iterator: A callable that returns an iterator over the modules. Defaults to _default_module_iterator.
        inspector: An instance of InspectionProvider for abstracting inspect calls. Defaults to None.

    Examples:
        If you have this model:
        ```python
        class Sentiment(BaseModel):
            label: Literal["POSITIVE", "NEGATIVE"]
        ```
        This scanner creates a Decision rule expecting both "POSITIVE" and "NEGATIVE"
        to appear in the `label` field during testing.
    """
    if registry is None:
        registry = get_registry()
    if config is None:
        config = get_config()
    if root_path is None:
        root_path = os.getcwd()
    if adapter_registry is None:
        adapter_registry = get_default_adapter_registry()
    if module_iterator is None:
        module_iterator = _default_module_iterator
    if inspector is None:
        inspector = InspectionProvider()

    existing_ids = {d.id for d in config.decisions}
    count = 0

    modules_snapshot = module_iterator()

    for mod_name, mod in modules_snapshot.items():
        if not hasattr(mod, "__file__") or not mod.__file__:
            continue
        mod_file = os.path.abspath(mod.__file__)

        # User code filter
        if not mod_file.startswith(root_path) or "site-packages" in mod_file:
            continue

        for name, val in list(vars(mod).items()):
            adapter = adapter_registry.get_adapter_for_class(val)
            if not adapter:
                continue

            try:
                # We use the injected inspector to get the file
                class_file = inspector.get_file(val)
                class_file = os.path.abspath(class_file)

                if (
                    not class_file.startswith(root_path)
                    or "site-packages" in class_file
                ):
                    continue

                try:
                    # We use the injected inspector for the lines of code
                    lines, start_line = inspector.get_source_lines(val)
                except Exception as e:
                    logger.warning(e, exc_info=True)
                    start_line = 0
            except Exception as e:
                logger.warning(e, exc_info=True)
                # If get_file fails (e.g. built-in or dynamic class without mocks), we skip
                continue

            try:
                fields_map = adapter.get_fields(val)
                _analyze_model_fields(
                    val, fields_map, config, existing_ids, class_file, start_line
                )
                count += 1
            except Exception as e:
                logger.warning(f"[AgentCover][STRUCT] Skip {name}: {e}", exc_info=True)

scan_raw_string_prompts(registry=None, root_path=None, module_iterator=None, source_reader=None)

Scans for raw string prompts in Python modules and registers them.

It iterates through all loaded modules in sys.modules that reside within root_path. For each string variable matching the configured prefixes/suffixes, it: 1. Calculates a regex pattern for runtime matching. 2. Registers it in the AgentRegistry.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	An optional AgentRegistry instance. If not provided, it defaults to the global registry.	None
root_path	Optional[str]	An optional root path to start the scan from. If not provided, it defaults to the current working directory.	None
module_iterator	Optional[Callable[[], Dict[str, Any]]]	An optional callable for iterating through loaded modules. Defaults to _default_module_iterator.	None
source_reader	Optional[Callable[[str], List[str]]]	An optional callable for reading source file contents. Defaults to _default_source_reader.	None

Source code in src/agent_cover/instrumentation/raw_strings/scanner.py

def scan_raw_string_prompts(
    registry: Optional[AgentRegistry] = None,
    root_path: Optional[str] = None,
    module_iterator: Optional[Callable[[], Dict[str, Any]]] = None,
    source_reader: Optional[Callable[[str], List[str]]] = None,
) -> None:
    """Scans for raw string prompts in Python modules and registers them.

    It iterates through all loaded modules in `sys.modules` that reside within
    `root_path`. For each string variable matching the configured prefixes/suffixes,
    it:
    1.  Calculates a regex pattern for runtime matching.
    2.  Registers it in the [`AgentRegistry`][agent_cover.registry.AgentRegistry].

    Args:
        registry: An optional AgentRegistry instance. If not provided, it
            defaults to the global registry.
        root_path: An optional root path to start the scan from. If not
            provided, it defaults to the current working directory.
        module_iterator: An optional callable for iterating through
            loaded modules.  Defaults to _default_module_iterator.
        source_reader: An optional callable for reading source file
            contents.  Defaults to _default_source_reader.
    """
    if registry is None:
        registry = get_registry()
    if root_path is None:
        root_path = os.getcwd()  # Default Prod
    if module_iterator is None:
        module_iterator = _default_module_iterator
    if source_reader is None:
        source_reader = _default_source_reader

    active_prefixes = tuple(DEFAULT_PREFIXES + _custom_prefixes)
    active_suffixes = tuple(DEFAULT_SUFFIXES + _custom_suffixes)

    modules_snapshot = module_iterator()

    for mod_name, mod in modules_snapshot.items():
        if not hasattr(mod, "__file__") or not mod.__file__:
            continue
        mod_file = os.path.abspath(mod.__file__)

        if not mod_file.startswith(root_path) or "site-packages" in mod_file:
            continue

        file_lines = None

        for name, val in list(vars(mod).items()):
            if not isinstance(val, str):
                continue

            match_prefix = name.startswith(active_prefixes)
            match_suffix = name.endswith(active_suffixes)

            if not (match_prefix or match_suffix):
                continue

            if len(val) < 10:
                continue

            if file_lines is None:
                file_lines = source_reader(mod_file)

            line_num = _find_variable_line_number(name, file_lines)
            if line_num == 0:
                continue

            clean_content = val.strip()
            regex_pattern = _create_robust_regex(clean_content)

            raw_id = f"RAW:{mod_file}::{name}"
            canonical_id = registry.get_canonical_id(clean_content, raw_id)

            if canonical_id not in registry.definitions:
                logger.debug(
                    f"[AgentCover][SCAN] Found {name} at line {line_num} in {mod_file}"
                )
                registry.register_definition(
                    key=canonical_id,
                    kind="PROMPT",
                    metadata={
                        "class": "StringConstant",
                        "preview": clean_content[:40],
                        "raw_content": clean_content,
                        "regex_pattern": regex_pattern,
                        "line_number": line_num,
                    },
                )

scan_static_definitions(registry=None, config=None, root_path=None, module_iterator=None, source_reader=None, inspector=None)

Scans the codebase for static definitions.

This function triggers static analysis to identify prompts, raw strings, and data structures (like Pydantic models) without executing the code.

Parameters:

Name	Type	Description	Default
registry	Optional[AgentRegistry]	The agent registry to store findings.	None
config	Optional[AgentCoverConfig]	The configuration object.	None
root_path	Optional[str]	The root directory to scan.	None
module_iterator	Optional[Callable]	Optional function to iterate over modules.	None
source_reader	Optional[Callable]	Optional function to read source files.	None
inspector	Optional[InspectionProvider]	Optional provider for code inspection.	None

Source code in src/agent_cover/instrumentation/__init__.py

def scan_static_definitions(
    registry: Optional[AgentRegistry] = None,
    config: Optional[AgentCoverConfig] = None,
    root_path: Optional[str] = None,
    module_iterator: Optional[Callable] = None,
    source_reader: Optional[Callable] = None,
    inspector: Optional[InspectionProvider] = None,
):
    """Scans the codebase for static definitions.

    This function triggers static analysis to identify prompts, raw strings,
    and data structures (like Pydantic models) without executing the code.

    Args:
        registry: The agent registry to store findings.
        config: The configuration object.
        root_path: The root directory to scan.
        module_iterator: Optional function to iterate over modules.
        source_reader: Optional function to read source files.
        inspector: Optional provider for code inspection.
    """
    if registry is None:
        registry = get_registry()
    if config is None:
        config = get_config()
    if root_path is None:
        root_path = os.getcwd()

    register_existing_prompts(
        registry=registry, root_path=root_path, module_iterator=module_iterator
    )

    scan_raw_string_prompts(
        registry=registry,
        root_path=root_path,
        module_iterator=module_iterator,
        source_reader=source_reader,
    )

    try:
        scan_pydantic_models(
            registry=registry,
            config=config,
            root_path=root_path,
            module_iterator=module_iterator,
            inspector=inspector,
        )
    except Exception as e:
        logger.warning(f"Struct scan error: {e}", exc_info=True)