sdk

agent_cover.sdk

Main entry point for the AgentCover SDK.

This module provides high-level functions to manually control the coverage process outside of a standard pytest environment (e.g., in a custom script or a notebook).

It is designed to handle complex execution models, including multi-process environments like Microsoft PromptFlow. It orchestrates the initialization of instrumentors, manages signals and hooks for distributed workers, and handles data persistence via JSON fragments and consolidated reports.

Exposed Functions

• start_coverage: Initializes and starts tracking.
• generate_report: Dumps the results to HTML/XML.

SDK Usage Example

If you want to run coverage for a script and automatically generate a report on exit:

from agent_cover.sdk import start_coverage

if __name__ == "__main__":
    # 1. Start coverage (registers atexit hooks for reporting)
    start_coverage(
        source_dir="src",
        report_dir="reports/agent-cover",
        auto_save=True
    )

    # 2. Run your code
    run_my_agent_logic()

    # When the script ends, reports are generated automatically.

Functions

arm_worker_handlers()

Arms signal handlers and atexit hooks for the current process.

This ensures that coverage data is flushed to disk when the process receives a termination signal (SIGTERM) or exits normally. This is critical for PromptFlow workers on Linux/WSL.

Source code in src/agent_cover/sdk.py

def arm_worker_handlers() -> None:
    """Arms signal handlers and atexit hooks for the current process.

    This ensures that coverage data is flushed to disk when the process
    receives a termination signal (SIGTERM) or exits normally. This is
    critical for PromptFlow workers on Linux/WSL.
    """
    global _HANDLERS_ARMED
    if _HANDLERS_ARMED:
        return
    _HANDLERS_ARMED = True

    def _termination_handler(signum: int, frame: Any) -> None:
        """Handles external termination signals."""
        logger.debug(
            f"Process {os.getpid()} received signal {signum}. Flushing coverage..."
        )
        _save_report_logic()
        # Exit to allow the parent orchestrator to continue
        sys.exit(0)

    # Register handlers for common termination signals used by orchestrators
    try:
        signal.signal(signal.SIGTERM, _termination_handler)
        signal.signal(signal.SIGINT, _termination_handler)
    except (ValueError, OSError):
        # signal.signal only works in the main thread.
        pass

    # Also register for normal python exit
    atexit.register(_save_report_logic)

generate_report(output_dir='coverage_report', xml_file='coverage.xml', source_dir=None)

Generates the final coverage reports (HTML and Cobertura XML).

Parameters:

Name	Type	Description	Default
output_dir	str	Directory where the reports will be saved.	'coverage_report'
xml_file	str	Filename for the XML Cobertura report.	'coverage.xml'
source_dir	str	If provided, triggers a re-scan of definitions.	None

Source code in src/agent_cover/sdk.py

def generate_report(
    output_dir: str = "coverage_report",
    xml_file: str = "coverage.xml",
    source_dir: Optional[str] = None,
) -> None:
    """Generates the final coverage reports (HTML and Cobertura XML).

    Args:
        output_dir (str): Directory where the reports will be saved.
        xml_file (str): Filename for the XML Cobertura report.
        source_dir (str, optional): If provided, triggers a re-scan of definitions.
    """
    if source_dir:
        try:
            scan_static_definitions(root_path=source_dir)
        except Exception:
            pass

    config = get_config()
    os.makedirs(output_dir, exist_ok=True)

    # Generate HTML summary
    generate_html_report(
        registry.definitions,
        registry.executions,
        output_dir,
        config,
        registry.decision_hits,
    )

    # Generate Cobertura XML for CI/CD integration
    xml_path = os.path.join(output_dir, xml_file)
    generate_cobertura_xml(
        registry.definitions,
        registry.executions,
        config,
        registry.decision_hits,
        xml_path,
    )

start_coverage(source_dir=None, report_dir='agent_coverage_report', auto_save=True, **kwargs)

