Function bodies 217 total

async def post_tool_use_hook(
    hook_input: PostToolUseHookInput,
    block_reason: str | None,
    hook_context: HookContext,
) -> SyncHookJSONOutput:
    """Hook called after a tool is executed.

    Currently just continues execution. Could be extended to log results.

    Args:
        hook_input: Input data including tool_name, tool_input, and tool_response.
        block_reason: Reason if the tool was blocked (unused).
        hook_context: Hook context (unused).

    Returns:
        SyncHookJSONOutput to continue execution.
    """
    # For minimal output, we don't log post-tool results
    # This hook is here for potential future enhancements
    return SyncHookJSONOutput(continue_=True)

def _extract_key_param(tool_name: str, tool_input: dict[str, Any]) -> str:
    """Extract the most relevant parameter from tool input for logging.

    Args:
        tool_name: Name of the tool.
        tool_input: Tool input parameters.

    Returns:
        A string representing the key parameter for this tool.
    """
    if tool_name == "Bash":
        return tool_input.get("command", "")
    elif tool_name == "Skill":
        return tool_input.get("skill", "")
    elif tool_name == "WebFetch":
        return tool_input.get("url", "")
    elif tool_name == "Read" or tool_name == "Write" or tool_name == "Edit":
        return tool_input.get("file_path", "")
    else:
        # For unknown tools, try to get a reasonable representation
        if tool_input:
            # Get first non-empty value
            for value in tool_input.values():
                if value:
                    return str(value)[:100]
        return ""

def get_ergast_client() -> ergast.Ergast:
    """Get initialized Ergast API client.

    Returns:
        Ergast API client instance
    """
    return ergast.Ergast()

def extract_round_from_response(response: Any, fallback: int | None = None) -> int:
    """Extract actual round number from Ergast response description.

    Args:
        response: Ergast API response object
        fallback: Fallback round number if extraction fails

    Returns:
        Extracted round number or fallback value
    """
    description = response.description
    if len(description) > 0:
        return int(description.iloc[0]["round"])
    return fallback

def parse_driver_standings_response(
    response: Any,
    year: int,
    round_number: int | None,
) -> dict:
    """Parse Ergast driver standings response into standard format.

    Handles Timestamp conversion, NaN handling, and data extraction.

    Args:
        response: Ergast API response from get_driver_standings
        year: Championship year
        round_number: Requested round number (None for final standings)

    Returns:
        Dictionary with parsed driver standings and metadata
    """
    # Extract actual round from response
    actual_round = extract_round_from_response(response, round_number)

    # Parse standings data
    standings_data = []
    if len(response.content) > 0:
        standings_df = response.content[0]

        for _, row in standings_df.iterrows():
            # Handle date of birth - convert Timestamp to string
            date_of_birth = row.get("dateOfBirth")
            if pd.isna(date_of_birth):
                date_of_birth = None
       

