"""Frictionless diagnose service — NL time extraction + layered log search. This module is the public interface for the diagnose package. Full implementation lives here so that patch("app.services.diagnose._HAS_DATEPARSER") and patch("app.services.diagnose._search_dates") continue to target the correct namespace, preserving backward compatibility with existing tests. The verbatim original is preserved in legacy.py for reference. """ from __future__ import annotations import asyncio import dataclasses import logging import os import re from collections.abc import AsyncGenerator from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from app.context.retriever import retrieve_context, format_context_block from app.services.llm import summarize from app.services.search import SearchResult, entries_in_window, search from app.services.diagnose.pipeline import run_pipeline logger = logging.getLogger(__name__) try: from dateparser.search import search_dates as _search_dates # type: ignore[import] _HAS_DATEPARSER = True except ImportError: _search_dates = None # type: ignore[assignment] _HAS_DATEPARSER = False _RELATIVE_RE = re.compile( r"\b(?:last|past)\s+(?:(?P\d+)|(?Pa\s+few|few|couple(?:\s+of)?|several))?\s*(?Pminute|hour|day|week)s?\b", re.IGNORECASE, ) _RELATIVE_UNITS = {"minute": 1, "hour": 60, "day": 1440, "week": 10080} # Fuzzy quantifiers map to a reasonable span so "last few hours" → 3h window _APPROX_N = 3 def _relative_window(match: re.Match) -> tuple[str, str]: """Convert a relative time match to (since_iso, until_iso).""" n_str = match.group("n") approx = match.group("approx") unit = match.group("unit").lower() n = int(n_str) if n_str else (_APPROX_N if approx else 1) minutes = n * _RELATIVE_UNITS[unit] return _last_n_minutes(minutes), _now_iso() def parse_time_window(query: str) -> tuple[str | None, str | None, str]: """Extract a time window from a natural-language query string. Returns (since_iso, until_iso, keywords) where keywords is the query with the matched time phrase stripped. Falls back to last-60-min window. """ # Handle relative expressions first ("last hour", "past 30 minutes", etc.) # dateparser misinterprets these as absolute times. m = _RELATIVE_RE.search(query) if m: since, until = _relative_window(m) keywords = re.sub(r"\s{2,}", " ", query[: m.start()] + query[m.end() :]).strip() return since, until, keywords or query if _HAS_DATEPARSER and _search_dates is not None: # Tell dateparser what timezone the user is in so "3:35 am" means local time. # PREFER_DAY_OF_MONTH is unused here but PREFER_DATES_FROM=past ensures # "3:35 am" resolves to the most recent past occurrence, not a future one. local_offset = datetime.now().astimezone().utcoffset() offset_h = int((local_offset.total_seconds() if local_offset else 0) / 3600) tz_str = f"UTC{'+' if offset_h >= 0 else ''}{offset_h}" try: results = _search_dates( query, languages=["en"], settings={ "PREFER_DATES_FROM": "past", "TIMEZONE": tz_str, "RETURN_AS_TIMEZONE_AWARE": True, }, ) except Exception as e: logger.warning( "dateparser failed (%s) on query %r — falling back to 60-min window", type(e).__name__, query, ) results = None if results: phrase, dt = results[0] if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) else: dt = dt.astimezone( timezone.utc ) # normalise to UTC for SQLite string compare since = (dt - timedelta(minutes=30)).isoformat() until = (dt + timedelta(minutes=30)).isoformat() keywords = re.sub(r"\s{2,}", " ", query.replace(phrase, " ").strip()) return since, until, keywords or query return _last_n_minutes(60), _now_iso(), query def diagnose( db_path: Path, query: str, since: str | None = None, until: str | None = None, source_filter: str | None = None, llm_url: str | None = None, llm_model: str | None = None, llm_api_key: str | None = None, ) -> dict[str, Any]: """Run layered log search with NL time extraction. Returns summary + entries.""" time_detected = since is not None and until is not None if not time_detected: parsed_since, parsed_until, keywords = parse_time_window(query) since = since or parsed_since until = until or parsed_until time_detected = keywords != query else: keywords = query keyword_hits = search( db_path, query=keywords, since=since, until=until, source_filter=source_filter, limit=150, or_mode=True, ) window_hits = entries_in_window( db_path, since=since, until=until, source_filter=source_filter, limit=50, per_source_cap=15, ) seen: set[str] = set() merged: list[SearchResult] = [] for r in keyword_hits + window_hits: if r.entry_id not in seen: seen.add(r.entry_id) merged.append(r) combined = sorted(merged, key=lambda r: (r.timestamp_iso or "\xff", r.sequence))[ :200 ] by_severity: dict[str, int] = {"CRITICAL": 0, "ERROR": 0, "WARN": 0, "INFO": 0} by_source: dict[str, int] = {} for r in combined: sev = (r.severity or "INFO").upper() if sev in by_severity: by_severity[sev] += 1 by_source[r.source_id] = by_source.get(r.source_id, 0) + 1 reasoning: str | None = None if llm_url and llm_model: reasoning = summarize( query, combined, llm_url=llm_url, llm_model=llm_model, api_key=llm_api_key ) return { "summary": { "total": len(combined), "window_start": since, "window_end": until, "time_detected": time_detected, "by_severity": by_severity, "by_source": by_source, }, "reasoning": reasoning, "entries": combined, } async def diagnose_stream( db_path: Path, query: str, since: str | None = None, until: str | None = None, source_filter: str | None = None, llm_url: str | None = None, llm_model: str | None = None, llm_api_key: str | None = None, context_db_path: Path | None = None, tech_level: str = "sysadmin", pattern_domain: dict[str, str] | None = None, ) -> AsyncGenerator[dict[str, Any], None]: """Async generator yielding SSE event dicts for the diagnose pipeline. Yields events in order: {"type":"status","message":"…"} — pipeline progress {"type":"summary","data":{…}} — window + severity counts (fast, from DB) {"type":"entries","data":[…]} — log entries (fast, from DB) {"type":"reasoning","text":"…"} — LLM analysis (slow, optional) {"type":"done"} """ keywords = query.strip() source_browse = not keywords and source_filter is not None if source_browse: # No keyword — browsing a source directly. Use 24h window; skip FTS entirely. yield {"type": "status", "message": f"Loading {source_filter}…"} since = since or _last_n_minutes(60 * 24) until = until or _now_iso() time_detected = False else: yield {"type": "status", "message": "Parsing time window…"} time_detected = since is not None and until is not None if not time_detected: parsed_since, parsed_until, keywords = await asyncio.to_thread( parse_time_window, query ) since = since or parsed_since until = until or parsed_until time_detected = keywords != query yield {"type": "status", "message": "Loading environment context…"} _ctx_db = context_db_path or db_path ctx = await asyncio.to_thread(lambda: retrieve_context(_ctx_db, query)) yield { "type": "context", "facts": ctx.facts, "chunks": ctx.chunks, } yield {"type": "status", "message": "Searching logs…"} if source_browse: keyword_hits: list[SearchResult] = [] window_hits = await asyncio.to_thread( lambda: entries_in_window( db_path, since, until, source_filter=source_filter, limit=200, ) ) else: keyword_hits, window_hits = await asyncio.gather( asyncio.to_thread( lambda: search( db_path, keywords, source_filter=source_filter, since=since, until=until, limit=150, or_mode=True, ) ), asyncio.to_thread( lambda: entries_in_window( db_path, since, until, source_filter=source_filter, limit=50, per_source_cap=15, ) ), ) seen: set[str] = set() merged: list[SearchResult] = [] for r in keyword_hits + window_hits: if r.entry_id not in seen: seen.add(r.entry_id) merged.append(r) combined = sorted(merged, key=lambda r: (r.timestamp_iso or "\xff", r.sequence))[ :200 ] by_severity: dict[str, int] = {"CRITICAL": 0, "ERROR": 0, "WARN": 0, "INFO": 0} by_source: dict[str, int] = {} for r in combined: sev = (r.severity or "INFO").upper() if sev in by_severity: by_severity[sev] += 1 by_source[r.source_id] = by_source.get(r.source_id, 0) + 1 by_domain: dict[str, int] = {} if pattern_domain: for r in combined: seen: set[str] = set() for tag in (r.matched_patterns or []): d = pattern_domain.get(tag, "") if d and d not in seen: seen.add(d) by_domain[d] = by_domain.get(d, 0) + 1 yield { "type": "summary", "data": { "total": len(combined), "window_start": since, "window_end": until, "time_detected": time_detected, "by_severity": by_severity, "by_source": by_source, "by_domain": by_domain, }, } yield {"type": "entries", "data": [dataclasses.asdict(r) for r in combined]} if MULTI_AGENT_ENABLED: async for event in run_pipeline( db_path=db_path, entries=combined, ctx=ctx, query=query, since=since, until=until, llm_url=llm_url, llm_model=llm_model, llm_api_key=llm_api_key, tech_level=tech_level, ): yield event return # pipeline emits its own "done" event if llm_url and llm_model and combined: # Only compute context_block in the legacy path — pipeline uses ctx directly. context_block = format_context_block(ctx) yield {"type": "status", "message": "Analyzing with LLM…"} reasoning = await asyncio.to_thread( lambda: summarize( query, combined, llm_url, llm_model, llm_api_key, context_block=context_block, ) ) if reasoning: yield {"type": "reasoning", "text": reasoning} yield {"type": "done"} def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _last_n_minutes(n: int) -> str: return (datetime.now(timezone.utc) - timedelta(minutes=n)).isoformat() __all__ = [ "diagnose", "diagnose_stream", "parse_time_window", ] # Feature flag for Task 6 MULTI_AGENT_ENABLED = ( os.getenv("TURNSTONE_MULTI_AGENT_DIAGNOSE", "false").lower() == "true" )