Add dashboard-triggered replay jobs and detector deltas

Closes the last ROADMAP §1 item. The replay/replay-graylog CLI backend
already existed and was already safe (a temporary, auto-cleaned SQLite
baseline, never touching live state) — the gap was purely that a human had
to SSH in and run it by hand. Extracts the core logic from cli.py into
run_local_replay()/run_graylog_replay() in replay.py, reused by both the
CLI (now thin wrappers) and a new dashboard panel (Diagnostics -> Replay &
Detector Deltas). Runs as a background job via an in-memory job store,
since the dashboard already runs ThreadingHTTPServer and this session
directly observed live MCP polling take anywhere from 9 minutes to over an
hour — a synchronous request/response was never viable. Deliberately
serialized to one job at a time given this box's GPU/CPU constraints.

Also removed a redundant "Detector Precision" panel that duplicated (with
a weaker, current-poll-cycle-only computation) what the new Detector
Health panel already does correctly with real historical data.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

README.md

@@ -516,6 +516,15 @@ signalscope replay-graylog --range-seconds 86400 --compare-config-file exports/c
 The comparison reports detector-count, field-finding, and source-anomaly deltas
 using the same fetched event window.
 
+Both replay modes are also available from `Diagnostics -> Replay & Detector
+Deltas` in the dashboard, so this no longer requires SSH access. A replay
+runs as a background job (only one at a time — this is a diagnostic tool,
+not something to run concurrently with itself on an already-busy box) and
+the panel polls for completion; results show the same detector counts and,
+when a comparison config is given, the same deltas as the CLI. Like the CLI
+commands, this only ever touches a temporary baseline — it cannot affect
+live detection state, and is safe to run while the monitor is polling.
+
 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