Initializes and starts the coverage tracking session.

This function sets up the configuration, instruments supported libraries, and performs static discovery of project assets (Prompts, Tools, Flows). It is suitable for manual instrumentation in scripts or distributed worker processes.

Parameters:

Name	Type	Description	Default
source_dir	Optional[str]	The root directory of the source code to cover. Defaults to the current working directory.	None
report_dir	str	The directory where coverage reports and fragments will be saved. Defaults to "agent_coverage_report".	'agent_coverage_report'
auto_save	bool	If True, enables immediate flushing on hits and registers atexit/signal hooks to generate reports automatically when the process finishes.	True
atexit_registrar	Optional[Callable]	Dependency injection for testing exit registration.	required
**kwargs	Any	Additional parameters like is_worker to identify sub-processes in multiprocessing environments (e.g., PromptFlow workers).	{}

Examples:

Basic Usage:

from agent_cover.sdk import start_coverage

# Start tracking and auto-save on exit
start_coverage(source_dir="./src")

# Run your agent logic...

Source code in src/agent_cover/sdk.py

def start_coverage(
    source_dir: Optional[str] = None,
    report_dir: str = "agent_coverage_report",
    auto_save: bool = True,
    **kwargs: Any,
) -> None:
    """Initializes and starts the coverage tracking session.

    This function sets up the configuration, instruments supported libraries,
    and performs static discovery of project assets (Prompts, Tools, Flows).
    It is suitable for manual instrumentation in scripts or distributed worker processes.

    Args:
        source_dir (Optional[str]): The root directory of the source code to cover.
            Defaults to the current working directory.
        report_dir (str): The directory where coverage reports and fragments will be saved.
            Defaults to "agent_coverage_report".
        auto_save (bool): If True, enables immediate flushing on hits and registers
            atexit/signal hooks to generate reports automatically when the process finishes.
        atexit_registrar (Optional[Callable]): Dependency injection for testing exit registration.
        **kwargs: Additional parameters like `is_worker` to identify sub-processes
            in multiprocessing environments (e.g., PromptFlow workers).

    Examples:
        **Basic Usage:**
        ```python
        from agent_cover.sdk import start_coverage

        # Start tracking and auto-save on exit
        start_coverage(source_dir="./src")

        # Run your agent logic...
        ```
    """
    # 1. Update internal configuration store
    if source_dir:
        _CONFIG_STORE["source_dir"] = os.path.abspath(source_dir)
    if report_dir:
        _CONFIG_STORE["report_dir"] = os.path.abspath(report_dir)

    root = _CONFIG_STORE["source_dir"]
    if root not in sys.path:
        sys.path.insert(0, root)

    # 2. Setup environment and instrument targets
    load_config(root)
    instrument_all()

    # 3. PromptFlow specific initialization
    # Workers need to scan definitions to map runtime Jinja2 hashes to filenames.
    from .instrumentation.promptflow import (
        instrument_promptflow,
        scan_promptflow_definitions,
    )

    try:
        scan_promptflow_definitions(root_path=root)
        instrument_promptflow()
    except Exception as e:
        logger.warning(f"PromptFlow initialization failed: {e}")

    # 4. General Static Discovery
    try:
        discover_repository_modules(root_path=root)
        scan_static_definitions(root_path=root)
    except Exception as e:
        logger.debug(f"Static discovery warning: {e}")

    # 5. Multiprocessing and persistence setup
    if auto_save:
        # If running in a bootstrapped subprocess, enable immediate flush.
        # This prevents data loss if a worker is killed abruptly.
        if "AGENT_COVER_BOOTSTRAPPED" in os.environ or kwargs.get("is_worker"):
            registry.on_hit_callback = _save_report_logic
            logger.debug(
                f"AgentCover: Immediate flush enabled for worker {os.getpid()}"
            )

        arm_worker_handlers()
        logger.info(
            f"AgentCover session started. Reports will be saved to {report_dir}"
        )