Index

agent_cover.instrumentation.prompts

Prompt Instrumentation.

This module intercepts the initialization and formatting of Prompt Templates.

🔗 Architectural Relationships

Prompts are unique because they are often defined globally but used locally. This module links these two states.

• Populates: AgentRegistry (Both definitions and executions).
• Depends on: AgentContextManager (Only tracks formatting if inside an agent loop).
• Complementary to: Raw Strings Scanner (This module handles Objects; the scanner handles Strings).

⚙️ How it works

1. Init Strategy: Patches __init__. Calculates a hash of the template text to create a stable ID, linking runtime objects to static files.
2. Execution Strategy: Patches format. Registers usage when the prompt is filled with variables.

Usage

from agent_cover.instrumentation.prompts import instrument_prompts

instrument_prompts()

Classes

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

Functions

instrument_prompts(registry=None)

Backward compatibility function for instrumenting prompts.

Parameters:

Name	Type	Description	Default
registry		The AgentRegistry.	None

Returns:

Type	Description
	The PromptInstrumentor instance.

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

def instrument_prompts(registry=None):
    """Backward compatibility function for instrumenting prompts.

    Args:
        registry: The AgentRegistry.

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

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)