Continued the Detection Quality roadmap.

README.md

@@ -80,6 +80,21 @@ Graylog event is not learned repeatedly. Related anomalies, profile deviations,
 and multi-stream correlations are grouped into investigation incidents with a
 compact evidence timeline.
 
+With a stream profile in place, SignalScope also builds independent burst
+baselines for authentication failures, DNS queries, and deny/block actions when
+those events are present. These are evaluated per configured entity, so a Windows
+account, DNS client, or firewall source is compared to its own history.
+
+Replay a historic JSONL or Graylog export without changing the live baseline:
+
+```bash
+signalscope replay --logs exports/windows-history.jsonl --stream-id <configured-stream-id>
+```
+
+Replay uses a temporary SQLite baseline and evaluates events in timestamp order.
+It reports detector counts and the findings that would have been generated. Use
+the configured stream ID so the export is evaluated with that stream's profile.
+
 The current MCP endpoint is `http://<graylog-host>:9000/api/mcp`. Enable it in
 Graylog under `System -> Configurations -> MCP` and use stream IDs internally;
 the fgAI stream picker resolves titles in the UI.

ROADMAP.md

@@ -25,11 +25,12 @@ Goal: make findings more accurate before adding more integrations.
 - [x] Add baseline confidence based on sample count and time-bucket coverage.
 - [x] Add generic event-rate burst detection per stream/entity.
 - [x] Add rare-value detection with a minimum historical observation threshold.
-- [ ] Add detector-specific authentication failure, DNS volume, and denied-traffic burst thresholds.
+- [x] Add detector-specific authentication failure, DNS volume, and denied-traffic burst thresholds.
 - [ ] Add configurable per-field detector weights.
 - [ ] Add sequence detection, for example DNS lookup -> outbound connection -> authentication event.
-- [ ] Add per-stream detector enablement and thresholds in the UI.
-- [ ] Add a dry-run replay command to evaluate detector changes against a selected historic Graylog time range.
+- [x] Add per-stream detector enablement and thresholds in the UI.
+- [x] Add a dry-run replay command for historic JSONL or Graylog exports using temporary baselines.
+- [ ] Add direct Graylog MCP time-range replay and result comparison against saved detector configurations.
 
 Acceptance: each finding shows its detector, confidence, baseline sample count, current value, expected value, and a bounded set of raw-event references.
 

src/fgai/baseline.py

@@ -11,6 +11,7 @@ from statistics import mean, pstdev
 from .logs import THREAT_ACTIONS, is_utm_event
 from .models import LogEvent
 from .entities import profile_entity
+from .detectors import DETECTOR_MINIMUMS, event_detector_categories
 
 
 def _number(value: str | None) -> int:
@@ -78,6 +79,15 @@ class BaselineStore:
                   events integer not null, numeric_sum real not null, numeric_sum_squares real not null,
                   primary key (stream_id, entity, field, weekday, hour, bucket_start)
                 );
