Kraft102's picture
Initial deployment - WidgeTDC Cortex Backend v2.1.0
529090e
/**
* ╔═══════════════════════════════════════════════════════════════════════════╗
* β•‘ AUDITORY PERCEPTION SERVICE β•‘
* ║═══════════════════════════════════════════════════════════════════════════║
* β•‘ The system's "ears" - listens to log streams, detects anomalies, β•‘
* β•‘ and interprets system "sounds" (events, errors, patterns) β•‘
* β•šβ•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•β•
*/
import { EventEmitter } from 'events';
import { neo4jAdapter } from '../adapters/Neo4jAdapter.js';
import { v4 as uuidv4 } from 'uuid';
// ═══════════════════════════════════════════════════════════════════════════
// Types
// ═══════════════════════════════════════════════════════════════════════════
export interface AudioSignal {
id: string;
type: 'LOG' | 'ERROR' | 'WARNING' | 'ANOMALY' | 'HEARTBEAT' | 'VOICE';
source: string;
content: string;
volume: 'WHISPER' | 'NORMAL' | 'LOUD' | 'ALARM';
frequency: number; // occurrences per minute
timestamp: string;
metadata?: Record<string, unknown>;
}
export interface AnomalyPattern {
id: string;
pattern: string;
description: string;
severity: 'LOW' | 'MEDIUM' | 'HIGH' | 'CRITICAL';
occurrences: number;
firstSeen: string;
lastSeen: string;
isActive: boolean;
}
export interface ListeningSession {
id: string;
source: string;
startedAt: string;
signalsReceived: number;
anomaliesDetected: number;
isActive: boolean;
}
// ═══════════════════════════════════════════════════════════════════════════
// Auditory Service
// ═══════════════════════════════════════════════════════════════════════════
class AuditoryService extends EventEmitter {
private static instance: AuditoryService;
private sessions: Map<string, ListeningSession> = new Map();
private signalBuffer: AudioSignal[] = [];
private anomalyPatterns: Map<string, AnomalyPattern> = new Map();
private frequencyTracker: Map<string, number[]> = new Map();
// Anomaly detection thresholds
private readonly ANOMALY_THRESHOLD = 10; // signals per minute
private readonly BUFFER_MAX_SIZE = 1000;
private readonly FREQUENCY_WINDOW_MS = 60000; // 1 minute
// Known error patterns
private readonly ERROR_PATTERNS = [
{ regex: /ECONNREFUSED/i, severity: 'HIGH' as const, description: 'Connection refused - service down' },
{ regex: /ETIMEDOUT/i, severity: 'MEDIUM' as const, description: 'Connection timeout' },
{ regex: /OutOfMemory|heap/i, severity: 'CRITICAL' as const, description: 'Memory exhaustion' },
{ regex: /ENOSPC/i, severity: 'CRITICAL' as const, description: 'Disk space exhausted' },
{ regex: /EACCES|EPERM/i, severity: 'HIGH' as const, description: 'Permission denied' },
{ regex: /deadlock/i, severity: 'CRITICAL' as const, description: 'Database deadlock detected' },
{ regex: /rate.?limit/i, severity: 'MEDIUM' as const, description: 'Rate limiting triggered' },
{ regex: /authentication.?fail/i, severity: 'HIGH' as const, description: 'Authentication failure' },
{ regex: /ssl|tls|certificate/i, severity: 'HIGH' as const, description: 'SSL/TLS issue' },
{ regex: /crash|fatal|panic/i, severity: 'CRITICAL' as const, description: 'Critical failure' },
];
private constructor() {
super();
this.startBackgroundListening();
}
public static getInstance(): AuditoryService {
if (!AuditoryService.instance) {
AuditoryService.instance = new AuditoryService();
}
return AuditoryService.instance;
}
// ═══════════════════════════════════════════════════════════════════════
// Core Listening Functions
// ═══════════════════════════════════════════════════════════════════════
/**
* Start listening to a specific source
*/
public startListening(source: string): ListeningSession {
const session: ListeningSession = {
id: `listen-${uuidv4()}`,
source,
startedAt: new Date().toISOString(),
signalsReceived: 0,
anomaliesDetected: 0,
isActive: true
};
this.sessions.set(session.id, session);
console.error(`[Auditory] πŸ‘‚ Started listening to: ${source}`);
return session;
}
/**
* Stop listening to a source
*/
public stopListening(sessionId: string): void {
const session = this.sessions.get(sessionId);
if (session) {
session.isActive = false;
console.error(`[Auditory] πŸ”‡ Stopped listening: ${session.source}`);
}
}
/**
* Process an incoming signal (log, event, etc.)
*/
public processSignal(input: {
source: string;
content: string;
type?: AudioSignal['type'];
metadata?: Record<string, unknown>;
}): AudioSignal {
const signal: AudioSignal = {
id: `sig-${uuidv4()}`,
type: input.type || this.classifySignalType(input.content),
source: input.source,
content: input.content,
volume: this.calculateVolume(input.content),
frequency: this.calculateFrequency(input.source),
timestamp: new Date().toISOString(),
metadata: input.metadata
};
// Add to buffer
this.signalBuffer.push(signal);
if (this.signalBuffer.length > this.BUFFER_MAX_SIZE) {
this.signalBuffer.shift();
}
// Track frequency
this.trackFrequency(input.source);
// Check for anomalies
this.detectAnomalies(signal);
// Update session stats
for (const session of this.sessions.values()) {
if (session.isActive && session.source === input.source) {
session.signalsReceived++;
}
}
// Emit event for real-time listeners
this.emit('signal', signal);
return signal;
}
/**
* Classify signal type based on content
*/
private classifySignalType(content: string): AudioSignal['type'] {
const lower = content.toLowerCase();
if (/error|exception|fail|crash/i.test(lower)) return 'ERROR';
if (/warn|caution|attention/i.test(lower)) return 'WARNING';
if (/heartbeat|ping|alive|health/i.test(lower)) return 'HEARTBEAT';
return 'LOG';
}
/**
* Calculate volume (severity/importance) of signal
*/
private calculateVolume(content: string): AudioSignal['volume'] {
const lower = content.toLowerCase();
if (/critical|fatal|crash|emergency|panic/i.test(lower)) return 'ALARM';
if (/error|fail|exception/i.test(lower)) return 'LOUD';
if (/warn|caution/i.test(lower)) return 'NORMAL';
return 'WHISPER';
}
/**
* Calculate signal frequency (signals per minute from source)
*/
private calculateFrequency(source: string): number {
const timestamps = this.frequencyTracker.get(source) || [];
const now = Date.now();
const recentTimestamps = timestamps.filter(t => now - t < this.FREQUENCY_WINDOW_MS);
return recentTimestamps.length;
}
/**
* Track frequency for anomaly detection
*/
private trackFrequency(source: string): void {
const timestamps = this.frequencyTracker.get(source) || [];
timestamps.push(Date.now());
// Keep only recent timestamps
const now = Date.now();
const recentTimestamps = timestamps.filter(t => now - t < this.FREQUENCY_WINDOW_MS);
this.frequencyTracker.set(source, recentTimestamps);
}
// ═══════════════════════════════════════════════════════════════════════
// Anomaly Detection
// ═══════════════════════════════════════════════════════════════════════
/**
* Detect anomalies in incoming signals
*/
private detectAnomalies(signal: AudioSignal): void {
// Check against known error patterns
for (const pattern of this.ERROR_PATTERNS) {
if (pattern.regex.test(signal.content)) {
this.registerAnomaly({
pattern: pattern.regex.source,
description: pattern.description,
severity: pattern.severity,
signal
});
}
}
// Check for frequency anomalies
if (signal.frequency > this.ANOMALY_THRESHOLD) {
this.registerAnomaly({
pattern: `HIGH_FREQUENCY:${signal.source}`,
description: `Abnormal signal frequency from ${signal.source}: ${signal.frequency}/min`,
severity: 'MEDIUM',
signal
});
}
// Check for sudden volume increase
if (signal.volume === 'ALARM') {
this.registerAnomaly({
pattern: `ALARM:${signal.type}`,
description: `Alarm-level signal: ${signal.content.substring(0, 100)}`,
severity: 'HIGH',
signal
});
}
}
/**
* Register a detected anomaly
*/
private async registerAnomaly(params: {
pattern: string;
description: string;
severity: AnomalyPattern['severity'];
signal: AudioSignal;
}): Promise<void> {
const existing = this.anomalyPatterns.get(params.pattern);
if (existing) {
existing.occurrences++;
existing.lastSeen = new Date().toISOString();
existing.isActive = true;
} else {
const anomaly: AnomalyPattern = {
id: `anomaly-${uuidv4()}`,
pattern: params.pattern,
description: params.description,
severity: params.severity,
occurrences: 1,
firstSeen: new Date().toISOString(),
lastSeen: new Date().toISOString(),
isActive: true
};
this.anomalyPatterns.set(params.pattern, anomaly);
// Persist to Neo4j
await this.persistAnomaly(anomaly, params.signal);
}
// Update session stats
for (const session of this.sessions.values()) {
if (session.isActive && session.source === params.signal.source) {
session.anomaliesDetected++;
}
}
// Emit anomaly event
this.emit('anomaly', { anomaly: this.anomalyPatterns.get(params.pattern), signal: params.signal });
console.error(`[Auditory] 🚨 Anomaly detected: ${params.description}`);
}
/**
* Persist anomaly to Neo4j
*/
private async persistAnomaly(anomaly: AnomalyPattern, signal: AudioSignal): Promise<void> {
try {
await neo4jAdapter.executeQuery(`
CREATE (a:Anomaly {
id: $id,
pattern: $pattern,
description: $description,
severity: $severity,
occurrences: $occurrences,
firstSeen: $firstSeen,
lastSeen: $lastSeen,
signalSource: $signalSource,
signalContent: $signalContent
})
`, {
id: anomaly.id,
pattern: anomaly.pattern,
description: anomaly.description,
severity: anomaly.severity,
occurrences: anomaly.occurrences,
firstSeen: anomaly.firstSeen,
lastSeen: anomaly.lastSeen,
signalSource: signal.source,
signalContent: signal.content.substring(0, 500)
});
} catch (error) {
console.warn('[Auditory] Failed to persist anomaly:', error);
}
}
// ═══════════════════════════════════════════════════════════════════════
// Query Functions
// ═══════════════════════════════════════════════════════════════════════
/**
* Get recent signals
*/
public getRecentSignals(params: {
source?: string;
type?: AudioSignal['type'];
volume?: AudioSignal['volume'];
limit?: number;
} = {}): AudioSignal[] {
let signals = [...this.signalBuffer];
if (params.source) {
signals = signals.filter(s => s.source === params.source);
}
if (params.type) {
signals = signals.filter(s => s.type === params.type);
}
if (params.volume) {
signals = signals.filter(s => s.volume === params.volume);
}
return signals.slice(-(params.limit || 50));
}
/**
* Get active anomalies
*/
public getActiveAnomalies(): AnomalyPattern[] {
return Array.from(this.anomalyPatterns.values()).filter(a => a.isActive);
}
/**
* Get all anomaly patterns
*/
public getAllAnomalies(): AnomalyPattern[] {
return Array.from(this.anomalyPatterns.values());
}
/**
* Get listening sessions
*/
public getListeningSessions(): ListeningSession[] {
return Array.from(this.sessions.values());
}
/**
* Get auditory system status
*/
public getStatus(): {
activeSessions: number;
totalSignals: number;
activeAnomalies: number;
signalBuffer: number;
frequencySources: number;
} {
return {
activeSessions: Array.from(this.sessions.values()).filter(s => s.isActive).length,
totalSignals: this.signalBuffer.length,
activeAnomalies: this.getActiveAnomalies().length,
signalBuffer: this.signalBuffer.length,
frequencySources: this.frequencyTracker.size
};
}
/**
* Acknowledge/dismiss an anomaly
*/
public acknowledgeAnomaly(pattern: string): boolean {
const anomaly = this.anomalyPatterns.get(pattern);
if (anomaly) {
anomaly.isActive = false;
return true;
}
return false;
}
// ═══════════════════════════════════════════════════════════════════════
// Background Listening
// ═══════════════════════════════════════════════════════════════════════
/**
* Start background listening for system events
*/
private startBackgroundListening(): void {
// Intercept console.error for system-wide listening
const originalError = console.error;
console.error = (...args: unknown[]) => {
originalError.apply(console, args);
// Don't process our own logs
const content = args.map(a => String(a)).join(' ');
if (!content.includes('[Auditory]')) {
this.processSignal({
source: 'console.error',
content,
type: 'ERROR'
});
}
};
// Periodic cleanup of old frequency data
setInterval(() => {
const now = Date.now();
for (const [source, timestamps] of this.frequencyTracker.entries()) {
const recent = timestamps.filter(t => now - t < this.FREQUENCY_WINDOW_MS);
if (recent.length === 0) {
this.frequencyTracker.delete(source);
} else {
this.frequencyTracker.set(source, recent);
}
}
}, 30000); // Every 30 seconds
console.error('[Auditory] πŸ‘‚ Background listening started');
}
/**
* Analyze log content for insights
*/
public analyzeLogContent(logs: string[]): {
summary: string;
errorCount: number;
warningCount: number;
patterns: string[];
recommendations: string[];
} {
let errorCount = 0;
let warningCount = 0;
const patterns: Set<string> = new Set();
const recommendations: string[] = [];
for (const log of logs) {
if (/error|exception|fail/i.test(log)) errorCount++;
if (/warn|caution/i.test(log)) warningCount++;
// Detect patterns
for (const pattern of this.ERROR_PATTERNS) {
if (pattern.regex.test(log)) {
patterns.add(pattern.description);
}
}
}
// Generate recommendations
if (errorCount > 10) {
recommendations.push('High error rate detected - consider investigating root cause');
}
if (patterns.has('Memory exhaustion')) {
recommendations.push('Memory issues detected - consider increasing heap size or optimizing memory usage');
}
if (patterns.has('Connection refused - service down')) {
recommendations.push('Service connectivity issues - check if dependent services are running');
}
return {
summary: `Analyzed ${logs.length} logs: ${errorCount} errors, ${warningCount} warnings`,
errorCount,
warningCount,
patterns: Array.from(patterns),
recommendations
};
}
}
export const auditoryService = AuditoryService.getInstance();