tatlock/src/core/tool_tracking.py

"""
Tool call tracking and benchmarking.

Tracks which tools are recommended by the Steward versus which tools
are actually used by Tatlock, recording benchmarks for analysis.
"""
from datetime import datetime, timezone
from typing import Optional

from src.core.benchmarks import PerformanceBenchmark, get_benchmark_store
from src.core.logging_config import get_logger

logger = get_logger(__name__)


class ToolCallTracker:
    """
    Tracks tool calls for benchmarking and accuracy analysis.

    Compares Steward's recommendations with Tatlock's actual tool usage
    to measure recommendation accuracy.
    """

    def __init__(
        self,
        recommended_capabilities: list[str],
        conversation_id: Optional[str] = None
    ):
        """
        Initialize tool call tracker.

        Args:
            recommended_capabilities: List of capability names recommended by Steward
            conversation_id: Optional conversation ID for tracking
        """
        self.recommended_capabilities = set(recommended_capabilities)
        self.actual_calls: dict[str, list[float]] = {}  # tool_name -> [durations]
        self.conversation_id = conversation_id

        logger.debug(
            "tool_tracker_initialized",
            recommended=list(self.recommended_capabilities),
            conversation_id=conversation_id,
        )

    async def track_call(self, tool_name: str, duration: float):
        """
        Record a tool call with timing.

        Args:
            tool_name: Name of the tool that was called
            duration: Duration of the call in seconds
        """
        # Record the call
        if tool_name not in self.actual_calls:
            self.actual_calls[tool_name] = []
        self.actual_calls[tool_name].append(duration)

        # Check if tool was recommended
        was_recommended = tool_name in self.recommended_capabilities

        if not was_recommended:
            logger.warning(
                "tool_call_not_recommended",
                tool_name=tool_name,
                duration=duration,
                recommended=list(self.recommended_capabilities),
            )

        # Record benchmark to Redis
        benchmark = PerformanceBenchmark(
            timestamp=datetime.now(timezone.utc),
            operation="tool_call",
            duration_seconds=duration,
            success=True,  # If we got here, the call succeeded
            tool_name=tool_name,
            was_recommended=was_recommended,
            was_actually_used=True,
            conversation_id=self.conversation_id,
            metadata={
                "recommended_capabilities": list(self.recommended_capabilities),
            },
        )

        await get_benchmark_store().record(benchmark)

        logger.debug(
            "tool_call_tracked",
            tool_name=tool_name,
            duration=duration,
            was_recommended=was_recommended,
        )

    async def finalize(self):
        """
        Finalize tracking and log unused recommended tools.

        Called after Tatlock completes its response to identify
        tools that were recommended but never used.
        """
        # Find tools that were recommended but not used
        unused_tools = self.recommended_capabilities - set(self.actual_calls.keys())

        if unused_tools:
            logger.info(
                "recommended_tools_unused",
                unused=list(unused_tools),
                used=list(self.actual_calls.keys()),
                conversation_id=self.conversation_id,
            )

            # Record benchmarks for unused recommendations
            for tool_name in unused_tools:
                benchmark = PerformanceBenchmark(
                    timestamp=datetime.now(timezone.utc),
                    operation="tool_call",
                    duration_seconds=0.0,  # Not used
                    success=True,
                    tool_name=tool_name,
                    was_recommended=True,
                    was_actually_used=False,
                    conversation_id=self.conversation_id,
                    metadata={
                        "recommended_capabilities": list(self.recommended_capabilities),
                        "reason": "recommended_but_unused",
                    },
                )
                await get_benchmark_store().record(benchmark)

        # Log summary
        total_calls = sum(len(durations) for durations in self.actual_calls.values())
        logger.info(
            "tool_tracking_finalized",
            total_calls=total_calls,
            unique_tools_used=len(self.actual_calls),
            recommended_count=len(self.recommended_capabilities),
            unused_count=len(unused_tools),
        )

    def get_summary(self) -> dict:
        """
        Get tracking summary for debugging.

        Returns:
            Dict with tracking statistics
        """
        total_calls = sum(len(durations) for durations in self.actual_calls.values())
        unused = self.recommended_capabilities - set(self.actual_calls.keys())

        return {
            "recommended_capabilities": list(self.recommended_capabilities),
            "tools_used": list(self.actual_calls.keys()),
            "tools_unused": list(unused),
            "total_calls": total_calls,
            "accuracy": {
                "recommended_and_used": len(
                    self.recommended_capabilities & set(self.actual_calls.keys())
                ),
                "recommended_but_unused": len(unused),
                "not_recommended_but_used": len(
                    set(self.actual_calls.keys()) - self.recommended_capabilities
                ),
            },
        }