+                create table if not exists profile_detector_buckets (
+                  stream_id text not null, entity text not null, detector text not null, bucket_start integer not null,
+                  events integer not null, primary key (stream_id, entity, detector, bucket_start)
+                );
+                create table if not exists profile_detector_temporal_buckets (
+                  stream_id text not null, entity text not null, detector text not null,
+                  weekday integer not null, hour integer not null, bucket_start integer not null,
+                  events integer not null, primary key (stream_id, entity, detector, weekday, hour, bucket_start)
+                );
                 """
             )
 
@@ -124,6 +134,8 @@ class BaselineStore:
         observed_at = observed_at or int(time.time())
         pending: dict[tuple[str, str, str, int], list[float]] = defaultdict(lambda: [0, 0.0, 0.0])
         temporal_pending: dict[tuple[str, str, str, int, int, int], list[float]] = defaultdict(lambda: [0, 0.0, 0.0])
+        detector_pending: Counter[tuple[str, str, str, int]] = Counter()
+        detector_temporal_pending: Counter[tuple[str, str, str, int, int, int]] = Counter()
         pending_values: Counter[tuple[str, str, str, str]] = Counter()
         with self._connect() as connection:
             for event in events:
@@ -140,6 +152,9 @@ class BaselineStore:
                 timestamp = _event_epoch(event, observed_at)
                 bucket = timestamp - (timestamp % self.bucket_seconds)
                 moment = datetime.fromtimestamp(timestamp, tz=timezone.utc)
+                for detector in event_detector_categories(event):
+                    detector_pending[(stream_id, entity, detector, bucket)] += 1
+                    detector_temporal_pending[(stream_id, entity, detector, moment.weekday(), moment.hour, bucket)] += 1
                 fields = [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())]
                 numeric = {str(field).lower() for field in getattr(profile, "numeric_fields", ())}
                 for field in fields:
@@ -165,11 +180,18 @@ class BaselineStore:
             for key, count in pending_values.items():
                 connection.execute("""insert into profile_values values (?, ?, ?, ?, ?)
                   on conflict(stream_id, entity, field, value) do update set seen_count=seen_count+excluded.seen_count""", (*key, count))
+            for (stream_id, entity, detector, bucket), count in detector_pending.items():
+                connection.execute("""insert into profile_detector_buckets values (?, ?, ?, ?, ?)
+                  on conflict(stream_id, entity, detector, bucket_start) do update set events=events+excluded.events""", (stream_id, entity, detector, bucket, count))
+            for (stream_id, entity, detector, weekday, hour, bucket), count in detector_temporal_pending.items():
+                connection.execute("""insert into profile_detector_temporal_buckets values (?, ?, ?, ?, ?, ?, ?)
+                  on conflict(stream_id, entity, detector, weekday, hour, bucket_start) do update set events=events+excluded.events""", (stream_id, entity, detector, weekday, hour, bucket, count))
         return len(pending)
 
     def profile_deviations(self, events: list[LogEvent], profiles: dict[str, object]) -> dict[str, list[dict[str, object]]]:
         current: dict[tuple[str, str, str], list[float]] = defaultdict(lambda: [0, 0.0])
         entity_events: dict[tuple[str, str], list[LogEvent]] = defaultdict(list)
+        detector_current: Counter[tuple[str, str, str]] = Counter()
         for event in events:
             profile = profiles.get(event.fields.get("fgai_stream_id", ""))
             if not profile:
@@ -178,6 +200,8 @@ class BaselineStore:
             if not entity:
                 continue
             entity_events[(event.fields.get("fgai_stream_id", ""), entity)].append(event)
+            for detector in event_detector_categories(event):
+                detector_current[(event.fields.get("fgai_stream_id", ""), entity, detector)] += 1
             numeric = {str(field).lower() for field in getattr(profile, "numeric_fields", ())}
             for field in [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())]:
                 key = (event.fields.get("fgai_stream_id", ""), entity, str(field).lower())
@@ -239,6 +263,36 @@ class BaselineStore:
                 score = min(30, (15 if confidence == "high" else 12 if confidence == "medium" else 8) + int(z_score))
                 samples = [{"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")), "source": event.src_ip or event.fields.get("source", ""), "destination": event.dst_ip or "", "action": event.action, "severity": event.severity, "service": event.fields.get("service", ""), "value": "", "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]]
                 output[entity].append({"detector": "event_rate_burst", "field": "event_rate", "stream_id": stream, "score": score, "confidence": confidence, "baseline_samples": len(rows), "baseline_scope": baseline_scope, "reason": f"event rate burst above its {baseline_scope} baseline (z={z_score:.1f})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [], "sample_events": samples})
+
+            for (stream, entity, detector), current_value in detector_current.items():
+                profile = profiles.get(stream)
+                settings = getattr(profile, "detectors", {}).get(detector, {}) if profile else {}
+                if not settings.get("enabled", True):
+                    continue
+                minimum = int(settings.get("minimum", DETECTOR_MINIMUMS[detector]))
+                z_threshold = float(settings.get("z_threshold", 3.0))
+                if current_value < minimum:
+                    continue
+                matching = [event for event in entity_events[(stream, entity)] if detector in event_detector_categories(event)]
+                current_timestamp = _event_epoch(matching[-1], int(time.time()))
+                moment = datetime.fromtimestamp(current_timestamp, tz=timezone.utc)
+                rows = connection.execute("select events from profile_detector_temporal_buckets where stream_id=? and entity=? and detector=? and weekday=? and hour=? order by bucket_start desc limit 25", (stream, entity, detector, moment.weekday(), moment.hour)).fetchall()
+                temporal = True
+                baseline_scope = "same weekday/hour"
+                if len(rows) < 12:
+                    rows = connection.execute("select events from profile_detector_buckets where stream_id=? and entity=? and detector=? order by bucket_start desc limit 25", (stream, entity, detector)).fetchall()
+                    temporal = False
+                    baseline_scope = "all observed periods"
+                if len(rows) < 12:
+                    continue
+                history = [row[0] for row in rows]
+                z_score = (current_value - mean(history)) / (pstdev(history) or 1.0)
+                if z_score < z_threshold:
+                    continue
+                confidence = _baseline_confidence(len(rows), temporal=temporal)
+                score = min(35, (18 if detector == "auth_failure" else 15 if detector == "deny_action" else 12) + int(z_score))
+                samples = [{"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")), "source": event.src_ip or event.fields.get("source", ""), "destination": event.dst_ip or event.fields.get("query_domain", ""), "action": event.action, "severity": event.severity, "service": event.fields.get("service", event.fields.get("query_type", "")), "value": detector, "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]]
+                output[entity].append({"detector": f"{detector}_burst", "field": detector, "stream_id": stream, "score": score, "confidence": confidence, "baseline_samples": len(rows), "baseline_scope": baseline_scope, "reason": f"{detector.replace('_', ' ')} burst above its {baseline_scope} baseline (z={z_score:.1f}, minimum={minimum})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [detector], "sample_events": samples})
             # Detect selected categorical values that have not appeared for this entity in prior data.
             for event in events:
                 profile = profiles.get(event.fields.get("fgai_stream_id", ""))

src/fgai/cli.py

@@ -15,6 +15,9 @@ from .recommendations import build_recommendations
 from .syslog_server import listen_udp_syslog
 from .monitor import monitor_loop
 from .threat_intel import enrich_ips, is_public_ip
+from .config import ConfigStore
+from .replay import replay_events
+from .stream_profiles import parse_profiles
 
 
 def _print_json(data: object) -> None:
@@ -215,6 +218,15 @@ def run_dashboard(args: argparse.Namespace) -> int:
     return 0
 
 
+def replay_history(args: argparse.Namespace) -> int:
+    config = ConfigStore(args.config_file).read()
+    profiles = parse_profiles(config.get("graylog_stream_profiles", []))
+    stream_name = next((str(item.get("title", "")) for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") == args.stream_id), args.stream_id)
+    result = replay_events(read_events(args.logs), profiles, stream_id=args.stream_id, stream_name=stream_name, bucket_seconds=args.bucket_seconds)
+    _print_json(result)
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Local FortiGate AI/ML inspection tool")
     subparsers = parser.add_subparsers(required=True)
@@ -305,6 +317,13 @@ def build_parser() -> argparse.ArgumentParser:
     dashboard.add_argument("--config-file", default="state/fgai-config.json", help="Local runtime configuration JSON")
     dashboard.set_defaults(func=run_dashboard)
 
+    replay = subparsers.add_parser("replay", help="Replay a historical log export against temporary baselines")
+    replay.add_argument("--logs", required=True, help="Historic JSONL or key/value log export")
+    replay.add_argument("--config-file", default="state/fgai-config.json", help="Stream profile configuration")
+    replay.add_argument("--stream-id", default="", help="Apply this configured Graylog stream profile to exported events")
+    replay.add_argument("--bucket-seconds", type=int, default=300, help="Replay baseline bucket size")
+    replay.set_defaults(func=replay_history)
+
     return parser
 
 

src/fgai/dashboard.py

@@ -80,7 +80,7 @@ HTML = """<!doctype html>
     <div data-view="overview" class="active"><section class="split"><div class="panel"><h2>Events and Anomalies</h2><canvas id="trendChart" class="chart"></canvas></div><div class="panel"><h2>Baseline and Stream Health</h2><div id="health"></div></div></section><section class="split"><div class="panel"><h2>Correlation Map</h2><canvas id="correlationGraph" class="graph"></canvas><div id="correlationGraphInfo" class="muted"></div></div><div class="panel"><h2>AI Assessment</h2><div id="llmAssessment" class="muted">LLM assessment disabled.</div></div></section><section class="panel"><h2>Investigation Incidents</h2><div id="incidents"></div></section><section class="split"><div class="panel"><h2>Anomalies</h2><div id="anomalies"></div></div><div class="panel"><h2>Recommendations</h2><div id="recommendations"></div></div></section></div>
     <div data-view="findings"><section class="panel"><h2>Field Baseline Deviations</h2><div id="feedbackNotice" class="muted" role="status"></div><div id="fieldDeviations"></div></section><section class="panel"><h2>Related Activity Across Sources</h2><div id="relatedActivity"></div></section><section class="split"><div class="panel"><h2>Block Candidates</h2><div id="blocks"></div></div><div class="panel"><h2>Threat Intelligence</h2><div id="reputation"></div></div></section><section class="panel"><h2>Policy Findings</h2><div id="policies"></div></section></div>
     <div data-view="diagnostics"><section class="panel"><h2>Diagnostics</h2><div id="diagnostics"></div></section></div>
