feat(insights): OpenSearch AD and Elasticsearch ML anomaly pipeline

.env.example

@@ -9,10 +9,20 @@ LLM_MODEL=gpt-5.4-codex
 
 # Logging:
 LOG_LEVEL=info
+# matrix-js-sdk verbosity (TRACE|DEBUG|INFO|WARN|ERROR). Default WARN.
+# MATRIX_SDK_LOG_LEVEL=ERROR
 
 # OpenSearch:
 OPENSEARCH_URL=https://localhost:9200
 OPENSEARCH_USERNAME=admin
+# e2e/kind: must satisfy OpenSearch OPENSEARCH_INITIAL_ADMIN_PASSWORD (>=8, A-Z, a-z, digit, symbol)
 OPENSEARCH_PASSWORD=""
 OPENSEARCH_TLS_INSECURE=true
 OPENSEARCH_INDEX_PATTERN=elastiflow-flow-codex-*
+
+# Optional insight polling (seconds). Default 300.
+# KAYTOO_POLL_INTERVAL_SECONDS=60
+
+# Optional lookback for OpenSearch AD / Elasticsearch ML anomaly fetches (minutes).
+# Use when historical AD results are older than pollInterval/60 + 10 (e.g. e2e after detectors/_start over days).
+# KAYTOO_AD_FETCH_MINUTES_BACK=10080

e2e/README.md