@@ -93,7 +93,17 @@ Goal: make findings more accurate before adding more integrations.
 - [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.
 - [x] Add direct Graylog MCP time-range replay and result comparison against saved detector configurations.
-- [ ] Add dashboard controls for launching safe replay jobs and viewing detector deltas.
+- [x] Add dashboard controls for launching safe replay jobs and viewing
+  detector deltas. The backend already existed (`replay`/`replay-graylog`
+  CLI commands, both against a temporary, auto-cleaned SQLite baseline that
+  never touches live state) — the gap was that a human had to SSH in and
+  run it by hand. Now runs from `Diagnostics -> Replay & Detector Deltas` as
+  a background job (the dashboard already runs `ThreadingHTTPServer`, so a
+  replay job doesn't block normal status polling), deliberately serialized
+  to one job at a time since this box is already resource-constrained
+  during a live poll. Along the way, removed a redundant, misleading old
+  "Detector Precision" panel that duplicated (with a weaker, current-cycle-only
+  computation) what the new Detector Health panel already does correctly.
 - [x] Add baseline confidence tooling: per detector learning state, expected
   false-positive rate, and why a deviation crossed threshold. The prior
   "detector precision" metric only existed in the Prometheus endpoint and
@@ -215,13 +225,12 @@ Acceptance: adding a new Graylog stream normally requires a profile template and
 
 ## Suggested Next Milestone
 
-Section 0 (Structural Rework) is complete and verified against the real
+Sections 0 (Structural Rework) and 1 (Detection Quality) are both complete.
-25-stream production deployment that motivated it: the `context`
+Section 0 is verified against the real 25-stream production deployment
-relationship flood dropped from 91 deviations to 0, total field deviations
+that motivated it: the `context` relationship flood dropped from 91
-from 117 to 17, and triage severity now differentiates (10 critical / 8
+deviations to 0, total field deviations from 117 to 17, and triage
-medium / 7 high, versus 25/25 critical before). See
+severity now differentiates (10 critical / 8 medium / 7 high, versus
-[`ARCHITECTURE.md`](ARCHITECTURE.md) for the full before/after. Sections 1
+25/25 critical before) — see [`ARCHITECTURE.md`](ARCHITECTURE.md) for the
-(Detection Quality, in particular baseline confidence tooling) and 6
+full before/after. Section 6 (Dashboard Usability, in particular compact
-(Dashboard Usability, compact incident cards) were previously blocked on
+incident cards) was previously blocked on section 0 and is the natural
-this and are the natural next targets, now that findings are clustered and
+next target, now that findings are clustered and scored correctly.
-scored correctly.

src/fgai/cli.py

@@ -15,13 +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 .baseline import BaselineStore, DEFAULT_MAX_VALUES_PER_FIELD, DEFAULT_RETENTION_DAYS, DEFAULT_VALUE_RETENTION_DAYS
 from .exports import investigation_report, investigation_report_markdown
-from .graylog_mcp import GraylogMcpClient
+from .replay import run_graylog_replay, run_local_replay
-from .graylog_source import GraylogStreamSource
-from .replay import replay_comparison, replay_events
-from .stream_profiles import parse_profiles
 
 
 def _print_json(data: object) -> None:
@@ -251,100 +247,14 @@ def baseline_maintenance(args: argparse.Namespace) -> int:
     return 0
 
 
-def _stream_name(config: dict[str, object], stream_id: str) -> str:
-    return next((str(item.get("title", "")) for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") == stream_id), stream_id)
-
-
-def _profile_fields(profile: object | None) -> tuple[str, ...]:
-    if not profile:
-        return ()
-    return tuple(
-        field for field in (
-            str(getattr(profile, "entity_field", "")),
-            *tuple(str(item) for item in getattr(profile, "entity_fields", ())),
-            str(getattr(profile, "timestamp_field", "")),
-            *tuple(str(item) for item in getattr(profile, "categorical_fields", ())),
-            *tuple(str(item) for item in getattr(profile, "numeric_fields", ())),
-            *tuple(str(getattr(relation, "left", "")) for relation in getattr(profile, "relationship_fields", ())),
-            *tuple(str(getattr(relation, "right", "")) for relation in getattr(profile, "relationship_fields", ())),
-        )
-        if field
-    )
-
-
-def _configured_streams(config: dict[str, object], selected: list[str] | None = None) -> list[dict[str, str]]:
-    selected_order = list(dict.fromkeys(item for item in selected or [] if item))
-    selected_set = set(selected_order)
-    configured = [
-        {"id": str(item.get("id", "")), "title": str(item.get("title", ""))}
-        for item in config.get("graylog_streams", [])
-        if isinstance(item, dict) and item.get("id") and (item.get("enabled") or selected_set)
-    ]
-    if selected_set:
-        configured = [item for item in configured if item["id"] in selected_set]
-        by_id = {item["id"]: item for item in configured}
-        configured = [by_id.get(stream_id, {"id": stream_id, "title": _stream_name(config, stream_id)}) for stream_id in selected_order]
-    if not configured and str(config.get("graylog_stream", "")):
-        stream_id = str(config.get("graylog_stream", ""))
-        configured = [{"id": stream_id, "title": _stream_name(config, stream_id)}]
-    return configured
-
-
 def replay_history(args: argparse.Namespace) -> int:
-    config = ConfigStore(args.config_file).read()
+    result = run_local_replay(args.config_file, args.logs, stream_id=args.stream_id, bucket_seconds=args.bucket_seconds, compare_config_file=args.compare_config_file)
-    profiles = parse_profiles(config.get("graylog_stream_profiles", []))
-    stream_name = _stream_name(config, args.stream_id)
-    events = read_events(args.logs)
-    result = replay_events(events, profiles, stream_id=args.stream_id, stream_name=stream_name, bucket_seconds=args.bucket_seconds)
-    if args.compare_config_file:
-        candidate_config = ConfigStore(args.compare_config_file).read()
-        candidate_profiles = parse_profiles(candidate_config.get("graylog_stream_profiles", []))
-        candidate = replay_events(events, candidate_profiles, stream_id=args.stream_id, stream_name=_stream_name(candidate_config, args.stream_id), bucket_seconds=args.bucket_seconds)
-        result = {"current": result, "candidate": candidate, "comparison": replay_comparison(result, candidate), "candidate_config_file": args.compare_config_file}
     _print_json(result)
     return 0
 
 
 def replay_graylog(args: argparse.Namespace) -> int:
-    config = ConfigStore(args.config_file).read()
+    result = run_graylog_replay(args.config_file, stream_ids=args.stream_id, range_seconds=args.range_seconds, max_events_per_stream=args.max_events_per_stream, bucket_seconds=args.bucket_seconds, compare_config_file=args.compare_config_file)
-    url, token = str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", ""))
-    if not url or not token:
-        raise RuntimeError("graylog_mcp_not_configured")
-    profiles = parse_profiles(config.get("graylog_stream_profiles", []))
-    candidate_profiles = {}
-    if args.compare_config_file:
-        candidate_config = ConfigStore(args.compare_config_file).read()
-        candidate_profiles = parse_profiles(candidate_config.get("graylog_stream_profiles", []))
-    streams = _configured_streams(config, args.stream_id)
-    if not streams:
-        raise RuntimeError("no_graylog_streams_selected")
-    events = []
-    stream_statuses = []
-    for stream in streams:
-        stream_id = stream["id"]
-        profile_fields = tuple(dict.fromkeys([*_profile_fields(profiles.get(stream_id)), *_profile_fields(candidate_profiles.get(stream_id))]))
-        stream_events, status = GraylogStreamSource(
-            GraylogMcpClient(url, token),
-            stream_id,
-            str(config.get("graylog_query", "*")),
-            str(config.get("graylog_field_mapping", "")),
-            stream.get("title") or stream_id,
-            profile_fields,
-        ).fetch(max_events=args.max_events_per_stream, range_seconds=args.range_seconds)
-        events.extend(stream_events)
-        stream_statuses.append({"stream_id": stream_id, "stream_name": stream.get("title") or stream_id, **status})
-    result = {
-        "source": "graylog_mcp",
-        "range_seconds": args.range_seconds,
-        "streams": stream_statuses,
-        "replay": replay_events(events, profiles, bucket_seconds=args.bucket_seconds),
-    }
-    if args.compare_config_file:
-        candidate = replay_events(events, candidate_profiles, bucket_seconds=args.bucket_seconds)
-        current = result["replay"]
-        result["candidate_config_file"] = args.compare_config_file
-        result["candidate_replay"] = candidate
-        result["comparison"] = replay_comparison(current, candidate)
     _print_json(result)
     return 0
 

src/fgai/dashboard.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import json
+import threading
+import time
+import uuid
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib import error, request
@@ -12,6 +15,65 @@ from .graylog_mcp import GraylogMcpClient
 from .metrics import prometheus_metrics
 from .feedback import DetectorPrecisionStore, FeedbackStore
 from .incidents import IncidentStore
+from .replay import run_graylog_replay, run_local_replay
+
+_replay_jobs: dict[str, dict[str, object]] = {}
+_replay_lock = threading.Lock()
+
+
+def _run_replay_job(job_id: str, params: dict[str, object]) -> None:
+    try:
+        if params.get("mode") == "graylog":
+            result = run_graylog_replay(
+                str(params.get("config_file") or "state/fgai-config.json"),
+                stream_ids=[str(item) for item in params.get("stream_ids", []) if item],
+                range_seconds=int(params.get("range_seconds", 24 * 3600) or 24 * 3600),
+                max_events_per_stream=int(params.get("max_events_per_stream", 5_000) or 5_000),
+                bucket_seconds=int(params.get("bucket_seconds", 300) or 300),
+                compare_config_file=str(params.get("compare_config_file") or ""),
+            )
+        else:
+            result = run_local_replay(
+                str(params.get("config_file") or "state/fgai-config.json"),
+                str(params.get("logs") or ""),
+                stream_id=str(params.get("stream_id") or ""),
+                bucket_seconds=int(params.get("bucket_seconds", 300) or 300),
+                compare_config_file=str(params.get("compare_config_file") or ""),
+            )
+        with _replay_lock:
+            _replay_jobs[job_id].update({"status": "done", "result": result, "finished_at": int(time.time())})
+    except Exception as exc:
+        with _replay_lock:
+            _replay_jobs[job_id].update({"status": "error", "error": str(exc), "finished_at": int(time.time())})
+
+
+def _start_replay_job(params: dict[str, object]) -> dict[str, object]:
+    """Starts a replay job in a background thread. Only one job runs at a
+    time — this box is already resource-constrained during a live monitor
+    poll, so replay jobs are deliberately serialized rather than queued or
+    run concurrently. replay.py's functions only ever touch a temporary
+    baseline, so this never affects the live baseline or monitor state.
+    """
+    with _replay_lock:
+        running = any(job.get("status") == "running" for job in _replay_jobs.values())
+        if running:
+            return {"error": "a replay job is already running"}
+        job_id = uuid.uuid4().hex
+        _replay_jobs[job_id] = {"status": "running", "started_at": int(time.time()), "params": params}
+    thread = threading.Thread(target=_run_replay_job, args=(job_id, params), daemon=True)
+    thread.start()
+    return {"job_id": job_id}
+
+
+def _replay_job_status(job_id: str = "") -> dict[str, object] | None:
+    with _replay_lock:
+        if job_id:
+            job = _replay_jobs.get(job_id)
+            return dict(job) if job else None
+        if not _replay_jobs:
+            return None
+        latest_id = max(_replay_jobs, key=lambda key: _replay_jobs[key].get("started_at", 0))
+        return {"job_id": latest_id, **_replay_jobs[latest_id]}
 
 
 def _ollama_models(host: str = "http://127.0.0.1:11434") -> dict[str, object]:
@@ -249,6 +311,46 @@ function table(rows, columns, id = '') {
   const body = sorted.map(row => `<tr>${columns.map(c => `<td>${c.render ? c.render(row) : esc(row[c.key])}</td>`).join('')}</tr>`).join('');
   return `<div class="table-wrap"${id ? ` data-table-id="${esc(id)}"` : ''}><table><thead><tr>${head}</tr></thead><tbody>${body}</tbody></table></div>`;
 }
+function renderReplayJob(job) {
+  const statusEl = document.getElementById('replayStatus');
+  const resultEl = document.getElementById('replayResult');
+  if (!statusEl || !resultEl) return;
+  if (!job) { statusEl.textContent = ''; resultEl.innerHTML = ''; return; }
+  if (job.status === 'running') {
+    statusEl.textContent = `Running (started ${new Date((job.started_at || 0) * 1000).toLocaleTimeString()})... this can take a while for a Graylog window with many streams.`;
+    resultEl.innerHTML = '';
+    return;
+  }
+  if (job.status === 'error') {
+    statusEl.innerHTML = `<span class="sev-high">Failed: ${esc(job.error || 'unknown error')}</span>`;
+    resultEl.innerHTML = '';
+    return;
+  }
+  statusEl.textContent = `Finished ${new Date((job.finished_at || 0) * 1000).toLocaleTimeString()}.`;
+  const result = job.result || {};
+  const replay = result.replay || result;
+  const detectorCounts = Object.entries(replay.field_detector_counts || {}).map(([detector, count]) => ({detector, count}));
+  let html = `<p class="muted">${replay.events || 0} events, ${replay.buckets || 0} buckets, ${(replay.source_anomalies || []).length} source anomalies.</p>` + table(detectorCounts, [{label:'Detector', key:'detector'}, {label:'Findings', key:'count'}], 'replay-detector-counts');
+  if (result.comparison) {
+    const deltas = Object.entries(result.comparison.detector_count_delta || {}).map(([detector, delta]) => ({detector, delta}));
+    html += `<p class="muted">Compared against ${esc(result.candidate_config_file || '')}: ${result.comparison.events_delta >= 0 ? '+' : ''}${result.comparison.events_delta} events, ${result.comparison.field_findings_delta >= 0 ? '+' : ''}${result.comparison.field_findings_delta} findings.</p>` + table(deltas, [{label:'Detector', key:'detector'}, {label:'Delta', key:'delta'}], 'replay-detector-deltas');
+  }
+  resultEl.innerHTML = html;
+}
+function pollReplayJob(jobId) {
+  window.replayPolling = true;
+  const poll = async () => {
+    const response = await fetch(`/api/replay/status?job_id=${encodeURIComponent(jobId)}`);
+    const job = await response.json().catch(() => null);
+    renderReplayJob(job);
+    if (job && job.status === 'running') {
+      setTimeout(poll, 3000);
+    } else {
+      window.replayPolling = false;
+    }
+  };
+  poll();
+}
 function capability(label, state, detail) { return `<span class="capability ${state}">${esc(label)}: ${esc(detail)}</span>`; }
 function mcpCapability(mcp, configuration, pollRunningSeconds=0) {
   if (configuration.log_source !== 'graylog_mcp') return {state: 'warn', detail: 'not selected'};
@@ -1096,17 +1198,6 @@ async function refresh() {
       [{label:'Scope', key:'scope'}, {label:'Active rules', key:'count'}],
       'feedback-rules-by-scope'
     ) + `<p class="muted">${rawFieldRows.filter(item => ['expected', 'false_positive'].includes(item.feedback || '')).length} finding(s) suppressed by feedback this poll.</p>`) +
-    collapsibleSection('detector-precision', 'Detector Precision', table(
-      Object.entries(rawFieldRows.reduce((acc, item) => {
-        if (!['confirmed', 'false_positive'].includes(item.feedback || '')) return acc;
-        const key = item.detector || 'unknown';
-        acc[key] = acc[key] || {confirmed: 0, false_positive: 0};
-        acc[key][item.feedback] += 1;
-        return acc;
-      }, {})).map(([detector, counts]) => ({detector, confirmed: counts.confirmed, false_positive: counts.false_positive, precision: (counts.confirmed + counts.false_positive) ? `${(counts.confirmed / (counts.confirmed + counts.false_positive) * 100).toFixed(1)}%` : '-'})),
-      [{label:'Detector', key:'detector'}, {label:'Confirmed', key:'confirmed'}, {label:'False positive', key:'false_positive'}, {label:'Precision', key:'precision'}],
-      'detector-precision'
-    )) +
     collapsibleSection('security-event-samples', 'Security Event Samples', table(context.security_event_samples || [], [{label:'Entity', key:'entity', render:r => entityCell(r)}, {label:'Type', key:'type'}, {label:'Action', key:'action'}, {label:'Severity', key:'severity'}, {label:'Destination', key:'dst'}, {label:'Service', key:'service'}], 'security-event-samples')) +
     collapsibleSection('top-sources', 'Top Sources', table(d.top_source_ips || [], [{label:'Value', key:'value', render:r => entityCell({entity:r.value})}, {label:'Count', key:'count'}], 'top-sources')) +
     collapsibleSection('top-destinations', 'Top Destinations', table(d.top_destination_ips || [], [{label:'Value', key:'value'}, {label:'Count', key:'count'}], 'top-destinations')) +
@@ -1114,7 +1205,8 @@ async function refresh() {
     collapsibleSection('top-destination-ports', 'Top Destination Ports', table(d.top_destination_ports || [], [{label:'Value', key:'value'}, {label:'Count', key:'count'}], 'top-destination-ports')) +
     collapsibleSection('top-source-ports', 'Top Source Ports', table(d.top_source_ports || [], [{label:'Value', key:'value'}, {label:'Count', key:'count'}], 'top-source-ports')) +
     collapsibleSection('top-services', 'Top Services', table(d.top_services || [], [{label:'Value', key:'value'}, {label:'Count', key:'count'}], 'top-services')) +
-    collapsibleSection('local-in-failures', 'Local-in Failures', table(d.local_in_failures || [], [{label:'Source', key:'src_ip', render:r => entityCell(r, 'src_ip')}, {label:'Service', key:'service'}, {label:'Policy', key:'policy'}, {label:'Count', key:'count'}], 'local-in-failures'));
+    collapsibleSection('local-in-failures', 'Local-in Failures', table(d.local_in_failures || [], [{label:'Source', key:'src_ip', render:r => entityCell(r, 'src_ip')}, {label:'Service', key:'service'}, {label:'Policy', key:'policy'}, {label:'Count', key:'count'}], 'local-in-failures')) +
+    collapsibleSection('replay', 'Replay & Detector Deltas', `<p class="muted">Replays historical events against a temporary baseline — never touches the live baseline or monitor state, so this is safe to run alongside normal monitoring. Only one replay job runs at a time.</p><form id="replayForm" class="toolbar"><label>Mode<br><select id="replayMode"><option value="local">Local log file</option><option value="graylog">Graylog MCP window</option></select></label><label id="replayLogsField">Log file path<br><input id="replayLogs" type="text" placeholder="logs/fg_syslog.jsonl"></label><label id="replayStreamIdField">Stream (optional)<br><select id="replayStreamId"><option value="">(none)</option>${(configuration.graylog_streams || []).map(s => `<option value="${esc(s.id)}">${esc(s.title || s.id)}</option>`).join('')}</select></label><label id="replayStreamIdsField" hidden>Streams (default: all enabled)<br><select id="replayStreamIds" multiple size="4">${(configuration.graylog_streams || []).filter(s => s.enabled).map(s => `<option value="${esc(s.id)}">${esc(s.title || s.id)}</option>`).join('')}</select></label><label id="replayRangeField" hidden>Range (seconds)<br><input id="replayRangeSeconds" type="number" min="60" step="60" value="86400"></label><label id="replayMaxEventsField" hidden>Max events per stream<br><input id="replayMaxEvents" type="number" min="1" step="500" value="5000"></label><label>Compare against config file (optional)<br><input id="replayCompareConfig" type="text" placeholder="state/other-config.json"></label></form><p><button type="button" id="replayRun">Run replay</button></p><div id="replayStatus" class="muted"></div><div id="replayResult"></div>`);
   document.querySelectorAll('details[data-detail-id]').forEach(item => {
     if (openDetailIds.has(item.dataset.detailId)) item.open = true;
     item.addEventListener('toggle', () => {
@@ -1127,6 +1219,44 @@ async function refresh() {
     wrap.addEventListener('scroll', () => { tableScrollLeft[tid] = wrap.scrollLeft; });
   });
   document.querySelectorAll('[data-sort-table]').forEach(button => button.addEventListener('click', () => { const current=tableSort[button.dataset.sortTable]; tableSort[button.dataset.sortTable]={key:button.dataset.sortKey,direction:current && current.key===button.dataset.sortKey ? -current.direction : 1}; refresh(); }));
+  const replayModeSelect = document.getElementById('replayMode');
+  if (replayModeSelect) {
+    const applyReplayMode = () => {
+      const graylogMode = replayModeSelect.value === 'graylog';
+      document.getElementById('replayLogsField').hidden = graylogMode;
+      document.getElementById('replayStreamIdField').hidden = graylogMode;
+      document.getElementById('replayStreamIdsField').hidden = !graylogMode;
+      document.getElementById('replayRangeField').hidden = !graylogMode;
+      document.getElementById('replayMaxEventsField').hidden = !graylogMode;
+    };
+    replayModeSelect.onchange = applyReplayMode;
+    applyReplayMode();
+    document.getElementById('replayRun').onclick = async () => {
+      const mode = replayModeSelect.value;
+      const payload = {
+        mode,
+        logs: document.getElementById('replayLogs').value,
+        stream_id: document.getElementById('replayStreamId').value,
+        stream_ids: Array.from(document.getElementById('replayStreamIds').selectedOptions).map(o => o.value),
+        range_seconds: Number(document.getElementById('replayRangeSeconds').value || 86400),
+        max_events_per_stream: Number(document.getElementById('replayMaxEvents').value || 5000),
+        compare_config_file: document.getElementById('replayCompareConfig').value,
+      };
+      const response = await fetch('/api/replay/start', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(payload)});
+      const result = await response.json().catch(() => ({}));
+      if (!response.ok) {
+        document.getElementById('replayStatus').textContent = result.error || 'Could not start replay job.';
+        return;
+      }
+      pollReplayJob(result.job_id);
+    };
+    if (!window.replayPolling) {
+      fetch('/api/replay/status').then(r => r.ok ? r.json() : null).then(job => {
+        if (job && job.status === 'running') pollReplayJob(job.job_id);
+        else if (job) renderReplayJob(job);
+      }).catch(() => {});
+    }
+  }
   } catch (error) {
     showRefreshError(error);
   } finally {
@@ -1556,6 +1686,16 @@ def serve_dashboard(host: str, port: int, status_file: str, *, image_dir: str |
                 else:
                     self._send(200, "text/markdown; charset=utf-8", investigation_report_markdown(report).encode("utf-8"))
                 return
+            if self.path.startswith("/api/replay/status"):
+                parsed = urlparse(self.path)
+                params = parse_qs(parsed.query)
+                job_id = params.get("job_id", [""])[0]
+                job = _replay_job_status(job_id)
+                if job is None:
+                    self._send(404, "application/json", json.dumps({"error": "no replay job found"}).encode("utf-8"))
+                else:
+                    self._send(200, "application/json", json.dumps(job).encode("utf-8"))
+                return
             if self.path == "/metrics":
                 status = read_status()
                 self._send(200, "text/plain; version=0.0.4; charset=utf-8", prometheus_metrics(status).encode("utf-8"))
@@ -1635,6 +1775,17 @@ def serve_dashboard(host: str, port: int, status_file: str, *, image_dir: str |
                 except (ValueError, json.JSONDecodeError) as exc:
                     self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
                 return
+            if self.path == "/api/replay/start" and self._is_loopback_client():
+                try:
+                    payload = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", "0")), 16_384)).decode("utf-8"))
+                    if not isinstance(payload, dict):
+                        raise ValueError("replay parameters must be an object")
+                    result = _start_replay_job(payload)
+                    status_code = 409 if "error" in result else 200
+                    self._send(status_code, "application/json", json.dumps(result).encode("utf-8"))
+                except (ValueError, json.JSONDecodeError) as exc:
+                    self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
+                return
             if self.path != "/api/config" or not self._is_loopback_client():
                 self._send(403, "application/json", b'{"error":"configuration is local-only"}')
                 return

src/fgai/replay.py

@@ -6,8 +6,13 @@ from pathlib import Path
 
 from .anomaly import anomaly_summary, detect_source_anomalies
 from .baseline import BaselineStore
+from .config import ConfigStore
 from .field_shapes import parse_iso_timestamp
+from .graylog_mcp import GraylogMcpClient
+from .graylog_source import GraylogStreamSource
+from .logs import read_events
 from .models import LogEvent
+from .stream_profiles import parse_profiles
 
 
 def replay_comparison(current: dict[str, object], candidate: dict[str, object]) -> dict[str, object]:
@@ -77,3 +82,109 @@ def replay_events(events: list[LogEvent], profiles: dict[str, object], *, stream
             for item in source_findings
         ],
     }
+
+
+def _stream_name(config: dict[str, object], stream_id: str) -> str:
+    return next((str(item.get("title", "")) for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") == stream_id), stream_id)
+
+
+def _profile_fields(profile: object | None) -> tuple[str, ...]:
+    if not profile:
+        return ()
+    return tuple(
+        field for field in (
+            str(getattr(profile, "entity_field", "")),
+            *tuple(str(item) for item in getattr(profile, "entity_fields", ())),
+            str(getattr(profile, "timestamp_field", "")),
+            *tuple(str(item) for item in getattr(profile, "categorical_fields", ())),
+            *tuple(str(item) for item in getattr(profile, "numeric_fields", ())),
+            *tuple(str(getattr(relation, "left", "")) for relation in getattr(profile, "relationship_fields", ())),
+            *tuple(str(getattr(relation, "right", "")) for relation in getattr(profile, "relationship_fields", ())),
+        )
+        if field
+    )
+
+
+def _configured_streams(config: dict[str, object], selected: list[str] | None = None) -> list[dict[str, str]]:
+    selected_order = list(dict.fromkeys(item for item in selected or [] if item))
+    selected_set = set(selected_order)
+    configured = [
+        {"id": str(item.get("id", "")), "title": str(item.get("title", ""))}
+        for item in config.get("graylog_streams", [])
+        if isinstance(item, dict) and item.get("id") and (item.get("enabled") or selected_set)
+    ]
+    if selected_set:
+        configured = [item for item in configured if item["id"] in selected_set]
+        by_id = {item["id"]: item for item in configured}
+        configured = [by_id.get(stream_id, {"id": stream_id, "title": _stream_name(config, stream_id)}) for stream_id in selected_order]
+    if not configured and str(config.get("graylog_stream", "")):
+        stream_id = str(config.get("graylog_stream", ""))
+        configured = [{"id": stream_id, "title": _stream_name(config, stream_id)}]
+    return configured
+
+
+def run_local_replay(config_path: str, logs_path: str, *, stream_id: str = "", bucket_seconds: int = 300, compare_config_file: str = "") -> dict[str, object]:
+    """Replay a local JSONL/key-value log export against a temporary baseline.
+
+    Safe to call from the dashboard: replay_events() never touches the live
+    baseline or monitor state, so this can run concurrently with a live poll.
+    """
+    config = ConfigStore(config_path).read()
+    profiles = parse_profiles(config.get("graylog_stream_profiles", []))
+    stream_name = _stream_name(config, stream_id)
+    events = read_events(logs_path)
+    result = replay_events(events, profiles, stream_id=stream_id, stream_name=stream_name, bucket_seconds=bucket_seconds)
+    if compare_config_file:
+        candidate_config = ConfigStore(compare_config_file).read()
+        candidate_profiles = parse_profiles(candidate_config.get("graylog_stream_profiles", []))
+        candidate = replay_events(events, candidate_profiles, stream_id=stream_id, stream_name=_stream_name(candidate_config, stream_id), bucket_seconds=bucket_seconds)
+        result = {"current": result, "candidate": candidate, "comparison": replay_comparison(result, candidate), "candidate_config_file": compare_config_file}
+    return result
+
+
+def run_graylog_replay(config_path: str, *, stream_ids: list[str] | None = None, range_seconds: int = 24 * 3600, max_events_per_stream: int = 5_000, bucket_seconds: int = 300, compare_config_file: str = "") -> dict[str, object]:
+    """Replay a Graylog MCP time window against a temporary baseline.
+
+    Safe to call from the dashboard for the same reason as run_local_replay —
+    a temporary baseline only, and this only reads from Graylog MCP.
+    """
+    config = ConfigStore(config_path).read()
+    url, token = str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", ""))
+    if not url or not token:
+        raise RuntimeError("graylog_mcp_not_configured")
+    profiles = parse_profiles(config.get("graylog_stream_profiles", []))
+    candidate_profiles = {}
+    if compare_config_file:
+        candidate_config = ConfigStore(compare_config_file).read()
+        candidate_profiles = parse_profiles(candidate_config.get("graylog_stream_profiles", []))
+    streams = _configured_streams(config, stream_ids)
+    if not streams:
+        raise RuntimeError("no_graylog_streams_selected")
+    events: list[LogEvent] = []
+    stream_statuses = []
+    for stream in streams:
+        stream_id = stream["id"]
+        profile_fields = tuple(dict.fromkeys([*_profile_fields(profiles.get(stream_id)), *_profile_fields(candidate_profiles.get(stream_id))]))
+        stream_events, status = GraylogStreamSource(
+            GraylogMcpClient(url, token),
+            stream_id,
+            str(config.get("graylog_query", "*")),
+            str(config.get("graylog_field_mapping", "")),
+            stream.get("title") or stream_id,
+            profile_fields,
+        ).fetch(max_events=max_events_per_stream, range_seconds=range_seconds)
+        events.extend(stream_events)
+        stream_statuses.append({"stream_id": stream_id, "stream_name": stream.get("title") or stream_id, **status})
+    result = {
+        "source": "graylog_mcp",
+        "range_seconds": range_seconds,
+        "streams": stream_statuses,
+        "replay": replay_events(events, profiles, bucket_seconds=bucket_seconds),
+    }
+    if compare_config_file:
+        candidate = replay_events(events, candidate_profiles, bucket_seconds=bucket_seconds)
+        current = result["replay"]
+        result["candidate_config_file"] = compare_config_file
+        result["candidate_replay"] = candidate
+        result["comparison"] = replay_comparison(current, candidate)
+    return result

tests/test_cli.py

@@ -1,39 +0,0 @@
-import unittest
-
-from fgai.cli import _configured_streams
-
-
-class CliTests(unittest.TestCase):
-    def test_configured_streams_uses_enabled_streams_by_default(self):
-        config = {
-            "graylog_streams": [
-                {"id": "fortigate", "title": "Fortigate", "enabled": True},
-                {"id": "adguard", "title": "Adguard", "enabled": False},
-            ]
-        }
-        self.assertEqual(_configured_streams(config), [{"id": "fortigate", "title": "Fortigate"}])
-
-    def test_configured_streams_can_select_disabled_stream(self):
-        config = {
-            "graylog_streams": [
-                {"id": "fortigate", "title": "Fortigate", "enabled": True},
-                {"id": "adguard", "title": "Adguard", "enabled": False},
-            ]
-        }
-        self.assertEqual(_configured_streams(config, ["adguard"]), [{"id": "adguard", "title": "Adguard"}])
-
-    def test_configured_streams_preserves_selected_order(self):
-        config = {
-            "graylog_streams": [
-                {"id": "fortigate", "title": "Fortigate", "enabled": True},
-                {"id": "adguard", "title": "Adguard", "enabled": False},
-            ]
-        }
-        self.assertEqual(
-            _configured_streams(config, ["adguard", "fortigate"]),
-            [{"id": "adguard", "title": "Adguard"}, {"id": "fortigate", "title": "Fortigate"}],
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_dashboard.py

@@ -1,5 +1,6 @@
 import json
 import inspect
+import time
 import unittest
 from unittest.mock import patch
 
@@ -255,6 +256,85 @@ class DashboardTests(unittest.TestCase):
         self.assertIn('name="auto_apply_profile_suggestions"', HTML)
         self.assertIn("field_confidence_threshold: config.field_confidence_threshold ?? 60", HTML)
 
+    def test_dashboard_exposes_replay_controls(self):
+        self.assertIn("/api/replay/start", HTML)
+        self.assertIn("/api/replay/status", HTML)
+        self.assertIn("Replay", HTML)
+
+
+class ReplayJobTests(unittest.TestCase):
+    def setUp(self):
+        dashboard._replay_jobs.clear()
+
+    def tearDown(self):
+        dashboard._replay_jobs.clear()
+
+    def test_run_replay_job_records_success(self):
+        with patch("fgai.dashboard.run_local_replay", return_value={"events": 5}):
+            dashboard._replay_jobs["job-1"] = {"status": "running", "started_at": 100}
+            dashboard._run_replay_job("job-1", {"mode": "local"})
+
+        job = dashboard._replay_jobs["job-1"]
+        self.assertEqual(job["status"], "done")
+        self.assertEqual(job["result"], {"events": 5})
+        self.assertIn("finished_at", job)
+
+    def test_run_replay_job_records_failure_without_crashing(self):
+        with patch("fgai.dashboard.run_local_replay", side_effect=RuntimeError("boom")):
+            dashboard._replay_jobs["job-1"] = {"status": "running", "started_at": 100}
+            dashboard._run_replay_job("job-1", {"mode": "local"})
+
+        job = dashboard._replay_jobs["job-1"]
+        self.assertEqual(job["status"], "error")
+        self.assertEqual(job["error"], "boom")
+
+    def test_run_replay_job_dispatches_graylog_mode(self):
+        with patch("fgai.dashboard.run_graylog_replay", return_value={"events": 9}) as mocked:
+            dashboard._replay_jobs["job-1"] = {"status": "running", "started_at": 100}
+            dashboard._run_replay_job("job-1", {"mode": "graylog", "stream_ids": ["a", "b"], "range_seconds": 600})
+
+        mocked.assert_called_once()
+        self.assertEqual(dashboard._replay_jobs["job-1"]["result"], {"events": 9})
+
+    def test_start_replay_job_rejects_a_second_concurrent_job(self):
+        dashboard._replay_jobs["existing"] = {"status": "running", "started_at": 100}
+
+        result = dashboard._start_replay_job({"mode": "local"})
+
+        self.assertIn("error", result)
+        self.assertEqual(len(dashboard._replay_jobs), 1)
+
+    def test_start_replay_job_starts_a_thread_when_idle(self):
+        with patch("fgai.dashboard.run_local_replay", return_value={"events": 1}):
+            result = dashboard._start_replay_job({"mode": "local"})
+            self.assertIn("job_id", result)
+            job_id = result["job_id"]
+            for _ in range(50):
+                if dashboard._replay_jobs.get(job_id, {}).get("status") != "running":
+                    break
+                time.sleep(0.02)
+
+        self.assertEqual(dashboard._replay_jobs[job_id]["status"], "done")
+
+    def test_replay_job_status_returns_job_by_id(self):
+        dashboard._replay_jobs["job-1"] = {"status": "done", "started_at": 100, "result": {"events": 1}}
+
+        status = dashboard._replay_job_status("job-1")
+
+        self.assertEqual(status["status"], "done")
+
+    def test_replay_job_status_returns_latest_job_when_no_id_given(self):
+        dashboard._replay_jobs["older"] = {"status": "done", "started_at": 100}
+        dashboard._replay_jobs["newer"] = {"status": "running", "started_at": 200}
+
+        status = dashboard._replay_job_status()
+
+        self.assertEqual(status["job_id"], "newer")
+
+    def test_replay_job_status_returns_none_when_no_jobs_exist(self):
+        self.assertIsNone(dashboard._replay_job_status())
+        self.assertIsNone(dashboard._replay_job_status("missing"))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_replay.py

@@ -1,7 +1,10 @@
+import json
+import tempfile
 import unittest
+from pathlib import Path
 
 from fgai.logs import parse_log_line
-from fgai.replay import replay_comparison, replay_events
+from fgai.replay import _configured_streams, replay_comparison, replay_events, run_graylog_replay, run_local_replay
 from fgai.stream_profiles import parse_profiles
 
 
@@ -29,3 +32,87 @@ class ReplayTests(unittest.TestCase):
         self.assertEqual(result["source_anomalies_delta"], 1)
         self.assertEqual(result["detector_count_delta"]["auth_failure_burst"], -1)
         self.assertEqual(result["detector_count_delta"]["rare_value"], 3)
+
+    def test_configured_streams_uses_enabled_streams_by_default(self):
+        config = {
+            "graylog_streams": [
+                {"id": "fortigate", "title": "Fortigate", "enabled": True},
+                {"id": "adguard", "title": "Adguard", "enabled": False},
+            ]
+        }
+        self.assertEqual(_configured_streams(config), [{"id": "fortigate", "title": "Fortigate"}])
+
+    def test_configured_streams_can_select_disabled_stream(self):
+        config = {
+            "graylog_streams": [
+                {"id": "fortigate", "title": "Fortigate", "enabled": True},
+                {"id": "adguard", "title": "Adguard", "enabled": False},
+            ]
+        }
+        self.assertEqual(_configured_streams(config, ["adguard"]), [{"id": "adguard", "title": "Adguard"}])
+
+    def test_configured_streams_preserves_selected_order(self):
+        config = {
+            "graylog_streams": [
+                {"id": "fortigate", "title": "Fortigate", "enabled": True},
+                {"id": "adguard", "title": "Adguard", "enabled": False},
+            ]
+        }
+        self.assertEqual(
+            _configured_streams(config, ["adguard", "fortigate"]),
+            [{"id": "adguard", "title": "Adguard"}, {"id": "fortigate", "title": "Fortigate"}],
+        )
+
+    def test_run_local_replay_matches_replay_events_output(self):
+        with tempfile.TemporaryDirectory() as directory:
+            config_path = Path(directory) / "config.json"
+            config_path.write_text(json.dumps({"graylog_stream_profiles": [{"stream_id": "windows", "entity_field": "username", "categorical_fields": ["action"]}]}), encoding="utf-8")
+            logs_path = Path(directory) / "events.log"
+            lines = [
+                f"timestamp={1_700_000_000 + index * 300} username=alice action=failed event={index}"
+                for index in range(12)
+            ]
+            lines.extend(f"timestamp={1_700_003_600} username=alice action=failed burst={index}" for index in range(5))
+            logs_path.write_text("\n".join(lines), encoding="utf-8")
+
+            result = run_local_replay(str(config_path), str(logs_path), stream_id="windows")
+
+            self.assertEqual(result["events"], 17)
+            self.assertGreaterEqual(result["field_detector_counts"].get("auth_failure_burst", 0), 1)
+
+    def test_run_local_replay_with_compare_config_reports_comparison(self):
+        with tempfile.TemporaryDirectory() as directory:
+            config_path = Path(directory) / "config.json"
+            config_path.write_text(json.dumps({"graylog_stream_profiles": [{"stream_id": "windows", "entity_field": "username", "categorical_fields": ["action"]}]}), encoding="utf-8")
+            candidate_path = Path(directory) / "candidate.json"
+            candidate_path.write_text(json.dumps({"graylog_stream_profiles": [{"stream_id": "windows", "entity_field": "username", "categorical_fields": []}]}), encoding="utf-8")
+            logs_path = Path(directory) / "events.log"
+            lines = [
+                f"timestamp={1_700_000_000 + index * 300} username=alice action=failed event={index}"
+                for index in range(12)
+            ]
+            lines.extend(f"timestamp={1_700_003_600} username=alice action=failed burst={index}" for index in range(5))
+            logs_path.write_text("\n".join(lines), encoding="utf-8")
+
+            result = run_local_replay(str(config_path), str(logs_path), stream_id="windows", compare_config_file=str(candidate_path))
+
+            self.assertIn("comparison", result)
+            self.assertIn("current", result)
+            self.assertIn("candidate", result)
+            self.assertEqual(result["candidate_config_file"], str(candidate_path))
+
+    def test_run_graylog_replay_requires_mcp_configuration(self):
+        with tempfile.TemporaryDirectory() as directory:
+            config_path = Path(directory) / "config.json"
+            config_path.write_text(json.dumps({}), encoding="utf-8")
+
+            with self.assertRaises(RuntimeError):
+                run_graylog_replay(str(config_path))
+
+    def test_run_graylog_replay_requires_selected_streams(self):
+        with tempfile.TemporaryDirectory() as directory:
+            config_path = Path(directory) / "config.json"
+            config_path.write_text(json.dumps({"graylog_mcp_url": "https://graylog.example", "graylog_mcp_token": "token"}), encoding="utf-8")
+
+            with self.assertRaises(RuntimeError):
+                run_graylog_replay(str(config_path))