-    <div data-view="settings"><section class="panel"><h2>Runtime Configuration</h2><form id="settingsForm"><div class="grid"><label>Log source<br><select name="log_source"><option value="local_syslog">Local syslog</option><option value="graylog_mcp">Graylog MCP</option></select></label><label>Graylog MCP URL<br><input name="graylog_mcp_url" type="url" placeholder="https://graylog.example/api/mcp"></label><label>Graylog streams<br><button type="button" id="loadStreams">Load streams</button><div id="streamPicker" class="muted">Load streams after URL and token are saved.</div></label><label>Discover fields<br><button type="button" id="loadFields">Load selected stream fields</button><div id="fieldPicker" class="muted">Select a stream first.</div></label><label>Profile name<br><input name="profile_name" placeholder="Example: Windows login behavior"></label><label>Graylog query<br><input name="graylog_query" placeholder="*"></label><label>Graylog field mapping (JSON)<br><textarea name="graylog_field_mapping" placeholder='{"srcip":"client_ip","dstip":"server_ip","action":"event_action"}'></textarea></label><label>Graylog MCP token<br><input name="graylog_mcp_token" type="password" placeholder="Leave blank to keep current token"></label><label>Ollama model<br><input name="llm_model" placeholder="llama3.1"></label><label><input name="llm_enabled" type="checkbox"> Enable Ollama analysis</label><label><input name="threat_intel_enabled" type="checkbox"> Enable threat intelligence</label></div><p><button type="submit">Save configuration</button> <span id="settingsResult" class="muted"></span></p></form></section></div>
+    <div data-view="settings"><section class="panel"><h2>Runtime Configuration</h2><form id="settingsForm"><div class="grid"><label>Log source<br><select name="log_source"><option value="local_syslog">Local syslog</option><option value="graylog_mcp">Graylog MCP</option></select></label><label>Graylog MCP URL<br><input name="graylog_mcp_url" type="url" placeholder="https://graylog.example/api/mcp"></label><label>Graylog streams<br><button type="button" id="loadStreams">Load streams</button><div id="streamPicker" class="muted">Load streams after URL and token are saved.</div></label><label>Discover fields<br><button type="button" id="loadFields">Load selected stream fields</button><div id="fieldPicker" class="muted">Select a stream first.</div></label><label>Profile name<br><input name="profile_name" placeholder="Example: Windows login behavior"></label><label>Detector thresholds (JSON)<br><textarea name="profile_detectors" placeholder='{"auth_failure":{"enabled":true,"minimum":5,"z_threshold":3}}'></textarea></label><label>Graylog query<br><input name="graylog_query" placeholder="*"></label><label>Graylog field mapping (JSON)<br><textarea name="graylog_field_mapping" placeholder='{"srcip":"client_ip","dstip":"server_ip","action":"event_action"}'></textarea></label><label>Graylog MCP token<br><input name="graylog_mcp_token" type="password" placeholder="Leave blank to keep current token"></label><label>Ollama model<br><input name="llm_model" placeholder="llama3.1"></label><label><input name="llm_enabled" type="checkbox"> Enable Ollama analysis</label><label><input name="threat_intel_enabled" type="checkbox"> Enable threat intelligence</label></div><p><button type="submit">Save configuration</button> <span id="settingsResult" class="muted"></span></p></form></section></div>
   </main>
 <script>
 function esc(value) {
@@ -255,6 +255,7 @@ document.getElementById('loadFields').addEventListener('click', async () => {
   const payload = await (await fetch(`/api/graylog/fields?stream_id=${encodeURIComponent(selected.dataset.id)}`)).json();
   const profile = (window.streamProfiles || []).find(item => item.stream_id === selected.dataset.id) || {};
   document.querySelector('[name="profile_name"]').value = profile.name || `${selected.dataset.title} profile`;
+  document.querySelector('[name="profile_detectors"]').value = Object.keys(profile.detectors || {}).length ? JSON.stringify(profile.detectors, null, 2) : '';
   const rows = (payload.fields || []).map(field => { const name=field.name||field.field, type=(field.type||{}).type||'', props=(field.type||{}).properties||[]; return `<div class="field-row"><code>${esc(name)}</code><span>${esc(type)}</span><span>${esc(props.join(', '))}</span><div class="field-controls"><label><input type="radio" name="profile_entity" value="${esc(name)}" ${profile.entity_field===name?'checked':''}> Entity</label><label><input type="radio" name="profile_timestamp" value="${esc(name)}" ${profile.timestamp_field===name?'checked':''}> Time</label>${props.includes('enumerable')?`<label><input class="profile-categorical" type="checkbox" value="${esc(name)}" ${(profile.categorical_fields||[]).includes(name)?'checked':''}> Categorical</label>`:''}${props.includes('numeric')?`<label><input class="profile-numeric" type="checkbox" value="${esc(name)}" ${(profile.numeric_fields||[]).includes(name)?'checked':''}> Numeric</label>`:''}</div></div>`; }).join('');
   document.getElementById('fieldPicker').innerHTML = rows ? `<div class="field-header"><span>Field</span><span>Type</span><span>Capabilities</span><span>Use In Profile</span></div>${rows}` : esc(payload.error || 'No fields found.');
   window.activeProfileStream = selected.dataset.id;
@@ -266,7 +267,7 @@ document.getElementById('settingsForm').addEventListener('submit', async event =
   values.llm_enabled = form.elements.llm_enabled.checked;
   values.threat_intel_enabled = form.elements.threat_intel_enabled.checked;
   values.graylog_streams = [...document.querySelectorAll('.graylog-stream')].map(item => ({id:item.dataset.id, title:item.dataset.title, enabled:item.checked}));
-  if (window.activeProfileStream) { const selectedStream=document.querySelector('.graylog-stream:checked'); const profile={stream_id:window.activeProfileStream,name:form.elements.profile_name.value.trim() || `${selectedStream?.dataset.title || window.activeProfileStream} profile`,entity_field:form.querySelector('[name="profile_entity"]:checked')?.value||'',timestamp_field:form.querySelector('[name="profile_timestamp"]:checked')?.value||'timestamp',categorical_fields:[...form.querySelectorAll('.profile-categorical:checked')].map(item=>item.value),numeric_fields:[...form.querySelectorAll('.profile-numeric:checked')].map(item=>item.value)}; values.graylog_stream_profiles=[...(window.streamProfiles||[]).filter(item=>item.stream_id!==profile.stream_id),profile]; }
+  if (window.activeProfileStream) { let detectors={}; try { detectors=form.elements.profile_detectors.value.trim() ? JSON.parse(form.elements.profile_detectors.value) : {}; } catch { document.getElementById('settingsResult').textContent='Detector thresholds must be valid JSON.'; return; } const selectedStream=document.querySelector('.graylog-stream:checked'); const profile={stream_id:window.activeProfileStream,name:form.elements.profile_name.value.trim() || `${selectedStream?.dataset.title || window.activeProfileStream} profile`,entity_field:form.querySelector('[name="profile_entity"]:checked')?.value||'',timestamp_field:form.querySelector('[name="profile_timestamp"]:checked')?.value||'timestamp',categorical_fields:[...form.querySelectorAll('.profile-categorical:checked')].map(item=>item.value),numeric_fields:[...form.querySelectorAll('.profile-numeric:checked')].map(item=>item.value),detectors}; values.graylog_stream_profiles=[...(window.streamProfiles||[]).filter(item=>item.stream_id!==profile.stream_id),profile]; }
   const response = await fetch('/api/config', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(values)});
   document.getElementById('settingsResult').textContent = response.ok ? 'Saved. Monitor applies supported settings on its next cycle.' : 'Could not save configuration.';
   if (response.ok) loadSettings();

src/fgai/detectors.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from .logs import THREAT_ACTIONS
+from .models import LogEvent
+
+
+DETECTOR_MINIMUMS = {
+    "auth_failure": 5,
+    "dns_query": 20,
+    "deny_action": 10,
+}
+
+
+def event_detector_categories(event: LogEvent) -> tuple[str, ...]:
+    fields = event.fields
+    action = event.action
+    event_id = fields.get("eventid", fields.get("event_id", fields.get("winlog_event_id", "")))
+    categories: list[str] = []
+    if action in {"fail", "failed", "failure", "login_failed", "logon_failed", "authentication_failed"} or event_id == "4625":
+        categories.append("auth_failure")
+    if fields.get("query_domain") or fields.get("qh") or fields.get("dns_query"):
+        categories.append("dns_query")
+    if action in THREAT_ACTIONS:
+        categories.append("deny_action")
+    return tuple(categories)

src/fgai/graylog_source.py

@@ -17,7 +17,7 @@ DEFAULT_FIELD_MAP = {
     "action": ("action", "event_action", "disposition"),
 }
 
-DEFAULT_FIELDS = ["timestamp", "source", "srcip", "src_ip", "source_ip", "client_ip", "ip", "dstip", "dst_ip", "destination_ip", "upstream", "query_domain", "qh", "qt", "query_type", "srcport", "dstport", "service", "action", "severity", "policyid", "subtype", "type", "sentbyte", "rcvdbyte", "hitcount", "elapsed", "message"]
+DEFAULT_FIELDS = ["timestamp", "source", "srcip", "src_ip", "source_ip", "client_ip", "ip", "dstip", "dst_ip", "destination_ip", "upstream", "query_domain", "qh", "qt", "query_type", "dns_query", "srcport", "dstport", "service", "action", "status", "eventid", "event_id", "winlog_event_id", "severity", "policyid", "subtype", "type", "sentbyte", "rcvdbyte", "hitcount", "elapsed", "message"]
 
 
 def _records(value: object) -> Iterable[dict[str, object]]:

src/fgai/monitor.py

@@ -130,7 +130,7 @@ def build_status(
         "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "new_events_recorded": baseline_events, "profile_fields_recorded": profile_baseline_fields},
         "capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status},
         "configuration": runtime_config,
-        "stream_profiles": [{"stream_id": item.stream_id, "name": item.name, "entity_field": item.entity_field, "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields)} for item in stream_profiles.values()],
+        "stream_profiles": [{"stream_id": item.stream_id, "name": item.name, "entity_field": item.entity_field, "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields), "detectors": item.detectors} for item in stream_profiles.values()],
         "profile_readiness": profile_readiness,
         "diagnostics": {
             "top_source_ips": top_field_values(events, "srcip", limit=10),

src/fgai/replay.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import tempfile
+from collections import Counter, defaultdict
+from datetime import datetime
+from pathlib import Path
+
+from .anomaly import anomaly_summary, detect_source_anomalies
+from .baseline import BaselineStore
+from .models import LogEvent
+
+
+def _timestamp(event: LogEvent, fallback: int) -> int:
+    value = event.fields.get("eventtime", event.fields.get("timestamp", ""))
+    try:
+        return int(float(value))
+    except (TypeError, ValueError):
+        try:
+            return int(datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp())
+        except ValueError:
+            return fallback
+
+
+def _with_stream(events: list[LogEvent], stream_id: str, stream_name: str) -> list[LogEvent]:
+    if not stream_id:
+        return events
+    return [LogEvent(event.raw, {**event.fields, "fgai_stream_id": stream_id, "fgai_stream": stream_name or stream_id}) for event in events]
+
+
+def replay_events(events: list[LogEvent], profiles: dict[str, object], *, stream_id: str = "", stream_name: str = "", bucket_seconds: int = 300) -> dict[str, object]:
+    """Evaluate historical events using a temporary baseline without touching runtime state."""
+    normalized = _with_stream(events, stream_id, stream_name)
+    buckets: dict[int, list[LogEvent]] = defaultdict(list)
+    for index, event in enumerate(normalized):
+        timestamp = _timestamp(event, index * bucket_seconds)
+        buckets[timestamp - (timestamp % bucket_seconds)].append(event)
+
+    field_findings: list[dict[str, object]] = []
+    source_findings = []
+    with tempfile.TemporaryDirectory(prefix="signalscope-replay-") as directory:
+        baseline = BaselineStore(str(Path(directory) / "baseline.sqlite3"), bucket_seconds=bucket_seconds)
+        for bucket, batch in sorted(buckets.items()):
+            profiles_by_source = baseline.profiles({event.src_ip for event in batch if event.src_ip})
+            deviations = baseline.profile_deviations(batch, profiles)
+            anomalies = detect_source_anomalies(batch, baselines=profiles_by_source, field_deviations=deviations)
+            for entity, findings in deviations.items():
+                for finding in findings:
+                    field_findings.append({"bucket_start": bucket, "entity": entity, **finding})
+            source_findings.extend(anomalies)
+            baseline.ingest(batch, observed_at=bucket)
+            baseline.ingest_profile_fields(batch, profiles, observed_at=bucket)
+
+    detector_counts = Counter(str(item.get("detector", "unknown")) for item in field_findings)
+    return {
+        "events": len(normalized),
+        "buckets": len(buckets),
+        "field_findings": field_findings,
+        "field_detector_counts": dict(sorted(detector_counts.items())),
+        "source_anomaly_summary": anomaly_summary(source_findings),
+        "source_anomalies": [
+            {"subject": item.subject, "score": item.score, "severity": item.severity, "confidence": item.confidence, "reasons": item.reasons, "evidence": item.evidence}
+            for item in source_findings
+        ],
+    }

src/fgai/stream_profiles.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 
 @dataclass(frozen=True)
@@ -11,6 +11,23 @@ class StreamProfile:
     timestamp_field: str
     categorical_fields: tuple[str, ...] = ()
     numeric_fields: tuple[str, ...] = ()
+    detectors: dict[str, dict[str, object]] = field(default_factory=dict)
+
+
+def _detectors(value: object) -> dict[str, dict[str, object]]:
+    if not isinstance(value, dict):
+        return {}
+    output: dict[str, dict[str, object]] = {}
+    for name, settings in value.items():
+        if str(name) not in {"auth_failure", "dns_query", "deny_action"} or not isinstance(settings, dict):
+            continue
+        try:
+            minimum = max(1, int(settings.get("minimum", 1)))
+            z_threshold = max(1.0, float(settings.get("z_threshold", 3.0)))
+        except (TypeError, ValueError):
+            continue
+        output[str(name)] = {"enabled": bool(settings.get("enabled", True)), "minimum": minimum, "z_threshold": z_threshold}
+    return output
 
 
 def parse_profiles(value: object) -> dict[str, StreamProfile]:
@@ -23,9 +40,11 @@ def parse_profiles(value: object) -> dict[str, StreamProfile]:
         timestamp = str(item.get("timestamp_field", "timestamp")).strip()
         if not stream_id or not entity:
             continue
+        detectors = _detectors(item.get("detectors", {}))
         profiles[stream_id] = StreamProfile(
             stream_id, str(item.get("name", "")).strip() or stream_id, entity, timestamp,
             tuple(str(field) for field in item.get("categorical_fields", []) if field),
             tuple(str(field) for field in item.get("numeric_fields", []) if field),
+            detectors,
         )
     return profiles

tests/test_baseline.py

@@ -45,3 +45,19 @@ class BaselineTests(unittest.TestCase):
             rate = next(item for item in deviations if item["detector"] == "event_rate_burst")
             self.assertEqual(rate["field"], "event_rate")
             self.assertEqual(rate["baseline_samples"], 12)
+
+    def test_profile_authentication_failure_burst(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = BaselineStore(str(Path(directory) / "baseline.sqlite3"))
+            profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "categorical_fields": ["action"]}])
+            for index in range(12):
+                event = parse_log_line(f"fgai_stream_id=windows username=alice action=failed baseline={index}")
+                store.ingest_profile_fields([event], profiles, observed_at=1_700_000_000 + index * 300)
+
+            burst = [
+                parse_log_line(f"fgai_stream_id=windows username=alice action=failed burst={index}")
+                for index in range(5)
+            ]
+            deviations = store.profile_deviations(burst, profiles)["alice"]
+            auth = next(item for item in deviations if item["detector"] == "auth_failure_burst")
+            self.assertEqual(auth["current"], 5)

tests/test_detectors.py

@@ -0,0 +1,11 @@
+import unittest
+
+from fgai.detectors import event_detector_categories
+from fgai.logs import parse_log_line
+
+
+class DetectorCategoryTests(unittest.TestCase):
+    def test_classifies_windows_auth_failure_dns_and_deny(self):
+        self.assertEqual(event_detector_categories(parse_log_line("eventid=4625")), ("auth_failure",))
+        self.assertEqual(event_detector_categories(parse_log_line("qh=example.org")), ("dns_query",))
+        self.assertEqual(event_detector_categories(parse_log_line("action=deny")), ("deny_action",))

tests/test_replay.py

@@ -0,0 +1,21 @@
+import unittest
+
+from fgai.logs import parse_log_line
+from fgai.replay import replay_events
+from fgai.stream_profiles import parse_profiles
+
+
+class ReplayTests(unittest.TestCase):
+    def test_replay_uses_temporary_profile_baseline(self):
+        profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "categorical_fields": ["action"]}])
+        events = [
+            parse_log_line(f"timestamp={1_700_000_000 + index * 300} username=alice action=failed event={index}")
+            for index in range(12)
+        ]
+        events.extend(
+            parse_log_line(f"timestamp={1_700_003_600} username=alice action=failed burst={index}")
+            for index in range(5)
+        )
+        result = replay_events(events, profiles, stream_id="windows")
+        self.assertEqual(result["events"], 17)
+        self.assertGreaterEqual(result["field_detector_counts"].get("auth_failure_burst", 0), 1)

tests/test_stream_profiles.py

@@ -9,3 +9,7 @@ class StreamProfileTests(unittest.TestCase):
         self.assertEqual(profiles["dns"].name, "DNS client behavior")
         self.assertEqual(profiles["dns"].entity_field, "IP")
         self.assertEqual(profiles["dns"].numeric_fields, ("Elapsed",))
+
+    def test_parses_detector_thresholds(self):
+        profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "detectors": {"auth_failure": {"enabled": False, "minimum": 7, "z_threshold": 4.5}}}])
+        self.assertEqual(profiles["windows"].detectors["auth_failure"], {"enabled": False, "minimum": 7, "z_threshold": 4.5})