@@ -31,7 +31,7 @@ Local stack: **Mermin** + OpenSearch ([`k8s/mermin-stack-values.yaml`](k8s/mermi
 
 | Variable | `e2e:up` | Notes |
 | --- | --- | --- |
-| `OPENSEARCH_PASSWORD` | required | Mermin/OpenSearch admin; also baked into rendered Kaytoo values |
+| `OPENSEARCH_PASSWORD` | required | Mermin/OpenSearch admin; also baked into rendered Kaytoo values. Must satisfy OpenSearch 2.12+ rules (length ≥8, include uppercase, lowercase, digit, and a non-alphanumeric symbol); hex-only values such as `openssl rand -hex` are rejected by the image. |
 | `LLM_BASE_URL` | required | OpenAI-compatible URL |
 | `LLM_API_KEY` | required | |
 | `LLM_MODEL` | optional | |
@@ -77,14 +77,20 @@ kubectl -n elastiflow logs deploy/kaytoo -c kaytoo -f
 
 ## Verify
 
-With forwards up and OpenSearch reachable on the host:
+Requires **`e2e:up`** (forwards on **9200** / **18080**) and repo **`.env`** with the same OpenSearch URL/creds as Kaytoo (see Configuration).
 
 ```bash
 export KUBECONFIG="$(pwd)/e2e/.generated/kubeconfig-kind-kaytoo-e2e"
 npm run e2e:verify
 ```
 
-`verify` starts a temporary chat port-forward if needed.
+**What it checks**
+
+- Chat: `e2e/chat.mjs` hits `/health` and `/chat`; OpenSearch-backed top talkers; Kaytoo logs show `topTalkersByBytes` ran.
+- **Anomaly Detection (real plugin):** stats 200; `detectors/_search` until 200; Kaytoo detector name present; no AD-plugin-missing line in Kaytoo logs; **`POST .../detectors/{id}/_start`** with a **72h historical window**; poll **`POST .../detectors/results/_search`** until **`anomaly_grade` > 0** for that detector/task; then Kaytoo must pick up **`opensearch_anomaly`** findings and post (see insights bullet). E2e Helm sets **`KAYTOO_AD_FETCH_MINUTES_BACK`** (via `config.behavior.adFetchMinutesBack`) so fetches reach historical AD results, not only the default poll-based lookback.
+- Insights: after AD historical results exist, Kaytoo must log **`posted findings`** with **`includesOpensearchAnomaly":true`** (LLM summarize must succeed; use a capable model in `.env`).
+
+Temporary `kubectl port-forward` for chat is started only if **18080** is not already serving `/health`.
 
 ## Host Kaytoo (optional)
 

e2e/chat.mjs

@@ -130,6 +130,7 @@ async function topTalkers() {
   const topIps = (agg.aggregations?.by_src?.buckets ?? []).map((b) => String(b.key)).filter(Boolean);
   if (!topIps.length) die('aggregation returned no talker buckets');
   console.error(`${p} top talker IPs from OpenSearch: ${topIps.join(', ')}`);
+  await chatPostJson('/chat', { text: 'reset' });
   const q =
     e.TOP_TALKERS_QUESTION ??
     'What is the pod name (from flow records) for each of the top 5 talkers by total bytes in the last 7 days? Use the topTalkersByBytes tool first, then answer. List each talker IP, total bytes, and pod name if present.';

e2e/cli.sh

@@ -247,8 +247,11 @@ cmd_verify() {
       -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}' 2>/dev/null | tail -n 1
   }
   kubectl -n "$ns" get svc kaytoo-chat &>/dev/null || e2e_die "svc/kaytoo-chat missing"
-  local PF_CHAT=0
-  cleanup_pf() { kill "${PF_CHAT:-0}" 2>/dev/null || true; }
+  PF_CHAT=""
+  cleanup_pf() {
+    local pid="${PF_CHAT:-}"
+    [[ -n "$pid" ]] && kill "$pid" 2>/dev/null || true
+  }
   trap cleanup_pf EXIT
   local CHAT_BASE="http://127.0.0.1:${CHAT_PF_LOCAL}"
   if ! e2e_wait_http "${CHAT_BASE}/health" 2; then
@@ -272,6 +275,126 @@ cmd_verify() {
   e2e_log "expect topTalkersByBytes in logs"
   logs_grep 'agent tool finished' || { kubectl -n "$ns" logs "$pod" -c kaytoo --tail=120; exit 1; }
   logs_grep '"tool":"topTalkersByBytes"' || { kubectl -n "$ns" logs "$pod" -c kaytoo --tail=120; exit 1; }
+
+  e2e_log "OpenSearch Anomaly Detection: plugin stats (expect HTTP 200)"
+  local ad_stats os_base="${OS_URL%/}"
+  ad_stats="$(curl -k -sS -o /dev/null -w '%{http_code}' -u "${OS_USER}:${OS_PASS}" \
+    -X GET "${os_base}/_plugins/_anomaly_detection/stats" 2>/dev/null || echo "000")"
+  [[ "$ad_stats" == "200" ]] || e2e_die "AD stats returned HTTP ${ad_stats} (plugin missing or auth?)"
+
+  e2e_log "OpenSearch AD: wait for detectors/_search HTTP 200 (system index after Kaytoo seed)"
+  local ad_http ad_json i
+  ad_http="000"
+  for ((i = 0; i < 45; i++)); do
+    ad_http="$(curl -k -sS -o /dev/null -w '%{http_code}' -u "${OS_USER}:${OS_PASS}" \
+      -X POST "${os_base}/_plugins/_anomaly_detection/detectors/_search" \
+      -H 'Content-Type: application/json' \
+      -d '{"query":{"match_all":{}},"size":1}' 2>/dev/null || echo "000")"
+    [[ "$ad_http" == "200" ]] && break
+    sleep 2
+  done
+  [[ "$ad_http" == "200" ]] || e2e_die "AD detectors/_search stayed non-200 (last HTTP ${ad_http}; index warming?)"
+
+  e2e_log "OpenSearch AD: expect Kaytoo egress detector in list (seed/adopt)"
+  ad_json="$(curl -k -sS -u "${OS_USER}:${OS_PASS}" \
+    -X POST "${os_base}/_plugins/_anomaly_detection/detectors/_search" \
+    -H 'Content-Type: application/json' \
+    -d '{"query":{"match_all":{}},"size":50}' 2>/dev/null || true)"
+  echo "$ad_json" | grep -q 'Kaytoo flow egress' ||
+    e2e_die "AD detector list missing Kaytoo egress detector (native pipeline seed/adopt)"
+
+  e2e_log "Kaytoo logs: native AD must not report plugin 404"
+  if logs_grep 'OpenSearch Anomaly Detection plugin not available (404)' 25000; then
+    kubectl -n "$ns" logs "$pod" -c kaytoo --tail=200 >&2
+    e2e_die "Kaytoo logged AD plugin unavailable (404)"
+  fi
+
+  e2e_log "OpenSearch AD: extract detector id for historical analysis"
+  local det_id
+  det_id="$(printf '%s' "$ad_json" | node "${REPO_ROOT}/e2e/extract-kaytoo-ad-detector-id.mjs")" ||
+    e2e_die "could not extract Kaytoo AD detector id"
+  [[ -n "$det_id" ]] || e2e_die "empty Kaytoo AD detector id"
+
+  local hist_payload hist_http htmp stopped
+  stopped=0
+  hist_payload="$(python3 -c "import time, json; e=int(time.time()*1000); s=e-72*3600*1000; print(json.dumps({'start_time':s,'end_time':e}))")"
+  htmp="$(mktemp "${TMPDIR:-/tmp}/e2e-ad-hist.XXXXXX")"
+  hist_http="$(curl -k -sS -o "$htmp" -w '%{http_code}' -u "${OS_USER}:${OS_PASS}" \
+    -X POST "${os_base}/_plugins/_anomaly_detection/detectors/${det_id}/_start" \
+    -H 'Content-Type: application/json' \
+    -d "$hist_payload")"
+  if [[ "$hist_http" != "200" && "$hist_http" != "201" ]]; then
+    e2e_log "WARN: AD historical _start HTTP ${hist_http}; retrying after _stop"
+    curl -k -sS -o /dev/null -u "${OS_USER}:${OS_PASS}" \
+      -X POST "${os_base}/_plugins/_anomaly_detection/detectors/${det_id}/_stop" || true
+    stopped=1
+    sleep 3
+    hist_http="$(curl -k -sS -o "$htmp" -w '%{http_code}' -u "${OS_USER}:${OS_PASS}" \
+      -X POST "${os_base}/_plugins/_anomaly_detection/detectors/${det_id}/_start" \
+      -H 'Content-Type: application/json' \
+      -d "$hist_payload")"
+  fi
+  [[ "$hist_http" == "200" || "$hist_http" == "201" ]] || {
+    cat "$htmp" >&2
+    rm -f "$htmp"
+    e2e_die "OpenSearch AD historical _start failed (HTTP ${hist_http})"
+  }
+  local task_id
+  task_id="$(python3 -c "import json,sys; d=json.load(open(sys.argv[1])); print(d.get('_id',''))" "$htmp")"
+  rm -f "$htmp"
+  [[ -n "$task_id" ]] || e2e_die "OpenSearch AD historical _start response missing _id (task id)"
+  local ad_use_task
+  ad_use_task=1
+  [[ "$task_id" == "$det_id" ]] && {
+    e2e_log "WARN: AD historical _id equals detector id; omitting task_id in results/_search"
+    ad_use_task=0
+  }
+  if [[ "$stopped" == "1" ]]; then
+    e2e_log "OpenSearch AD: restart real-time detector after _stop"
+    curl -k -sS -o /dev/null -u "${OS_USER}:${OS_PASS}" \
+      -X POST "${os_base}/_plugins/_anomaly_detection/detectors/${det_id}/_start" \
+      -H 'Content-Type: application/json' -d '{}' || true
+    sleep 5
+  fi
+
+  e2e_log "OpenSearch AD: wait for historical anomaly results (results/_search)"
+  local hits j
+  hits=0
+  for ((j = 0; j < 60; j++)); do
+    if [[ "$ad_use_task" == "1" ]]; then
+      hits="$(node "${REPO_ROOT}/e2e/count-ad-historical-hits.mjs" "$os_base" "$OS_USER" "$OS_PASS" "$det_id" "$task_id" | tr -d '\n')"
+    else
+      hits="$(node "${REPO_ROOT}/e2e/count-ad-historical-hits.mjs" "$os_base" "$OS_USER" "$OS_PASS" "$det_id" '' | tr -d '\n')"
+    fi
+    [[ "${hits:-0}" =~ ^[0-9]+$ ]] || hits=0
+    [[ "${hits:-0}" -gt 0 ]] && break
+    if [[ "$j" -eq 35 && "$ad_use_task" == "1" ]]; then
+      e2e_log "retrying AD results/_search without task_id filter"
+      ad_use_task=0
+    fi
+    sleep 10
+  done
+  [[ "${hits:-0}" -gt 0 ]] || e2e_die "no AD historical hits with anomaly_grade>0 (detector=${det_id} task=${task_id})"
+
+  e2e_log "Kaytoo: wait for poll to ingest AD results and post (includesOpensearchAnomaly)"
+  sleep 45
+  pod="$(kaytoo_pod)"
+  [[ -n "$pod" ]] || e2e_die "no Running kaytoo pod (after AD historical)"
+  local found k
+  found=0
+  for ((k = 0; k < 48; k++)); do
+    if logs_grep 'includesOpensearchAnomaly":true' 80000; then
+      found=1
+      break
+    fi
+    sleep 5
+  done
+  if [[ "$found" != "1" ]]; then
+    kubectl -n "$ns" logs "$pod" -c kaytoo --tail=250 >&2
+    e2e_die "Kaytoo did not log posted findings with includesOpensearchAnomaly:true (check LLM summarizeFindings in .env)"
+  fi
+  e2e_log "Kaytoo posted OpenSearch AD finding(s) to console pipeline"
+
   e2e_log "OK verify"
 }
 

e2e/count-ad-historical-hits.mjs

@@ -0,0 +1,59 @@
+#!/usr/bin/env node
+/* eslint-disable no-undef -- e2e OpenSearch AD results/_search hit count */
+import http from 'node:http';
+import https from 'node:https';
+import { buffer } from 'node:stream/consumers';
+
+const [, , base, user, pass, det, task] = process.argv;
+if (!base || !user || !pass || !det) {
+  console.error('usage: count-ad-historical-hits <osBaseUrl> <user> <pass> <detectorId> [taskId]');
+  process.exit(2);
+}
+
+const auth = Buffer.from(`${user}:${pass}`).toString('base64');
+const u = new URL(base);
+const filters = [{ term: { detector_id: det } }, { range: { anomaly_grade: { gt: 0 } } }];
+if (task) filters.splice(1, 0, { term: { task_id: task } });
+const body = JSON.stringify({
+  size: 1,
+  query: { bool: { filter: filters } },
+});
+
+const opts = {
+  hostname: u.hostname,
+  port: u.port || (u.protocol === 'https:' ? 443 : 80),
+  path: '/_plugins/_anomaly_detection/detectors/results/_search',
+  method: 'POST',
+  headers: {
+    authorization: `Basic ${auth}`,
+    'content-type': 'application/json',
+    'content-length': Buffer.byteLength(body),
+  },
+  rejectUnauthorized: false,
+};
+
+const mod = u.protocol === 'https:' ? https : http;
+
+const req = mod.request(opts, async (res) => {
+  const bin = await buffer(res);
+  const t = bin.toString('utf8');
+  if ((res.statusCode ?? 0) >= 400) {
+    console.error(t.slice(0, 500));
+    process.stdout.write('0');
+    process.exit(0);
+  }
+  try {
+    const j = JSON.parse(t);
+    const tot = j.hits?.total;
+    const n = typeof tot === 'object' && tot !== null ? Number(tot.value ?? 0) : Number(tot ?? 0);
+    const hits = Array.isArray(j.hits?.hits) ? j.hits.hits.length : 0;
+    process.stdout.write(String(n > 0 ? n : hits));
+  } catch {
+    process.stdout.write('0');
+  }
+});
+req.on('error', () => {
+  process.stdout.write('0');
+});
+req.write(body);
+req.end();

e2e/extract-kaytoo-ad-detector-id.mjs

@@ -0,0 +1,44 @@
+#!/usr/bin/env node
+/* eslint-disable no-undef -- stdin JSON helper for e2e/cli.sh */
+import fs from 'node:fs';
+
+const want = 'Kaytoo flow egress by source';
+const raw = fs.readFileSync(0, 'utf8');
+let j;
+try {
+  j = JSON.parse(raw);
+} catch {
+  console.error('invalid JSON on stdin');
+  process.exit(1);
+}
+
+if (Array.isArray(j.detectorList)) {
+  for (const d of j.detectorList) {
+    if (!d || typeof d !== 'object') continue;
+    const n = d.name;
+    const id = d.id;
+    if (typeof id !== 'string' || !id) continue;
+    if (n === want || (typeof n === 'string' && n.includes('Kaytoo flow egress'))) {
+      process.stdout.write(id);
+      process.exit(0);
+    }
+  }
+}
+
+const hits = j.hits?.hits;
+if (Array.isArray(hits)) {
+  for (const h of hits) {
+    if (!h || typeof h !== 'object') continue;
+    const id = typeof h._id === 'string' ? h._id : '';
+    const src = h._source && typeof h._source === 'object' ? h._source : {};
+    const n = src.name;
+    if (!id) continue;
+    if (n === want || (typeof n === 'string' && n.includes('Kaytoo flow egress'))) {
+      process.stdout.write(id);
+      process.exit(0);
+    }
+  }
+}
+
+console.error('no Kaytoo AD detector id in detectors/_search response');
+process.exit(1);

e2e/k8s/kaytoo-values.yaml

@@ -26,6 +26,7 @@ config:
     tlsInsecure: "true"
   behavior:
     pollIntervalSeconds: "15"
+    adFetchMinutesBack: "10080"
   thresholds:
     egressMinBytes: "5000"
     egressMultiplier: "2"

helm/kaytoo/templates/configmap.yaml

@@ -25,6 +25,16 @@ data:
   {{- end }}
 
   LOG_LEVEL: {{ .Values.logging.level | quote }}
+  {{- if .Values.logging.matrixSdkLevel }}
+  MATRIX_SDK_LOG_LEVEL: {{ .Values.logging.matrixSdkLevel | quote }}
+  {{- end }}
+
+  {{- if and .Values.config.behavior .Values.config.behavior.pollIntervalSeconds }}
+  KAYTOO_POLL_INTERVAL_SECONDS: {{ .Values.config.behavior.pollIntervalSeconds | quote }}
+  {{- end }}
+  {{- if and .Values.config.behavior .Values.config.behavior.adFetchMinutesBack }}
+  KAYTOO_AD_FETCH_MINUTES_BACK: {{ .Values.config.behavior.adFetchMinutesBack | quote }}
+  {{- end }}
 
   {{- if .Values.config.matrix.homeserver }}
   MATRIX_HOMESERVER: {{ .Values.config.matrix.homeserver | quote }}

helm/kaytoo/values.yaml

@@ -28,6 +28,9 @@ resources:
     memory: 512Mi
 
 config:
+  behavior:
+    pollIntervalSeconds: ""
+    adFetchMinutesBack: ""
   slack:
     channelId: ""
   matrix:
@@ -69,6 +72,8 @@ secrets:
 
 logging:
   level: "info"
+  # matrix-js-sdk (Room echo, sync). Default WARN when unset.
+  matrixSdkLevel: "ERROR"
 
 networkPolicy:
   enabled: false

src/agent/tools/handlers/common.ts

@@ -9,12 +9,22 @@ export function clampMinutesBack(minutes: number, policy: AgentPolicy): number {
   return Math.min(Math.floor(minutes), maxM);
 }
 
+function indexAllowedOrDefault(candidate: string, defaultIndex: string, policy: AgentPolicy): string {
+  try {
+    assertIndexAllowed(candidate, policy);
+    return candidate;
+  } catch {
+    assertIndexAllowed(defaultIndex, policy);
+    return defaultIndex;
+  }
+}
+
 export async function resolveToolIndexAndFields(opts: {
   ctx: { client: SearchClient; policy: AgentPolicy; defaultIndex: string };
   args: Record<string, unknown>;
 }): Promise<{ index: string; fields: FieldPreference }> {
-  const index = typeof opts.args.index === 'string' ? opts.args.index : opts.ctx.defaultIndex;
-  assertIndexAllowed(index, opts.ctx.policy);
+  const fromArgs = typeof opts.args.index === 'string' ? opts.args.index : undefined;
+  const index = indexAllowedOrDefault(fromArgs ?? opts.ctx.defaultIndex, opts.ctx.defaultIndex, opts.ctx.policy);
   const fields = await chooseFields({ client: opts.ctx.client, index });
   return { index, fields };
 }

src/chat/adapters/matrix.ts

@@ -8,7 +8,7 @@ import {
   RoomEvent,
 } from 'matrix-js-sdk';
 import { getLogger, logErr } from '../../logging/logger.js';
-import { createMatrixJsSdkLogger, type MatrixSdkLevel } from '../../logging/matrixSdkLogger.js';
+import { configureMatrixJsSdkLogging, type MatrixSdkLevel } from '../../logging/matrixSdkLogger.js';
 import type { ChatEvent } from '../types.js';
 
 const log = getLogger({ component: 'chat.matrix' });
@@ -36,7 +36,7 @@ export async function startMatrixAdapter(opts: {
   defaultRoomId?: string;
   onEvent: (evt: ChatEvent) => Promise<void>;
 }): Promise<{ stop: () => Promise<void>; client: MatrixClient }> {
-  const logger = createMatrixJsSdkLogger(opts.matrixSdkLevel);
+  const logger = configureMatrixJsSdkLogging(opts.matrixSdkLevel);
   const userId =
     'accessToken' in opts.auth ? await whoamiUserId(opts.homeserverUrl, opts.auth.accessToken) : undefined;