feat: add evidence-backed log explanations

README.md

@@ -64,6 +64,15 @@ flowchart LR
 
 Supported formats include ISO-8601 application logs, syslog, nginx access logs, and generic level-prefixed lines. Detection is an explainable statistical heuristic; it is not a trained model and no accuracy claim is made without a labeled evaluation corpus.
 
+## Evidence-backed reasoning
+
+The optional `--explain` flag turns existing detector output into concise, user-facing evidence statements. Each statement identifies its direct or statistical basis: parsed error level, message-length z-score with its configured threshold, or observed error count/rate in a complete analysis window. LogSight does not infer an incident root cause, use an LLM, send logs externally, or report a model-confidence score.
+
+```bash
+logsight analyze application.log --window 200 --spike-threshold 0.20 --explain
+cat application.log | logsight stdin --explain
+```
+
 ## Quick start
 
 ```bash

logsight/analyzer.py

@@ -26,11 +26,33 @@ def error_rate(self) -> float:
         return self.error_count / self.total if self.total else 0.0
 
 
+@dataclass(frozen=True)
+class AnomalyEvidence:
+    """Detection reason for one anomalous log entry."""
+
+    entry: LogEntry
+    reasons: tuple[str, ...]
+    message_length_zscore: float | None = None
+
+
+@dataclass(frozen=True)
+class ErrorRateSpike:
+    """Observed error rate for one complete analysis window."""
+
+    start: int
+    end: int
+    error_count: int
+    total: int
+    error_rate: float
+    threshold: float
+
+
 @dataclass
 class AnomalyReport:
     """Result of an anomaly scan over a sequence of log entries."""
 
     anomalies: list[LogEntry] = field(default_factory=list)
+    evidence: list[AnomalyEvidence] = field(default_factory=list)
     stats: WindowStats = field(default_factory=WindowStats)
     zscore_threshold: float = 2.5
 
@@ -105,21 +127,31 @@ def detect_anomalies(
     std = pstdev(lengths)
 
     anomalous: list[LogEntry] = []
+    evidence: list[AnomalyEvidence] = []
     seen_ids: set[int] = set()
 
     for entry in entries:
-        flagged = False
+        reasons: list[str] = []
+        zscore: float | None = None
         if flag_errors and entry.is_error:
-            flagged = True
+            reasons.append("error_level")
         if std > 0:
-            z = abs(len(entry.message) - mean) / std
-            if z > zscore_threshold:
-                flagged = True
-        if flagged and id(entry) not in seen_ids:
+            zscore = abs(len(entry.message) - mean) / std
+            if zscore > zscore_threshold:
+                reasons.append("message_length_zscore")
+        if reasons and id(entry) not in seen_ids:
             anomalous.append(entry)
+            evidence.append(
+                AnomalyEvidence(
+                    entry=entry,
+                    reasons=tuple(reasons),
+                    message_length_zscore=zscore,
+                )
+            )
             seen_ids.add(id(entry))
 
     report.anomalies = anomalous
+    report.evidence = evidence
     return report
 
 
@@ -148,12 +180,42 @@ def error_rate_spike(
     if not 0 <= spike_threshold <= 1:
         raise ValueError("spike_threshold must be between 0 and 1")
 
-    spike_starts: list[int] = []
-    n = len(entries)
-    for start in range(0, n - window_size + 1, window_size):
+    return [
+        spike.start
+        for spike in error_rate_spike_details(
+            entries,
+            window_size=window_size,
+            spike_threshold=spike_threshold,
+        )
+    ]
+
+
+def error_rate_spike_details(
+    entries: Sequence[LogEntry],
+    window_size: int = 100,
+    spike_threshold: float = 0.25,
+) -> list[ErrorRateSpike]:
+    """Return the measured evidence for every complete error-rate spike window."""
+
+    if window_size <= 0:
+        raise ValueError("window_size must be greater than zero")
+    if not 0 <= spike_threshold <= 1:
+        raise ValueError("spike_threshold must be between 0 and 1")
+
+    spikes: list[ErrorRateSpike] = []
+    for start in range(0, len(entries) - window_size + 1, window_size):
         window = entries[start : start + window_size]
-        errors = sum(1 for e in window if e.is_error)
-        rate = errors / window_size
-        if rate >= spike_threshold:
-            spike_starts.append(start)
-    return spike_starts
+        errors = sum(1 for entry in window if entry.is_error)
+        error_rate = errors / window_size
+        if error_rate >= spike_threshold:
+            spikes.append(
+                ErrorRateSpike(
+                    start=start,
+                    end=start + window_size - 1,
+                    error_count=errors,
+                    total=window_size,
+                    error_rate=error_rate,
+                    threshold=spike_threshold,
+                )
+            )
+    return spikes

logsight/cli.py

@@ -6,10 +6,12 @@
 
 import click
 from rich.console import Console
+from rich.markup import escape
 from rich.table import Table
 
-from logsight.analyzer import AnomalyReport, detect_anomalies, error_rate_spike
+from logsight.analyzer import AnomalyReport, detect_anomalies, error_rate_spike_details
 from logsight.parser import parse_file, parse_lines
+from logsight.reasoning import EvidenceExplanation, explain_report
 
 console = Console()
 
@@ -29,22 +31,36 @@ def _print_report(report: AnomalyReport, show_anomalies: bool) -> None:
         table.add_column("Count", justify="right", style="cyan", no_wrap=True)
         table.add_column("Message")
         for msg, cnt in stats.top_messages:
-            table.add_row(str(cnt), msg)
+            table.add_row(str(cnt), escape(msg))
         console.print(table)
 
     if show_anomalies and report.has_anomalies:
         console.print(f"\n[bold red]Anomalies detected: {len(report.anomalies)}[/bold red]")
         for entry in report.anomalies[:20]:
             level_style = "red" if entry.is_error else "yellow"
             console.print(
-                f"  [[{level_style}]{entry.level.value}[/{level_style}]] {entry.message[:200]}"
+                f"  [[{level_style}]{entry.level.value}[/{level_style}]] {escape(entry.message[:200])}"
             )
         if len(report.anomalies) > 20:
             console.print(f"  … and {len(report.anomalies) - 20} more.")
     elif show_anomalies:
         console.print("\n[bold green]No anomalies detected.[/bold green]")
 
 
+def _print_explanations(explanations: list[EvidenceExplanation]) -> None:
+    """Render concise detector evidence without making causal claims."""
+
+    if not explanations:
+        return
+    console.print("\n[bold]Evidence-backed findings[/bold]")
+    for explanation in explanations:
+        evidence = ", ".join(explanation.evidence)
+        console.print(
+            f"  [{explanation.evidence_strength}] {escape(explanation.summary)} "
+            f"([dim]{escape(evidence)}[/dim])"
+        )
+
+
 @click.group()
 @click.version_option()
 def main() -> None:
@@ -83,12 +99,19 @@ def main() -> None:
     show_default=True,
     help="Error-rate fraction that constitutes a spike.",
 )
+@click.option(
+    "--explain",
+    is_flag=True,
+    default=False,
+    help="Show evidence-backed detector explanations.",
+)
 def analyze_cmd(
     logfile: str,
     threshold: float,
     no_anomalies: bool,
     window: int,
     spike_threshold: float,
+    explain: bool,
 ) -> None:
     """Analyze LOGFILE and report anomalies."""
     try:
@@ -104,12 +127,18 @@ def analyze_cmd(
     report = detect_anomalies(entries, zscore_threshold=threshold)
     _print_report(report, show_anomalies=not no_anomalies)
 
-    spikes = error_rate_spike(entries, window_size=window, spike_threshold=spike_threshold)
+    spikes = error_rate_spike_details(
+        entries,
+        window_size=window,
+        spike_threshold=spike_threshold,
+    )
     if spikes:
         console.print(
             f"\n[bold red]Error-rate spikes at windows starting at lines: "
-            f"{', '.join(str(s) for s in spikes)}[/bold red]"
+            f"{', '.join(str(spike.start) for spike in spikes)}[/bold red]"
         )
+    if explain:
+        _print_explanations(explain_report(report, spikes))
 
 
 @main.command("stdin")
@@ -121,7 +150,13 @@ def analyze_cmd(
     show_default=True,
     help="Z-score threshold for anomaly detection.",
 )
-def stdin_cmd(threshold: float) -> None:
+@click.option(
+    "--explain",
+    is_flag=True,
+    default=False,
+    help="Show evidence-backed detector explanations.",
+)
+def stdin_cmd(threshold: float, explain: bool) -> None:
     """Read log lines from stdin and report anomalies."""
     lines = sys.stdin.readlines()
     entries = parse_lines(lines)
@@ -130,6 +165,8 @@ def stdin_cmd(threshold: float) -> None:
         return
     report = detect_anomalies(entries, zscore_threshold=threshold)
     _print_report(report, show_anomalies=True)
+    if explain:
+        _print_explanations(explain_report(report))
 
 
 @main.command("health")

logsight/reasoning.py

@@ -0,0 +1,85 @@
+"""Evidence-backed explanations for deterministic LogSight findings.
+
+This module does not infer a root cause or use a language model.  Each explanation
+is derived only from the detector's recorded reason, threshold, and observed value.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+
+from logsight.analyzer import AnomalyReport, ErrorRateSpike
+
+EvidenceStrength = Literal["direct", "statistical"]
+
+
+@dataclass(frozen=True)
+class EvidenceExplanation:
+    """Concise user-facing explanation with machine-readable evidence strings."""
+
+    category: str
+    summary: str
+    evidence_strength: EvidenceStrength
+    evidence: tuple[str, ...]
+
+
+def explain_report(
+    report: AnomalyReport,
+    spikes: list[ErrorRateSpike] | None = None,
+) -> list[EvidenceExplanation]:
+    """Explain detector output without claiming a cause beyond observed evidence."""
+
+    explanations: list[EvidenceExplanation] = []
+    for finding in report.evidence:
+        entry = finding.entry
+        if "error_level" in finding.reasons:
+            explanations.append(
+                EvidenceExplanation(
+                    category="error-level",
+                    summary=(
+                        f"Entry was flagged because its parsed level is {entry.level.value}; "
+                        "no root cause is inferred."
+                    ),
+                    evidence_strength="direct",
+                    evidence=(
+                        f"level={entry.level.value}",
+                        f"message_length={len(entry.message)}",
+                    ),
+                )
+            )
+        if "message_length_zscore" in finding.reasons and finding.message_length_zscore is not None:
+            explanations.append(
+                EvidenceExplanation(
+                    category="message-length-outlier",
+                    summary=(
+                        "Entry was flagged because its message length exceeded the configured "
+                        "statistical threshold; no root cause is inferred."
+                    ),
+                    evidence_strength="statistical",
+                    evidence=(
+                        f"zscore={finding.message_length_zscore:.3f}",
+                        f"threshold={report.zscore_threshold:.3f}",
+                        f"message_length={len(entry.message)}",
+                    ),
+                )
+            )
+
+    for spike in spikes or []:
+        explanations.append(
+            EvidenceExplanation(
+                category="error-rate-spike",
+                summary=(
+                    f"Entries {spike.start}-{spike.end} crossed the configured error-rate "
+                    "threshold; no root cause is inferred."
+                ),
+                evidence_strength="statistical",
+                evidence=(
+                    f"errors={spike.error_count}",
+                    f"entries={spike.total}",
+                    f"error_rate={spike.error_rate:.3f}",
+                    f"threshold={spike.threshold:.3f}",
+                ),
+            )
+        )
+    return explanations

tests/test_analyzer.py

@@ -9,6 +9,7 @@
     compute_stats,
     detect_anomalies,
     error_rate_spike,
+    error_rate_spike_details,
 )
 from logsight.parser import LogEntry, LogLevel, parse_line
 
@@ -127,3 +128,25 @@ def test_fewer_than_window(self):
         entries = [_entry("ERROR", "fail")] * 50
         spikes = error_rate_spike(entries, window_size=100, spike_threshold=0.25)
         assert spikes == []
+
+
+class TestEvidenceDetails:
+    def test_records_direct_error_reason(self):
+        report = detect_anomalies([_entry("ERROR", "database connection failed")])
+        assert report.evidence[0].reasons == ("error_level",)
+        assert report.evidence[0].message_length_zscore is None
+
+    def test_records_zscore_reason(self):
+        normal = [_entry("INFO", "normal log line")] * 50
+        outlier = _entry("INFO", "x" * 500)
+        report = detect_anomalies(normal + [outlier], zscore_threshold=2.0, flag_errors=False)
+        assert report.evidence[0].reasons == ("message_length_zscore",)
+        assert report.evidence[0].message_length_zscore is not None
+
+    def test_spike_details_include_observed_rate(self):
+        entries = [_entry("ERROR", "fail")] * 2 + [_entry("INFO", "ok")] * 2
+        spikes = error_rate_spike_details(entries, window_size=4, spike_threshold=0.5)
+        assert spikes[0].start == 0
+        assert spikes[0].end == 3
+        assert spikes[0].error_count == 2
+        assert spikes[0].error_rate == 0.5

tests/test_cli.py

@@ -60,3 +60,12 @@ def test_empty_stdin(self):
         result = CliRunner().invoke(main, ["stdin"], input="\n")
         assert result.exit_code == 0
         assert "No log entries" in result.output
+
+    def test_explain_uses_detector_evidence(self, tmp_path):
+        log_file = tmp_path / "events.log"
+        log_file.write_text("ERROR database connection failed\n", encoding="utf-8")
+        result = CliRunner().invoke(main, ["analyze", str(log_file), "--explain"])
+        assert result.exit_code == 0
+        assert "Evidence-backed findings" in result.output
+        assert "level=ERROR" in result.output
+        assert "no root cause is inferred" in " ".join(result.output.split())