def parse_constructor_standings_response(
    response: Any,
    year: int,
    round_number: int | None,
) -> dict:
    """Parse Ergast constructor standings response into standard format.

    Args:
        response: Ergast API response from get_constructor_standings
        year: Championship year
        round_number: Requested round number (None for final standings)

    Returns:
        Dictionary with parsed constructor standings and metadata
    """
    # Extract actual round from response
    actual_round = extract_round_from_response(response, round_number)

    # Parse standings data
    standings_data = []
    if len(response.content) > 0:
        standings_df = response.content[0]

        for _, row in standings_df.iterrows():
            standings_data.append(
                {
                    "position": int(row["position"]),
                    "points": float(row["points"]),
                    "wins": int(row["wins"]),
                    "constructor_id": row["c

def get_fastf1_cache_dir() -> Path:
    """Get the shared FastF1 cache directory path.

    Returns:
        Path to the FastF1 cache directory in the user's home directory.

    Examples:
        >>> cache_dir = get_fastf1_cache_dir()
        >>> print(cache_dir)
        /home/user/.pitlane/cache/fastf1
    """
    return Path.home() / ".pitlane" / "cache" / "fastf1"

def setup_fastf1_cache() -> None:
    """Initialize FastF1 cache with standard shared directory.

    Uses get_fastf1_cache_dir() from utils to ensure all commands
    share the same cache directory (~/.pitlane/cache/fastf1/).
    """
    cache_dir = get_fastf1_cache_dir()
    cache_dir.mkdir(parents=True, exist_ok=True)
    fastf1.Cache.enable_cache(str(cache_dir))

def load_session(
    year: int,
    gp: str,
    session_type: str,
    telemetry: bool = False,
    weather: bool = False,
    messages: bool = False,
) -> Session:
    """Load FastF1 session with standard configuration.

    Automatically sets up cache before loading session data.

    Args:
        year: Season year (e.g., 2024)
        gp: Grand Prix name (e.g., "Monaco")
        session_type: Session identifier (R, Q, FP1, FP2, FP3, S, SQ)
        telemetry: Whether to load telemetry data (default: False)
        weather: Whether to load weather data (default: False)
        messages: Whether to load messages data (default: False)

    Returns:
        Loaded FastF1 session object

    Raises:
        Exception: If session loading fails
    """
    # Ensure cache is set up
    setup_fastf1_cache()

    # Get session
    session = fastf1.get_session(year, gp, session_type)

    # Load session data with specified options
    session.load(telemetry=telemetry, weather=weather, messag

def load_testing_session(
    year: int,
    test_number: int,
    session_number: int,
    telemetry: bool = False,
    weather: bool = False,
    messages: bool = False,
) -> Session:
    """Load FastF1 testing session with standard configuration.

    Testing sessions require a separate API from regular GP sessions.
    FastF1's get_session() cannot load testing events — it will fuzzy-match
    to an incorrect GP or reject round 0.

    Args:
        year: Season year (e.g., 2026)
        test_number: Testing event number (e.g., 1 or 2)
        session_number: Session within testing event (e.g., 1, 2, or 3)
        telemetry: Whether to load telemetry data (default: False)
        weather: Whether to load weather data (default: False)
        messages: Whether to load messages data (default: False)

    Returns:
        Loaded FastF1 session object

    Raises:
        Exception: If session loading fails
    """
    setup_fastf1_cache()

    session = fastf1.get_testing_session(year

def load_session_or_testing(
    year: int,
    gp: str | None,
    session_type: str | None,
    test_number: int | None = None,
    session_number: int | None = None,
    telemetry: bool = False,
    weather: bool = False,
    messages: bool = False,
) -> Session:
    """Load either a regular GP session or a testing session.

    Dispatches to load_testing_session when test_number/session_number are
    provided, otherwise falls back to load_session.

    Args:
        year: Season year
        gp: Grand Prix name (for regular sessions)
        session_type: Session identifier (for regular sessions)
        test_number: Testing event number (for testing sessions)
        session_number: Session within testing event (for testing sessions)
        telemetry: Whether to load telemetry data
        weather: Whether to load weather data
        messages: Whether to load messages data

    Returns:
        Loaded FastF1 session object
    """
    if test_number is not None and session_numb

def validate_session_or_test(
    gp: str | None,
    session: str | None,
    test_number: int | None,
    session_number: int | None,
) -> tuple[bool, bool]:
    """Validate that either --gp/--session or --test/--day is provided, not both.

    Returns:
        Tuple of (has_gp, has_test) booleans.

    Raises:
        click.UsageError: If validation fails.
    """
    has_gp = gp is not None and session is not None
    has_test = test_number is not None and session_number is not None
    if not has_gp and not has_test:
        raise click.UsageError("Must provide either --gp and --session, or --test and --day")
    if has_gp and has_test:
        raise click.UsageError("Cannot use --gp/--session with --test/--day")
    return has_gp, has_test

def build_chart_path(
    workspace_dir: Path,
    chart_type: str,
    year: int,
    gp: str,
    session_type: str,
    drivers: list[str] | None = None,
    test_number: int | None = None,
    session_number: int | None = None,
    extension: str = "png",
) -> Path:
    """Construct standardized chart output path.

    Handles driver list formatting to prevent overly long filenames.
    Creates charts/ subdirectory within workspace if it doesn't exist.

    Args:
        workspace_dir: Workspace base directory
        chart_type: Type of chart (e.g., "lap_times", "tyre_strategy")
        year: Season year
        gp: Grand Prix name (ignored for testing sessions)
        session_type: Session identifier (ignored for testing sessions)
        drivers: Optional list of driver abbreviations (for driver-specific charts)
        test_number: Testing event number (for testing sessions)
        session_number: Session within testing event (for testing sessions)
        extension: File ext

def build_data_path(
    workspace_dir: Path,
    data_type: str,
    year: int | None = None,
    gp: str | None = None,
    session_type: str | None = None,
    round_number: int | None = None,
    driver_code: str | None = None,
    season: int | None = None,
    test_number: int | None = None,
    session_number: int | None = None,
) -> Path:
    """Construct standardized data output path.

    Creates unique filenames for fetch command outputs to prevent overwrites
    when fetching different sessions. Follows similar pattern to build_chart_path().

    Args:
        workspace_dir: Workspace base directory
        data_type: Type of data (e.g., "session_info", "driver_standings")
        year: Season year (for session-scoped or year-scoped data)
        gp: Grand Prix name (for session-scoped data)
        session_type: Session identifier (for session-scoped data)
        round_number: Round number (for round-scoped standings)
        driver_code: Driver code (for driver-scoped da

def format_lap_time(lap_time: pd.Timedelta | None) -> str | None:
    """Format a lap time Timedelta as M:SS.mmm, or None if not available.

    Uses explicit arithmetic instead of the str()[10:18] slice hack so the
    result is always 3 decimal places with no leading zero on minutes.
    """
    if lap_time is None or pd.isna(lap_time):
        return None
    total_seconds = lap_time.total_seconds()
    minutes = int(total_seconds // 60)
    seconds = total_seconds % 60
    return f"{minutes}:{seconds:06.3f}"

def format_sector_time(sector_time: pd.Timedelta | None) -> str | None:
    """Format a sector time Timedelta as SS.mmm or M:SS.mmm, or None if not available.

    Sectors under one minute are returned without a minutes prefix (e.g. "28.341").
    """
    if sector_time is None or pd.isna(sector_time):
        return None
    total_seconds = sector_time.total_seconds()
    minutes = int(total_seconds // 60)
    secs = total_seconds % 60
    if minutes > 0:
        return f"{minutes}:{secs:06.3f}"
    return f"{secs:.3f}"

def pick_lap_by_spec(driver_laps, spec: str | int) -> Lap:
    """Pick a lap by specification string or number.

    Args:
        driver_laps: FastF1 LapsDataFrame filtered to a single driver
        spec: Either "best" (picks fastest lap) or an integer lap number

    Returns:
        The matching FastF1 Lap object

    Raises:
        ValueError: If the specified lap number is not found, with available laps listed
    """
    if str(spec) == "best":
        return driver_laps.pick_fastest()
    n = int(spec)
    matching = driver_laps[driver_laps["LapNumber"] == n]
    if matching.empty:
        available = sorted(driver_laps["LapNumber"].dropna().astype(int).unique().tolist())
        raise ValueError(f"Lap {n} not found for this driver. Available lap numbers: {available}")
    return matching.iloc[0]

def get_merged_telemetry(lap: Lap, required_channels: list[str] | None = None) -> Telemetry:
    """Get merged telemetry with validation.

    Uses FastF1's get_telemetry() which merges position and car data with proper
    interpolation to handle different sampling rates between telemetry channels.

    This prevents gaps in visualizations caused by misaligned timestamps when
    manually merging get_pos_data() and get_car_data() with inner joins.

    Args:
        lap: FastF1 Lap object
        required_channels: Optional list of required column names to validate

    Returns:
        DataFrame-like telemetry object with merged data (X, Y, Speed, nGear, etc.)

    Raises:
        ValueError: If telemetry is empty or required channels are missing

    Example:
        >>> telemetry = get_merged_telemetry(fastest_lap, required_channels=["X", "Y", "nGear"])
        >>> print(telemetry[["X", "Y", "nGear"]].head())
    """
    telemetry = lap.get_telemetry()

    if telemetry.empty:
    

def sanitize_filename(text: str) -> str:
    """Sanitize text for use in filenames.

    Converts text to lowercase, strips Unicode diacritics, and replaces
    spaces and special characters with underscores to create
    ASCII-safe filenames.

    Args:
        text: Input text (e.g., GP name like "Abu Dhabi" or "São Paulo")

    Returns:
        Sanitized string safe for filenames (e.g., "abu_dhabi", "sao_paulo")

    Examples:
        >>> sanitize_filename("Monaco")
        'monaco'
        >>> sanitize_filename("Abu Dhabi")
        'abu_dhabi'
        >>> sanitize_filename("Emilia-Romagna")
        'emilia_romagna'
        >>> sanitize_filename("São Paulo")
        'sao_paulo'
    """
    # Normalize to NFD (decomposed form) and strip diacritics
    text = unicodedata.normalize("NFD", text)
    text = text.encode("ascii", "ignore").decode("ascii")
    # Convert to lowercase
    text = text.lower()
    # Replace any non-word characters (not letters, digits, underscore) with undersco

def setup_plot_style():
    """Configure matplotlib for F1-style dark theme.

    Uses MATPLOTLIB_DARK_THEME configuration from utils.constants.
    This function should be called before creating any matplotlib figures
    to ensure consistent styling across all visualizations.
    """
    plt.style.use("dark_background")
    plt.rcParams.update(MATPLOTLIB_DARK_THEME)

def save_figure(
    fig: plt.Figure,
    output_path: Path,
    dpi: int = DEFAULT_DPI,
    bbox_inches: str = "tight",
) -> None:
    """Save matplotlib figure with consistent settings.

    Handles directory creation, proper DPI settings, and memory cleanup.

    Args:
        fig: Matplotlib figure to save
        output_path: Path where figure should be saved
        dpi: Dots per inch for output (default: DEFAULT_DPI from constants)
        bbox_inches: Bounding box setting (default: "tight")
    """
    # Ensure output directory exists
    output_path.parent.mkdir(parents=True, exist_ok=True)

    # Apply tight layout and save
    fig.tight_layout()
    fig.savefig(
        str(output_path),
        dpi=dpi,
        facecolor=fig.get_facecolor(),
        edgecolor="none",
        bbox_inches=bbox_inches,
    )

    # Clean up to free memory
    plt.close(fig)

def get_driver_color_safe(
    driver_abbr: str,
    session: fastf1.core.Session,
    fallback: str | None = None,
) -> str | None:
    """Get driver color with exception handling.

    Attempts to retrieve the official FastF1 driver color, falling back
    to a specified color or None if not available.

    Args:
        driver_abbr: Driver abbreviation (e.g., "VER", "HAM")
        session: FastF1 session object
        fallback: Optional fallback color if driver color not found

    Returns:
        Driver color hex code or fallback value
    """
    try:
        return fastf1.plotting.get_driver_color(driver_abbr, session)
    except Exception:
        return fallback

def get_driver_team(driver_abbr: str, session: fastf1.core.Session) -> str | None:
    """Get team name for a driver from session results.

    Args:
        driver_abbr: Driver abbreviation (e.g., "VER")
        session: FastF1 session object

    Returns:
        Team name string or None if not found
    """
    try:
        driver_info = session.results[session.results["Abbreviation"] == driver_abbr]
        if not driver_info.empty:
            return str(driver_info.iloc[0]["TeamName"])
    except Exception:
        pass
    return None

def ensure_color_contrast(hex_color: str, min_lightness: float = MIN_COLOR_LIGHTNESS) -> str:
    """Ensure a hex color has sufficient lightness for dark backgrounds.

    If the color's HLS lightness is below min_lightness, it is raised to that threshold
    while preserving hue and saturation.

    Args:
        hex_color: Hex color string (e.g., "#1a1a2e")
        min_lightness: Minimum lightness value (0.0-1.0)

    Returns:
        Adjusted hex color string
    """
    hex_color = hex_color.lstrip("#")
    r, g, b = (int(hex_color[i : i + 2], 16) / 255.0 for i in (0, 2, 4))
    h, lightness, s = colorsys.rgb_to_hls(r, g, b)
    if lightness < min_lightness:
        lightness = min_lightness
        r, g, b = colorsys.hls_to_rgb(h, lightness, s)
    return f"#{int(r * 255):02x}{int(g * 255):02x}{int(b * 255):02x}"

def get_circuit_length_km(session: Session) -> float | None:
    """Compute circuit lap distance in kilometres from telemetry.

    Uses the fastest lap's car data with distance channel to determine
    the total circuit length. Returns None if telemetry is unavailable.

    Args:
        session: FastF1 session object with laps loaded

    Returns:
        Circuit length in km rounded to 3 decimal places, or None
    """
    try:
        fastest = session.laps.pick_fastest()
        if fastest is None:
            return None
        telemetry = fastest.get_car_data().add_distance()
        if telemetry.empty or "Distance" not in telemetry.columns:
            return None
        return round(telemetry["Distance"].max() / 1000, 3)
    except Exception:
        return None

def compute_driver_position_stats(driver_abbr: str, session: Session) -> DriverPositionStats | None:
    """Compute position statistics for a single driver.

    Args:
        driver_abbr: Driver abbreviation (e.g., 'VER', 'HAM')
        session: FastF1 session object with laps loaded

    Returns:
        Dictionary with driver position statistics, or None if no data available
    """
    driver_laps = session.laps.pick_drivers(driver_abbr)

    if driver_laps.empty:
        return None

    position_data = driver_laps[["LapNumber", "Position"]].copy()
    position_data = position_data.dropna(subset=["Position"])

    if position_data.empty:
        return None

    positions = position_data["Position"].values
    start_position = float(positions[0])
    finish_position = float(positions[-1])
    position_changes = np.diff(positions)

    overtakes = int(np.sum(position_changes < 0))
    times_overtaken = int(np.sum(position_changes > 0))
    volatility = float(np.std(positions))
    

def compute_race_summary_stats(session: Session) -> RaceSummaryStats | None:
    """Compute aggregate race statistics from a loaded session.

    Args:
        session: FastF1 session object with laps loaded

    Returns:
        Dictionary with aggregate stats, or None if laps data is unavailable
    """
    try:
        laps = session.laps
        if laps.empty:
            return None
    except DataNotLoadedError:
        return None

    driver_abbrs = [session.get_driver(d)["Abbreviation"] for d in session.drivers]

    stats = []
    for abbr in driver_abbrs:
        driver_stats = compute_driver_position_stats(abbr, session)
        if driver_stats is not None:
            stats.append(driver_stats)

    if not stats:
        return None

    total_overtakes = sum(s["overtakes"] for s in stats)
    total_position_changes = sum(abs(s["net_change"]) for s in stats)
    avg_volatility = sum(s["volatility"] for s in stats) / len(stats)
    mean_pit_stops = sum(s["pit_stops"] for s 

def _validate_telemetry_columns(
    telemetry: pd.DataFrame,
    required: list[str],
) -> None:
    """Raise ``ValueError`` if *telemetry* is empty or missing columns."""
    if telemetry.empty:
        raise ValueError("Telemetry DataFrame is empty")
    missing = [c for c in required if c not in telemetry.columns]
    if missing:
        raise ValueError(f"Missing required telemetry channels: {missing}")

def detect_lift_and_coast_zones(
    telemetry: pd.DataFrame,
    *,
    min_duration: float = LIFT_COAST_MIN_DURATION,
    throttle_threshold: float = LIFT_COAST_THROTTLE_THRESHOLD,
    brake_threshold: int = LIFT_COAST_BRAKE_THRESHOLD,
) -> list[LiftAndCoastZone]:
    """Detect lift-and-coast zones in *telemetry*.

    A zone is a contiguous region where throttle is near-zero, the brake
    is off, and speed is declining (drag deceleration).

    Args:
        telemetry: DataFrame with columns Distance, Speed, Throttle,
            Brake, RPM, nGear, Time.
        min_duration: Minimum zone duration in seconds.
        throttle_threshold: Maximum throttle % considered "off".
        brake_threshold: Maximum brake value considered "off" (0 or 1).

    Returns:
        List of :class:`LiftAndCoastZone` dicts ordered by distance.
    """
    _validate_telemetry_columns(telemetry, _LIFT_COAST_COLUMNS)

    coasting = (telemetry["Throttle"] <= throttle_threshold) & (telemetry["Brake"] <= 

def detect_super_clipping_zones(
    telemetry: pd.DataFrame,
    *,
    min_duration: float = SUPER_CLIP_MIN_DURATION,
    throttle_threshold: float = SUPER_CLIP_THROTTLE_THRESHOLD,
    speed_tolerance: float = SUPER_CLIP_SPEED_TOLERANCE,
    speed_slope_threshold: float = SUPER_CLIP_SPEED_SLOPE_THRESHOLD,
    rpm_stutter_threshold: float = SUPER_CLIP_RPM_STUTTER_THRESHOLD,
    min_gear: int = SUPER_CLIP_MIN_GEAR,
    accel_lookback: int = SUPER_CLIP_ACCEL_LOOKBACK,
    min_speed_gain: float = SUPER_CLIP_MIN_SPEED_GAIN,
) -> list[SuperClippingZone]:
    """Detect super-clipping zones in *telemetry*.

    A zone is a contiguous region where throttle is pinned at 100 %,
    speed plateaus (low rolling standard deviation and no upward trend),
    and RPM stops climbing — all in a high gear.  The zone must be
    preceded by a rising-speed phase (acceleration) to distinguish
    genuine MGU-K depletion from steady-state cruising.

    Args:
        telemetry: DataFrame with columns Distan

def analyze_telemetry(
    telemetry: pd.DataFrame,
    *,
    lc_min_duration: float = LIFT_COAST_MIN_DURATION,
    lc_throttle_threshold: float = LIFT_COAST_THROTTLE_THRESHOLD,
    lc_brake_threshold: int = LIFT_COAST_BRAKE_THRESHOLD,
    sc_min_duration: float = SUPER_CLIP_MIN_DURATION,
    sc_throttle_threshold: float = SUPER_CLIP_THROTTLE_THRESHOLD,
    sc_speed_tolerance: float = SUPER_CLIP_SPEED_TOLERANCE,
    sc_speed_slope_threshold: float = SUPER_CLIP_SPEED_SLOPE_THRESHOLD,
    sc_rpm_stutter_threshold: float = SUPER_CLIP_RPM_STUTTER_THRESHOLD,
    sc_min_gear: int = SUPER_CLIP_MIN_GEAR,
    sc_accel_lookback: int = SUPER_CLIP_ACCEL_LOOKBACK,
    sc_min_speed_gain: float = SUPER_CLIP_MIN_SPEED_GAIN,
) -> TelemetryAnalysisResult:
    """Run all telemetry detectors and return an aggregated result.

    Parameters prefixed with ``lc_`` are forwarded to
    :func:`detect_lift_and_coast_zones`; those prefixed with ``sc_`` are
    forwarded to :func:`detect_super_clipping_zones`.

    def __init__(self, max_size: int = AGENT_CACHE_MAX_SIZE):
        """Initialize the agent cache.

        Args:
            max_size: Maximum number of agents to cache
        """
        self._cache: OrderedDict[str, F1Agent] = OrderedDict()
        self._max_size = max_size
        self._lock = asyncio.Lock()

    async def get_or_create(self, workspace_id: str) -> F1Agent:
        """Get cached agent or create new one for workspace.

        Implements LRU-style eviction when cache is full.
        When accessing an existing agent, it's moved to the end (most recently used).

        This method is async-safe and uses a lock to prevent race conditions
        in concurrent access scenarios.

        Args:
            workspace_id: Workspace identifier for the agent

        Returns:
            F1Agent instance for the workspace
        """
        async with self._lock:
            if workspace_id in self._cache:
                logger.debug(f"Using cached agent for workspace: {workspace_id}")
                # Move to end to mark as recently used (LRU) - atomic operation
                self._cache.move_to_end(workspace_id)
                return self._cache[workspace_id]

            # Evict oldest entry if cache is full
            if len(self._cache) >= self._max_size:
                

    async def evict(self, workspace_id: str) -> None:
        """Manually evict agent from cache.

        This method is async-safe and uses a lock to prevent race conditions.

        Args:
            workspace_id: Workspace identifier to evict
        """
        async with self._lock:
            if workspace_id in self._cache:
                del self._cache[workspace_id]
                logger.info(f"Manually evicted agent for workspace: {workspace_id}")

    def size(self) -> int:
        """Return current cache size.

        Returns:
            Number of agents currently in cache
        """
        return len(self._cache)

async def index(
    request: Request,
    session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """Render the home page with session management."""
    # Note: session_id in this context refers to the workspace identifier (stored in browser cookie),
    # which is distinct from the agent_session_id (Claude SDK session for conversation resumption)

    # Validate existing session (with timing attack protection)
    is_valid, validated_session = validate_session_safely(session)

    if is_valid and validated_session:
        session_id = validated_session
        needs_new_session = False
        logger.info(f"Index page loaded with existing session: {session_id}")
        # Update last accessed time with proper error handling
        update_workspace_metadata_safe(session_id)
    else:
        # Create new session — workspace ID doubles as the web session ID
        session_id = generate_workspace_id()
        needs_new_session = True
        logger.info(f"Index page lo

async def chat(
    request: Request,
    question: str = Form(...),
    session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """Process a user question and return an HTML response.

    Uses F1Agent with Claude Agent SDK to analyze F1 data.
    """
    # Validate existing session (with timing attack protection)
    is_valid, validated_session = validate_session_safely(session)

    if is_valid and validated_session:
        session_id = validated_session
        needs_new_session = False
        logger.info(f"Using existing session: {session_id}")
        # Update last accessed time with proper error handling
        update_workspace_metadata_safe(session_id)
    else:
        # Create new session — workspace ID doubles as the web session ID
        session_id = generate_workspace_id()
        needs_new_session = True
        logger.info(f"Creating new session: {session_id}")

    try:
        # Check for active conversation to resume
        active_conv = get_active_

async def serve_chart(
    request: Request,
    session_id: str,
    filename: str,
    current_session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """Serve chart files from session workspace with security validation.

    Security checks:
    1. Validate session ID format (UUID)
    2. Verify session ownership (matches cookie)
    3. Ensure workspace exists
    4. Validate filename (no path traversal)
    5. Check file exists within workspace
    6. Validate file extension
    """
    logger.info(f"Chart request: {filename} for session {session_id}")
    logger.debug(f"Chart serving - URL session: {session_id}, Cookie session: {current_session}")

    # 1. Validate session ID format
    if not is_valid_session_id(session_id):
        logger.warning(f"Invalid session ID format: {session_id}")
        raise HTTPException(status_code=400, detail="Invalid session ID")

    # 2. Verify session ownership (case-insensitive comparison for UUID)
    if session_id.lower() != 

async def list_conversations(
    request: Request,
    session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """List all conversations for the current session."""
    is_valid, validated_session = validate_session_safely(session)
    if not is_valid or not validated_session:
        raise HTTPException(status_code=401, detail="Invalid session")

    conversations_data = load_conversations(validated_session)

    return templates.TemplateResponse(
        request,
        "partials/conversation_list.html",
        {
            "conversations": conversations_data["conversations"],
            "active_id": conversations_data.get("active_conversation_id"),
        },
    )

async def new_conversation(
    request: Request,
    session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """Start a new conversation (clears active, doesn't delete history)."""
    is_valid, validated_session = validate_session_safely(session)
    if not is_valid or not validated_session:
        raise HTTPException(status_code=401, detail="Invalid session")

    # Clear active conversation
    set_active_conversation(validated_session, None)

    # Evict cached agent to force fresh SDK context
    await _agent_cache.evict(validated_session)

    logger.info(f"Started new conversation for session: {validated_session}")

    return templates.TemplateResponse(
        request,
        "partials/conversation_status.html",
        {"status": "new", "message": "Ready for new conversation"},
    )

async def resume_conversation(
    request: Request,
    conversation_id: str,
    session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """Resume a specific conversation."""
    is_valid, validated_session = validate_session_safely(session)
    if not is_valid or not validated_session:
        raise HTTPException(status_code=401, detail="Invalid session")

    # Validate conversation belongs to this session
    conversations_data = load_conversations(validated_session)
    conversation = None
    for conv in conversations_data["conversations"]:
        if conv["id"] == conversation_id:
            conversation = conv
            break

    if not conversation:
        raise HTTPException(status_code=404, detail="Conversation not found")

    # Set as active conversation
    set_active_conversation(validated_session, conversation_id)

    # Evict cached agent to force new context with resume
    await _agent_cache.evict(validated_session)

    logger.info(f"Resumed conv

async def get_conversation_messages(
    request: Request,
    conversation_id: str,
    session: str | None = Cookie(None, alias=SESSION_COOKIE_NAME),
):
    """Return rendered HTML of all messages for a conversation (newest first)."""
    is_valid, validated_session = validate_session_safely(session)
    if not is_valid or not validated_session:
        raise HTTPException(status_code=401, detail="Invalid session")

    # Verify conversation belongs to this session
    conversations_data = load_conversations(validated_session)
    if not any(c["id"] == conversation_id for c in conversations_data["conversations"]):
        raise HTTPException(status_code=404, detail="Conversation not found")

    messages = load_messages(validated_session, conversation_id)

    return templates.TemplateResponse(
        request,
        "partials/conversation_history.html",
        {
            "messages": list(reversed(messages)),  # newest first, matching live chat display order
            "sessio

def get_default_reload() -> bool:
    """Determine if reload should be enabled based on environment.

    Returns:
        True if PITLANE_ENV is 'development', False otherwise
    """
    env = os.getenv("PITLANE_ENV", "production")
    return env == "development"

def main(
    host: str,
    port: int,
    reload: bool | None,
    log_level: str,
    env: str | None,
) -> None:
    """Run the PitLane Web application.

    This starts a FastAPI web server for F1 data analysis powered by AI.

    Environment Variables:
        PITLANE_ENV                 - Environment mode (development/production/test)
        PITLANE_TRACING_ENABLED     - Enable OpenTelemetry tracing (0/1)
        PITLANE_HTTPS_ENABLED       - Enable secure cookies (true/false)
        PITLANE_SESSION_MAX_AGE     - Session cookie max age in seconds
        PITLANE_RATE_LIMIT_ENABLED  - Enable rate limiting (true/false)

    Examples:
        # Start with defaults (development mode auto-reload)
        uvx pitlane-web

        # Custom port
        uvx pitlane-web --port 3000

        # Production mode on all interfaces
        uvx pitlane-web --host 0.0.0.0 --env production --no-reload

        # Development with tracing
        PITLANE_TRACING_ENABLED=1 uvx pitlane-web --env de

def rewrite_workspace_paths(text: str, session_id: str) -> str:
    """Rewrite absolute workspace paths to web-relative URLs.

    Transforms:
      /Users/.../.pitlane/workspaces/{session-id}/charts/lap_times.png
      → /charts/{session-id}/lap_times.png

    Also handles bare /charts/filename references (missing session ID) that the
    LLM may produce when it doesn't use the full workspace path:
      /charts/lap_times.png → /charts/{session-id}/lap_times.png

    Security: Only rewrites paths for the current session to prevent data leaks.

    Args:
        text: Response text containing absolute paths
        session_id: Current session ID

    Returns:
        Text with rewritten paths
    """
    workspace_base = str(Path.home() / ".pitlane" / "workspaces")
    escaped_base = re.escape(workspace_base)

    # Pattern: /path/to/workspaces/{uuid}/{charts|data}/{filename}
    # Support both upper and lowercase UUIDs, stop at whitespace, quotes, or parens
    pattern = rf"{escaped_b

def html_charts_to_iframes(text: str) -> str:
    """Convert .html chart references from markdown image syntax to iframe tags.

    Transforms:
      ![label](/charts/{session-id}/chart.html)
      → <iframe src="/charts/{session-id}/chart.html" ...></iframe>

    Also converts img tags with .html src (in case markdown ran first):
      <img ... src="/charts/{session-id}/chart.html" ... />
      → <iframe src="/charts/{session-id}/chart.html" ...></iframe>

    Non-.html references (e.g. .png) are left unchanged.

    Args:
        text: Text potentially containing HTML chart references

    Returns:
        Text with .html chart references converted to iframes
    """
    # Markdown image syntax: ![alt](/charts/session/file.html)
    md_pattern = r"!\[([^\]]*)\]\((/charts/[^\s\)]+\.html)\)"
    text = re.sub(
        md_pattern,
        r'<div class="chart-container">'
        r'<a class="chart-open-link" href="\2" target="_blank" rel="noopener noreferrer">&#8599; Open in new tab</a>'

def md_to_html(text: str) -> str:
    """Convert markdown to HTML.

    Args:
        text: Markdown text to convert

    Returns:
        HTML string
    """
    return markdown.markdown(text, extensions=["fenced_code", "tables"])

def timeago(iso_timestamp: str) -> str:
    """Convert ISO timestamp to human-readable relative time.

    Args:
        iso_timestamp: ISO 8601 formatted timestamp string

    Returns:
        Human-readable relative time (e.g., "2 hours ago", "yesterday")
    """
    try:
        # Parse ISO timestamp (handle trailing Z)
        timestamp_str = iso_timestamp.rstrip("Z")
        dt = datetime.fromisoformat(timestamp_str).replace(tzinfo=UTC)
        now = datetime.now(UTC)
        delta = now - dt

        seconds = delta.total_seconds()

        if seconds < 60:
            return "just now"
        elif seconds < 3600:
            minutes = int(seconds / 60)
            return f"{minutes} minute{'s' if minutes != 1 else ''} ago"
        elif seconds < 86400:
            hours = int(seconds / 3600)
            return f"{hours} hour{'s' if hours != 1 else ''} ago"
        elif seconds < 172800:
            return "yesterday"
        elif seconds < 604800:
            days = int(seconds

def register_filters(templates: Jinja2Templates) -> None:
    """Register custom Jinja2 filters with the templates instance.

    Args:
        templates: Jinja2Templates instance to register filters with
    """
    templates.env.filters["markdown"] = md_to_html
    templates.env.filters["rewrite_paths"] = rewrite_workspace_paths
    templates.env.filters["html_charts_to_iframes"] = html_charts_to_iframes
    templates.env.filters["timeago"] = timeago

def is_valid_session_id(session_id: str) -> bool:
    """Validate that session_id is a valid UUID.

    Args:
        session_id: The session ID to validate

    Returns:
        True if session_id is a valid UUID, False otherwise
    """
    if not isinstance(session_id, str):
        return False
    try:
        uuid.UUID(session_id)
        return True
    except (ValueError, AttributeError, TypeError):
        return False